Add element create

pyseed/seed_client.py

@@ -256,7 +256,7 @@ def create_organization(self, org_name: str, allow_exist: bool = False) -> dict:
         org = self.client.post(endpoint="organizations", json=payload)
         return org
 
-    def get_buildings(self) -> list[dict]:
+    def get_buildings(self, filters: dict = {}) -> list[dict]:
         total_qry = self.client.list(endpoint="properties", data_name="pagination", per_page=100)
 
         # step through each page of the results
@@ -268,6 +268,7 @@ def get_buildings(self) -> list[dict]:
                 per_page=100,
                 page=i,
                 cycle=self.cycle_id,
+                **filters,
             )
         # print(f"number of buildings retrieved: {len(buildings)}")
 
@@ -1335,7 +1336,14 @@ def get_meter(self, property_view_id: int, meter_type: str, source: str, source_
                 return meter
         return None
 
-    def get_or_create_meter(self, property_view_id: int, meter_type: str, source: str, source_id: str) -> Optional[dict[Any, Any]]:
+    def get_or_create_meter(
+        self,
+        property_view_id: int,
+        meter_type: str,
+        source: str,
+        source_id: str,
+        connection_type="Imported",
+    ) -> Optional[dict[Any, Any]]:
         """get or create a meter for a property view.
 
         Args:
@@ -1357,6 +1365,7 @@ def get_or_create_meter(self, property_view_id: int, meter_type: str, source: st
                 "type": meter_type,
                 "source": source,
                 "source_id": source_id,
+                "connection_type": connection_type,
             }
 
             meter = self.client.post(endpoint="properties_meters", url_args={"PK": property_view_id}, json=payload)
@@ -1394,6 +1403,24 @@ def upsert_meter_readings_bulk(self, property_view_id: int, meter_id: int, data:
         )
         return readings
 
+    def create_element(self, property_id: int, data: dict[Any, Any]) -> dict:
+        """Upsert an element for a property.
+
+        Args:
+            property_id (int): property id
+            data (dict[Any, Any]): dictionary of element data
+
+        Returns:
+            dict: element object
+        """
+        # get the element data for the property
+        element = self.client.post(
+            endpoint="properties_elements",
+            url_args={"PK": property_id},
+            json=data,
+        )
+        return element
+
     def get_meter_data(self, property_id, interval: str = "Exact", excluded_meter_ids: list = []):
         """Return the meter data from the property.
 

pyseed/seed_client_base.py

@@ -80,6 +80,7 @@
         "properties_analyses": "/api/v3/properties/PK/analyses/",
         "properties_meter_usage": "/api/v3/properties/PK/meter_usage/",
         "properties_meters": "/api/v3/properties/PK/meters/",
+        "properties_elements": "/api/v3/properties/PK/elements/",
         # GET & POST with replaceable keys
         "properties_meters_reading": "/api/v3/properties/PK/meters/METER_PK/readings/",
         # DELETES with replaceable keys

tests/test_seed_client.py

@@ -631,6 +631,63 @@ def setup_class(cls):
     def teardown_class(cls):
         cls.seed_client.delete_inventory()
 
+    def test_create_element(self):
+        # Test the create_element method by creating an element for a property
+
+        # First, create a property to attach the element to
+        completion_date = "02/02/2023"
+        year = "2023"
+        cycle = self.seed_client.get_or_create_cycle(
+            "pyseed-element-test",
+            date(int(year), 1, 1),
+            date(int(year), 12, 31),
+            set_cycle_id=True,
+        )
+
+        state = {
+            "organization_id": self.seed_client.client.org_id,
+            "custom_id_1": "element-test-building-123ABC",
+            "address_line_1": "456 Element Test St",
+            "city": "Element City",
+            "state": "CA",
+            "postal_code": "90211",
+            "property_name": "Element Test Building",
+            "extra_data": {"pathway": "new", "completion_date": completion_date},
+        }
+
+        params = {"state": state, "cycle_id": cycle["id"]}
+
+        # Create the building first
+        result = self.seed_client.create_building(params=params)
+        assert result["status"] == "success"
+        assert result["view"]["id"] is not None
+        property_view_id = result["view"]["id"]
+
+        # Get the property ID (not view ID) for the element creation
+        property_result = self.seed_client.get_property_view(property_view_id)
+        property_id = property_result["property_id"]
+
+        # Create test element data
+        # Element Data needs an id, a "code", and then extra data for the element itself
+        # Note: extra_data must be a flat JSON object, not a list
+        element_data = {
+            "id": "RTU-123",
+            "code": "D3050",
+            "extra_data": {
+                "name": "Rooftop Unit 123",
+                "refrigerant_type": "R-410A",
+                "capacity_tons": 15.0,
+                "efficiency_eer": 12.2,
+            },
+        }
+
+        # Test the create_element method
+        element_result = self.seed_client.create_element(property_id, element_data)
+
+        # Verify the element was created successfully
+        assert element_result is not None
+        # Additional assertions can be added based on the actual API response structure
+
     def test_upload_multiple_cycles_and_read_back(self):
         # Get/create the new cycle and upload the data. Make sure to set the cycle ID so that the
         # data end up in the correct cycle