Expand capability of geojson class

geojson_modelica_translator/geojson/urbanopt_geojson.py

@@ -1,40 +1,19 @@
 # :copyright (c) URBANopt, Alliance for Sustainable Energy, LLC, and other contributors.
 # See also https://github.com/urbanopt/geojson-modelica-translator/blob/develop/LICENSE.md
 
+import json
 import logging
 from pathlib import Path
 
 import geojson
 from jsonpath_ng.ext import parse
 
 from geojson_modelica_translator.geojson.schemas import Schemas
+from geojson_modelica_translator.geojson.urbanopt_load import GeoJsonValidationError, UrbanOptLoad
 
 _log = logging.getLogger(__name__)
 
 
-class GeoJsonValidationError(Exception):
-    pass
-
-
-# TODO: Inherit from GeoJSON Feature class, move to its own file
-class UrbanOptLoad:
-    """An UrbanOptLoad is a container for holding Building-related data in a dictionary. This object
-    does not do much work on the GeoJSON definition of the data at the moment, rather it creates
-    an isolation layer between the GeoJSON data and the GMT.
-    """
-
-    def __init__(self, feature):
-        self.feature = feature
-        self.id = feature.get("properties", {}).get("id", None)
-
-        # do some validation
-        if self.id is None:
-            raise GeoJsonValidationError("GeoJSON feature requires an ID property but value was null")
-
-    def __str__(self):
-        return f"ID: {self.id}"
-
-
 class UrbanOptGeoJson:
     """Root class for parsing an URBANopt GeoJSON file. This class simply reads and parses
     URBANopt GeoJSON files.
@@ -47,6 +26,8 @@ def __init__(self, filename, building_ids=None, skip_validation=False):
         :param building_ids: list[str | int] | None, optional, list of GeoJSON building
             IDs to parse from the file. If None or an empty list, parse all buildings.
         """
+
+        self._filename = Path(filename).resolve()
         if not Path(filename).exists():
             raise GeoJsonValidationError(f"URBANopt GeoJSON file does not exist: {filename}")
 
@@ -62,7 +43,7 @@ def __init__(self, filename, building_ids=None, skip_validation=False):
             if feature["properties"]["type"] == "Building":
                 building = UrbanOptLoad(feature)
                 if not building_ids or building.id in building_ids:
-                    # Ignore validation failures for features with 'detailed_model_filename' in the properties
+                    # Do not attempt validation for features with 'detailed_model_filename' in the properties
                     # Buildings defined by an osm don't have all keys in geojson, therefore will always fail validation
                     if "detailed_model_filename" not in feature["properties"]:
                         errors = self.schemas.validate("building", building.feature.properties)
@@ -97,6 +78,8 @@ def get_feature_by_id(self, feature_id=None):
         for feature in self.data.features:
             if feature["properties"]["id"] == str(feature_id):
                 return feature
+        if feature_id not in self.data.features:
+            raise KeyError(f"No matches found for id {feature_id}")
 
     def get_feature(self, jsonpath):
         """Return the parameter(s) from a jsonpath.
@@ -118,7 +101,160 @@ def get_feature(self, jsonpath):
             # If only one value, then return that value and not a list of values
             results = results[0]
         elif len(results) == 0:
-            return print(f"No matches found for jsonpath {jsonpath}")
+            raise KeyError(f"No matches found for jsonpath {jsonpath}")
 
         # otherwise return the list of values
         return results
+
+    # TODO: test the following methods
+    def get_building_paths(self, scenario_name: str) -> list[Path]:
+        """Return a list of Path objects for the building GeoJSON files"""
+        result = []
+        for feature in self.data["features"]:
+            if feature["properties"]["type"] == "Building":
+                building_path = self._filename.parent / "run" / scenario_name / feature["properties"]["id"]
+                result.append(building_path)
+                # result.append(Path(feature["properties"]["file"]))
+
+        # verify that the paths exist
+        for path in result:
+            if not path.exists():
+                raise FileNotFoundError(f"File not found: {path}")
+
+        return result
+
+    def get_building_ids(self) -> list:
+        """Return a list of building names"""
+        result = []
+        for feature in self.data["features"]:
+            if "type" in feature["properties"] and feature["properties"]["type"] == "Building":
+                result.append(feature["properties"]["id"])
+            elif "name" in feature["properties"] and feature["properties"]["name"] == "Site Origin":
+                pass
+            else:
+                # need to implement a reasonable logger.
+                pass
+                # print(f"Feature does not have a type Building: {feature}")
+                # print("Did you forget to call the `update_geojson_from_seed_data` method?")
+
+        return result
+
+    def get_building_names(self) -> list:
+        """Return a list of building names. Typically this field is only used for visual display name only."""
+        result = []
+        for feature in self.data["features"]:
+            if feature["properties"]["type"] == "Building":
+                result.append(feature["properties"]["name"])
+
+        return result
+
+    def get_buildings(self, ids: list[str] | None = None) -> list:
+        """Return a list of all the properties of type Building"""
+        result = []
+        for feature in self.data["features"]:
+            if feature["properties"]["type"] == "Building" and (ids is None or feature["properties"]["id"] in ids):
+                # TODO: eventually add a list of building ids to keep, for now it
+                # will be all buildings.
+                result.append(feature)
+
+        return result
+
+    def get_building_properties_by_id(self, building_id: str) -> dict:
+        """Get the list of building ids in the GeoJSON file. The Building id is what
+        is used in URBANopt as the identifier. It is common that this is used to name
+        the building, more than the GeoJSON's building name field.
+
+        Args:
+            building_id (str): building id, this is the property.id values in the geojson's feature
+
+        Returns:
+            dict: building properties
+        """
+        result = {}
+        for feature in self.data["features"]:
+            if feature["properties"]["type"] == "Building" and feature["properties"]["id"] == building_id:
+                result = feature["properties"]
+
+        return result
+
+    def get_meters_for_building(self, building_id: str) -> list:
+        """Return a list of meters for the building_id"""
+        result = []
+        for feature in self.data["features"]:
+            if feature["properties"]["type"] == "Building" and feature["properties"]["id"] == building_id:
+                for meter in feature["properties"].get("meters", []):
+                    result.append(meter["type"])
+
+        if not result:
+            raise KeyError(f"No meters found for building {building_id}")
+        return result
+
+    def get_meter_readings_for_building(self, building_id: str, meter_type: str) -> list:
+        """Return a list of meter readings for the building_id"""
+        result = []
+        for feature in self.data["features"]:
+            if feature["properties"]["type"] == "Building" and feature["properties"]["id"] == building_id:
+                for meter in feature["properties"].get("meters", []):
+                    if meter["type"] == meter_type:
+                        result = meter["readings"]
+
+        if not result:
+            raise KeyError(f"No meter readings found for building {building_id}")
+        return result
+
+    def get_monthly_readings(self, building_id: str, meter_type: str = "Electricity") -> list:
+        """Return a list of monthly electricity consumption for the building_id"""
+        result = []
+        for feature in self.data["features"]:
+            if (
+                feature["properties"]["type"] == "Building"
+                and feature["properties"]["id"] == building_id
+                and meter_type == "Electricity"
+            ):
+                result = feature["properties"].get("monthly_electricity")
+
+        if not result:
+            raise KeyError(f"No monthly readings found for building {building_id}")
+        return result
+
+    def set_property_on_building_id(
+        self, building_id: str, property_name: str, property_value: str, overwrite=True
+    ) -> None:
+        """Set a property on a building_id.
+
+        Note this method does not change the GeoJSON file, it only changes the in-memory data."""
+        for feature in self.data["features"]:
+            if (
+                feature["properties"]["type"] == "Building"
+                and feature["properties"]["id"] == building_id
+                and (overwrite or property_name not in feature["properties"])
+            ):
+                feature["properties"][property_name] = property_value
+
+    def get_property_by_building_id(self, building_id: str, property_name: str) -> str | None:
+        """Get a property on a building_id"""
+        for feature in self.data["features"]:
+            if feature["properties"]["type"] == "Building" and feature["properties"]["id"] == building_id:
+                return feature["properties"].get(property_name, None)
+        return None
+
+    def get_site_lat_lon(self) -> list | None:
+        """Return the site's latitude and longitude
+
+        Rounds to 6 decimal places, if the geojson file has more than 6 decimal places.
+        Returns None if the site origin is not found."""
+        for feature in self.data["features"]:
+            if feature["properties"]["name"] == "Site Origin":
+                # reverse the order of the coordinates
+                return feature["geometry"]["coordinates"][::-1]
+        _log.warning("Site Origin not found in GeoJSON file")
+        return None
+
+    def save(self) -> None:
+        """Save the GeoJSON file"""
+        self.save_as(self._filename)
+
+    def save_as(self, filename: Path) -> None:
+        """Save the GeoJSON file"""
+        with open(filename, "w") as f:
+            json.dump(self.data, f, indent=2)

geojson_modelica_translator/geojson/urbanopt_load.py

@@ -0,0 +1,20 @@
+class GeoJsonValidationError(Exception):
+    pass
+
+
+class UrbanOptLoad:
+    """An UrbanOptLoad is a container for holding Building-related data in a dictionary. This object
+    does not do much work on the GeoJSON definition of the data at the moment, rather it creates
+    an isolation layer between the GeoJSON data and the GMT.
+    """
+
+    def __init__(self, feature):
+        self.feature = feature
+        self.id = feature.get("properties", {}).get("id", None)
+
+        # do some validation
+        if self.id is None:
+            raise GeoJsonValidationError("GeoJSON feature requires an ID property but value was null")
+
+    def __str__(self):
+        return f"ID: {self.id}"

geojson_modelica_translator/model_connectors/model_base.py

@@ -62,7 +62,7 @@ def __init__(self, system_parameters, template_dir):
                 }
             # Get access to loop order output from ThermalNetwork package.
             if "fifth_generation" in district_params and "ghe_parameters" in district_params["fifth_generation"]:
-                self.loop_order: list = load_loop_order(self.system_parameters.filename)
+                self.loop_order = load_loop_order(self.system_parameters.filename)
 
     def ft2_to_m2(self, area_in_ft2: float) -> float:
         """Converts square feet to square meters

tests/geojson/test_geojson.py

@@ -36,3 +36,124 @@ def test_validate(self):
         filename = self.data_dir / "geojson_1_invalid.json"
         with pytest.raises(GeoJsonValidationError, match="is not valid under any of the given schemas"):
             UrbanOptGeoJson(filename)
+
+    def test_get_all_features(self):
+        filename = self.data_dir / "geojson_1.json"
+        json = UrbanOptGeoJson(filename)
+        feature_properties = json.get_feature("$.features.[*].properties")
+        assert len(feature_properties) == 4
+        # Check that the first feature has the expected properties
+        assert feature_properties[0]["floor_height"] == 9
+
+    def test_get_feature(self):
+        filename = self.data_dir / "geojson_1.json"
+        json = UrbanOptGeoJson(filename)
+        feature = json.get_feature("$.features[1]")
+        assert feature["properties"]["floor_height"] == 3
+
+    def test_get_feature_invalid(self):
+        filename = self.data_dir / "geojson_1.json"
+        json = UrbanOptGeoJson(filename)
+        with pytest.raises(KeyError, match="No matches found"):
+            json.get_feature("$.features[4]")
+
+    def test_get_feature_by_id(self):
+        filename = self.data_dir / "geojson_1.json"
+        json = UrbanOptGeoJson(filename)
+        feature = json.get_feature_by_id("5a7229e737f4de77124f946d")
+        assert feature["properties"]["footprint_area"] == 8612
+
+    def test_get_feature_by_id_invalid(self):
+        filename = self.data_dir / "geojson_1.json"
+        json = UrbanOptGeoJson(filename)
+        with pytest.raises(KeyError, match="No matches found"):
+            json.get_feature_by_id("non-existent-id")
+
+    def test_get_feature_by_id_missing(self):
+        filename = self.data_dir / "geojson_1.json"
+        json = UrbanOptGeoJson(filename)
+        with pytest.raises(SystemExit):
+            json.get_feature_by_id()
+
+    def test_get_building_paths(self):
+        filename = self.data_dir / "geojson_1.json"
+        json = UrbanOptGeoJson(filename)
+        building_paths = json.get_building_paths(scenario_name="baseline_test")
+        assert len(building_paths) == 3
+        # Check that the building paths end with the dir of the building_id
+        assert building_paths[0].stem == "5a6b99ec37f4de7f94020090"
+        assert building_paths[1].stem == "5a72287837f4de77124f946a"
+        assert building_paths[2].stem == "5a7229e737f4de77124f946d"
+        # Check that the correct error is raised if the path doesn't exist
+        with pytest.raises(FileNotFoundError, match="File not found"):
+            json.get_building_paths(scenario_name="baseline")
+
+    def test_get_building_ids(self):
+        filename = self.data_dir / "geojson_1.json"
+        json = UrbanOptGeoJson(filename)
+        building_names = json.get_building_names()
+        assert len(building_names) == 3
+        assert building_names[0] == "Medium Office"
+
+    def test_get_buildings(self):
+        filename = self.data_dir / "geojson_1.json"
+        json = UrbanOptGeoJson(filename)
+        buildings = json.get_buildings(ids=None)
+        assert len(buildings) == 3
+        assert buildings[2]["properties"]["floor_area"] == 34448
+
+    def test_get_building_properties_by_id(self):
+        filename = self.data_dir / "geojson_1.json"
+        json = UrbanOptGeoJson(filename)
+        building_properties = json.get_building_properties_by_id("5a72287837f4de77124f946a")
+        assert building_properties["floor_area"] == 24567
+
+    def test_get_meters_for_building(self):
+        filename = self.data_dir / "geojson_1.json"
+        json = UrbanOptGeoJson(filename)
+        with pytest.raises(KeyError, match="No meters found"):
+            json.get_meters_for_building("5a72287837f4de77124f946a")
+
+    def test_get_meter_readings_for_building(self):
+        filename = self.data_dir / "geojson_1.json"
+        json = UrbanOptGeoJson(filename)
+        with pytest.raises(KeyError, match="No meter readings found"):
+            json.get_meter_readings_for_building(building_id="5a72287837f4de77124f946a", meter_type="Electricity")
+
+    def test_get_monthly_readings(self):
+        filename = self.data_dir / "geojson_1.json"
+        json = UrbanOptGeoJson(filename)
+        with pytest.raises(KeyError, match="No monthly readings found"):
+            json.get_monthly_readings(building_id="5a72287837f4de77124f946a")
+
+    def test_set_property_on_building_id(self):
+        filename = self.data_dir / "geojson_1.json"
+        json = UrbanOptGeoJson(filename)
+        building_id = "5a72287837f4de77124f946a"
+        property_name = "floor_area"
+        property_value = 12345
+        json.set_property_on_building_id(building_id, property_name, property_value)
+        assert json.get_building_properties_by_id(building_id)[property_name] == property_value
+
+    def test_get_property_by_building_id(self):
+        filename = self.data_dir / "geojson_1.json"
+        json = UrbanOptGeoJson(filename)
+        building_id = "5a72287837f4de77124f946a"
+        property_name = "building_type"
+        assert json.get_property_by_building_id(building_id, property_name) == "Retail other than mall"
+
+    def test_get_site_lat_lon_none(self):
+        filename = self.data_dir / "geojson_1.json"
+        json = UrbanOptGeoJson(filename)
+        assert json.get_site_lat_lon() is None
+
+    def test_get_site_lat_lon(self):
+        filename = (
+            self.data_dir.parent.parent
+            / "model_connectors"
+            / "data"
+            / "sdk_output_skeleton_13_buildings"
+            / "exportGeo.json"
+        )
+        json = UrbanOptGeoJson(filename)
+        assert json.get_site_lat_lon() == [42.816772, -78.849485]

tests/model_connectors/test_district_multi_ghe.py

@@ -43,7 +43,7 @@ def setUp(self):
         sys_params = SystemParameters(sys_param_filename)
 
         # read the loop order and create building groups
-        loop_order: list = load_loop_order(sys_param_filename)
+        loop_order = load_loop_order(sys_param_filename)
 
         # create ambient water stub
         ambient_water_stub = NetworkDistributionPump(sys_params)