fix: add more hint type

Makefile

@@ -5,7 +5,7 @@ check_dirs := lekin examples tests
 # run checks on all files and potentially modifies some of them
 
 style:
-	black --preview $(check_dirs)
+	black $(check_dirs)
 	isort $(check_dirs)
 	flake8 $(check_dirs)
 	pre-commit run --files $(check_dirs)

lekin/dashboard/gantt.py

@@ -3,6 +3,7 @@
 """
 
 import logging
+from typing import Any, Callable, Dict, Generator, List, Optional, Tuple, Type, Union
 
 from matplotlib import ticker
 import matplotlib.patches as patches

lekin/dashboard/pages.py

@@ -1 +1,3 @@
+from typing import Any, Callable, Dict, Generator, List, Optional, Tuple, Type, Union
+
 import streamlit as st

lekin/lekin_struct/job.py

@@ -2,17 +2,6 @@
 Struct Job/订单作业
     - a job could finish one product while finished
     - job/mo/operation/activity
-
-property
-    - 已完成活动
-    - 待完成活动
-    - processing time
-    - due date
-    - weight
-    - slack time remaining
-    - critical ratio
-    - priority
-    - 属于哪个订单
 """
 
 from datetime import datetime
@@ -35,17 +24,17 @@ def __init__(
         earliest_start_time=None,
         assigned_route_id=None,
         assigned_bom_id=None,
-        **kwargs,
-    ):
-        self.job_id = job_id
-        self.priority = priority
-        self.demand_date = demand_date
-        self.quantity = quantity
-        self.job_type = job_type
-        self.earliest_start_time = earliest_start_time  # Material constraint
-        self.assigned_route_id = assigned_route_id  # Route object assigned to this job
-        self.assigned_bom_id = assigned_bom_id
-        self._operations_sequence = []  # List of Operation objects for this job
+        **kwargs: Dict,
+    ) -> None:
+        self.job_id: str = job_id
+        self.priority: int = priority
+        self.demand_date: datetime = demand_date
+        self.quantity: int = quantity
+        self.job_type: str = job_type
+        self.earliest_start_time: datetime = earliest_start_time  # Material constraint
+        self.assigned_route_id: str = assigned_route_id  # Route object assigned to this job
+        self.assigned_bom_id: str = assigned_bom_id
+        self._operations_sequence: List[Operation] = []  # List of Operation objects for this job
 
         for key, value in kwargs.items():
             setattr(self, key, value)
@@ -100,6 +89,16 @@ def get_job_by_id(self, job_id: str):
                 return job
         return None
 
+    def sort_jobs(self, jobs):
+        # Custom sorting function based on priority and continuity
+        def custom_sort(job):
+            priority_weight = (job.priority, job.demand_date)
+            # continuity_weight = -self.calculate_gap_time(job)
+            return priority_weight  # + continuity_weight
+
+        # jobs = sorted(jobs, key=custom_sort, reverse=True)
+        return [i[0] for i in sorted(enumerate(jobs), key=lambda x: custom_sort(x[1]), reverse=False)]
+
     def get_schedule(self):
         schedule = {}
 

lekin/lekin_struct/resource.py

@@ -3,8 +3,7 @@
 """
 
 import math
-
-import pandas as pd
+from typing import Any, Callable, Dict, Generator, List, Optional, Tuple, Type, Union
 
 from lekin.lekin_struct.timeslot import TimeSlot
 
@@ -233,7 +232,7 @@ def __next__(self):
             raise StopIteration("Stop")
 
     def add_resource_dict(self, resource: Resource):
-        self.resources.update(resource)
+        self.resources.update({resource.resource_id: resource})
 
     def get_resource_by_id(self, resource_id):
         return self.resources.get(resource_id)

lekin/lekin_struct/timeslot.py

@@ -3,6 +3,7 @@
 """
 
 from datetime import datetime, timedelta
+from typing import Any, Callable, Dict, Generator, List, Optional, Tuple, Type, Union
 
 import pandas as pd
 

lekin/solver/__init__.py

@@ -6,9 +6,12 @@
 
 from datetime import datetime, timedelta
 import heapq
+from typing import Any, Callable, Dict, Generator, List, Optional, Tuple, Type, Union
 
 import pandas as pd
 
+from lekin.lekin_struct.operation import Operation
+
 
 class CTPSolver(object):
     def __init__(self, optimizers=None):
@@ -90,10 +93,10 @@ def get_operation_time_constraint():
 
 
 def find_best_resource_and_timeslot_for_operation(
-    operation, earliest_start_time=None, latest_end_time=None, allowed_conflict=False
+    operation: Operation, earliest_start_time=None, latest_end_time=None, allowed_conflict=False
 ):
     # assign operation
-    resource_timeslots_pq = []  # Create a priority queue to store the possible resource-time slot pairs
+    resource_timeslots_pq: List = []  # Create a priority queue to store the possible resource-time slot pairs
     available_resource = operation.available_resource
     # if operation.required_resource_priority is not None:
     #     required_resource.sort()  # Sort by resource priority

lekin/solver/meta_heuristics/critical_path.py

@@ -1,6 +1,7 @@
 """Critical path"""
 
 import copy
+from typing import Any, Callable, Dict, Generator, List, Optional, Tuple, Type, Union
 
 
 class CriticalPathScheduler:
@@ -26,7 +27,7 @@ def schedule(self):
         return critical_path_schedule
 
     def forward_pass(self):
-        forward_pass_times = {}
+        forward_pass_times: Dict[str, Dict[str, float]] = {}
         for job in self.job_collector.jobs:
             forward_pass_times[job.id] = {}
             for operation in job.route.operations:
@@ -42,7 +43,7 @@ def forward_pass(self):
         return forward_pass_times
 
     def backward_pass(self):
-        backward_pass_times = {}
+        backward_pass_times: Dict[str, Dict[str, float]] = {}
         for job in self.job_collector.jobs:
             backward_pass_times[job.id] = {}
             for operation in reversed(job.route.operations):

lekin/solver/meta_heuristics/nsga3.py

@@ -1,4 +1,5 @@
 import random
+from typing import Any, Callable, Dict, Generator, List, Optional, Tuple, Type, Union
 
 
 class NSGA3Scheduler:
@@ -18,7 +19,7 @@ def initialize_population(self):
 
     def generate_random_solution(self):
         # Generate a random solution representing start times for operations
-        solution = {}
+        solution: Dict = {}
         for job in self.jobs:
             solution[job] = {}
             for operation in job.route.operations: