Merge branch 'dev/jab' of https://github.com/DataIntegrationGroup/DataIntegrationEngine into dev/jab

Author: jacob-a-brown

backend/persister.py

@@ -17,7 +17,7 @@
 import io
 import os
 import shutil
-from pprint import pprint   
+from pprint import pprint
 import json
 
 import pandas as pd
@@ -37,6 +37,7 @@ class BasePersister(Loggable):
     Class to persist the data to a file or cloud storage.
     If persisting to a file, the output directory is created by config._make_output_path()
     """
+
     add_extension: str = "csv"
 
     def __init__(self):
@@ -107,13 +108,14 @@ def add_extension(self, path: str):
 
     def _write(self, path: str, records):
         raise NotImplementedError
-    
+
     def _dump_timeseries(self, path: str, timeseries: list):
         raise NotImplementedError
 
     def _make_output_directory(self, output_directory: str):
         os.mkdir(output_directory)
 
+
 def write_csv_file(path, func, records):
     with open(path, "w", newline="") as f:
         func(csv.writer(f), records)
@@ -221,19 +223,25 @@ def _write(self, path: str, records: list):
                 "type": "Feature",
                 "geometry": {
                     "type": "Point",
-                    "coordinates": [record.get("longitude"), record.get("latitude"), record.get("elevation")],
+                    "coordinates": [
+                        record.get("longitude"),
+                        record.get("latitude"),
+                        record.get("elevation"),
+                    ],
+                },
+                "properties": {
+                    k: record.get(k)
+                    for k in record.keys
+                    if k not in ["latitude", "longitude", "elevation"]
                 },
-                "properties": {k: record.get(k) for k in record.keys if k not in ["latitude", "longitude", "elevation"]},
             }
             for record in records
         ]
         feature_collection["features"].extend(features)
 
-
         with open(path, "w") as f:
             json.dump(feature_collection, f, indent=4)
 
-
     def _get_gdal_type(self, dtype):
         """
         Map pandas dtypes to GDAL-compatible types for the schema.
@@ -249,6 +257,7 @@ def _get_gdal_type(self, dtype):
         else:
             return "str"  # Default to string for unsupported types
 
+
 # class ST2Persister(BasePersister):
 #     extension = "st2"
 #

backend/record.py

@@ -30,37 +30,37 @@ def to_csv(self):
 
     def __init__(self, payload):
         self._payload = payload
-    
+
     def get(self, attr):
-            # v = self._payload.get(attr)
-            # if v is None and self.defaults:
-            #     v = self.defaults.get(attr)
-            v = self.__getattr__(attr)
-
-            field_sigfigs = [
-                ("elevation", 2),
-                ("well_depth", 2),
-                ("latitude", 6),
-                ("longitude", 6),
-                ("min", 2),
-                ("max", 2),
-                ("mean", 2),
-            ]
-
-            # both analyte and water level tables have the same fields, but the
-            # rounding should only occur for water level tables
-            if isinstance(self, WaterLevelRecord):
-                field_sigfigs.append((PARAMETER_VALUE, 2))
-
-            for field, sigfigs in field_sigfigs:
-                if v is not None and field == attr:
-                    try:
-                        v = round(v, sigfigs)
-                    except TypeError as e:
-                        print(field, attr)
-                        raise e
-                    break
-            return v
+        # v = self._payload.get(attr)
+        # if v is None and self.defaults:
+        #     v = self.defaults.get(attr)
+        v = self.__getattr__(attr)
+
+        field_sigfigs = [
+            ("elevation", 2),
+            ("well_depth", 2),
+            ("latitude", 6),
+            ("longitude", 6),
+            ("min", 2),
+            ("max", 2),
+            ("mean", 2),
+        ]
+
+        # both analyte and water level tables have the same fields, but the
+        # rounding should only occur for water level tables
+        if isinstance(self, WaterLevelRecord):
+            field_sigfigs.append((PARAMETER_VALUE, 2))
+
+        for field, sigfigs in field_sigfigs:
+            if v is not None and field == attr:
+                try:
+                    v = round(v, sigfigs)
+                except TypeError as e:
+                    print(field, attr)
+                    raise e
+                break
+        return v
 
     def to_row(self):
         return [self.get(k) for k in self.keys]

backend/unifier.py

@@ -114,7 +114,9 @@ def _perister_factory(config):
 #     persister.save(config.output_path)
 
 
-def _site_wrapper(site_source, parameter_source, sites_summary_persister, timeseries_persister, config):
+def _site_wrapper(
+    site_source, parameter_source, sites_summary_persister, timeseries_persister, config
+):
 
     try:
         # TODO: fully develop checks/discoveries below
@@ -203,16 +205,25 @@ def _site_wrapper(site_source, parameter_source, sites_summary_persister, timese
                         # num_sites_to_remove from the length of the list
                         # to remove the last num_sites_to_remove sites
                         if use_summarize:
-                            sites_summary_persister.records = sites_summary_persister.records[
-                                : len(sites_summary_persister.records) - num_sites_to_remove
-                            ]
+                            sites_summary_persister.records = (
+                                sites_summary_persister.records[
+                                    : len(sites_summary_persister.records)
+                                    - num_sites_to_remove
+                                ]
+                            )
                         else:
-                            timeseries_persister.timeseries = timeseries_persister.timeseries[
-                                : len(timeseries_persister.timeseries) - num_sites_to_remove
-                            ]
-                            sites_summary_persister.sites = sites_summary_persister.sites[
-                                : len(sites_summary_persister.sites) - num_sites_to_remove
-                            ]
+                            timeseries_persister.timeseries = (
+                                timeseries_persister.timeseries[
+                                    : len(timeseries_persister.timeseries)
+                                    - num_sites_to_remove
+                                ]
+                            )
+                            sites_summary_persister.sites = (
+                                sites_summary_persister.sites[
+                                    : len(sites_summary_persister.sites)
+                                    - num_sites_to_remove
+                                ]
+                            )
                         break
 
     except BaseException:
@@ -230,7 +241,13 @@ def _unify_parameter(
     sites_summary_persister = _perister_factory(config)
     timeseries_persister = CSVPersister()
     for site_source, parameter_source in sources:
-        _site_wrapper(site_source, parameter_source, sites_summary_persister, timeseries_persister, config)
+        _site_wrapper(
+            site_source,
+            parameter_source,
+            sites_summary_persister,
+            timeseries_persister,
+            config,
+        )
 
     if config.output_summary:
         sites_summary_persister.dump_summary(config.output_path)

tests/test_sources/__init__.py

@@ -14,10 +14,14 @@
 EXCLUDED_GEOJSON_KEYS = ["latitude", "longitude", "elevation"]
 
 SUMMARY_RECORD_CSV_HEADERS = list(SummaryRecord.keys)
-SUMMARY_RECORD_GEOJSON_KEYS = [k for k in SUMMARY_RECORD_CSV_HEADERS if k not in EXCLUDED_GEOJSON_KEYS]
+SUMMARY_RECORD_GEOJSON_KEYS = [
+    k for k in SUMMARY_RECORD_CSV_HEADERS if k not in EXCLUDED_GEOJSON_KEYS
+]
 
 SITE_RECORD_CSV_HEADERS = list(SiteRecord.keys)
-SITE_RECORD_GEOJSON_KEYS = [k for k in SITE_RECORD_CSV_HEADERS if k not in EXCLUDED_GEOJSON_KEYS]
+SITE_RECORD_GEOJSON_KEYS = [
+    k for k in SITE_RECORD_CSV_HEADERS if k not in EXCLUDED_GEOJSON_KEYS
+]
 
 PARAMETER_RECORD_HEADERS = list(ParameterRecord.keys)
 
@@ -96,7 +100,9 @@ def _check_summary_file(self, extension: str):
                 for feature in summary["features"]:
                     assert feature["geometry"]["type"] == "Point"
                     assert len(feature["geometry"]["coordinates"]) == 3
-                    assert sorted(feature["properties"].keys()) == sorted(SUMMARY_RECORD_GEOJSON_KEYS)
+                    assert sorted(feature["properties"].keys()) == sorted(
+                        SUMMARY_RECORD_GEOJSON_KEYS
+                    )
                 assert summary["features"][0]["type"] == "Feature"
         else:
             raise ValueError(f"Unsupported file extension: {extension}")
@@ -122,7 +128,9 @@ def _check_sites_file(self, extension: str):
                 for feature in sites["features"]:
                     assert feature["geometry"]["type"] == "Point"
                     assert len(feature["geometry"]["coordinates"]) == 3
-                    assert sorted(feature["properties"].keys()) == sorted(SITE_RECORD_GEOJSON_KEYS)
+                    assert sorted(feature["properties"].keys()) == sorted(
+                        SITE_RECORD_GEOJSON_KEYS
+                    )
                 assert sites["features"][0]["type"] == "Feature"
         else:
             raise ValueError(f"Unsupported file extension: {extension}")
@@ -236,7 +244,6 @@ def test_timeseries_separated_geojson(self):
 
         for timeseries_file in timeseries_dir.iterdir():
             self._check_timeseries_file(timeseries_dir, timeseries_file.name)
-   
 
     @pytest.mark.skip(reason="test_date_range not implemented yet")
     def test_date_range(self):