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}")
 
@@ -122,3 +103,140 @@ def get_feature(self, 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"])
+
+        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"]
+
+        return result
+
+    def get_monthly_readings(self, building_id: str, meter_type: str) -> 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:
+                result = feature["properties"]["monthly_electricity"]
+
+        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"""
+        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_on_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) -> tuple | None:
+        """Return the site's latitude and longitude"""
+        for feature in self.data["features"]:
+            if feature["properties"]["name"] == "Site Origin":
+                # reverse the order of the coordinates
+                return feature["geometry"]["coordinates"][::-1]
+        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/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)

tests/model_connectors/test_district_single_ghe.py

@@ -36,7 +36,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 loop stub
         ambient_water_stub = NetworkDistributionPump(sys_params)