Compare commits

..

3 Commits

Author SHA1 Message Date
majidr
045eae249c file dump 2024-03-28 10:30:54 -04:00
Majid Rezaei
35c5f19abb fixing isolated graph issues 2024-02-06 14:25:26 -05:00
Majid Rezaei
e76ace02aa assigning building demands to the graph 2024-02-04 18:58:46 -05:00
307 changed files with 1880513 additions and 302400 deletions

BIN
.DS_Store vendored Normal file

Binary file not shown.

14
.gitignore vendored
View File

@ -1,13 +1 @@
.idea
*.idf
*.bnd
*.eio
*.end
*.err
*.eso
*.expidf
*.mtr
*.rvaudit
*.shd
*.csv
*.htm
.idea

View File

@ -0,0 +1,187 @@
import json
import geopandas as gpd
import matplotlib.pyplot as plt
from shapely.geometry import Polygon, Point, LineString, MultiPoint
import networkx as nx
from scipy.spatial import cKDTree
import hub.helpers.constants as cte
class DistrictHeatingNetworkCreator:
def __init__(self, buildings_file, roads_file, central_plant_longitude, central_plant_latitude):
self.buildings_file = buildings_file
self.roads_file = roads_file
self.central_plant_longitude = central_plant_longitude
self.central_plant_latitude = central_plant_latitude
def run(self):
self._load_and_process_data()
self._find_nearest_roads()
self._process_intersections()
network_graph = self._create_network_graph()
return network_graph
def _load_and_process_data(self):
self.gdf_road = gpd.read_file(self.roads_file)
with open(self.buildings_file, 'r') as file:
city = json.load(file)
centroids = []
building_ids = []
for building in city['features']:
coordinates = building['geometry']['coordinates'][0]
building_polygon = Polygon(coordinates)
centroid = building_polygon.centroid
centroids.append(centroid)
building_ids.append(building['id'])
centroids.append(Point(self.central_plant_longitude, self.central_plant_latitude))
building_ids.append(1)
self.centroids_gdf = gpd.GeoDataFrame({
'geometry': [Point(centroid.x, centroid.y) for centroid in centroids],
'building_id': building_ids,
'type': ['centroid' for _ in centroids]
}, crs='EPSG:4326')
def _find_nearest_roads(self):
self.centroids_gdf = self.centroids_gdf.to_crs(self.gdf_road.crs)
self.gdf_clean = gpd.GeoDataFrame(
{'geometry': [LineString([coord for coord in line.coords]) for line in self.gdf_road.geometry]})
self.closest_linestrings = []
self.nearest_points = []
for centroid in self.centroids_gdf.geometry:
closest_road = min(self.gdf_clean.geometry, key=lambda x: x.distance(centroid))
self.closest_linestrings.append(closest_road)
nearest_point = closest_road.interpolate(closest_road.project(centroid))
self.nearest_points.append(nearest_point)
def _process_intersections(self):
self.gdf_pts = gpd.GeoDataFrame(
{'geometry': [Point(coord) for line in self.gdf_clean.geometry for coord in line.coords]})
self.gdf_pts2 = gpd.GeoDataFrame({'geometry': self.nearest_points})
self.gdf_pts3 = gpd.GeoDataFrame({'geometry': self.nearest_points + list(self.gdf_pts.geometry)})
intersects = []
for geom in self.gdf_clean.geometry:
intersecting_points = []
if geom is not None:
for y in range(len(self.gdf_pts2)):
point_geom = self.gdf_pts2.geometry[y]
if point_geom is not None and point_geom.distance(geom) <= 1.0:
intersecting_points.append(y)
intersects.append(intersecting_points)
self.gdf_clean["intersect"] = intersects
self.gdf_cleaner = self.gdf_clean[self.gdf_clean["intersect"].apply(len).gt(0)].reset_index(drop=True)
self.test_list = []
for idx, row in self.gdf_cleaner.iterrows():
for i in range(len(row["intersect"]) + 1):
if i == 0:
self.test_list.append(
LineString([row['geometry'].coords[0], self.gdf_pts3.geometry[row['intersect'][i]]]))
elif i < len(row['intersect']):
self.test_list.append(LineString(
[self.gdf_pts3.geometry[row['intersect'][i - 1]], self.gdf_pts3.geometry[row['intersect'][i]]]))
else:
self.test_list.append(
LineString([self.gdf_pts3.geometry[row['intersect'][i - 1]], row['geometry'].coords[-1]]))
self.gdf_cleanest = gpd.GeoDataFrame({'geometry': self.test_list})
points = [coord for geom in self.gdf_cleanest.geometry for coord in geom.coords]
gdf_pts_cnt = self.gdf_pts.copy()
gdf_pts_cnt["count"] = gdf_pts_cnt.geometry.apply(lambda x: points.count(x.coords[0]))
gdf_pts_reset = gdf_pts_cnt[gdf_pts_cnt["count"] > 1].reset_index(drop=True)
gdf_pts_drop = gdf_pts_cnt[gdf_pts_cnt["count"] <= 1].reset_index(drop=True)
for idx, geom in self.gdf_cleanest.iterrows():
for coord in geom.geometry.coords:
if coord in [pt.coords[0] for pt in gdf_pts_drop.geometry]:
self.gdf_cleanest = self.gdf_cleanest.drop(idx)
self.gdf_cleanest.reset_index(drop=True, inplace=True)
def _create_network_graph(self):
g = nx.Graph()
# Add nodes with geometry attribute
for idx, row in self.centroids_gdf.iterrows():
building_name = f"Building_{idx}"
g.add_node((row.geometry.x, row.geometry.y),
geometry=row.geometry, # Include geometry attribute
type='centroid',
name=building_name,
building_id=str(row.get('building_id')))
for point in self.nearest_points:
g.add_node((point.x, point.y),
geometry=point, # Include geometry attribute
type='nearest_point')
for line in self.gdf_cleanest.geometry:
length = line.length
if isinstance(line.boundary, MultiPoint):
points = list(line.boundary.geoms)
for i in range(len(points) - 1):
start_point = points[i]
end_point = points[i + 1]
g.add_edge((start_point.x, start_point.y), (end_point.x, end_point.y), weight=length)
else:
start_point, end_point = line.boundary
g.add_edge((start_point.x, start_point.y), (end_point.x, end_point.y), weight=length)
for point, centroid in zip(self.nearest_points, self.centroids_gdf.geometry):
distance = point.distance(centroid)
g.add_edge((point.x, point.y), (centroid.x, centroid.y), weight=distance)
# Check and connect isolated components
components = list(nx.connected_components(g))
if len(components) > 1:
components.sort(key=len)
main_component = components[-1]
for comp in components[:-1]:
self._connect_component_to_main(g, comp, main_component)
return g
def _connect_component_to_main(self, graph, component, main_component):
main_component_nodes = [graph.nodes[node] for node in main_component if 'geometry' in graph.nodes[node]]
# Create cKDTree for efficient nearest neighbor search
tree = cKDTree([(node['geometry'].x, node['geometry'].y) for node in main_component_nodes])
# For each node in the component, find the closest street node in the main component
for node in component:
if 'geometry' in graph.nodes[node]: # Check for geometry attribute
node_geometry = graph.nodes[node]['geometry']
distance, idx = tree.query((node_geometry.x, node_geometry.y))
closest_node_geometry = main_component_nodes[idx]['geometry']
# Add edge to the graph
graph.add_edge((node_geometry.x, node_geometry.y),
(closest_node_geometry.x, closest_node_geometry.y), weight=distance)
def plot_network_graph(network_graph, central_plant_id=1):
plt.figure(figsize=(12, 12))
pos = {node: (node[0], node[1]) for node in network_graph.nodes()}
# Node colors based on type
node_colors = ['red' if data.get('building_id') == str(central_plant_id) else 'green' if data.get(
'type') == 'centroid' else 'blue' for node, data in network_graph.nodes(data=True)]
# Node sizes, larger for central plant
node_sizes = [100 if data.get('building_id') == str(central_plant_id) else 50 for node, data in
network_graph.nodes(data=True)]
nx.draw_networkx_nodes(network_graph, pos, node_color=node_colors, node_size=node_sizes)
nx.draw_networkx_edges(network_graph, pos, edge_color='gray', width=1)
plt.title('District Heating Network Graph')
plt.axis('off')
plt.savefig('network_graph_visualization.png', format='png', dpi=300) # Save as PNG with high dpi for clarity
plt.show()

76
ThermalModeling.py Normal file
View File

@ -0,0 +1,76 @@
import networkx as nx
import numpy as np
class ThermalModeling:
def __init__(self, graph, T_initial=70, Tg=3, cp=4200, rho=980, U=500, dx=20, delta_t=60):
"""
Initialize the ThermalModeling class with a networkx graph.
:param graph: A directed networkx graph where edges have 'length', 'diameter', and 'mass_flow_rate_actual'.
:param T_initial: Initial temperature at each node in Celsius.
:param Tg: Ground temperature in Celsius.
:param cp: Isobaric specific heat capacity of water at 60 C in J/(kg*K).
:param rho: Density of water in kg/m3.
:param U: Heat transfer coefficient for all pipes.
:param dx: Number of segments for calculating temperature drops in a pipe.
:param delta_t: Time step for the calculation.
"""
self.graph = graph
self.T_initial = T_initial
self.Tg = Tg
self.cp = cp
self.rho = rho
self.U = U
self.dx = dx
self.delta_t = delta_t
# Initialize node temperatures and flow rates
for node in self.graph.nodes():
self.graph.nodes[node]['temperature'] = T_initial
self.graph.nodes[node]['total_mass_flow_in'] = 0
self.graph.nodes[node]['weighted_temp_sum'] = 0
def adjust_flow_directions(self):
"""
Adjust the directions of flow based on the mass flow rate. If the mass flow rate is negative,
the direction of the flow is reversed.
"""
to_reverse = [(u, v) for u, v, d in self.graph.edges(data=True) if d['mass_flow_rate_actual'] < 0]
for u, v in to_reverse:
attrs = self.graph.edges[u, v]
self.graph.remove_edge(u, v)
self.graph.add_edge(v, u, **attrs)
# Update the mass flow rate to be positive after reversing
self.graph.edges[v, u]['mass_flow_rate_actual'] = abs(attrs['mass_flow_rate_actual'])
def calculate_temperatures(self):
"""
Calculate and update temperatures for all nodes based on the network graph, considering weighted averages
for nodes with multiple incoming temperatures.
"""
self.adjust_flow_directions() # Adjust flow directions based on mass flow rates
# Calculate weighted temperatures for incoming flows
for u, v, d in self.graph.edges(data=True):
length = d['weight']
diameter = d['Diameter']
A = np.pi * diameter**2 / 4
delta_x = length / self.dx
mass_flow_rate = abs(d['mass_flow_rate_actual'])
C1 = 2 * self.delta_t * self.U / (A * self.rho * self.cp)
C2 = 2 * mass_flow_rate * self.delta_t / (self.rho * A * delta_x)
C = 1 / (1 + C1 + C2)
T_in = self.graph.nodes[u]['temperature']
T_out = C * (T_in + C1 * self.Tg) # Simplified model for demonstration
# Update weighted temperature sum and total mass flow for the target node
self.graph.nodes[v]['total_mass_flow_in'] += mass_flow_rate
self.graph.nodes[v]['weighted_temp_sum'] += T_out * mass_flow_rate
# Calculate final temperatures based on weighted averages
for node in self.graph.nodes():
if self.graph.nodes[node]['total_mass_flow_in'] > 0: # To avoid division by zero
weighted_average_temp = self.graph.nodes[node]['weighted_temp_sum'] / self.graph.nodes[node]['total_mass_flow_in']
self.graph.nodes[node]['temperature'] = weighted_average_temp

Binary file not shown.

Binary file not shown.

BIN
data/MTLBLD_V4.geojson Normal file

Binary file not shown.

16
enrich.py Normal file
View File

@ -0,0 +1,16 @@
def enrich(graph, city):
"""
Enrich the graph nodes with hourly building demand data.
:param graph: The networkx graph of the district heating network.
:param buildings: A list of building objects, each with a 'name' and 'heating_demand' attribute.
"""
for node in graph.nodes:
node_data = graph.nodes[node]
# Check if the node has a 'building_id' attribute before comparing
if 'building_id' in node_data:
for building in city.buildings:
if node_data['building_id'] == building.name:
# Assuming `building.heating_demand` is properly structured for direct assignment
graph.nodes[node]["Demand"] = building.heating_demand[cte.HOUR]
graph.nodes[node]["Demand"] = building.heating_peak_load[cte.YEAR]

View File

@ -14,12 +14,12 @@ class ElectricalStorageSystem(EnergyStorageSystem):
Energy Storage System Class
"""
def __init__(self, storage_id, type_energy_stored=None, model_name=None, manufacturer=None, storage_type=None,
def __init__(self, storage_id, model_name=None, manufacturer=None, storage_type=None,
nominal_capacity=None, losses_ratio=None, rated_output_power=None, nominal_efficiency=None,
battery_voltage=None, depth_of_discharge=None, self_discharge_rate=None):
super().__init__(storage_id, model_name, manufacturer, nominal_capacity, losses_ratio)
self._type_energy_stored = type_energy_stored
self._type_energy_stored = 'electrical'
self._storage_type = storage_type
self._rated_output_power = rated_output_power
self._nominal_efficiency = nominal_efficiency

View File

@ -25,7 +25,7 @@ class NonPvGenerationSystem(GenerationSystem):
maximum_cooling_supply_temperature=None, minimum_cooling_supply_temperature=None, heat_output_curve=None,
heat_fuel_consumption_curve=None, heat_efficiency_curve=None, cooling_output_curve=None,
cooling_fuel_consumption_curve=None, cooling_efficiency_curve=None,
distribution_systems=None, energy_storage_systems=None, dual_supply_capability=False):
distribution_systems=None, energy_storage_systems=None):
super().__init__(system_id=system_id, name=name, model_name=model_name, manufacturer=manufacturer, fuel_type=fuel_type,
distribution_systems=distribution_systems, energy_storage_systems=energy_storage_systems)
self._system_type = system_type
@ -53,7 +53,6 @@ class NonPvGenerationSystem(GenerationSystem):
self._cooling_output_curve = cooling_output_curve
self._cooling_fuel_consumption_curve = cooling_fuel_consumption_curve
self._cooling_efficiency_curve = cooling_efficiency_curve
self._dual_supply_capability = dual_supply_capability
@property
def system_type(self):
@ -255,14 +254,6 @@ class NonPvGenerationSystem(GenerationSystem):
"""
return self._cooling_efficiency_curve
@property
def dual_supply_capability(self):
"""
Get dual supply capability
:return: bool
"""
return self._dual_supply_capability
def to_dictionary(self):
"""Class content to dictionary"""
_distribution_systems = [_distribution_system.to_dictionary() for _distribution_system in
@ -303,8 +294,7 @@ class NonPvGenerationSystem(GenerationSystem):
'cooling fuel consumption curve': self.cooling_fuel_consumption_curve,
'cooling efficiency curve': self.cooling_efficiency_curve,
'distribution systems connected': _distribution_systems,
'storage systems connected': _energy_storage_systems,
'dual supply capability': self.dual_supply_capability
'storage systems connected': _energy_storage_systems
}
}
return content

View File

@ -14,7 +14,7 @@ class PvGenerationSystem(GenerationSystem):
Electricity Generation system class
"""
def __init__(self, system_id, name, system_type, model_name=None, manufacturer=None, electricity_efficiency=None,
def __init__(self, system_id, name, model_name=None, manufacturer=None, electricity_efficiency=None,
nominal_electricity_output=None, nominal_ambient_temperature=None, nominal_cell_temperature=None,
nominal_radiation=None, standard_test_condition_cell_temperature=None,
standard_test_condition_maximum_power=None, cell_temperature_coefficient=None, width=None, height=None,
@ -22,7 +22,7 @@ class PvGenerationSystem(GenerationSystem):
super().__init__(system_id=system_id, name=name, model_name=model_name,
manufacturer=manufacturer, fuel_type='renewable', distribution_systems=distribution_systems,
energy_storage_systems=energy_storage_systems)
self._system_type = system_type
self._system_type = 'PV system'
self._electricity_efficiency = electricity_efficiency
self._nominal_electricity_output = nominal_electricity_output
self._nominal_ambient_temperature = nominal_ambient_temperature

View File

@ -15,12 +15,12 @@ class ThermalStorageSystem(EnergyStorageSystem):
Energy Storage System Class
"""
def __init__(self, storage_id, type_energy_stored=None, model_name=None, manufacturer=None, storage_type=None,
def __init__(self, storage_id, model_name=None, manufacturer=None, storage_type=None,
nominal_capacity=None, losses_ratio=None, volume=None, height=None, layers=None,
maximum_operating_temperature=None, storage_medium=None):
super().__init__(storage_id, model_name, manufacturer, nominal_capacity, losses_ratio)
self._type_energy_stored = type_energy_stored
self._type_energy_stored = 'thermal'
self._storage_type = storage_type
self._volume = volume
self._height = height

View File

@ -69,10 +69,8 @@ class MontrealCustomCatalog(Catalog):
storage_system = ThermalStorageSystem(equipment_id)
storage_systems = [storage_system]
if model_name == 'PV system':
system_type = 'Photovoltaic'
generation_system = PvGenerationSystem(equipment_id,
name=None,
system_type=system_type,
model_name=model_name,
electricity_efficiency=electricity_efficiency,
energy_storage_systems=storage_systems
@ -86,9 +84,8 @@ class MontrealCustomCatalog(Catalog):
heat_efficiency=heating_efficiency,
cooling_efficiency=cooling_efficiency,
electricity_efficiency=electricity_efficiency,
energy_storage_systems=storage_systems,
dual_supply_capability=False
)
energy_storage_systems=storage_systems
)
_equipments.append(generation_system)
return _equipments

View File

@ -1,539 +0,0 @@
"""
Montreal future energy system catalog
SPDX - License - Identifier: LGPL - 3.0 - or -later
Copyright © 2022 Concordia CERC group
Project Coder Saeed Ranjbar saeed.ranjbar@concordia.ca
"""
import xmltodict
from pathlib import Path
from hub.catalog_factories.catalog import Catalog
from hub.catalog_factories.data_models.energy_systems.distribution_system import DistributionSystem
from hub.catalog_factories.data_models.energy_systems.emission_system import EmissionSystem
from hub.catalog_factories.data_models.energy_systems.system import System
from hub.catalog_factories.data_models.energy_systems.content import Content
from hub.catalog_factories.data_models.energy_systems.non_pv_generation_system import NonPvGenerationSystem
from hub.catalog_factories.data_models.energy_systems.pv_generation_system import PvGenerationSystem
from hub.catalog_factories.data_models.energy_systems.thermal_storage_system import ThermalStorageSystem
from hub.catalog_factories.data_models.energy_systems.performance_curves import PerformanceCurves
from hub.catalog_factories.data_models.energy_systems.archetype import Archetype
from hub.catalog_factories.data_models.construction.material import Material
from hub.catalog_factories.data_models.construction.layer import Layer
class MontrealFutureSystemCatalogue(Catalog):
"""
North america energy system catalog class
"""
def __init__(self, path):
path = str(path / 'montreal_future_systems.xml')
with open(path, 'r', encoding='utf-8') as xml:
self._archetypes = xmltodict.parse(xml.read(),
force_list=['pv_generation_component', 'templateStorages', 'demand'])
self._storage_components = self._load_storage_components()
self._generation_components = self._load_generation_components()
self._energy_emission_components = self._load_emission_equipments()
self._distribution_components = self._load_distribution_equipments()
self._systems = self._load_systems()
self._system_archetypes = self._load_archetypes()
self._content = Content(self._system_archetypes,
self._systems,
generations=self._generation_components,
distributions=self._distribution_components)
def _load_generation_components(self):
generation_components = []
non_pv_generation_components = self._archetypes['EnergySystemCatalog']['energy_generation_components'][
'non_pv_generation_component']
if non_pv_generation_components is not None:
for non_pv in non_pv_generation_components:
system_id = non_pv['system_id']
name = non_pv['name']
system_type = non_pv['system_type']
model_name = non_pv['model_name']
manufacturer = non_pv['manufacturer']
fuel_type = non_pv['fuel_type']
distribution_systems = non_pv['distribution_systems']
energy_storage_systems = None
if non_pv['energy_storage_systems'] is not None:
storage_component = non_pv['energy_storage_systems']['storage_id']
storage_systems = self._search_storage_equipment(self._load_storage_components(), storage_component)
energy_storage_systems = storage_systems
nominal_heat_output = non_pv['nominal_heat_output']
maximum_heat_output = non_pv['maximum_heat_output']
minimum_heat_output = non_pv['minimum_heat_output']
source_medium = non_pv['source_medium']
supply_medium = non_pv['supply_medium']
heat_efficiency = non_pv['heat_efficiency']
nominal_cooling_output = non_pv['nominal_cooling_output']
maximum_cooling_output = non_pv['maximum_cooling_output']
minimum_cooling_output = non_pv['minimum_cooling_output']
cooling_efficiency = non_pv['cooling_efficiency']
electricity_efficiency = non_pv['electricity_efficiency']
source_temperature = non_pv['source_temperature']
source_mass_flow = non_pv['source_mass_flow']
nominal_electricity_output = non_pv['nominal_electricity_output']
maximum_heat_supply_temperature = non_pv['maximum_heat_supply_temperature']
minimum_heat_supply_temperature = non_pv['minimum_heat_supply_temperature']
maximum_cooling_supply_temperature = non_pv['maximum_cooling_supply_temperature']
minimum_cooling_supply_temperature = non_pv['minimum_cooling_supply_temperature']
heat_output_curve = None
heat_fuel_consumption_curve = None
heat_efficiency_curve = None
cooling_output_curve = None
cooling_fuel_consumption_curve = None
cooling_efficiency_curve = None
if non_pv['heat_output_curve'] is not None:
curve_type = non_pv['heat_output_curve']['curve_type']
dependant_variable = non_pv['heat_output_curve']['dependant_variable']
parameters = non_pv['heat_output_curve']['parameters']
coefficients = list(non_pv['heat_output_curve']['coefficients'].values())
heat_output_curve = PerformanceCurves(curve_type, dependant_variable, parameters, coefficients)
if non_pv['heat_fuel_consumption_curve'] is not None:
curve_type = non_pv['heat_fuel_consumption_curve']['curve_type']
dependant_variable = non_pv['heat_fuel_consumption_curve']['dependant_variable']
parameters = non_pv['heat_fuel_consumption_curve']['parameters']
coefficients = list(non_pv['heat_fuel_consumption_curve']['coefficients'].values())
heat_fuel_consumption_curve = PerformanceCurves(curve_type, dependant_variable, parameters, coefficients)
if non_pv['heat_efficiency_curve'] is not None:
curve_type = non_pv['heat_efficiency_curve']['curve_type']
dependant_variable = non_pv['heat_efficiency_curve']['dependant_variable']
parameters = non_pv['heat_efficiency_curve']['parameters']
coefficients = list(non_pv['heat_efficiency_curve']['coefficients'].values())
heat_efficiency_curve = PerformanceCurves(curve_type, dependant_variable, parameters, coefficients)
if non_pv['cooling_output_curve'] is not None:
curve_type = non_pv['cooling_output_curve']['curve_type']
dependant_variable = non_pv['cooling_output_curve']['dependant_variable']
parameters = non_pv['cooling_output_curve']['parameters']
coefficients = list(non_pv['cooling_output_curve']['coefficients'].values())
cooling_output_curve = PerformanceCurves(curve_type, dependant_variable, parameters, coefficients)
if non_pv['cooling_fuel_consumption_curve'] is not None:
curve_type = non_pv['cooling_fuel_consumption_curve']['curve_type']
dependant_variable = non_pv['cooling_fuel_consumption_curve']['dependant_variable']
parameters = non_pv['cooling_fuel_consumption_curve']['parameters']
coefficients = list(non_pv['cooling_fuel_consumption_curve']['coefficients'].values())
cooling_fuel_consumption_curve = PerformanceCurves(curve_type, dependant_variable, parameters, coefficients)
if non_pv['cooling_efficiency_curve'] is not None:
curve_type = non_pv['cooling_efficiency_curve']['curve_type']
dependant_variable = non_pv['cooling_efficiency_curve']['dependant_variable']
parameters = non_pv['cooling_efficiency_curve']['parameters']
coefficients = list(non_pv['cooling_efficiency_curve']['coefficients'].values())
cooling_efficiency_curve = PerformanceCurves(curve_type, dependant_variable, parameters, coefficients)
dual_supply_capability = None
if non_pv['dual_supply_capability'] is not None:
if non_pv['dual_supply_capability'] == 'True':
dual_supply_capability = True
else:
dual_supply_capability = False
non_pv_component = NonPvGenerationSystem(system_id=system_id,
name=name,
system_type=system_type,
model_name=model_name,
manufacturer=manufacturer,
fuel_type=fuel_type,
nominal_heat_output=nominal_heat_output,
maximum_heat_output=maximum_heat_output,
minimum_heat_output=minimum_heat_output,
source_medium=source_medium,
supply_medium=supply_medium,
heat_efficiency=heat_efficiency,
nominal_cooling_output=nominal_cooling_output,
maximum_cooling_output=maximum_cooling_output,
minimum_cooling_output=minimum_cooling_output,
cooling_efficiency=cooling_efficiency,
electricity_efficiency=electricity_efficiency,
source_temperature=source_temperature,
source_mass_flow=source_mass_flow,
nominal_electricity_output=nominal_electricity_output,
maximum_heat_supply_temperature=maximum_heat_supply_temperature,
minimum_heat_supply_temperature=minimum_heat_supply_temperature,
maximum_cooling_supply_temperature=maximum_cooling_supply_temperature,
minimum_cooling_supply_temperature=minimum_cooling_supply_temperature,
heat_output_curve=heat_output_curve,
heat_fuel_consumption_curve=heat_fuel_consumption_curve,
heat_efficiency_curve=heat_efficiency_curve,
cooling_output_curve=cooling_output_curve,
cooling_fuel_consumption_curve=cooling_fuel_consumption_curve,
cooling_efficiency_curve=cooling_efficiency_curve,
distribution_systems=distribution_systems,
energy_storage_systems=energy_storage_systems,
dual_supply_capability=dual_supply_capability)
generation_components.append(non_pv_component)
pv_generation_components = self._archetypes['EnergySystemCatalog']['energy_generation_components'][
'pv_generation_component']
if pv_generation_components is not None:
for pv in pv_generation_components:
system_id = pv['system_id']
name = pv['name']
system_type = pv['system_type']
model_name = pv['model_name']
manufacturer = pv['manufacturer']
electricity_efficiency = pv['electricity_efficiency']
nominal_electricity_output = pv['nominal_electricity_output']
nominal_ambient_temperature = pv['nominal_ambient_temperature']
nominal_cell_temperature = pv['nominal_cell_temperature']
nominal_radiation = pv['nominal_radiation']
standard_test_condition_cell_temperature = pv['standard_test_condition_cell_temperature']
standard_test_condition_maximum_power = pv['standard_test_condition_maximum_power']
cell_temperature_coefficient = pv['cell_temperature_coefficient']
width = pv['width']
height = pv['height']
distribution_systems = pv['distribution_systems']
energy_storage_systems = None
if pv['energy_storage_systems'] is not None:
storage_component = pv['energy_storage_systems']['storage_id']
storage_systems = self._search_storage_equipment(self._load_storage_components(), storage_component)
energy_storage_systems = storage_systems
pv_component = PvGenerationSystem(system_id=system_id,
name=name,
system_type=system_type,
model_name=model_name,
manufacturer=manufacturer,
electricity_efficiency=electricity_efficiency,
nominal_electricity_output=nominal_electricity_output,
nominal_ambient_temperature=nominal_ambient_temperature,
nominal_cell_temperature=nominal_cell_temperature,
nominal_radiation=nominal_radiation,
standard_test_condition_cell_temperature=
standard_test_condition_cell_temperature,
standard_test_condition_maximum_power=standard_test_condition_maximum_power,
cell_temperature_coefficient=cell_temperature_coefficient,
width=width,
height=height,
distribution_systems=distribution_systems,
energy_storage_systems=energy_storage_systems)
generation_components.append(pv_component)
return generation_components
def _load_distribution_equipments(self):
_equipments = []
distribution_systems = self._archetypes['EnergySystemCatalog']['distribution_systems']['distribution_system']
if distribution_systems is not None:
for distribution_system in distribution_systems:
system_id = None
model_name = None
system_type = None
supply_temperature = None
distribution_consumption_fix_flow = None
distribution_consumption_variable_flow = None
heat_losses = None
generation_systems = None
energy_storage_systems = None
emission_systems = None
distribution_equipment = DistributionSystem(system_id=system_id,
model_name=model_name,
system_type=system_type,
supply_temperature=supply_temperature,
distribution_consumption_fix_flow=distribution_consumption_fix_flow,
distribution_consumption_variable_flow=
distribution_consumption_variable_flow,
heat_losses=heat_losses,
generation_systems=generation_systems,
energy_storage_systems=energy_storage_systems,
emission_systems=emission_systems
)
_equipments.append(distribution_equipment)
return _equipments
def _load_emission_equipments(self):
_equipments = []
dissipation_systems = self._archetypes['EnergySystemCatalog']['dissipation_systems']['dissipation_system']
if dissipation_systems is not None:
for dissipation_system in dissipation_systems:
system_id = None
model_name = None
system_type = None
parasitic_energy_consumption = None
emission_system = EmissionSystem(system_id=system_id,
model_name=model_name,
system_type=system_type,
parasitic_energy_consumption=parasitic_energy_consumption)
_equipments.append(emission_system)
return _equipments
def _load_storage_components(self):
storage_components = []
thermal_storages = self._archetypes['EnergySystemCatalog']['energy_storage_components']['thermalStorages']
template_storages = self._archetypes['EnergySystemCatalog']['energy_storage_components']['templateStorages']
for tes in thermal_storages:
storage_id = tes['storage_id']
type_energy_stored = tes['type_energy_stored']
model_name = tes['model_name']
manufacturer = tes['manufacturer']
storage_type = tes['storage_type']
volume = tes['physical_characteristics']['volume']
height = tes['physical_characteristics']['height']
maximum_operating_temperature = tes['maximum_operating_temperature']
materials = self._load_materials()
insulation_material_id = tes['insulation']['material_id']
insulation_material = self._search_material(materials, insulation_material_id)
material_id = tes['physical_characteristics']['material_id']
tank_material = self._search_material(materials, material_id)
thickness = float(tes['insulation']['insulationThickness']) / 100 # from cm to m
insulation_layer = Layer(None, 'insulation', insulation_material, thickness)
thickness = float(tes['physical_characteristics']['tankThickness']) / 100 # from cm to m
tank_layer = Layer(None, 'tank', tank_material, thickness)
media = self._load_media()
media_id = tes['storage_medium']['medium_id']
medium = self._search_media(media, media_id)
layers = [insulation_layer, tank_layer]
nominal_capacity = tes['nominal_capacity']
losses_ratio = tes['losses_ratio']
storage_component = ThermalStorageSystem(storage_id=storage_id,
model_name=model_name,
type_energy_stored=type_energy_stored,
manufacturer=manufacturer,
storage_type=storage_type,
nominal_capacity=nominal_capacity,
losses_ratio=losses_ratio,
volume=volume,
height=height,
layers=layers,
maximum_operating_temperature=maximum_operating_temperature,
storage_medium=medium)
storage_components.append(storage_component)
for template in template_storages:
storage_id = template['storage_id']
storage_type = template['storage_type']
type_energy_stored = template['type_energy_stored']
maximum_operating_temperature = template['maximum_operating_temperature']
height = template['physical_characteristics']['height']
materials = self._load_materials()
insulation_material_id = template['insulation']['material_id']
insulation_material = self._search_material(materials, insulation_material_id)
material_id = template['physical_characteristics']['material_id']
tank_material = self._search_material(materials, material_id)
thickness = float(template['insulation']['insulationThickness']) / 100 # from cm to m
insulation_layer = Layer(None, 'insulation', insulation_material, thickness)
thickness = float(template['physical_characteristics']['tankThickness']) / 100 # from cm to m
tank_layer = Layer(None, 'tank', tank_material, thickness)
layers = [insulation_layer, tank_layer]
media = self._load_media()
media_id = template['storage_medium']['medium_id']
medium = self._search_media(media, media_id)
model_name = template['model_name']
manufacturer = template['manufacturer']
nominal_capacity = template['nominal_capacity']
losses_ratio = template['losses_ratio']
volume = template['physical_characteristics']['volume']
storage_component = ThermalStorageSystem(storage_id=storage_id,
model_name=model_name,
type_energy_stored=type_energy_stored,
manufacturer=manufacturer,
storage_type=storage_type,
nominal_capacity=nominal_capacity,
losses_ratio=losses_ratio,
volume=volume,
height=height,
layers=layers,
maximum_operating_temperature=maximum_operating_temperature,
storage_medium=medium)
storage_components.append(storage_component)
return storage_components
def _load_systems(self):
base_path = Path(Path(__file__).parent.parent.parent / 'data/energy_systems')
_catalog_systems = []
systems = self._archetypes['EnergySystemCatalog']['systems']['system']
for system in systems:
system_id = system['id']
name = system['name']
demands = system['demands']['demand']
generation_components = system['components']['generation_id']
generation_systems = self._search_generation_equipment(self._load_generation_components(), generation_components)
configuration_schema = Path(base_path / system['schema'])
energy_system = System(system_id=system_id,
name=name,
demand_types=demands,
generation_systems=generation_systems,
distribution_systems=None,
configuration_schema=configuration_schema)
_catalog_systems.append(energy_system)
return _catalog_systems
def _load_archetypes(self):
_system_archetypes = []
system_clusters = self._archetypes['EnergySystemCatalog']['system_archetypes']['system_archetype']
for system_cluster in system_clusters:
name = system_cluster['name']
systems = system_cluster['systems']['system_id']
integer_system_ids = [int(item) for item in systems]
_systems = []
for system_archetype in self._systems:
if int(system_archetype.id) in integer_system_ids:
_systems.append(system_archetype)
_system_archetypes.append(Archetype(name=name, systems=_systems))
return _system_archetypes
def _load_materials(self):
materials = []
_materials = self._archetypes['EnergySystemCatalog']['materials']['material']
for _material in _materials:
material_id = _material['material_id']
name = _material['name']
conductivity = _material['conductivity']
solar_absorptance = _material['solar_absorptance']
thermal_absorptance = _material['thermal_absorptance']
density = _material['density']
specific_heat = _material['specific_heat']
no_mass = _material['no_mass']
visible_absorptance = _material['visible_absorptance']
thermal_resistance = _material['thermal_resistance']
material = Material(material_id,
name,
solar_absorptance=solar_absorptance,
thermal_absorptance=thermal_absorptance,
density=density,
conductivity=conductivity,
thermal_resistance=thermal_resistance,
visible_absorptance=visible_absorptance,
no_mass=no_mass,
specific_heat=specific_heat)
materials.append(material)
return materials
@staticmethod
def _search_material(materials, material_id):
_material = None
for material in materials:
if int(material.id) == int(material_id):
_material = material
break
if _material is None:
raise ValueError(f'Material with the id = [{material_id}] not found in catalog ')
return _material
def _load_media(self):
media = []
_media = [self._archetypes['EnergySystemCatalog']['media']['medium']]
for _medium in _media:
medium_id = _medium['medium_id']
density = _medium['density']
name = _medium['name']
conductivity = _medium['conductivity']
solar_absorptance = _medium['solar_absorptance']
thermal_absorptance = _medium['thermal_absorptance']
specific_heat = _medium['specific_heat']
no_mass = _medium['no_mass']
visible_absorptance = _medium['visible_absorptance']
thermal_resistance = _medium['thermal_resistance']
medium = Material(material_id=medium_id,
name=name,
solar_absorptance=solar_absorptance,
thermal_absorptance=thermal_absorptance,
visible_absorptance=visible_absorptance,
no_mass=no_mass,
thermal_resistance=thermal_resistance,
conductivity=conductivity,
density=density,
specific_heat=specific_heat)
media.append(medium)
return media
@staticmethod
def _search_media(media, medium_id):
_medium = None
for medium in media:
if int(medium.id) == int(medium_id):
_medium = medium
break
if _medium is None:
raise ValueError(f'media with the id = [{medium_id}] not found in catalog ')
return _medium
@staticmethod
def _search_generation_equipment(generation_systems, generation_id):
_generation_systems = []
if isinstance(generation_id, list):
integer_ids = [int(item) for item in generation_id]
for generation in generation_systems:
if int(generation.id) in integer_ids:
_generation_systems.append(generation)
else:
integer_id = int(generation_id)
for generation in generation_systems:
if int(generation.id) == integer_id:
_generation_systems.append(generation)
if len(_generation_systems) == 0:
_generation_systems = None
raise ValueError(f'The system with the following id is not found in catalog [{generation_id}]')
return _generation_systems
@staticmethod
def _search_storage_equipment(storage_systems, storage_id):
_storage_systems = []
for storage in storage_systems:
if storage.id in storage_id:
_storage_systems.append(storage)
if len(_storage_systems) == 0:
_storage_systems = None
raise ValueError(f'The system with the following id is not found in catalog [{storage_id}]')
return _storage_systems
def names(self, category=None):
"""
Get the catalog elements names
:parm: optional category filter
"""
if category is None:
_names = {'archetypes': [], 'systems': [], 'generation_equipments': [], 'storage_equipments': []}
for archetype in self._content.archetypes:
_names['archetypes'].append(archetype.name)
for system in self._content.systems:
_names['systems'].append(system.name)
for equipment in self._content.generation_equipments:
_names['generation_equipments'].append(equipment.name)
else:
_names = {category: []}
if category.lower() == 'archetypes':
for archetype in self._content.archetypes:
_names[category].append(archetype.name)
elif category.lower() == 'systems':
for system in self._content.systems:
_names[category].append(system.name)
elif category.lower() == 'generation_equipments':
for system in self._content.generation_equipments:
_names[category].append(system.name)
else:
raise ValueError(f'Unknown category [{category}]')
return _names
def entries(self, category=None):
"""
Get the catalog elements
:parm: optional category filter
"""
if category is None:
return self._content
if category.lower() == 'archetypes':
return self._content.archetypes
if category.lower() == 'systems':
return self._content.systems
if category.lower() == 'generation_equipments':
return self._content.generation_equipments
raise ValueError(f'Unknown category [{category}]')
def get_entry(self, name):
"""
Get one catalog element by names
:parm: entry name
"""
for entry in self._content.archetypes:
if entry.name.lower() == name.lower():
return entry
for entry in self._content.systems:
if entry.name.lower() == name.lower():
return entry
for entry in self._content.generation_equipments:
if entry.name.lower() == name.lower():
return entry
raise IndexError(f"{name} doesn't exists in the catalog")

View File

@ -56,9 +56,6 @@ class NorthAmericaEnergySystemCatalog(Catalog):
boiler_maximum_heat_output = float(boiler['@maximumHeatOutput'])
boiler_minimum_heat_output = float(boiler['@minimumHeatOutput'])
boiler_heat_efficiency = float(boiler['@nominalEfficiency'])
dual_supply = False
if '@dual_supply' in boiler.keys() and boiler['@dual_supply'] == 'True':
dual_supply = True
boiler_component = NonPvGenerationSystem(boiler_id,
name=name,
system_type=system_type,
@ -68,8 +65,7 @@ class NorthAmericaEnergySystemCatalog(Catalog):
nominal_heat_output=boiler_nominal_thermal_output,
maximum_heat_output=boiler_maximum_heat_output,
minimum_heat_output=boiler_minimum_heat_output,
heat_efficiency=boiler_heat_efficiency,
dual_supply_capability=dual_supply)
heat_efficiency=boiler_heat_efficiency)
generation_components.append(boiler_component)
for heat_pump in heat_pumps:
heat_pump_id = heat_pump['@generation_id']
@ -93,9 +89,6 @@ class NorthAmericaEnergySystemCatalog(Catalog):
parameters = heat_pump['performance_curve']['parameters']
coefficients = list(heat_pump['performance_curve']['coefficients'].values())
cop_curve = PerformanceCurves(cop_curve_type, dependant_variable, parameters, coefficients)
dual_supply = False
if '@dual_supply' in heat_pump.keys() and heat_pump['@dual_supply'] == 'True':
dual_supply = True
heat_pump_component = NonPvGenerationSystem(heat_pump_id,
name=name,
@ -113,8 +106,7 @@ class NorthAmericaEnergySystemCatalog(Catalog):
minimum_heat_supply_temperature=heat_pump_minimum_heat_supply_temperature,
maximum_cooling_supply_temperature=heat_pump_maximum_cooling_supply_temperature,
minimum_cooling_supply_temperature=heat_pump_minimum_cooling_supply_temperature,
heat_efficiency_curve=cop_curve,
dual_supply_capability=dual_supply)
heat_efficiency_curve=cop_curve)
generation_components.append(heat_pump_component)
for pv in photovoltaics:
pv_id = pv['@generation_id']
@ -151,8 +143,6 @@ class NorthAmericaEnergySystemCatalog(Catalog):
for template in templates:
system_id = template['@generation_id']
system_name = template['@name']
if '@dual_supply' in template.keys() and template['@dual_supply'] == 'True':
dual_supply = True
if 'storage_id' in template.keys():
storage_component = template['storage_id']
storage_systems = self._search_storage_equipment(self._load_storage_components(), storage_component)
@ -168,8 +158,7 @@ class NorthAmericaEnergySystemCatalog(Catalog):
system_type=system_type,
fuel_type=fuel_type,
heat_efficiency=heat_efficiency,
energy_storage_systems=energy_storage_system,
dual_supply_capability=dual_supply)
energy_storage_systems=energy_storage_system)
generation_components.append(boiler_template)
elif "Heat Pump" in system_name:
system_type = 'heat pump'
@ -184,8 +173,7 @@ class NorthAmericaEnergySystemCatalog(Catalog):
supply_medium=supply_medium,
fuel_type=fuel_type,
heat_efficiency=heat_efficiency,
energy_storage_systems=energy_storage_system,
dual_supply_capability=dual_supply)
energy_storage_systems=energy_storage_system)
generation_components.append(heat_pump_template)
else:
electricity_efficiency = float(template['@nominalEfficiency'])

View File

@ -10,7 +10,6 @@ from typing import TypeVar
from hub.catalog_factories.energy_systems.montreal_custom_catalog import MontrealCustomCatalog
from hub.catalog_factories.energy_systems.north_america_energy_system_catalog import NorthAmericaEnergySystemCatalog
from hub.catalog_factories.energy_systems.montreal_future_system_catalogue import MontrealFutureSystemCatalogue
from hub.helpers.utils import validate_import_export_type
Catalog = TypeVar('Catalog')
@ -41,13 +40,6 @@ class EnergySystemsCatalogFactory:
"""
return NorthAmericaEnergySystemCatalog(self._path)
@property
def _montreal_future(self):
"""
Retrieve North American catalog
"""
return MontrealFutureSystemCatalogue(self._path)
@property
def catalog(self) -> Catalog:
"""

View File

@ -762,16 +762,13 @@ class Building(CityObject):
if demand_type.lower() == consumption_type.lower():
if consumption_type in (cte.HEATING, cte.DOMESTIC_HOT_WATER):
for generation_system in generation_systems:
if generation_system.heat_efficiency is not None:
coefficient_of_performance = float(generation_system.heat_efficiency)
coefficient_of_performance = generation_system.heat_efficiency
elif consumption_type == cte.COOLING:
for generation_system in generation_systems:
if generation_system.cooling_efficiency is not None:
coefficient_of_performance = float(generation_system.cooling_efficiency)
coefficient_of_performance = generation_system.cooling_efficiency
elif consumption_type == cte.ELECTRICITY:
for generation_system in generation_systems:
if generation_system.electricity_efficiency is not None:
coefficient_of_performance = float(generation_system.electricity_efficiency)
coefficient_of_performance = generation_system.electricity_efficiency
if coefficient_of_performance == 0:
values = [0]*len(demand)
final_energy_consumed = values
@ -802,22 +799,18 @@ class Building(CityObject):
if self.energy_systems is None:
return self._onsite_electrical_production
for energy_system in self.energy_systems:
for generation_system in energy_system.generation_systems:
if generation_system.system_type == cte.PHOTOVOLTAIC:
if generation_system.electricity_efficiency is not None:
_efficiency = float(generation_system.electricity_efficiency)
else:
_efficiency = 0
self._onsite_electrical_production = {}
for _key in self.roofs[0].global_irradiance.keys():
_results = [0 for _ in range(0, len(self.roofs[0].global_irradiance[_key]))]
for surface in self.roofs:
if _key in orientation_losses_factor:
_results = [x + y * _efficiency * surface.perimeter_area
* surface.solar_collectors_area_reduction_factor * z
for x, y, z in zip(_results, surface.global_irradiance[_key],
orientation_losses_factor[_key]['south'])]
self._onsite_electrical_production[_key] = _results
if energy_system.generation_systems[0].system_type == cte.PHOTOVOLTAIC:
_efficiency = energy_system.generation_systems[0].electricity_efficiency
self._onsite_electrical_production = {}
for _key in self.roofs[0].global_irradiance.keys():
_results = [0 for _ in range(0, len(self.roofs[0].global_irradiance[_key]))]
for surface in self.roofs:
if _key in orientation_losses_factor:
_results = [x + y * _efficiency * surface.perimeter_area
* surface.solar_collectors_area_reduction_factor * z
for x, y, z in zip(_results, surface.global_irradiance[_key],
orientation_losses_factor[_key]['south'])]
self._onsite_electrical_production[_key] = _results
return self._onsite_electrical_production
@property

Some files were not shown because too many files have changed in this diff Show More