Merge pull request #2 from HPI-Information-Systems:feat/reference-data

Support loading reference data

Author: lisehr

.gitignore

@@ -9,3 +9,4 @@ out/
 .idea/
 *.pyc
 __pycache__/
+.venv/

demo/getting_started.py

@@ -1,7 +1,7 @@
 from metis.dq_orchestrator import DQOrchestrator
 
 # No config file means default to console writer
-orchestrator = DQOrchestrator(writer_config="configs/writer/sqlite.json")
+orchestrator = DQOrchestrator(writer_config_path="configs/writer/sqlite.json")
 
 orchestrator.load(data_loader_configs=["data/adult.json"])
 

metis/dq_orchestrator.py

@@ -1,27 +1,29 @@
-from typing import List
-import pandas as pd
 import json
-import os
-import sqlite3
+from typing import Dict, List, Type
+
+import pandas as pd
 
-import metis.globals
+from metis.loader.csv_loader import CSVLoader
 from metis.metric import Metric
 from metis.utils.data_config import DataConfig
 from metis.utils.result import DQResult
-from metis.loader.csv_loader import CSVLoader
-from metis.writer.sqlite_writer import SQLiteWriter
-from metis.writer.postgres_writer import PostgresWriter
 from metis.writer.console_writer import ConsoleWriter
+from metis.writer.postgres_writer import PostgresWriter
+from metis.writer.sqlite_writer import SQLiteWriter
+
 
 class DQOrchestrator:
-    def __init__(self, writer_config=None) -> None:
-        self.dataframes = {}
-        self.data_paths = {}
-        self.results = {} #TODO: Decide what to do with these in memory results
+    def __init__(self, writer_config_path: str | None = None) -> None:
+        self.dataframes: Dict[str, pd.DataFrame] = {}
+        self.reference_dataframes: Dict[str, pd.DataFrame] = {}
+        self.data_paths: Dict[str, str] = {}
+        self.results: Dict[str, DQResult] = (
+            {}
+        )  # TODO: Decide what to do with these in memory results
 
         self.writer = ConsoleWriter({})
-        if writer_config:
-            with open(writer_config, 'r') as f:
+        if writer_config_path:
+            with open(writer_config_path, "r") as f:
                 writer_config = json.load(f)
             if not "writer_name" in writer_config:
                 raise ValueError("Writer config must include 'writer_name' field.")
@@ -32,36 +34,46 @@ def __init__(self, writer_config=None) -> None:
 
     def load(self, data_loader_configs: List[str]) -> None:
         for config_path in data_loader_configs:
-            with open(config_path, 'r') as f:
+            with open(config_path, "r") as f:
                 config_data = json.load(f)
                 config = DataConfig(config_data)
-                config.file_name = os.path.join(metis.globals.data_root, config.file_name)
+
                 if config.loader == "CSV":
                     loader = CSVLoader()
                     dataframe = loader.load(config)
                     self.dataframes[config.name] = dataframe
                     self.data_paths[config.name] = config_path
 
+                    if config.reference_file_name:
+                        reference_config = DataConfig(config_data)
+                        reference_config.file_name = config.reference_file_name
+                        reference_dataframe = loader.load(reference_config)
+                        self.reference_dataframes[config.name] = reference_dataframe
                 else:
-                    raise ValueError(f"Unsupported loader type: {config_data.get('loader', None)}")
-            
-    def assess(self, metrics: List[str], metric_configs: List[str]) -> None:
+                    raise ValueError(
+                        f"Unsupported loader type: {config_data.get('loader', None)}"
+                    )
+
+    def assess(self, metrics: List[str], metric_configs: List[str | None]) -> None:
         results = []
-        
+
         for metric, metric_config in zip(metrics, metric_configs):
-            metric_class = Metric.registry.get(metric)
+            metric_class: Type[Metric] | None = Metric.registry.get(metric)
             if not metric_class:
                 raise ValueError(f"Metric {metric} is not registered.")
-            metric_instance = metric_class()
+            metric_instance: Metric = metric_class()
             for df_name, df in self.dataframes.items():
-                incomplete_metric_results = metric_instance.assess(df, metric_config=metric_config) #TODO: Add reference data support
+                incomplete_metric_results = metric_instance.assess(
+                    data=df,
+                    reference=self.reference_dataframes.get(df_name),
+                    metric_config=metric_config,
+                )
                 for result in incomplete_metric_results:
                     result.tableName = df_name
                     result.dataset = self.data_paths[df_name]
                     results.append(result)
 
         self.writer.write(results)
 
-
-    def getDQResult(query: str) -> List[DQResult]:
-        pass
+    def get_dq_result(self, query: str) -> List[DQResult]:
+        return []

metis/loader/csv_loader.py

@@ -3,12 +3,13 @@
 from metis.loader.loader import DataLoader
 from metis.utils.data_config import DataConfig
 
+
 class CSVLoader(DataLoader):
     def load(self, config: DataConfig) -> pd.DataFrame:
         """
         Load data from a CSV file specified by the config.
-        
-        :param config: Path to the CSV file.
+
+        :param config: DataConfig object containing the CSV parsing details.
         :return: DataFrame containing the loaded data.
         """
 
@@ -22,4 +23,4 @@ def load(self, config: DataConfig) -> pd.DataFrame:
             parse_dates=config.parse_dates,
             decimal=config.decimals,
             thousands=config.thousands
-        )
+        )

metis/loader/loader.py

@@ -1,13 +1,15 @@
 from abc import ABC, abstractmethod
 import pandas as pd
 
+from metis.utils.data_config import DataConfig
+
 class DataLoader(ABC):
     @abstractmethod
-    def load(self, config: str) -> pd.DataFrame:
+    def load(self, config: DataConfig) -> pd.DataFrame:
         """
         Load data from a source defined by the config.
-        
-        :param config: Configuration string or path to the configuration file.
+
+        :param config: Configuration object containing the data source details.
         :return: DataFrame containing the loaded data.
         """
-        pass
+        pass

metis/utils/data_config.py

@@ -1,19 +1,33 @@
-import json
-from typing import Dict
+from pathlib import Path
+from typing import Dict, List
+
+import metis.globals
+
 
 class DataConfig:
     def __init__(
         self,
         config: Dict,
     ):
-        self.name = config.get('name')
-        self.file_name = config.get('file_name')
-        self.loader = config.get('loader')
-        self.delimiter = config.get('delimiter', ',')
-        self.encoding = config.get('encoding', 'utf-8')
-        self.header = config.get('header', 0)
-        self.nrows = config.get('nrows', None)
-        self.usecols = config.get('usecols', None)
-        self.parse_dates = config.get('parse_dates', False)
-        self.decimals = config.get('decimals', ".")
-        self.thousands = config.get('thousands', None)
+        if "file_name" not in config:
+            raise ValueError(f"Data config must include 'file_name' field.")
+        if "name" not in config:
+            raise ValueError(f"Data config must include 'name' field.")
+        self.name: str = config["name"]
+        self.file_name: str = Path(metis.globals.data_root) / config["file_name"]
+        self.reference_file_name: str | None = (
+            Path(metis.globals.data_root) / config["reference_file_name"]
+            if config.get("reference_file_name")
+            else None
+        )
+        self.loader: str | None = config.get("loader")
+        self.delimiter: str = config.get("delimiter", ",")
+        self.encoding: str = config.get("encoding", "utf-8")
+        self.header: int = config.get("header", 0)
+        self.nrows: int | None = config.get("nrows")
+        self.usecols: List[str] | None = config.get("usecols")
+        self.parse_dates: bool = config.get("parse_dates", False)
+        self.decimals: str = config.get("decimals", ".")
+        self.thousands: str | None = config.get("thousands")
+        self.decimals: str = config.get("decimals", ".")
+        self.thousands: str | None = config.get("thousands")