(WIP) feat: add result factory for archetype mapping

hub/city_model_structure/simplified_building.py

@@ -14,6 +14,10 @@ class SimplifiedBuilding:
     self._postal_code = postal_code
     self._city = city
     self._type = 'building'
+    self.heating_demand = []
+    self.cooling_demand = []
+    self.electricity_demand = []
+    self.appliance_demand = []
 
   @property
   def name(self):

hub/imports/results/archetype_based_demand.py

@@ -0,0 +1,152 @@
+import pandas as pd
+from sqlalchemy import create_engine, text
+
+
+class DemandEnricher:
+  """
+  DemandEnricher class to enrich buildings with demand data
+  """
+
+  def __init__(self, database_url):
+      # Create a SQLAlchemy engine using the provided database URL
+      self.engine = create_engine(database_url)
+      # Initialize the function mapping and cache
+      self.create_function_mapping()
+      self.archetype_cache = {}
+
+  def create_function_mapping(self):
+    # Define function mapping from city functions to archetype functions
+    self.function_mapping = {
+      'residential': 'Maison Unifamiliale',
+      'single family house': 'Maison Unifamiliale',
+      'multifamily house': 'Apartements partie 3 du code',
+      'medium office': 'Bureaux',
+      'office and administration': 'Bureaux',
+      'commercial': 'Commercial attaché',
+      'warehouse': 'Commercial détaché',  # Approximate
+      'restaurant': 'Commercial attaché',  # Approximate
+      'hotel': 'Commercial attaché',      # Approximate
+      # Add more mappings as needed
+    }
+
+  def get_vintage_range(self, year):
+    # Determine the vintage range based on the year of construction
+    if year <= 1947:
+      return 'avant 1947'
+    elif 1947 < year <= 1983:
+      return '1947-1983'
+    elif 1983 < year <= 2010:
+      return '1984-2010'
+    elif year > 2010:
+      return 'après 2010'
+    else:
+      return None
+
+  def get_archetype_demands(self, type_of_building):
+    # Check if the demands for this archetype are already cached
+    if type_of_building in self.archetype_cache:
+      return self.archetype_cache[type_of_building]
+
+    # Construct the SQL query
+    query = text("""
+      SELECT heating, cooling, equipment, lighting
+      FROM energy_data
+      WHERE type_of_building = :type_of_building
+      ORDER BY timestamp
+    """)
+
+    # Execute the query with parameter substitution
+    with self.engine.connect() as conn:
+      result = conn.execute(query, type_of_building=type_of_building)
+      demands = result.fetchall()
+      if not demands:
+        return None
+
+      # Convert the result to a DataFrame
+      demands_df = pd.DataFrame(demands, columns=['heating', 'cooling', 'equipment', 'lighting'])
+      # Convert columns to numeric types
+      for demand_column in ['heating', 'cooling', 'equipment', 'lighting']:
+        demands_df[demand_column] = pd.to_numeric(demands_df[demand_column], errors='coerce')
+        demands_df[demand_column].fillna(0, inplace=True)
+
+      # Cache the demands for future use
+      self.archetype_cache[type_of_building] = demands_df
+      return demands_df
+
+  def enrich_city(self, city):
+    # Enrich each building in the city with demand data
+    for building in city.buildings:
+      # Ensure the building has the necessary attributes
+      if (building.year_of_construction is not None and
+          building.function is not None and
+          building.total_floor_area is not None):
+        # Map the building's function to an archetype function
+        building_function_lower = building.function.lower()
+        mapped_function = None
+        for key in self.function_mapping:
+          if key in building_function_lower:
+            mapped_function = self.function_mapping[key]
+            break
+        if mapped_function:
+          # Determine the vintage range
+          vintage_range = self.get_vintage_range(building.year_of_construction)
+          if vintage_range:
+            # Construct the Type_of_building string
+            type_of_building = f"{mapped_function} {vintage_range}"
+            # Get the demands for this archetype
+            demands_df = self.get_archetype_demands(type_of_building)
+            if demands_df is not None:
+              # Check total_floor_area
+              total_floor_area = building.total_floor_area
+              if not isinstance(total_floor_area, (int, float)) or pd.isnull(total_floor_area):
+                print(f"Invalid total_floor_area for building {building.name}. Skipping.")
+                building.heating_demand = []
+                building.cooling_demand = []
+                building.electricity_demand = []
+                building.appliance_demand = []
+                continue
+              # Proceed with multiplication
+              try:
+                demands_df['Heating_total'] = demands_df['heating'] * total_floor_area
+                demands_df['Cooling_total'] = demands_df['cooling'] * total_floor_area
+                demands_df['Equipment_total'] = demands_df['equipment'] * total_floor_area
+                demands_df['Lighting_total'] = demands_df['lighting'] * total_floor_area
+                # Assign the total demand profiles to the building's attributes
+                building.heating_demand = demands_df['Heating_total'].tolist()
+                building.cooling_demand = demands_df['Cooling_total'].tolist()
+                building.electricity_demand = demands_df['Lighting_total'].tolist()
+                building.appliance_demand = demands_df['Equipment_total'].tolist()
+              except Exception as e:
+                print(f"Error calculating demands for building {building.name}: {e}")
+                building.heating_demand = []
+                building.cooling_demand = []
+                building.electricity_demand = []
+                building.appliance_demand = []
+            else:
+              # No data found for this Type_of_building
+              building.heating_demand = []
+              building.cooling_demand = []
+              building.electricity_demand = []
+              building.appliance_demand = []
+          else:
+            # Vintage range could not be determined
+            building.heating_demand = []
+            building.cooling_demand = []
+            building.electricity_demand = []
+            building.appliance_demand = []
+        else:
+          # Function mapping not found
+          building.heating_demand = []
+          building.cooling_demand = []
+          building.electricity_demand = []
+          building.appliance_demand = []
+      else:
+        # Missing necessary attributes
+        building.heating_demand = []
+        building.cooling_demand = []
+        building.electricity_demand = []
+        building.appliance_demand = []
+
+  def close_connection(self):
+    # Dispose of the engine to close the database connection
+    self.engine.dispose()

main.py

@@ -1,10 +1,24 @@
 from hub.imports.geometry_factory import GeometryFactory
 from hub.helpers.dictionaries import Dictionaries
-import os
+from hub.imports.results.archetype_based_demand import DemandEnricher
-import psycopg2
+import urllib.parse
-import csv
 
 input_file = "data/cmm_points_function_vintage_surface.csv"
+output_file = "output_buildings.csv"
+
+# Database credentials
+db_username = 'postgres'
+db_password = ''
+db_host = 'localhost'
+db_port = '5432'
+db_name = 'energydemanddb'
+
+# URL-encode username and password
+db_username_encoded = urllib.parse.quote_plus(db_username)
+db_password_encoded = urllib.parse.quote_plus(db_password)
+
+# Construct the database connection URL
+database_url = f'postgresql://{db_username_encoded}:{db_password_encoded}@{db_host}:{db_port}/{db_name}'
 
 # Initialize city object from GeometryFactory
 city = GeometryFactory(
@@ -17,3 +31,11 @@ city = GeometryFactory(
     function_to_hub=Dictionaries().montreal_function_to_hub_function,
     total_floor_area_field="supfi_etag"
 ).city
+
+# Create an instance of DemandEnricher using the database URL
+demand_enricher = DemandEnricher(database_url)
+demand_enricher.enrich_city(city)
+# Close the database connection when done
+demand_enricher.close_connection()
+
+print("done")