implement handling of JSON columns

requirements.txt

@@ -6,6 +6,7 @@ boto3
 category_encoders==2.6.3
 click
 Jinja2
+flatten_json
 keras==2.15.*
 lazy==1.4
 loguru

setup.cfg

@@ -33,6 +33,7 @@ install_requires =
     category_encoders==2.6.3
     click
     Jinja2
+    flatten_json
     keras==2.15.*
     lazy==1.4
     loguru

src/syngen/VERSION

@@ -1 +1 @@
-0.10.2
+0.10.3rc15

src/syngen/__init__.py

@@ -1,8 +1,6 @@
 import os
 import argparse
 
-from syngen.train import preprocess_data  # noqa: F401
-
 
 base_dir = os.path.dirname(__file__)
 version_file = os.path.join(base_dir, "VERSION")

src/syngen/infer.py

@@ -9,9 +9,9 @@
 from syngen.ml.utils import (
     setup_logger,
     set_log_path,
-    check_if_logs_available
+    check_if_logs_available,
+    validate_parameter_reports
 )
-from syngen.ml.utils import validate_parameter_reports
 from syngen.ml.validation_schema import ReportTypes
 
 
@@ -22,7 +22,12 @@
 
 
 @click.command()
-@click.option("--metadata_path", type=str, default=None, help="Path to the metadata file")
+@click.option(
+    "--metadata_path",
+    type=str,
+    default=None,
+    help="Path to the metadata file"
+)
 @click.option(
     "--size",
     default=100,

src/syngen/ml/config/configurations.py

@@ -10,7 +10,8 @@
 from slugify import slugify
 
 from syngen.ml.data_loaders import DataLoader, DataFrameFetcher
-from syngen.ml.utils import slugify_attribute
+from syngen.ml.utils import slugify_attribute, fetch_unique_root
+from syngen.ml.convertor import CSVConvertor
 
 
 @dataclass
@@ -25,6 +26,7 @@ class TrainConfig:
     row_limit: Optional[int]
     table_name: Optional[str]
     metadata: Dict
+    metadata_path: Optional[str]
     reports: List[str]
     batch_size: int
     loader: Optional[Callable[[str], pd.DataFrame]]
@@ -40,8 +42,6 @@ class TrainConfig:
 
     def __post_init__(self):
         self._set_paths()
-        self._remove_existed_artifacts()
-        self._prepare_dirs()
 
     def __getstate__(self) -> Dict:
         """
@@ -57,7 +57,6 @@ def __getstate__(self) -> Dict:
     def preprocess_data(self):
         self._extract_data()
         self._save_original_schema()
-        self.columns = list(self.data.columns)
         self._remove_empty_columns()
         self._mark_removed_columns()
         self._prepare_data()
@@ -100,31 +99,6 @@ def _check_reports(self):
         """
         self._check_sample_report()
 
-    def _remove_existed_artifacts(self):
-        """
-        Remove existed artifacts from previous train process
-        """
-        if os.path.exists(self.paths["resources_path"]):
-            shutil.rmtree(self.paths["resources_path"])
-            logger.info(
-                f"The artifacts located in the path - '{self.paths['resources_path']}' "
-                f"were removed"
-            )
-        if os.path.exists(self.paths["tmp_store_path"]):
-            shutil.rmtree(self.paths["tmp_store_path"])
-            logger.info(
-                f"The artifacts located in the path - '{self.paths['tmp_store_path']}' "
-                f"were removed"
-            )
-
-    def _prepare_dirs(self):
-        """
-        Create main directories for saving original, synthetic data and model artifacts
-        """
-        os.makedirs(self.paths["model_artifacts_path"], exist_ok=True)
-        os.makedirs(self.paths["state_path"], exist_ok=True)
-        os.makedirs(self.paths["tmp_store_path"], exist_ok=True)
-
     def _fetch_dataframe(self) -> Tuple[pd.DataFrame, Dict]:
         """
         Fetch the dataframe using the callback function
@@ -140,8 +114,11 @@ def _load_source(self) -> Tuple[pd.DataFrame, Dict]:
         """
         Return dataframe and schema of original data
         """
-        if self.loader is not None:
-            return self._fetch_dataframe()
+        if os.path.exists(self.paths["path_to_flatten_metadata"]):
+            data, schema = DataLoader(self.paths["input_data_path"]).load_data()
+            self.original_schema = DataLoader(self.paths["input_data_path"]).original_schema
+            schema = CSVConvertor.schema
+            return data, schema
         else:
             data_loader = DataLoader(self.source)
             self.original_schema = data_loader.original_schema
@@ -159,8 +136,9 @@ def _remove_empty_columns(self):
         self.data = self.data.dropna(how="all", axis=1)
 
         self.dropped_columns = data_columns - set(self.data.columns)
-        if len(self.dropped_columns) > 0:
-            logger.info(f"Empty columns - {', '.join(self.dropped_columns)} were removed")
+        list_of_dropped_columns = [f"'{column}'" for column in self.dropped_columns]
+        if len(list_of_dropped_columns) > 0:
+            logger.info(f"Empty columns - {', '.join(list_of_dropped_columns)} were removed")
 
     def _mark_removed_columns(self):
         """
@@ -189,6 +167,7 @@ def _extract_data(self):
         """
         self.data, self.schema = self._load_source()
         self.initial_data_shape = self.data.shape
+        self.columns = list(self.data.columns)
         self._check_if_data_is_empty()
 
     def _preprocess_data(self):
@@ -297,6 +276,9 @@ def _set_paths(self):
                                     f"merged_infer_{self.slugify_table_name}.csv",
             "no_ml_state_path":
                 f"model_artifacts/resources/{self.slugify_table_name}/no_ml/checkpoints/",
+            "path_to_flatten_metadata":
+                f"model_artifacts/tmp_store/flatten_configs/"
+                f"flatten_metadata_{fetch_unique_root(self.table_name, self.metadata_path)}.json",
             "losses_path": f"model_artifacts/tmp_store/losses/{slugify(losses_file_name)}.csv"
         }
 
@@ -449,4 +431,7 @@ def _set_paths(self):
             "fk_kde_path": f"model_artifacts/resources/{dynamic_name}/vae/checkpoints/stat_keys/",
             "path_to_no_ml":
                 f"model_artifacts/resources/{dynamic_name}/no_ml/checkpoints/kde_params.pkl",
+            "path_to_flatten_metadata":
+                f"model_artifacts/tmp_store/flatten_configs/"
+                f"flatten_metadata_{fetch_unique_root(self.table_name, self.metadata_path)}.json"
         }

src/syngen/ml/config/validation.py

@@ -9,6 +9,7 @@
 from loguru import logger
 from syngen.ml.data_loaders import MetadataLoader, DataLoader
 from syngen.ml.validation_schema import ValidationSchema, ReportTypes
+from syngen.ml.utils import fetch_unique_root
 
 
 @dataclass
@@ -286,9 +287,8 @@ def _fetch_existed_columns(self, table_name: str) -> List[str]:
         """
         metadata_of_table = self.merged_metadata[table_name]
         format_settings = metadata_of_table.get("format", {})
-        return DataLoader(
-            metadata_of_table["train_settings"]["source"]
-        ).get_columns(**format_settings)
+        path_to_source = self._fetch_path_to_source(table_name)
+        return DataLoader(path_to_source).get_columns(**format_settings)
 
     def _gather_existed_columns(self, table_name: str):
         """
@@ -298,16 +298,34 @@ def _gather_existed_columns(self, table_name: str):
         existed_columns = self._fetch_existed_columns(table_name)
         self.existed_columns_mapping[table_name] = existed_columns
 
-    def _run(self):
+    def preprocess_metadata(self):
         """
-        Run the validation process
+        Preprocess the metadata, set the metadata and the merged metadata
         """
         self._launch_validation_of_schema()
         self._define_mapping()
         self._merge_metadata()
         self.merged_metadata.pop("global", None)
         self.metadata.pop("global", None)
 
+    def _fetch_path_to_source(self, table_name):
+        """
+        Fetch the path to the source of the certain table
+        """
+        if os.path.exists(
+            f"{os.getcwd()}/model_artifacts/tmp_store/flatten_configs/flatten_metadata_"
+            f"{fetch_unique_root(table_name, self.metadata_path)}.json"
+        ):
+            return (f"{os.getcwd()}/model_artifacts/tmp_store/{slugify(table_name)}/"
+                    f"input_data_{slugify(table_name)}.pkl")
+        return self.metadata[table_name]["train_settings"]["source"]
+
+    def _run(self):
+        """
+        Run the validation process
+        """
+        self.preprocess_metadata()
+
         if self.type_of_process == "train" and self.validation_source:
             for table_name in self.merged_metadata.keys():
                 self._gather_existed_columns(table_name)

src/syngen/ml/convertor/convertor.py

@@ -106,7 +106,7 @@ class CSVConvertor(Convertor):
     """
     Class for supporting custom schema for csv files
     """
-    schema = {"fields": {}, "format": "CSV"}
+    schema: Dict = {"fields": {}, "format": "CSV"}
 
     def __init__(self, df):
         schema = {"fields": {}, "format": "CSV"}

src/syngen/ml/data_loaders/data_loaders.py

@@ -497,6 +497,10 @@ def _load_data(self) -> pd.DataFrame:
         with open(self.path, "rb") as f:
             return pkl.load(f)
 
+    def get_columns(self) -> List[str]:
+        data, schema = self.load_data()
+        return data.columns.tolist()
+
     def load_data(self) -> Tuple[pd.DataFrame, None]:
         return self._load_data(), None
 

src/syngen/ml/handlers/handlers.py

@@ -465,6 +465,15 @@ def _restore_empty_columns(self, df: pd.DataFrame) -> pd.DataFrame:
 
         return df
 
+    def _save_data(self, generated_data):
+        """
+        Save generated data to the path
+        """
+        DataLoader(self.paths["path_to_merged_infer"]).save_data(
+            generated_data,
+            format=get_context().get_config(),
+        )
+
     def handle(self, **kwargs):
         self._prepare_dir()
         list_of_reports = [f'"{report}"' for report in self.reports]
@@ -504,9 +513,7 @@ def handle(self, **kwargs):
         if tech_columns:
             prepared_data = prepared_data.drop(tech_columns, axis=1)
             logger.debug(
-                "Technical columns "
-                f"{tech_columns} were removed "
-                "from the generated table."
+                f"Technical columns {tech_columns} were removed from the generated table."
             )
             Report().unregister_reporters(self.table_name)
             logger.info(
@@ -525,27 +532,12 @@ def handle(self, **kwargs):
                 generated_data = generated_data[self.dataset.order_of_columns]
 
                 if generated_data is None:
-                    DataLoader(self.paths["path_to_merged_infer"]).save_data(
-                        prepared_data,
-                        schema=self.original_schema,
-                        format=get_context().get_config(),
-                    )
+                    self._save_data(prepared_data)
                 else:
-                    DataLoader(self.paths["path_to_merged_infer"]).save_data(
-                        generated_data,
-                        schema=self.original_schema,
-                        format=get_context().get_config(),
-                    )
+                    self._save_data(generated_data)
             else:
-                DataLoader(self.paths["path_to_merged_infer"]).save_data(
-                    prepared_data,
-                    schema=self.original_schema,
-                    format=get_context().get_config(),
-                )
+                self._save_data(prepared_data)
         if self.metadata_path is None:
             prepared_data = prepared_data[self.dataset.order_of_columns]
-            DataLoader(self.paths["path_to_merged_infer"]).save_data(
-                prepared_data,
-                schema=self.original_schema,
-                format=get_context().get_config(),
-            )
+
+            self._save_data(prepared_data)

src/syngen/ml/metrics/accuracy_test/accuracy_test.py

@@ -268,8 +268,12 @@ def _generate_report(
             utility_barplot=transform_to_base64(
                 f"{self.reports_path}/utility_barplot.svg"
             ),
-            utility_table=utility_result.to_html(),
-            is_data_available=False if utility_result.empty else True,
+            utility_table=utility_result.to_html() if utility_result is not None else None,
+            is_data_available=(
+                False
+                if utility_result is None or (utility_result is not None and utility_result.empty)
+                else True
+            ),
             table_name=self.table_name,
             training_config=train_config,
             inference_config=infer_config,
@@ -296,9 +300,14 @@ def report(self, *args, **kwargs):
         ) = metrics
         MlflowTracker().log_metrics(
             {
-                "Utility_avg": utility_result["Synth to orig ratio"].mean(),
-                "Clustering": clustering_result if clustering_result is not None
-                else np.NaN,
+                "Utility_avg": (
+                    utility_result["Synth to orig ratio"].mean()
+                    if utility_result is not None else None
+                ),
+                "Clustering": (
+                    clustering_result
+                    if clustering_result is not None else np.NaN
+                ),
                 "Accuracy": acc_median,
                 "Correlation": corr_result,
             }

src/syngen/ml/metrics/metrics_classes/metrics.py

@@ -1,4 +1,4 @@
-from typing import Union, List, Optional, Dict
+from typing import Union, List, Optional, Dict, Literal
 from abc import ABC
 from itertools import combinations
 from collections import Counter
@@ -1097,6 +1097,19 @@ def __init__(
 
         self.sample_size = sample_size
 
+    @staticmethod
+    def check_empty_df(df: pd.DataFrame, df_type: Literal["original", "synthetic"]) -> bool:
+        """
+        Check if the dataframe is empty after dropping rows with missing values
+        """
+        if df.empty:
+            logger.warning(
+                f"Utility metric calculation is skipped: the {df_type} dataframe is empty "
+                "after dropping rows with missing values (dropna() function is applied)"
+            )
+            return True
+        return False
+
     def calculate_all(self, categorical_columns: List[str], cont_columns: List[str]):
         logger.info("Calculating utility metric")
 
@@ -1115,9 +1128,14 @@ def calculate_all(self, categorical_columns: List[str], cont_columns: List[str])
         self.synthetic = self.synthetic[cont_columns + categorical_columns].apply(
             pd.to_numeric, axis=0, errors="ignore"
         )
-
         self.original = self.original.select_dtypes(include="number").dropna()
         self.synthetic = self.synthetic.select_dtypes(include="number").dropna()
+
+        if self.check_empty_df(self.original, "original"):
+            return
+        if self.check_empty_df(self.synthetic, "synthetic"):
+            return
+
         self.synthetic = self.synthetic[self.original.columns]
 
         excluded_cols = [

src/syngen/ml/processors/__init__.py

@@ -0,0 +1 @@
+from syngen.ml.processors.processors import PreprocessHandler, PostprocessHandler  # noqa: F401