🔥 implement Pits Change Requests

tracex_project/extraction/logic/modules/module_location_extractor.py

@@ -41,7 +41,7 @@ def execute(
         return df
 
     @staticmethod
-    def __classify_location(activity_label):
+    def __classify_location(activity_label: str) -> str:
         """Classify the location for a given activity."""
         messages = Prompt.objects.get(name="LOCATION_MESSAGES").text
         messages.append({"role": "user", "content": activity_label})

tracex_project/extraction/logic/modules/module_metrics_analyzer.py

@@ -1,5 +1,6 @@
 """This module measures the outpupt of the pipeline based on specified metrics."""
 from pathlib import Path
+from typing import Tuple
 import pandas as pd
 from django.conf import settings
 
@@ -60,7 +61,7 @@ def execute(
         return metrics_df
 
     @staticmethod
-    def __rate_activity_relevance(activity, condition):
+    def __rate_activity_relevance(activity, condition) -> str:
         category_mapping = {
             "No Relevance": 0,
             "Low Relevance": 1,
@@ -90,7 +91,7 @@ def __rate_activity_relevance(activity, condition):
 
         return category
 
-    def __rate_timestamps_correctness(self, activity, start, end):
+    def __rate_timestamps_correctness(self, activity, start, end) -> Tuple[str, float]:
         messages = Prompt.objects.get(name="METRIC_TIMESTAMP_MESSAGES").text
         messages.append(
             {

tracex_project/extraction/logic/modules/module_patient_journey_preprocessor.py

@@ -22,7 +22,7 @@ def __init__(self):
     @log_execution_time(Path(settings.BASE_DIR / "tracex/logs/execution_time.log"))
     def execute(
         self, _input=None, patient_journey=None, patient_journey_sentences=None
-    ):
+    ) -> str:
         """Preprocesses the patient input for better data quality."""
         super().execute(
             _input,
@@ -57,7 +57,7 @@ def execute(
         return preprocessed_text
 
     @staticmethod
-    def __apply_preprocessing_step(text, prompt_name):
+    def __apply_preprocessing_step(text: str, prompt_name: str) -> str:
         """Applies a preprocessing step based on the step name."""
         messages = Prompt.objects.get(name=f"PREPROCESSING_{prompt_name}").text
         new_user_message = {"role": "user", "content": text}

tracex_project/extraction/logic/modules/module_time_extractor.py

@@ -45,7 +45,7 @@ def execute(
 
         return df
 
-    def __extract_start_date(self, row):
+    def __extract_start_date(self, row: pd.Series) -> str:
         """Extract the start date for a given activity."""
         lower, upper = u.get_snippet_bounds(
             index=(int(row["sentence_id"])), length=len(self.patient_journey_sentences)
@@ -65,7 +65,7 @@ def __extract_start_date(self, row):
 
         return start
 
-    def __extract_end_date(self, row):
+    def __extract_end_date(self, row: pd.Series) -> str:
         """Extract the end date for a given activity."""
         lower, upper = u.get_snippet_bounds(
             index=(int(row["sentence_id"])), length=len(self.patient_journey_sentences)
@@ -88,7 +88,7 @@ def __extract_end_date(self, row):
         return end
 
     @staticmethod
-    def __calculate_duration(row):
+    def __calculate_duration(row: pd.Series) -> str:
         """Calculate the duration of an activity."""
         duration = row["time:end_timestamp"] - row["time:timestamp"]
         hours, remainder = divmod(duration.total_seconds(), 3600)
@@ -97,7 +97,7 @@ def __calculate_duration(row):
         return f"{int(hours):02d}:{int(minutes):02d}:{int(seconds):02d}"
 
     @staticmethod
-    def __post_processing(df):
+    def __post_processing(df: pd.DataFrame) -> pd.DataFrame:
         """Fill missing values for dates with default values."""
 
         def convert_to_datetime(df, column):