fix: add more hint type (#32)

* fix: add more hint type

* Update 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

@@ -24,19 +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
-        self.makespan = None  # finish of the job
-        self.tardiness = None  # delay of the 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)
@@ -91,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,7 +3,7 @@
 """
 
 import math
-
+from typing import Any, Callable, Dict, Generator, List, Optional, Tuple, Type, Union
 import numpy as np
 import pandas as pd
 
@@ -237,7 +237,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/construction_heuristics/backward.py

@@ -3,7 +3,12 @@
 import logging
 import math
 
-from lekin.lekin_struct import JobCollector, ResourceCollector, RouteCollector, TimeSlot
+from lekin.lekin_struct.job import Job, JobCollector
+from lekin.lekin_struct.operation import Operation
+from lekin.lekin_struct.resource import ResourceCollector
+from lekin.lekin_struct.route import RouteCollector
+from lekin.lekin_struct.timeslot import TimeSlot
+
 from lekin.solver.construction_heuristics.base import BaseScheduler
 
 
@@ -14,21 +19,22 @@ def __init__(
         resource_collector: ResourceCollector,
         route_collector: RouteCollector = None,
         **kwargs,
-    ):
+    ) -> None:
+
         self.job_collector = job_collector
         self.resource_collector = resource_collector
         self.route_collector = route_collector
 
         for key, value in kwargs.items():
             setattr(self, key, value)
 
-    def run(self):
+    def run(self) -> None:
         for i, job in enumerate(self.job_collector.job_list):
             self.scheduling_job(job, self.resource_collector, self.route_collector)
         logging.info("First Scheduling Done")
         return
 
-    def scheduling_job(self, job, resource_collector, route_collector):
+    def scheduling_job(self, job: Job, resource_collector, route_collector: RouteCollector) -> None:
         logging.info(f"\nAssign Job {job.job_id}")
 
         if route_collector is not None:
@@ -70,7 +76,7 @@ def scheduling_job(self, job, resource_collector, route_collector):
         return
 
     def find_best_resource_and_timeslot_for_operation(
-        self, operation, op_latest_end=None, op_earliest_start=None, **kwargs
+        self, operation: Operation, op_latest_end=None, op_earliest_start=None, **kwargs
     ):
         available_resource = operation.available_resource
 
@@ -90,7 +96,7 @@ def find_best_resource_and_timeslot_for_operation(
 
         return chosen_resource, chosen_hours
 
-    def assign_operation(self, operation, start_time, end_time, resources):
+    def assign_operation(self, operation: Operation, start_time, end_time, resources):
         timeslot = TimeSlot(start_time, end_time)
         self.timeslots.append(timeslot)
         for resource in resources:
@@ -99,7 +105,7 @@ def assign_operation(self, operation, start_time, end_time, resources):
         # Link operation to scheduled timeslot
         operation.scheduled_timeslot = timeslot
 
-    def select_resources(self, job, operation):
+    def select_resources(self, job: Job, operation: Operation):
         available_slots = self.find_available_timeslots(job, operation)
 
         selected_resources = []

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: