"""
Unified GeoJSON schema for district heating networks with layered data model.
Author: Dipl.-Ing. (FH) Jonas Pfeiffer
"""
import json
import logging
from datetime import datetime
from typing import Any
import geopandas as gpd
import pandas as pd
from shapely.geometry import LineString
from districtheatingsim.utilities.crs_utils import DEFAULT_CRS, crs_to_urn, epsg_from_urn
logger = logging.getLogger(__name__)
[docs]
class NetworkGeoJSONSchema:
"""
Unified GeoJSON schema for district heating networks with editable/protected data separation.
"""
VERSION = "2.0"
# Feature types
FEATURE_TYPE_FLOW = "network_line_flow"
FEATURE_TYPE_RETURN = "network_line_return"
FEATURE_TYPE_BUILDING = "building_connection"
FEATURE_TYPE_GENERATOR = "generator_connection"
# Edit levels
EDIT_LEVEL_EDITABLE = "editable"
EDIT_LEVEL_GENERATED = "generated"
EDIT_LEVEL_PROTECTED = "protected"
[docs]
@staticmethod
def create_network_line_feature(
geometry: LineString, layer: str, segment_id: str, color: str = None, calculated_data: dict = None
) -> dict[str, Any]:
"""
Create a network line feature (flow or return).
:param geometry: Line geometry
:type geometry: LineString
:param layer: Layer type ('flow' or 'return')
:type layer: str
:param segment_id: Unique segment identifier
:type segment_id: str
:param color: Hex color code (default: #59DB7F for flow, #0C350A for return)
:type color: str
:param calculated_data: Calculation results (diameter, flow rate, etc.)
:type calculated_data: Dict
:return: GeoJSON Feature with style and calculated properties
:rtype: Dict[str, Any]
"""
# Default colors
if color is None:
color = "#59DB7F" if layer == "flow" else "#0C350A"
# Calculate length from geometry
length_m = geometry.length
# Extract start/end elevations from 3-D geometry when available
coords = list(geometry.coords)
elevation_start_m = coords[0][2] if len(coords[0]) > 2 else None
elevation_end_m = coords[-1][2] if len(coords[-1]) > 2 else None
feature = {
"type": "Feature",
"properties": {
"feature_type": NetworkGeoJSONSchema.FEATURE_TYPE_FLOW
if layer == "flow"
else NetworkGeoJSONSchema.FEATURE_TYPE_RETURN,
"layer": layer,
"segment_id": segment_id,
"editable": True,
"style": {"color": color, "weight": 3, "opacity": 1},
"calculated": {
"length_m": length_m,
"diameter_mm": None,
"std_type": None,
"flow_rate_kg_s": None,
"pressure_loss_bar": None,
"velocity_m_s": None,
"elevation_start_m": elevation_start_m,
"elevation_end_m": elevation_end_m,
},
},
"geometry": geometry.__geo_interface__,
}
# Add calculated data if provided
if calculated_data:
feature["properties"]["calculated"].update(calculated_data)
return feature
[docs]
@staticmethod
def create_building_connection_feature(
geometry: LineString, connection_id: str, building_data: dict[str, Any]
) -> dict[str, Any]:
"""
Create a building connection feature with protected building data.
:param geometry: Connection line geometry
:type geometry: LineString
:param connection_id: Unique connection identifier
:type connection_id: str
:param building_data: Building data from CSV (protected)
:type building_data: Dict[str, Any]
:return: GeoJSON Feature with building metadata
:rtype: Dict[str, Any]
"""
feature = {
"type": "Feature",
"properties": {
"feature_type": NetworkGeoJSONSchema.FEATURE_TYPE_BUILDING,
"connection_id": connection_id,
"editable": False,
"building_data": {key: (None if pd.isna(value) else value) for key, value in building_data.items()},
"style": {"color": "#FF0000", "weight": 2, "opacity": 0.8},
},
"geometry": geometry.__geo_interface__,
}
return feature
[docs]
@staticmethod
def create_generator_connection_feature(
geometry: LineString, connection_id: str, generator_type: str = "main", location_index: int = 0
) -> dict[str, Any]:
"""
Create a generator connection feature.
**Coordinate convention** – the LineString geometry *must* follow
``[Vorlauf_point, Rücklauf_point]`` order, i.e.:
* ``coords[0]`` → supply / Vorlauf (VL) endpoint
* ``coords[1]`` → return / Rücklauf (RL) endpoint
This convention is required by
:func:`pp_net_initialisation_geojson.create_network` so that the
circulation pump is connected with ``flow_junction`` on the VL side
and ``return_junction`` on the RL side. All network generation
helpers (``osmnx_steiner_network``, ``import_and_create_layers``)
already follow this convention. If you create or edit generator
connections manually, ensure the line direction is preserved.
:param geometry: Connection line geometry (VL→RL direction).
:type geometry: LineString
:param connection_id: Unique connection identifier.
:type connection_id: str
:param generator_type: Generator type (``'main'`` or ``'secondary'``).
:type generator_type: str
:param location_index: Generator location index.
:type location_index: int
:return: GeoJSON Feature with generator metadata.
:rtype: Dict[str, Any]
"""
feature = {
"type": "Feature",
"properties": {
"feature_type": NetworkGeoJSONSchema.FEATURE_TYPE_GENERATOR,
"connection_id": connection_id,
"editable": False,
"generator_data": {"type": generator_type, "location_index": location_index},
"style": {"color": "#0000FF", "weight": 2, "opacity": 1},
},
"geometry": geometry.__geo_interface__,
}
return feature
[docs]
@staticmethod
def create_network_geojson(
flow_lines: gpd.GeoDataFrame,
return_lines: gpd.GeoDataFrame,
building_connections: gpd.GeoDataFrame,
generator_connections: gpd.GeoDataFrame,
state: str = "designed",
calculated_data: dict = None,
crs: str = DEFAULT_CRS,
) -> dict[str, Any]:
"""
Create unified network GeoJSON from separate components.
:param flow_lines: Supply line network
:type flow_lines: gpd.GeoDataFrame
:param return_lines: Return line network
:type return_lines: gpd.GeoDataFrame
:param building_connections: Building connections with data
:type building_connections: gpd.GeoDataFrame
:param generator_connections: Generator connections
:type generator_connections: gpd.GeoDataFrame
:param state: Network state
:type state: str
:param calculated_data: Calculation results indexed by segment_id
:type calculated_data: Dict
:param crs: Projected CRS for the output GeoJSON (default EPSG:25833)
:type crs: str
:return: Complete GeoJSON FeatureCollection
:rtype: Dict[str, Any]
"""
features = []
# Add flow lines
for idx, row in flow_lines.iterrows():
segment_id = f"flow_{idx:03d}"
# Extract calculated data from row if present
calc_data = {}
if "diameter_mm" in row and not pd.isna(row["diameter_mm"]):
calc_data["diameter_mm"] = float(row["diameter_mm"])
if "std_type" in row and not pd.isna(row["std_type"]):
calc_data["std_type"] = str(row["std_type"])
if "length_m" in row and not pd.isna(row["length_m"]):
calc_data["length_m"] = float(row["length_m"])
if "flow_rate_kg_s" in row and not pd.isna(row["flow_rate_kg_s"]):
calc_data["flow_rate_kg_s"] = float(row["flow_rate_kg_s"])
if "pressure_loss_bar" in row and not pd.isna(row["pressure_loss_bar"]):
calc_data["pressure_loss_bar"] = float(row["pressure_loss_bar"])
if "velocity_m_s" in row and not pd.isna(row["velocity_m_s"]):
calc_data["velocity_m_s"] = float(row["velocity_m_s"])
# Override with explicitly provided calculated_data if available
if calculated_data and segment_id in calculated_data:
calc_data.update(calculated_data[segment_id])
# Check if row has properties (after editing in Leaflet)
color = None
if "color" in row:
color = row["color"]
elif "properties" in row and isinstance(row["properties"], dict):
color = row["properties"].get("color")
feature = NetworkGeoJSONSchema.create_network_line_feature(
geometry=row.geometry,
layer="flow",
segment_id=segment_id,
color=color,
calculated_data=calc_data if calc_data else None,
)
features.append(feature)
# Add return lines
for idx, row in return_lines.iterrows():
segment_id = f"return_{idx:03d}"
# Extract calculated data from row if present
calc_data = {}
if "diameter_mm" in row and not pd.isna(row["diameter_mm"]):
calc_data["diameter_mm"] = float(row["diameter_mm"])
if "std_type" in row and not pd.isna(row["std_type"]):
calc_data["std_type"] = str(row["std_type"])
if "length_m" in row and not pd.isna(row["length_m"]):
calc_data["length_m"] = float(row["length_m"])
if "flow_rate_kg_s" in row and not pd.isna(row["flow_rate_kg_s"]):
calc_data["flow_rate_kg_s"] = float(row["flow_rate_kg_s"])
if "pressure_loss_bar" in row and not pd.isna(row["pressure_loss_bar"]):
calc_data["pressure_loss_bar"] = float(row["pressure_loss_bar"])
if "velocity_m_s" in row and not pd.isna(row["velocity_m_s"]):
calc_data["velocity_m_s"] = float(row["velocity_m_s"])
# Override with explicitly provided calculated_data if available
if calculated_data and segment_id in calculated_data:
calc_data.update(calculated_data[segment_id])
color = None
if "color" in row:
color = row["color"]
elif "properties" in row and isinstance(row["properties"], dict):
color = row["properties"].get("color")
feature = NetworkGeoJSONSchema.create_network_line_feature(
geometry=row.geometry,
layer="return",
segment_id=segment_id,
color=color,
calculated_data=calc_data if calc_data else None,
)
features.append(feature)
# Add building connections
for idx, row in building_connections.iterrows():
connection_id = f"hast_{idx:03d}"
# Extract building data from row
building_data = {col: row[col] for col in row.index if col != "geometry" and not pd.isna(row[col])}
feature = NetworkGeoJSONSchema.create_building_connection_feature(
geometry=row.geometry, connection_id=connection_id, building_data=building_data
)
features.append(feature)
# Add generator connections
for idx, row in generator_connections.iterrows():
connection_id = f"gen_{idx:03d}"
feature = NetworkGeoJSONSchema.create_generator_connection_feature(
geometry=row.geometry,
connection_id=connection_id,
generator_type="main" if idx == 0 else "secondary",
location_index=idx,
)
features.append(feature)
# Create complete GeoJSON
geojson = {
"type": "FeatureCollection",
"name": "Wärmenetz",
"crs": {"type": "name", "properties": {"name": crs_to_urn(crs)}},
"metadata": NetworkGeoJSONSchema.create_metadata(state),
"features": features,
}
print(f"Erstelltes GeoJSON mit {len(features)} Features.")
return geojson
[docs]
@staticmethod
def export_to_file(geojson: dict[str, Any], filepath: str) -> None:
"""
Export network GeoJSON to file.
:param geojson: Network GeoJSON dictionary
:type geojson: Dict[str, Any]
:param filepath: Output file path
:type filepath: str
"""
with open(filepath, "w", encoding="utf-8") as f:
json.dump(geojson, f, indent=2, ensure_ascii=False)
print(f"✓ Exported unified network GeoJSON: {filepath}")
[docs]
@staticmethod
def validate_version(geojson: dict[str, Any], *, filepath: str = "") -> str | None:
"""
Warn (do **not** raise) if the network GeoJSON schema version is missing or
newer than this app understands (BACKLOG D4).
The version lives inside the file as ``metadata.version`` (a semver-ish
string, e.g. ``"2.0"``) — its own convention, distinct from the int-based
``_meta`` block used by the JSON artifacts in ``utilities/schema.py`` (a
GeoJSON FeatureCollection carries its metadata inline, not in a sidecar).
Comparison is on the major version.
:param geojson: The loaded GeoJSON FeatureCollection dict.
:param filepath: Optional source path for clearer log messages.
:return: The version string found on disk, or ``None`` if absent.
:rtype: str | None
"""
where = f" ({filepath})" if filepath else ""
found = (geojson.get("metadata") or {}).get("version")
if found is None:
logger.warning("Network GeoJSON%s has no schema version (pre-2.0 file); loading best-effort", where)
return None
try:
found_major = int(str(found).split(".")[0])
current_major = int(NetworkGeoJSONSchema.VERSION.split(".")[0])
except (ValueError, IndexError):
logger.warning("Network GeoJSON%s has an unparseable schema version %r", where, found)
return found
if found_major > current_major:
logger.warning(
"Network GeoJSON%s is schema v%s, newer than this app (v%s); loading best-effort",
where,
found,
NetworkGeoJSONSchema.VERSION,
)
return found
[docs]
@staticmethod
def import_from_file(filepath: str) -> dict[str, Any]:
"""
Import network GeoJSON from file.
:param filepath: Input file path
:type filepath: str
:return: Network GeoJSON dictionary
:rtype: Dict[str, Any]
"""
with open(filepath, encoding="utf-8") as f:
geojson = json.load(f)
NetworkGeoJSONSchema.validate_version(geojson, filepath=filepath)
return geojson
[docs]
@staticmethod
def update_calculated_data(
geojson: dict[str, Any], flow_results: dict[str, dict], return_results: dict[str, dict]
) -> dict[str, Any]:
"""
Update calculated data in unified GeoJSON after network dimensioning.
:param geojson: Unified network GeoJSON
:type geojson: Dict[str, Any]
:param flow_results: Calculation results for flow lines {segment_id: {diameter_mm, ...}}
:type flow_results: Dict[str, Dict]
:param return_results: Calculation results for return lines
:type return_results: Dict[str, Dict]
:return: Updated GeoJSON with calculation results
:rtype: Dict[str, Any]
"""
updated_geojson = geojson.copy()
updated_geojson["metadata"]["state"] = "calculated"
for feature in updated_geojson["features"]:
segment_id = feature["properties"].get("segment_id")
if not segment_id:
continue
# Update flow line results
if segment_id.startswith("flow_") and segment_id in flow_results:
feature["properties"]["calculated"].update(flow_results[segment_id])
# Update return line results
elif segment_id.startswith("return_") and segment_id in return_results:
feature["properties"]["calculated"].update(return_results[segment_id])
return updated_geojson