🔥 implement pits change requests

tracex_project/extraction/logic/modules/module_metrics_analyzer.py

@@ -61,7 +61,7 @@ def execute(
         return metrics_df
 
     @staticmethod
-    def __rate_activity_relevance(activity, condition) -> str:
+    def __rate_activity_relevance(activity: str, condition: str | None) -> str:
         category_mapping = {
             "No Relevance": 0,
             "Low Relevance": 1,
@@ -91,7 +91,9 @@ def __rate_activity_relevance(activity, condition) -> str:
 
         return category
 
-    def __rate_timestamps_correctness(self, activity, start, end) -> Tuple[str, float]:
+    def __rate_timestamps_correctness(
+        self, activity: str, start, end
+    ) -> Tuple[str, float]:
         messages = Prompt.objects.get(name="METRIC_TIMESTAMP_MESSAGES").text
         messages.append(
             {

tracex_project/extraction/logic/modules/module_time_extractor.py

@@ -100,20 +100,20 @@ def __calculate_duration(row: pd.Series) -> str:
     def __post_processing(df: pd.DataFrame) -> pd.DataFrame:
         """Fill missing values for dates with default values."""
 
-        def convert_to_datetime(df, column):
+        def convert_to_datetime(df: pd.DataFrame, column: pd.Series):
             df[column] = pd.to_datetime(
                 df[column], format="%Y%m%dT%H%M", errors="coerce"
             )
 
             return df
 
-        def set_default_date_if_na(df, column):
+        def set_default_date_if_na(df: pd.DataFrame, column: pd.Series):
             if df[column].isna().all():
                 df[column] = df[column].fillna(pd.Timestamp("2020-01-01 00:00"))
 
             return df
 
-        def fill_missing_values(df, column):
+        def fill_missing_values(df: pd.DataFrame, column: pd.Series):
             df[column] = df[column].ffill().bfill()
 
             return df

tracex_project/extraction/logic/orchestrator.py

@@ -140,11 +140,11 @@ def set_db_objects_id(self, object_name: str, object_id: int):
         """Set the database id objects for the orchestrator instance."""
         self.db_objects_id[object_name] = object_id
 
-    def get_db_objects_id(self, object_name: str):
+    def get_db_objects_id(self, object_name: str) -> int:
         """Return the database id objects for the orchestrator instance."""
         return self.db_objects_id[object_name]
 
-    def reduce_modules_to(self, modules: List):
+    def reduce_modules_to(self, modules: List) -> ExtractionConfiguration:
         """Reduce the modules of the orchestrator instance to the modules in the keyword argument."""
         old_modules: Dict[str, Any] = self.get_configuration().modules
         updated_modules: Dict[str, Any] = {
@@ -212,7 +212,7 @@ def run(self, view=None) -> pd.DataFrame:
             self.get_data().insert(0, "case:concept:name", latest_id + 1)
             self.set_default_values()
 
-    def save_results_to_db(self):
+    def save_results_to_db(self) -> None:
         """Save the trace to the database."""
         patient_journey: PatientJourney = PatientJourney.manager.get(
             pk=self.get_db_objects_id("patient_journey")
@@ -251,7 +251,7 @@ def save_results_to_db(self):
         patient_journey.trace.add(trace)
         patient_journey.save()
 
-    def set_default_values(self):
+    def set_default_values(self) -> None:
         """Set default values for all modules not executed."""
         config_modules = self.get_configuration().modules
         data = self.get_data()