Skip to content
GitLab
Commit 7da1c336 authored by Eric Duminil's avatar Eric Duminil
Browse files

Splitting results in multiple files

parent b679af18
Hide whitespace changes
Inline Side-by-side
src/simstadt/results/__init__.py 0 → 100644
View file @ 7da1c336
from .base import SimStadtResults, detect_decimal, get_workflow_provider
from .green_water import GreenWaterResults
from .heat_demand import HeatDemandResults
from .kpi import KPI
from .load_profile import LoadProfileResults
from .photovoltaic import PhotovoltaicResults
from .providers import Providers
from .solar_potential import SolarPotentialResults
def create_simstadt_results(description: str, output_files: list) -> SimStadtResults:
"""Factory: detect workflow type from params.xml and return the correct subclass."""
provider = get_workflow_provider(output_files)
return SimStadtResults.create(provider, description=description, output_files=sorted(output_files))
__all__ = [
"KPI",
"Providers",
"SimStadtResults",
"HeatDemandResults",
"PhotovoltaicResults",
"LoadProfileResults",
"SolarPotentialResults",
"GreenWaterResults",
"create_simstadt_results",
"detect_decimal",
"get_workflow_provider",
]
src/simstadt/results.py → src/simstadt/results/base.py
View file @ 7da1c336
"""Parse SimStadt output files into typed result objects.
Supports HeatDemand, Photovoltaic, and GreenWater workflows.
Each result type exposes: .dataframe, .kpis, .diagrams, .csv_path,
.to_json() / .from_json(), and .zip_content().
"""
import io import io
import json import json
import logging import logging
import re
import zipfile import zipfile
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from dataclasses import asdict, dataclass, field from dataclasses import dataclass, field
from enum import Enum
from pathlib import Path from pathlib import Path
from typing import ClassVar, Dict, Sequence, Type from typing import ClassVar, Dict, Sequence, Type
from xml.etree import ElementTree as et from xml.etree import ElementTree as et
import matplotlib.pyplot as plt
import pandas as pd import pandas as pd
logger = logging.getLogger(__name__) from .kpi import KPI
@dataclass
class KPI:
"""Metric with name, value, unit and display precision."""
name: str
value: float
unit: str | None = None
precision: int = 2
NAME_AND_UNIT = re.compile(r"^(.*?) \[(.*?)\]$")
def __post_init__(self):
"""Move units embedded in column names (e.g. 'Area [m²]') into the unit field."""
if match := self.NAME_AND_UNIT.match(self.name):
self.name, self.unit = match.groups()
def to_dict(self) -> dict:
return asdict(self)
@property logger = logging.getLogger(__name__)
def rounded_value(self) -> float:
return round(self.value, self.precision)
def __str__(self):
s = f"{self.name:35s} : {self.value:.{self.precision}f}"
if self.unit:
s = f"{s} {self.unit}"
return s
def __repr__(self):
return str(self)
@classmethod
def from_dict(cls, data: dict) -> "KPI":
return cls(**data)
class Providers(str, Enum):
HEAT_DEMAND = "de.hftstuttgart.simstadtworkflows.energy.HeatDemandAnalysisWorkflowProvider"
HEAT_DEMAND_WITH_REFURBISHMENT = (
"de.hftstuttgart.simstadtworkflows.energy.HeatDemandAnalysisWithRefurbishmentStrategyWorkflowProvider"
)
HEAT_DEMAND_WITH_HISTORIC_REFURBISHMENT = (
"de.hftstuttgart.simstadtworkflows.energy.HeatDemandAnalysisWithHistoricAndFutureRefurbishmentWorkflowProvider"
)
HEAT_DEMAND_WITH_SHADOW = (
"de.hftstuttgart.simstadtworkflows.shadow.HeatDemandCalculationWithShadowProcessingProvider"
)
PHOTOVOLTAIC = "de.hftstuttgart.simstadtworkflows.energy.PhotovoltaicPotentialAnalysisWorkflowProvider"
PHOTOVOLTAIC_WITH_SHADOW = "de.hftstuttgart.simstadtworkflows.shadow.PVPotentialWithShadowProcessingProvider"
PHOTOVOLTAIC_FINANCE = (
"de.hftstuttgart.simstadtworkflows.economics.PhotovoltaicPotentialFinancialAnalysisWorkflowProvider"
)
GREEN_WATER = "de.hftstuttgart.simstadtworkflows.greenwater.GreenWaterWorkflowProvider"
LOAD_PROFILE = "de.hftstuttgart.simstadtworkflows.energy.LoadProfileProvider"
SOLAR_POTENTIAL = "de.hftstuttgart.simstadtworkflows.energy.SolarPotentialAnalysisWorkflowProvider"
def detect_decimal(csv_path: Path, line_number: int, check: str = "") -> str: def detect_decimal(csv_path: Path, line_number: int, check: str = "") -> str:
...@@ -307,307 +242,3 @@ class SimStadtResults(ABC): ...@@ -307,307 +242,3 @@ class SimStadtResults(ABC):
if provider not in cls._registry: if provider not in cls._registry:
raise ValueError(f"Unknown workflow provider: {provider}") raise ValueError(f"Unknown workflow provider: {provider}")
return cls._registry[provider](**kwargs) return cls._registry[provider](**kwargs)
@dataclass(repr=False)
class HeatDemandResults(SimStadtResults):
"""Results for Heat Demand (and Heat+Cool Demand) workflows."""
MAX_FLAT_ROOF_DIFFERENCE = 0.1 # [m]
TOTAL_DEMAND = "Total Yearly Heating + DHW demand"
supported_providers: ClassVar[list[str]] = [
Providers.HEAT_DEMAND,
Providers.HEAT_DEMAND_WITH_SHADOW,
Providers.HEAT_DEMAND_WITH_REFURBISHMENT,
Providers.HEAT_DEMAND_WITH_HISTORIC_REFURBISHMENT,
]
csv_identifier = "DIN18599"
def _parse_results(self) -> pd.DataFrame:
csv_decimal = detect_decimal(self.csv_path, 5, "Latitude")
df = pd.read_csv(self.csv_path, skiprows=list(range(19)) + [20], sep=";", decimal=csv_decimal)
df["has_flat_roof"] = (
df["Ridge/mean Height"] - df["Eaves/mean Height"] < self.MAX_FLAT_ROOF_DIFFERENCE
).fillna(False)
df.attrs["Heating"] = "Yearly Heating demand" in df.columns
df.attrs["Cooling"] = "Yearly Cooling demand" in df.columns
return df
@property
def kpis(self) -> list[KPI]:
df = self.dataframe
custom_kpis = [KPI("Number of buildings", df.shape[0], precision=0)]
sums = [("Heated area", "m²"), ("Footprint area", "m²")]
if df.attrs["Heating"]:
specific_heat_demand = df[self.TOTAL_DEMAND].sum() / df["Heated area"].sum()
heated_buildings = int((df[self.TOTAL_DEMAND] > 10_000).sum())
custom_kpis.extend(
[
KPI("Number of heated buildings", heated_buildings, precision=0),
KPI("Specific Heating Demand", specific_heat_demand, "kWh / (m² · a)", precision=0),
]
)
sums.extend(
[
("Yearly Heating demand", "kWh / a"),
(self.TOTAL_DEMAND, "kWh / a"),
]
)
if df.attrs["Cooling"]:
specific_cooling_demand = df["Yearly Cooling demand"].sum() / df["Heated area"].sum()
custom_kpis.append(KPI("Specific Cooling Demand", specific_cooling_demand, "kWh / (m² · a)", precision=0))
sums.append(("Yearly Cooling demand", "kWh / a"))
averages = [("Mean Uvalue", "W / (m² · K)", 1), ("Year of construction", None), ("Storey number", None)]
return custom_kpis + self._prepare_kpis(sums, averages)
@property
def diagrams(self) -> dict[str, Path]:
heat_png = self.workflow_path / "heating.png"
ax = self.monthly_df().plot.bar(
rot=0,
ylabel="[MWh]",
width=0.8,
color={"Monthly Heating Demand": "darkred", "Monthly Cooling Demand": "darkblue"},
)
plt.savefig(heat_png, bbox_inches="tight", dpi=300)
plt.close(ax.figure)
return {"Monthly demands": heat_png}
def monthly_df(self) -> pd.DataFrame:
months = ["Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"]
df = self.dataframe
monthly_df = pd.DataFrame([], index=pd.Index(months))
for mode in ["Heating", "Cooling"]:
search_str = f"{mode} demand"
found_cols = [
col for col in df.columns if search_str in col and "Yearly" not in col and "Specific" not in col
]
if found_cols:
monthly_df[f"Monthly {mode} Demand"] = (df[found_cols].sum() / 1000).values
return monthly_df
@dataclass(repr=False)
class PhotovoltaicResults(SimStadtResults):
"""Results for Photovoltaic (PV) workflows."""
supported_providers: ClassVar[list[str]] = [
Providers.PHOTOVOLTAIC,
Providers.PHOTOVOLTAIC_FINANCE,
Providers.PHOTOVOLTAIC_WITH_SHADOW,
]
csv_identifier = "_pv_potential"
def _count_header_lines(self) -> int:
count = 0
with open(self.csv_path) as csv:
for line in csv:
if line.startswith("Building ID"):
return count
count += 1
raise ValueError(f"Header not found in {self.csv_path}")
def _parse_results(self) -> pd.DataFrame:
csv_decimal = detect_decimal(self.csv_path, 4, "Latitude")
header_length = self._count_header_lines()
df = pd.read_csv(
self.csv_path,
skiprows=list(range(header_length)) + [header_length + 1],
sep=";",
decimal=csv_decimal,
)
df = df.rename(columns={"Area": "Roof area for PV"})
return df.dropna(axis=1, how="all")
@property
def kpis(self) -> list[KPI]:
sums = [("Roof area for PV", "m²"), ("PV potential nominal power", "kWp"), ("PV potential yield", "MWh / a")]
averages = [("Irradiance in module plane", "W / m²"), ("PV specific yield", "kWh / (kWp · a)")]
return self._prepare_kpis(sums, averages)
@dataclass(repr=False)
class LoadProfileResults(SimStadtResults):
"""Results for LoadProfile workflows.
The dataframe has one column per building (building GML ID as column name)
and 8760 rows of hourly energy demand [kWh/h].
"""
supported_providers: ClassVar[list[str]] = [Providers.LOAD_PROFILE]
csv_identifier = "_load_profile_Hourly"
def _parse_results(self) -> pd.DataFrame:
csv_decimal = detect_decimal(self.csv_path, 6, "Area")
df = pd.read_csv(self.csv_path, skiprows=range(1, 12), sep=";", decimal=csv_decimal)
# First column is the timestamp; remaining columns are per-building loads
return df.drop(columns=[df.columns[0]])
@property
def kpis(self) -> list[KPI]:
# TODO: Add sum, average. Add timestep too?
# TODO: Add people? Add total heated area?
df = self.dataframe
total = df.sum().sum()
return [
KPI("Number of buildings", df.shape[1], precision=0),
KPI("Total load", total, "kWh/a", precision=1),
KPI("Average load", total / 8760, "kWh/h", precision=1),
]
@dataclass(repr=False)
class SolarPotentialResults(SimStadtResults):
"""Results for SolarPotential workflows.
The output is a .prn weather file with 8760 hourly rows and three columns:
GHI (W/m²), DHI (W/m²), Ta (°C).
"""
supported_providers: ClassVar[list[str]] = [Providers.SOLAR_POTENTIAL]
csv_identifier = "_solar_potential" # no CSV output; csv_path will raise if called
def _parse_results(self) -> pd.DataFrame:
prn_paths = self.get_all_by_extension(".prn")
if len(prn_paths) != 1:
raise ValueError(f"Expected exactly one .prn file, found {len(prn_paths)}")
return pd.read_csv(prn_paths[0], sep=r"\s+", names=["GHI", "DHI", "Ta"], header=None)
@property
def kpis(self) -> list[KPI]:
df = self.dataframe
return [
KPI("Annual GHI", df["GHI"].sum() / 1000, "kWh/m²", precision=1),
KPI("Average GHI", df["GHI"].mean(), "W/m²", precision=1),
KPI("Annual DHI", df["DHI"].sum() / 1000, "kWh/m²", precision=1),
KPI("Average DHI", df["DHI"].mean(), "W/m²", precision=1),
KPI("Average temperature", float(df["Ta"].mean()), "°C", precision=1),
]
@dataclass(repr=False)
class GreenWaterResults(SimStadtResults):
"""Results for GreenWater workflows (hourly time series with tree species breakdown)."""
csv_identifier = "_greenwater"
supported_providers: ClassVar[list[str]] = [Providers.GREEN_WATER]
def _parse_results(self) -> pd.DataFrame:
csv_decimal = detect_decimal(self.csv_path, 5, "tree area")
df = pd.read_csv(
self.csv_path,
sep=";",
header=[0, 1],
skiprows=list(range(8)) + list(range(10, 13)),
decimal=csv_decimal,
)
# Flatten multi-level columns: species names are forward-filled across sub-columns
cols = df.columns.tolist()
new_cols = []
current_species = "All"
species = []
for col in cols:
level0, level1 = col
if pd.notna(level0) and not level0.startswith("Unnamed") and not level0.startswith("#"):
current_species = level0
species.append(current_species)
new_cols.append(level1 if current_species == "All" else f"{current_species} - {level1}")
df.columns = new_cols
# Parse header metadata (tree counts, areas per species)
with open(self.csv_path, encoding="utf-8") as csv:
line = ""
for _ in range(5):
line = next(csv)
tree_count = int(line.split(";")[1])
line = next(csv)
tree_area_header = re.split("[;,]", line.strip())
total_tree_area = float(tree_area_header[1].replace(",", "."))
if tree_count == 0:
raise ValueError(f"CityGML {self.citygml} does not have any tree!")
for _ in range(5):
line = next(csv)
species_count = [int(c) for c in re.split(";+", line.strip())[1:-1]]
line = next(csv)
species_area = [float(c.replace(",", ".")) for c in re.split(";+", line.strip())[1:-1]]
species.pop(0)
df.attrs = {
"species": {
name: {"count": count, "area": area}
for name, count, area in zip(species, species_count, species_area)
},
"tree count": tree_count,
"tree area": total_tree_area,
}
return df
def df_with_time(self) -> pd.DataFrame:
"""Return the dataframe with a DatetimeIndex (year 2005, hourly)."""
df = self.dataframe
times = pd.date_range(start="2005-01-01 00:00", freq="1h", periods=8760)
df = df.set_index(times)
return df.drop(columns=["#"])
@property
def kpis(self) -> list[KPI]:
df = self.dataframe
rain = df["Rain [mm]"].sum()
etc = df["Average ETc [mm]"].sum()
cooling = df["Evaporative cooling [kWh/h]"].sum() / 1000
irrigation_mm = df["Average irrigation [mm]"].sum()
tree_area = df.attrs["tree area"]
irrigation_m3 = irrigation_mm * tree_area / 1000
return [
KPI("Number of trees", df.attrs["tree count"], precision=0),
KPI("Projected tree area", tree_area, "m²", precision=1),
KPI("Rain", rain, "mm / a", precision=0),
KPI("Average ETc", etc, "mm / a", precision=0),
KPI("Irrigation", irrigation_m3, "m³ / a", precision=0),
KPI("Evaporative cooling", cooling, "MWh / a", precision=0),
]
@property
def more_info(self) -> dict[str, list[KPI]]:
df = self.dataframe
all_info = {}
for name, info in df.attrs["species"].items():
count = info["count"]
area = info["area"]
irrigation_mm = df[f"{name} - Average irrigation [mm]"].sum()
average_area = area / count
all_info[f"_{name}_"] = [
KPI("Number of trees", count, precision=0),
KPI("Average tree area", average_area, "m²", precision=1),
KPI("Average ETc", df[f"{name} - Average ETc [mm]"].sum(), "mm / a", precision=0),
KPI("Irrigation", irrigation_mm, "mm / a", precision=0),
KPI("Average irrigation", average_area * irrigation_mm / 1000, "m³ / a", precision=0),
]
return all_info
@property
def diagrams(self) -> dict[str, Path]:
rain_irrigation_png = self.workflow_path / "rain_and_irrigation.png"
df_months = self.df_with_time()[["Rain [mm]", "Average irrigation [mm]"]].resample("ME").sum()
fig, ax = plt.subplots()
df_months = df_months.rename(
columns={"Rain [mm]": "Niederschlag", "Average irrigation [mm]": "Künstliche Bewässerung"}
)
df_months.plot(kind="bar", stacked=True, ax=ax, ylabel="[mm]", figsize=(13, 6))
ax.set_xticklabels([x.strftime("%b") for x in df_months.index], rotation=0)
plt.savefig(rain_irrigation_png, bbox_inches="tight", dpi=300)
plt.close(fig)
return {"Rain and Irrigation": rain_irrigation_png}
def create_simstadt_results(description: str, output_files: list[Path]) -> SimStadtResults:
"""Factory: detect workflow type from params.xml and return the correct subclass."""
provider = get_workflow_provider(output_files)
return SimStadtResults.create(provider, description=description, output_files=sorted(output_files))
src/simstadt/results/green_water.py 0 → 100644
View file @ 7da1c336
import re
from dataclasses import dataclass
from pathlib import Path
from typing import ClassVar
import matplotlib.pyplot as plt
import pandas as pd
from .base import SimStadtResults, detect_decimal
from .kpi import KPI
from .providers import Providers
@dataclass(repr=False)
class GreenWaterResults(SimStadtResults):
"""Results for GreenWater workflows (hourly time series with tree species breakdown)."""
csv_identifier = "_greenwater"
supported_providers: ClassVar[list[str]] = [Providers.GREEN_WATER]
def _parse_results(self) -> pd.DataFrame:
csv_decimal = detect_decimal(self.csv_path, 5, "tree area")
df = pd.read_csv(
self.csv_path,
sep=";",
header=[0, 1],
skiprows=list(range(8)) + list(range(10, 13)),
decimal=csv_decimal,
)
# Flatten multi-level columns: species names are forward-filled across sub-columns
cols = df.columns.tolist()
new_cols = []
current_species = "All"
species = []
for col in cols:
level0, level1 = col
if pd.notna(level0) and not level0.startswith("Unnamed") and not level0.startswith("#"):
current_species = level0
species.append(current_species)
new_cols.append(level1 if current_species == "All" else f"{current_species} - {level1}")
df.columns = new_cols
# Parse header metadata (tree counts, areas per species)
with open(self.csv_path, encoding="utf-8") as csv:
line = ""
for _ in range(5):
line = next(csv)
tree_count = int(line.split(";")[1])
line = next(csv)
tree_area_header = re.split("[;,]", line.strip())
total_tree_area = float(tree_area_header[1].replace(",", "."))
if tree_count == 0:
raise ValueError(f"CityGML {self.citygml} does not have any tree!")
for _ in range(5):
line = next(csv)
species_count = [int(c) for c in re.split(";+", line.strip())[1:-1]]
line = next(csv)
species_area = [float(c.replace(",", ".")) for c in re.split(";+", line.strip())[1:-1]]
species.pop(0)
df.attrs = {
"species": {
name: {"count": count, "area": area}
for name, count, area in zip(species, species_count, species_area)
},
"tree count": tree_count,
"tree area": total_tree_area,
}
return df
def df_with_time(self) -> pd.DataFrame:
"""Return the dataframe with a DatetimeIndex (year 2005, hourly)."""
df = self.dataframe
times = pd.date_range(start="2005-01-01 00:00", freq="1h", periods=8760)
df = df.set_index(times)
return df.drop(columns=["#"])
@property
def kpis(self) -> list[KPI]:
df = self.dataframe
rain = df["Rain [mm]"].sum()
etc = df["Average ETc [mm]"].sum()
cooling = df["Evaporative cooling [kWh/h]"].sum() / 1000
irrigation_mm = df["Average irrigation [mm]"].sum()
tree_area = df.attrs["tree area"]
irrigation_m3 = irrigation_mm * tree_area / 1000
return [
KPI("Number of trees", df.attrs["tree count"], precision=0),
KPI("Projected tree area", tree_area, "m²", precision=1),
KPI("Rain", rain, "mm / a", precision=0),
KPI("Average ETc", etc, "mm / a", precision=0),
KPI("Irrigation", irrigation_m3, "m³ / a", precision=0),
KPI("Evaporative cooling", cooling, "MWh / a", precision=0),
]
@property
def more_info(self) -> dict[str, list[KPI]]:
df = self.dataframe
all_info = {}
for name, info in df.attrs["species"].items():
count = info["count"]
area = info["area"]
irrigation_mm = df[f"{name} - Average irrigation [mm]"].sum()
average_area = area / count
all_info[f"_{name}_"] = [
KPI("Number of trees", count, precision=0),
KPI("Average tree area", average_area, "m²", precision=1),
KPI("Average ETc", df[f"{name} - Average ETc [mm]"].sum(), "mm / a", precision=0),
KPI("Irrigation", irrigation_mm, "mm / a", precision=0),
KPI("Average irrigation", average_area * irrigation_mm / 1000, "m³ / a", precision=0),
]
return all_info
@property
def diagrams(self) -> dict[str, Path]:
rain_irrigation_png = self.workflow_path / "rain_and_irrigation.png"
df_months = self.df_with_time()[["Rain [mm]", "Average irrigation [mm]"]].resample("ME").sum()
fig, ax = plt.subplots()
df_months = df_months.rename(
columns={"Rain [mm]": "Niederschlag", "Average irrigation [mm]": "Künstliche Bewässerung"}
)
df_months.plot(kind="bar", stacked=True, ax=ax, ylabel="[mm]", figsize=(13, 6))
ax.set_xticklabels([x.strftime("%b") for x in df_months.index], rotation=0)
plt.savefig(rain_irrigation_png, bbox_inches="tight", dpi=300)
plt.close(fig)
return {"Rain and Irrigation": rain_irrigation_png}
src/simstadt/results/heat_demand.py 0 → 100644
View file @ 7da1c336
from dataclasses import dataclass
from pathlib import Path
from typing import ClassVar
import matplotlib.pyplot as plt
import pandas as pd
from .base import SimStadtResults, detect_decimal
from .kpi import KPI
from .providers import Providers
@dataclass(repr=False)
class HeatDemandResults(SimStadtResults):
"""Results for Heat Demand (and Heat+Cool Demand) workflows."""
MAX_FLAT_ROOF_DIFFERENCE = 0.1 # [m]
TOTAL_DEMAND = "Total Yearly Heating + DHW demand"
supported_providers: ClassVar[list[str]] = [
Providers.HEAT_DEMAND,
Providers.HEAT_DEMAND_WITH_SHADOW,
Providers.HEAT_DEMAND_WITH_REFURBISHMENT,
Providers.HEAT_DEMAND_WITH_HISTORIC_REFURBISHMENT,
]
csv_identifier = "DIN18599"
def _parse_results(self) -> pd.DataFrame:
csv_decimal = detect_decimal(self.csv_path, 5, "Latitude")
df = pd.read_csv(self.csv_path, skiprows=list(range(19)) + [20], sep=";", decimal=csv_decimal)
df["has_flat_roof"] = (
df["Ridge/mean Height"] - df["Eaves/mean Height"] < self.MAX_FLAT_ROOF_DIFFERENCE
).fillna(False)
df.attrs["Heating"] = "Yearly Heating demand" in df.columns
df.attrs["Cooling"] = "Yearly Cooling demand" in df.columns
return df
@property
def kpis(self) -> list[KPI]:
df = self.dataframe
custom_kpis = [KPI("Number of buildings", df.shape[0], precision=0)]
sums = [("Heated area", "m²"), ("Footprint area", "m²")]
if df.attrs["Heating"]:
specific_heat_demand = df[self.TOTAL_DEMAND].sum() / df["Heated area"].sum()
heated_buildings = int((df[self.TOTAL_DEMAND] > 10_000).sum())
custom_kpis.extend(
[
KPI("Number of heated buildings", heated_buildings, precision=0),
KPI("Specific Heating Demand", specific_heat_demand, "kWh / (m² · a)", precision=0),
]
)
sums.extend(
[
("Yearly Heating demand", "kWh / a"),
(self.TOTAL_DEMAND, "kWh / a"),
]
)
if df.attrs["Cooling"]:
specific_cooling_demand = df["Yearly Cooling demand"].sum() / df["Heated area"].sum()
custom_kpis.append(KPI("Specific Cooling Demand", specific_cooling_demand, "kWh / (m² · a)", precision=0))
sums.append(("Yearly Cooling demand", "kWh / a"))
averages = [("Mean Uvalue", "W / (m² · K)", 1), ("Year of construction", None), ("Storey number", None)]
return custom_kpis + self._prepare_kpis(sums, averages)
@property
def diagrams(self) -> dict[str, Path]:
heat_png = self.workflow_path / "heating.png"
ax = self.monthly_df().plot.bar(
rot=0,
ylabel="[MWh]",
width=0.8,
color={"Monthly Heating Demand": "darkred", "Monthly Cooling Demand": "darkblue"},
)
plt.savefig(heat_png, bbox_inches="tight", dpi=300)
plt.close(ax.figure)
return {"Monthly demands": heat_png}
def monthly_df(self) -> pd.DataFrame:
months = ["Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"]
df = self.dataframe
monthly_df = pd.DataFrame([], index=pd.Index(months))
for mode in ["Heating", "Cooling"]:
search_str = f"{mode} demand"
found_cols = [
col for col in df.columns if search_str in col and "Yearly" not in col and "Specific" not in col
]
if found_cols:
monthly_df[f"Monthly {mode} Demand"] = (df[found_cols].sum() / 1000).values
return monthly_df
src/simstadt/results/kpi.py 0 → 100644
View file @ 7da1c336
import re
from dataclasses import asdict, dataclass
@dataclass
class KPI:
"""Metric with name, value, unit and display precision."""
name: str
value: float
unit: str | None = None
precision: int = 2
NAME_AND_UNIT = re.compile(r"^(.*?) \[(.*?)\]$")
def __post_init__(self):
"""Move units embedded in column names (e.g. 'Area [m²]') into the unit field."""
if match := self.NAME_AND_UNIT.match(self.name):
self.name, self.unit = match.groups()
def to_dict(self) -> dict:
return asdict(self)
@property
def rounded_value(self) -> float:
return round(self.value, self.precision)
def __str__(self):
s = f"{self.name:35s} : {self.value:.{self.precision}f}"
if self.unit:
s = f"{s} {self.unit}"
return s
def __repr__(self):
return str(self)
@classmethod
def from_dict(cls, data: dict) -> "KPI":
return cls(**data)
src/simstadt/results/load_profile.py 0 → 100644
View file @ 7da1c336
from dataclasses import dataclass
from typing import ClassVar
import pandas as pd
from .base import SimStadtResults, detect_decimal
from .kpi import KPI
from .providers import Providers
@dataclass(repr=False)
class LoadProfileResults(SimStadtResults):
"""Results for LoadProfile workflows.
The dataframe has one column per building (building GML ID as column name)
and 8760 rows of hourly energy demand [kWh/h].
"""
supported_providers: ClassVar[list[str]] = [Providers.LOAD_PROFILE]
csv_identifier = "_load_profile_Hourly"
def _parse_results(self) -> pd.DataFrame:
csv_decimal = detect_decimal(self.csv_path, 6, "Area")
df = pd.read_csv(self.csv_path, skiprows=range(1, 12), sep=";", decimal=csv_decimal)
# First column is the timestamp; remaining columns are per-building loads
return df.drop(columns=[df.columns[0]])
@property
def kpis(self) -> list[KPI]:
# TODO: Add sum, average. Add timestep too?
# TODO: Add people? Add total heated area?
df = self.dataframe
total = df.sum().sum()
return [
KPI("Number of buildings", df.shape[1], precision=0),
KPI("Total load", total, "kWh/a", precision=1),
KPI("Average load", total / 8760, "kWh/h", precision=1),
]
src/simstadt/results/photovoltaic.py 0 → 100644
View file @ 7da1c336
from dataclasses import dataclass
from typing import ClassVar
import pandas as pd
from .base import SimStadtResults, detect_decimal
from .kpi import KPI
from .providers import Providers
@dataclass(repr=False)
class PhotovoltaicResults(SimStadtResults):
"""Results for Photovoltaic (PV) workflows."""
supported_providers: ClassVar[list[str]] = [
Providers.PHOTOVOLTAIC,
Providers.PHOTOVOLTAIC_FINANCE,
Providers.PHOTOVOLTAIC_WITH_SHADOW,
]
csv_identifier = "_pv_potential"
def _count_header_lines(self) -> int:
count = 0
with open(self.csv_path) as csv:
for line in csv:
if line.startswith("Building ID"):
return count
count += 1
raise ValueError(f"Header not found in {self.csv_path}")
def _parse_results(self) -> pd.DataFrame:
csv_decimal = detect_decimal(self.csv_path, 4, "Latitude")
header_length = self._count_header_lines()
df = pd.read_csv(
self.csv_path,
skiprows=list(range(header_length)) + [header_length + 1],
sep=";",
decimal=csv_decimal,
)
df = df.rename(columns={"Area": "Roof area for PV"})
return df.dropna(axis=1, how="all")
@property
def kpis(self) -> list[KPI]:
sums = [("Roof area for PV", "m²"), ("PV potential nominal power", "kWp"), ("PV potential yield", "MWh / a")]
averages = [("Irradiance in module plane", "W / m²"), ("PV specific yield", "kWh / (kWp · a)")]
return self._prepare_kpis(sums, averages)
src/simstadt/results/providers.py 0 → 100644
View file @ 7da1c336
from enum import Enum
class Providers(str, Enum):
HEAT_DEMAND = "de.hftstuttgart.simstadtworkflows.energy.HeatDemandAnalysisWorkflowProvider"
HEAT_DEMAND_WITH_REFURBISHMENT = (
"de.hftstuttgart.simstadtworkflows.energy.HeatDemandAnalysisWithRefurbishmentStrategyWorkflowProvider"
)
HEAT_DEMAND_WITH_HISTORIC_REFURBISHMENT = (
"de.hftstuttgart.simstadtworkflows.energy.HeatDemandAnalysisWithHistoricAndFutureRefurbishmentWorkflowProvider"
)
HEAT_DEMAND_WITH_SHADOW = (
"de.hftstuttgart.simstadtworkflows.shadow.HeatDemandCalculationWithShadowProcessingProvider"
)
PHOTOVOLTAIC = "de.hftstuttgart.simstadtworkflows.energy.PhotovoltaicPotentialAnalysisWorkflowProvider"
PHOTOVOLTAIC_WITH_SHADOW = "de.hftstuttgart.simstadtworkflows.shadow.PVPotentialWithShadowProcessingProvider"
PHOTOVOLTAIC_FINANCE = (
"de.hftstuttgart.simstadtworkflows.economics.PhotovoltaicPotentialFinancialAnalysisWorkflowProvider"
)
GREEN_WATER = "de.hftstuttgart.simstadtworkflows.greenwater.GreenWaterWorkflowProvider"
LOAD_PROFILE = "de.hftstuttgart.simstadtworkflows.energy.LoadProfileProvider"
SOLAR_POTENTIAL = "de.hftstuttgart.simstadtworkflows.energy.SolarPotentialAnalysisWorkflowProvider"
src/simstadt/results/solar_potential.py 0 → 100644
View file @ 7da1c336
from dataclasses import dataclass
from typing import ClassVar
import pandas as pd
from .base import SimStadtResults
from .kpi import KPI
from .providers import Providers
@dataclass(repr=False)
class SolarPotentialResults(SimStadtResults):
"""Results for SolarPotential workflows.
The output is a .prn weather file with 8760 hourly rows and three columns:
GHI (W/m²), DHI (W/m²), Ta (°C).
"""
supported_providers: ClassVar[list[str]] = [Providers.SOLAR_POTENTIAL]
csv_identifier = "_solar_potential" # no CSV output; csv_path will raise if called
def _parse_results(self) -> pd.DataFrame:
prn_paths = self.get_all_by_extension(".prn")
if len(prn_paths) != 1:
raise ValueError(f"Expected exactly one .prn file, found {len(prn_paths)}")
return pd.read_csv(prn_paths[0], sep=r"\s+", names=["GHI", "DHI", "Ta"], header=None)
@property
def kpis(self) -> list[KPI]:
df = self.dataframe
return [
KPI("Annual GHI", df["GHI"].sum() / 1000, "kWh/m²", precision=1),
KPI("Average GHI", df["GHI"].mean(), "W/m²", precision=1),
KPI("Annual DHI", df["DHI"].sum() / 1000, "kWh/m²", precision=1),
KPI("Average DHI", df["DHI"].mean(), "W/m²", precision=1),
KPI("Average temperature", float(df["Ta"].mean()), "°C", precision=1),
]
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment

Dies ist die Gitlab-Instanz des Transferportals der Hochschule für Technik Stuttgart. Hier geht es zurück zum Portal