chore: add yaml definitions

Group reusable yaml under a definitions key

Author: jsolaas

src/libecalc/presentation/yaml/yaml_models/pyyaml_yaml_model.py

@@ -26,7 +26,7 @@
 from libecalc.presentation.yaml.yaml_models.exceptions import DuplicateKeyError, FileContext, YamlError
 from libecalc.presentation.yaml.yaml_models.yaml_model import YamlConfiguration, YamlValidator
 from libecalc.presentation.yaml.yaml_node import YamlDict, YamlList
-from libecalc.presentation.yaml.yaml_types.components.yaml_asset import YamlAsset
+from libecalc.presentation.yaml.yaml_types.components.yaml_asset import YamlAsset, YamlDefinitions
 from libecalc.presentation.yaml.yaml_types.components.yaml_installation import YamlInstallation
 from libecalc.presentation.yaml.yaml_types.facility_model.yaml_facility_model import YamlFacilityModel
 from libecalc.presentation.yaml.yaml_types.fuel_type.yaml_fuel_type import YamlFuelType
@@ -46,6 +46,7 @@
 from libecalc.presentation.yaml.yaml_validation_context import YamlModelValidationContext
 
 # Top-level YAML keywords used only by experimental/new sections
+_DEFINITIONS_KEY = "DEFINITIONS"
 _PROCESS_UNITS_KEY = "PROCESS_UNITS"
 _PROCESS_PIPELINES_KEY = "PROCESS_PIPELINES"
 _INLET_STREAMS_KEY = "INLET_STREAMS"
@@ -54,10 +55,6 @@
 _ECALC_EVENTS_KEY = "ECALC_EVENTS"
 _PROCESS_EVENTS_KEY = "PROCESS_EVENTS"
 _PUMP_PROCESS_SIMULATIONS_KEY = "PUMP_PROCESS_SIMULATIONS"
-_NEW_SECTIONS_WITH_FILE_REFS: tuple[str, ...] = (
-    "PROCESS_UNITS",
-    "PROCESS_PIPELINES",
-)
 dt_adapter = TypeAdapter(datetime.datetime)
 
 
@@ -323,8 +320,14 @@ def facility_resource_names(self) -> list[str]:
             data.get(EcalcYamlKeywords.file) for data in resource_data if data.get(EcalcYamlKeywords.file) is not None
         ]
 
-        # Pick up FILE references nested in the new YAML sections (PROCESS_UNITS, PROCESS_PIPELINES, ...).
-        for section in _NEW_SECTIONS_WITH_FILE_REFS:
+        # Pick up FILE references nested in the new YAML sections (DEFINITIONS.PROCESS_UNITS, PROCESS_PIPELINES, ...).
+        definitions = (
+            self._internal_datamodel.get(_DEFINITIONS_KEY, {}) if isinstance(self._internal_datamodel, dict) else {}
+        )
+        if isinstance(definitions, dict):
+            for section in ("PROCESS_UNITS",):
+                resource_names.extend(_find_file_references(definitions.get(section)))
+        for section in ("PROCESS_PIPELINES",):
             resource_names.extend(_find_file_references(self._internal_datamodel.get(section)))
 
         # Dedup while preserving order — the same CSV may be referenced by multiple charts.
@@ -451,16 +454,19 @@ def fluid_models(self) -> dict[str, YamlFluidModel]:
         return fluid_models
 
     @property
-    def process_units(self) -> dict[str, YamlProcessUnitDefinition]:
+    def definitions(self) -> YamlDefinitions:
         process_units: dict[str, YamlProcessUnitDefinition] = {}
-        raw = self._get_yaml_dict_or_empty(_PROCESS_UNITS_KEY)
+        definitions = (
+            self._internal_datamodel.get(_DEFINITIONS_KEY, {}) if isinstance(self._internal_datamodel, dict) else {}
+        )
+        raw = definitions.get(_PROCESS_UNITS_KEY, {}) if isinstance(definitions, dict) else {}
 
         for name, unit_data in raw.items():
             try:
                 process_units[name] = TypeAdapter(YamlProcessUnitDefinition).validate_python(unit_data)
             except PydanticValidationError:
                 pass
-        return process_units
+        return YamlDefinitions(process_units=process_units)
 
     @property
     def process_pipelines(self) -> dict[str, YamlProcessPipeline]:

src/libecalc/presentation/yaml/yaml_models/yaml_model.py

@@ -11,7 +11,7 @@
     ResourceStream,
 )
 from libecalc.presentation.yaml.yaml_keywords import EcalcYamlKeywords
-from libecalc.presentation.yaml.yaml_types.components.yaml_asset import YamlAsset
+from libecalc.presentation.yaml.yaml_types.components.yaml_asset import YamlAsset, YamlDefinitions
 from libecalc.presentation.yaml.yaml_types.components.yaml_installation import YamlInstallation
 from libecalc.presentation.yaml.yaml_types.facility_model.yaml_facility_model import YamlFacilityModel
 from libecalc.presentation.yaml.yaml_types.fuel_type.yaml_fuel_type import YamlFuelType
@@ -23,7 +23,6 @@
     YamlProcessSimulation,
     YamlPumpProcessSimulation,
 )
-from libecalc.presentation.yaml.yaml_types.process.yaml_process_units import YamlProcessUnitDefinition
 from libecalc.presentation.yaml.yaml_types.streams.yaml_inlet_stream import YamlInletStream
 from libecalc.presentation.yaml.yaml_types.time_series.yaml_time_series import (
     YamlTimeSeriesCollection,
@@ -90,7 +89,7 @@ def inlet_streams(self) -> dict[str, YamlInletStream]:
 
     @property
     @abc.abstractmethod
-    def process_units(self) -> dict[str, YamlProcessUnitDefinition]:
+    def definitions(self) -> YamlDefinitions:
         pass
 
     @property

src/libecalc/presentation/yaml/yaml_reference_service.py

@@ -37,7 +37,7 @@
 
 YamlModel = YamlConsumerModel | YamlFacilityModel
 
-ReferenceType = (
+type ReferenceType = (
     YamlModel
     | YamlFuelType
     | YamlInletStream
@@ -77,7 +77,7 @@ def _model_parsing_order(model: YamlModel) -> int:
         raise EcalcError(title="Invalid model", message=msg) from e
 
 
-Reference = str
+type Reference = str
 
 
 class YamlReferenceService(ReferenceService):
@@ -111,10 +111,8 @@ def __init__(
             references[stream.name] = stream
             reference_yaml_context[stream.name] = stream_path
 
-        process_units_path = YamlPath(keys=("PROCESS_UNITS",))
-        for process_unit_key, process_unit in configuration.process_units.items():
-            # TODO: We have both a name attribute (on compressor only) and they key in the dict
-            # It is unnecessary to have both atm, so just using key for now
+        process_units_path = YamlPath(keys=("DEFINITIONS", "PROCESS_UNITS"))
+        for process_unit_key, process_unit in configuration.definitions.process_units.items():
             process_unit_path = process_units_path.append(process_unit_key)
             references[process_unit_key] = process_unit
             reference_yaml_context[process_unit_key] = process_unit_path

src/libecalc/presentation/yaml/yaml_types/components/yaml_asset.py

@@ -22,13 +22,32 @@
 from libecalc.presentation.yaml.yaml_validation_context import YamlModelValidationContextNames
 
 
+class YamlDefinitions(YamlBase):
+    """Definitions section of an eCalc™ yaml file, containing reusable definitions."""
+
+    model_config = ConfigDict(
+        title="Definitions",
+    )
+
+    process_units: dict[str, YamlProcessUnitDefinition] = Field(
+        default_factory=dict,
+        title="PROCESS_UNITS",
+        description="Defines process units used in PROCESS_PIPELINES.",
+    )
+
+
 class YamlAsset(YamlBase):
     """An eCalc™ yaml file"""
 
     model_config = ConfigDict(
         title="Asset",
     )
 
+    definitions: YamlDefinitions = Field(
+        default_factory=YamlDefinitions,
+        title="DEFINITIONS",
+        description="Contains reusable definitions such as process units.",
+    )
     time_series: list[YamlTimeSeriesCollection] = Field(
         default_factory=list,
         title="TIME_SERIES",
@@ -69,11 +88,6 @@ class YamlAsset(YamlBase):
         description="Defines variables used in an energy usage model by means of expressions or constants."
         "\n\n$ECALC_DOCS_KEYWORDS_URL/VARIABLES",
     )
-    process_units: dict[str, YamlProcessUnitDefinition] = Field(
-        default_factory=dict,
-        title="PROCESS_UNITS",
-        description="Defines process units used in PROCESS_PIPELINES.",
-    )
     process_pipelines: dict[str, YamlProcessPipeline] = Field(
         default_factory=dict,
         title="PROCESS_PIPELINES",
@@ -188,8 +202,8 @@ def validate_unique_references(self):
         if self.process_pipelines is not None:
             references.extend(self.process_pipelines.keys())
 
-        if self.process_units is not None:
-            references.extend(self.process_units.keys())
+        if self.definitions and self.definitions.process_units is not None:
+            references.extend(self.definitions.process_units.keys())
 
         if self.process_simulations is not None:
             for process_simulation in self.process_simulations: