Implement routing rules

diploma_thesis/agents/workcenter/model/rule/ct.py

@@ -1,3 +1,4 @@
+import torch
 
 from .routing_rule import *
 
@@ -7,11 +8,11 @@ class CTRoutingRule(RoutingRule):
     Earliest Completion Time rule
     """
 
-    def __call__(self, job: Job, work_center_idx: int, machines: List['Machine']) -> 'Machine | None':
-        cumulative_processing_times = torch.FloatTensor(
-            [machine.cumulative_processing_time for machine in machines]
-        ) + job.operation_processing_time_in_work_center(work_center_idx, JobReductionStrategy.none)
+    @property
+    def selector(self):
+        return torch.argmin
 
-        idx = torch.argmin(cumulative_processing_times)
-
-        return machines[idx]
+    def criterion(self, job: Job, work_center: WorkCenter) -> torch.FloatTensor:
+        return torch.FloatTensor(
+            [machine.cumulative_processing_time for machine in work_center.machines]
+        ) + job.operation_processing_time_in_work_center(work_center.work_center_idx, JobReductionStrategy.none)

diploma_thesis/agents/workcenter/model/rule/ea.py

@@ -6,7 +6,9 @@ class EARoutingRule(RoutingRule):
     Earliest Available (EA) routing rule
     """
 
-    def __call__(self, job: Job, work_center_idx: int, machines: List['Machine']) -> 'Machine | None':
-        machine = min(machines, key=lambda machine: machine.time_till_available)
+    @property
+    def selector(self):
+        return torch.argmin
 
-        return machine
+    def criterion(self, job: Job, work_center: WorkCenter) -> torch.FloatTensor:
+        return torch.FloatTensor([machine.time_till_available for machine in work_center.machines])

diploma_thesis/agents/workcenter/model/rule/et.py

@@ -3,7 +3,9 @@
 
 class ETRoutingRule(RoutingRule):
 
-    def __call__(self, job: Job, work_center_idx: int, machines: List['Machine']) -> 'Machine | None':
-        operation_times = job.operation_processing_time_in_work_center(work_center_idx, JobReductionStrategy.none)
+    @property
+    def selector(self):
+        return torch.argmin
 
-        return machines[operation_times.argmin()]
+    def criterion(self, job: Job, work_center: WorkCenter) -> torch.FloatTensor:
+        return job.operation_processing_time_in_work_center(work_center.work_center_idx, JobReductionStrategy.none)

diploma_thesis/agents/workcenter/model/rule/random.py

@@ -6,7 +6,10 @@ class RandomRoutingRule(RoutingRule):
     Selects a machine at random
     """
 
-    def __call__(self, job: Job, work_center_idx: int, machines: List['Machine']) -> 'Machine | None':
-        idx = torch.randint(0, len(machines), (1,))
+    @property
+    def selector(self):
+        return lambda x: torch.randint(0, len(x), (1,))
+
+    def criterion(self, job: Job, work_center: WorkCenter) -> torch.FloatTensor:
+        return torch.zeros(len(work_center.machines))
 
-        return machines[idx]

diploma_thesis/agents/workcenter/model/rule/routing_rule.py

@@ -3,7 +3,7 @@
 import torch
 
 from abc import ABCMeta, abstractmethod
-from environment import Job, Machine, JobReductionStrategy
+from environment import Job, Machine, WorkCenter, JobReductionStrategy
 
 
 class RoutingRule(metaclass=ABCMeta):
@@ -12,6 +12,18 @@ class RoutingRule(metaclass=ABCMeta):
     """
 
     @abstractmethod
-    def __call__(self, job: Job, work_center_idx: int, machines: List[Machine]) -> Machine | None:
+    def __call__(self, job: Job, work_center: WorkCenter) -> Machine | None:
+        value = self.criterion(job, work_center)
+        selector = self.selector
+        idx = selector(value)
+
+        return work_center.machines[idx]
+
+    @property
+    @abstractmethod
+    def selector(self):
         pass
 
+    @abstractmethod
+    def criterion(self, job: Job, work_center: WorkCenter) -> torch.FloatTensor:
+        pass

diploma_thesis/agents/workcenter/model/rule/sq.py

@@ -6,7 +6,9 @@ class SQRoutingRule(RoutingRule):
     Selects machine with the shortest queue
     """
 
-    def __call__(self, job: Job, work_center_idx: int, machines: List['Machine']) -> 'Machine | None':
-        machine = min(machines, key=lambda machine: len(machine.state.queue))
+    @property
+    def selector(self):
+        return torch.argmin
 
-        return machine
+    def criterion(self, job: Job, work_center: WorkCenter) -> torch.FloatTensor:
+        return torch.FloatTensor([len(machine.queue) for machine in work_center.machines])

diploma_thesis/agents/workcenter/model/rule/tt.py

@@ -6,7 +6,9 @@ class TTRoutingRule(RoutingRule):
     Shortest Total Waiting Time (TT) routing rule
     """
 
-    def __call__(self, job: Job, work_center_idx: int, machines: List['Machine']) -> 'Machine | None':
-        machine = min(machines, key=lambda machine: machine.cumulative_processing_time)
+    @property
+    def selector(self):
+        return torch.argmin
 
-        return machine
+    def criterion(self, job: Job, work_center: WorkCenter) -> torch.FloatTensor:
+        return torch.FloatTensor([machine.cumulative_processing_time for machine in work_center.machines])

diploma_thesis/agents/workcenter/model/rule/ut.py

@@ -6,7 +6,9 @@ class UTRoutingRule(RoutingRule):
     Selects machine with the lowest utilization
     """
 
-    def __call__(self, job: Job, work_center_idx: int, machines: List['Machine']) -> 'Machine | None':
-        machine = min(machines, key=lambda machine: machine.cumulative_run_time)
+    @property
+    def selector(self):
+        return torch.argmin
 
-        return machine
+    def criterion(self, job: Job, work_center: WorkCenter) -> torch.FloatTensor:
+        return torch.FloatTensor([machine.cumulative_run_time for machine in work_center.machines])