pyyaml_yaml_model.py

import datetime
import re
from collections.abc import Iterable, Iterator, Sequence
from copy import deepcopy
from pathlib import Path
from typing import Any, Self, TextIO

import yaml
from pydantic import TypeAdapter
from pydantic import ValidationError as PydanticValidationError
from yaml import SafeLoader

from libecalc.common.errors.exceptions import ProgrammingError
from libecalc.common.time_utils import convert_date_to_datetime
from libecalc.dto.utils.validators import COMPONENT_NAME_ALLOWED_CHARS, COMPONENT_NAME_PATTERN
from libecalc.presentation.yaml.file_context import FileMark
from libecalc.presentation.yaml.mappers.yaml_path import YamlPath
from libecalc.presentation.yaml.model_validation_exception import ModelValidationException
from libecalc.presentation.yaml.validation_errors import (
    Location,
    ModelValidationError,
    custom_errors,
)
from libecalc.presentation.yaml.yaml_entities import ResourceStream
from libecalc.presentation.yaml.yaml_keywords import EcalcYamlKeywords
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_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
from libecalc.presentation.yaml.yaml_types.models import YamlConsumerModel, YamlFluidModel
from libecalc.presentation.yaml.yaml_types.process.yaml_process_pipeline import YamlProcessPipeline
from libecalc.presentation.yaml.yaml_types.process.yaml_process_simulation import (
    YamlEcalcEvent,
    YamlProcessEvent,
    YamlProcessSimulation,
    YamlPumpProcessSimulation,
)
from libecalc.presentation.yaml.yaml_types.process.yaml_process_units import YamlProcessUnit
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
from libecalc.presentation.yaml.yaml_types.yaml_default_datetime import YamlDefaultDatetime
from libecalc.presentation.yaml.yaml_types.yaml_variable import YamlVariable, YamlVariableReferenceId, YamlVariables
from libecalc.presentation.yaml.yaml_validation_context import YamlModelValidationContext

# Top-level YAML keywords used only by experimental/new sections
_PROCESS_UNITS_KEY = "PROCESS_UNITS"
_PROCESS_PIPELINES_KEY = "PROCESS_PIPELINES"
_INLET_STREAMS_KEY = "INLET_STREAMS"
_FLUID_MODELS_KEY = "FLUID_MODELS"
_PROCESS_SIMULATIONS_KEY = "PROCESS_SIMULATIONS"
_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)


def datetime_parser(key: str) -> datetime.datetime:
    return dt_adapter.validate_python(key)


class PyYamlYamlModel(YamlValidator, YamlConfiguration):
    """Implementation of yaml model using PyYaml library
    Keeping comments and horizontal lists on loading currently not supported!
    """

    @classmethod
    def get_validator(cls, *args, **kwargs) -> "YamlValidator":
        return cls.read(*args, **kwargs)

    def __init__(self, internal_datamodel: dict[str, Any], name: str, instantiated_through_read: bool = False):
        """To avoid mistakes, make sure that this is only instantiated through read method/named constructor
        :param instantiated_through_read: set to True to allow to use constructor.
        """
        if not instantiated_through_read:
            raise ProgrammingError(f"{self.__class__} can only be instantiated through read() method/named constructor")

        super().__init__(internal_datamodel=internal_datamodel, name=name)

    def dump(self) -> str:
        if self._internal_datamodel is None:
            raise ProgrammingError("You cannot dump a model without first reading one. Use read() to read a model.")

        return PyYamlYamlModel.dump_yaml(self._internal_datamodel)

    @classmethod
    def read(
        cls,
        main_yaml: ResourceStream,
        base_dir: Path | None = None,
        resources: dict[str, TextIO] | None = None,
        enable_include: bool = True,
    ) -> "PyYamlYamlModel":
        internal_datamodel = PyYamlYamlModel.read_yaml(
            main_yaml=main_yaml, resources=resources, base_dir=base_dir, enable_include=enable_include
        )
        self = cls(internal_datamodel=internal_datamodel, name=main_yaml.name, instantiated_through_read=True)
        return self

    class SafeLineLoader(SafeLoader):
        def construct_yaml_map(self, node):
            (obj,) = super().construct_yaml_map(node)
            return YamlDict(obj, start_mark=node.start_mark, end_mark=node.end_mark)

        def construct_yaml_seq(self, node):
            (obj,) = super().construct_yaml_seq(node)
            return YamlList(obj, start_mark=node.start_mark, end_mark=node.end_mark)

    SafeLineLoader.add_constructor("tag:yaml.org,2002:map", SafeLineLoader.construct_yaml_map)

    SafeLineLoader.add_constructor("tag:yaml.org,2002:seq", SafeLineLoader.construct_yaml_seq)

    class IncludeConstructor:
        """Add !include constructor to the yaml reader.
        This will include a separate yaml file into the position where the include keyword is placed.

        Example:
            SOME_KEY: !include some.yaml
        """

        def __init__(self, base_dir: Path | None = None, resources: dict[str, TextIO] | None = None):
            self._base_dir = base_dir
            self._resources = resources if resources else {}

        def __call__(self, loader: yaml.SafeLoader, node: yaml.ScalarNode):
            resource_name = str(loader.construct_scalar(node=node))
            if self._resources:
                yaml_resource = ResourceStream(stream=self._resources[resource_name], name=resource_name)
                yaml_data = PyYamlYamlModel._read_yaml_helper(
                    yaml_file=yaml_resource, resources=self._resources, loader=loader.__class__, enable_include=True
                )
            elif self._base_dir:
                resource_path = self._base_dir / str(resource_name)
                with open(resource_path) as resource_file:
                    yaml_resource = ResourceStream(name=resource_path.name, stream=resource_file)
                    yaml_data = PyYamlYamlModel._read_yaml_helper(
                        yaml_file=yaml_resource, loader=loader.__class__, enable_include=True, base_dir=self._base_dir
                    )
            else:
                raise ValueError(
                    f"Could not find the !include resource: {resource_name} in either attached resources nor bas_dir."
                )
            return yaml_data

    class IndentationDumper(yaml.Dumper):
        """In order to increase indentation of nested elements."""

        def increase_indent(self, flow=False, indentless=False):
            return super().increase_indent(flow, False)

    class YamlReader:
        def __init__(
            self,
            loader: type[yaml.SafeLoader],
            enable_include: bool = True,
            base_dir: Path | None = None,
            resources: dict[str, TextIO] | None = None,
        ):
            self.__loader = loader
            if enable_include and (base_dir or resources):
                loader.add_constructor(
                    "!include",
                    PyYamlYamlModel.IncludeConstructor(base_dir=base_dir, resources=resources),
                )

        def load(self, yaml_file: ResourceStream):
            class UniqueKeyLoader(self.__loader):  # type: ignore[name-defined]
                def construct_mapping(self, node, deep=False):
                    mapping = set()
                    for key_node, _ in node.value:
                        each_key = self.construct_object(key_node, deep=deep)
                        if each_key in mapping:
                            raise DuplicateKeyError(
                                key=each_key,
                                file_context=FileContext(
                                    name=yaml_file.name,
                                    start=FileMark(
                                        line_number=key_node.start_mark.line + 1,
                                        column=key_node.start_mark.column + 1,
                                    ),
                                ),
                            )
                        mapping.add(each_key)
                    return super().construct_mapping(node, deep)

            if re.search(COMPONENT_NAME_PATTERN, Path(yaml_file.name).stem) is None:
                raise YamlError(
                    problem=f"The model file, {yaml_file.name}, contains illegal special characters. "
                    f"Allowed characters are {COMPONENT_NAME_ALLOWED_CHARS}",
                )
            try:
                return yaml.load(yaml_file, Loader=UniqueKeyLoader)  # noqa: S506 - loader should be SafeLoader
            except KeyError as e:
                raise YamlError(problem=f"Error occurred while loading yaml file, key {e} not found") from e

        def dump_and_load(self, yaml_file: ResourceStream):
            return yaml.dump(self.load(yaml_file), Dumper=PyYamlYamlModel.IndentationDumper, sort_keys=False)

    @staticmethod
    def _read_yaml_helper(
        yaml_file: ResourceStream,
        loader: type[yaml.SafeLoader],
        enable_include: bool = True,
        base_dir: Path | None = None,
        resources: dict[str, TextIO] | None = None,
    ):
        """Read yaml helper for include functionality."""
        yaml_reader = PyYamlYamlModel.YamlReader(
            loader=loader, enable_include=enable_include, base_dir=base_dir, resources=resources
        )
        return yaml_reader.load(yaml_file)

    @staticmethod
    def dump_and_load_yaml(
        main_yaml: ResourceStream,
        enable_include: bool = True,
        base_dir: Path | None = None,
        resources: dict[str, TextIO] | None = None,
    ) -> str:
        yaml_reader = PyYamlYamlModel.YamlReader(
            loader=SafeLoader, enable_include=enable_include, base_dir=base_dir, resources=resources
        )
        return yaml_reader.dump_and_load(main_yaml)

    @staticmethod
    def dump_yaml(yaml_dict: dict) -> str:
        return yaml.dump(yaml_dict, Dumper=PyYamlYamlModel.IndentationDumper, sort_keys=False)

    @staticmethod
    def read_yaml(
        main_yaml: ResourceStream,
        enable_include: bool = True,
        base_dir: Path | None = None,
        resources: dict[str, TextIO] | None = None,
    ) -> YamlDict:
        try:
            read_yaml = PyYamlYamlModel._read_yaml_helper(
                yaml_file=main_yaml,
                loader=PyYamlYamlModel.SafeLineLoader,
                enable_include=enable_include,
                base_dir=base_dir,
                resources=resources,
            )
            if read_yaml is None:
                raise yaml.YAMLError("YAML is empty")
            if not isinstance(read_yaml, dict):
                raise yaml.YAMLError("Not a valid YAML object")
            return read_yaml
        except yaml.YAMLError as e:
            file_context = None
            if hasattr(e, "problem_mark"):
                mark = e.problem_mark
                if mark is not None:
                    file_context = FileContext(
                        name=main_yaml.name,
                        start=FileMark(
                            line_number=mark.line + 1,
                            column=mark.column + 1,
                        ),
                    )

            problem = "Invalid YAML file"
            if hasattr(e, "problem"):
                optional_problem = e.problem
                if optional_problem is not None:
                    problem = optional_problem

            raise YamlError(
                problem=problem,
                file_context=file_context,
            ) from e

    # start of validation/parsing methods
    @property
    def name(self):
        return self._name

    def _get_yaml_data_or_default(self, keyword, factory):
        """
        Function used to get data when we don't want validation to fail, only get the data if available.

        Args:
            keyword: keyword to get from the yaml
            factory: builtin type that should work as a factory and for typechecking

        Returns: data for keyword if available, else default created by factory

        """
        default = factory()
        if not isinstance(self._internal_datamodel, dict):
            return default

        data = self._internal_datamodel.get(keyword, default)

        if data is None or not isinstance(data, factory):
            return default

        return data

    def _get_yaml_list_or_empty(self, keyword: str) -> list:
        return self._get_yaml_data_or_default(keyword, list)

    def _get_yaml_dict_or_empty(self, keyword: str) -> dict:
        return self._get_yaml_data_or_default(keyword, dict)

    @property
    def facility_resource_names(self) -> list[str]:
        facility_input_data = self._get_yaml_list_or_empty(EcalcYamlKeywords.facility_inputs)
        model_curves_data = [
            model.get(model_curves)
            for model in self._get_yaml_list_or_empty(EcalcYamlKeywords.models)
            for model_curves in [EcalcYamlKeywords.consumer_chart_curves, EcalcYamlKeywords.consumer_chart_curve]
            if isinstance(model.get(model_curves), dict)
        ]
        resource_data = facility_input_data + model_curves_data
        resource_names: 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:
            resource_names.extend(_find_file_references(self._internal_datamodel.get(section)))

        # Dedup while preserving order — the same CSV may be referenced by multiple charts.
        return list(dict.fromkeys(resource_names))

    @property
    def timeseries_resource_names(self) -> list[str]:
        timeseries_resources = []
        for resource in self._get_yaml_list_or_empty(EcalcYamlKeywords.time_series):
            resource_name = resource.get(EcalcYamlKeywords.file)
            if resource_name is not None:
                timeseries_resources.append(resource_name)
        return timeseries_resources

    @property
    def variables(self) -> YamlVariables:
        """
        Get variables, invalid variable definitions will be skipped.

        Returns: valid variables

        """
        variables = self._get_yaml_dict_or_empty(EcalcYamlKeywords.variables)

        valid_variables: YamlVariables = {}
        for reference, variable in variables.items():
            try:
                reference = TypeAdapter(YamlVariableReferenceId).validate_python(reference)
                variable = TypeAdapter(YamlVariable).validate_python(variable)
                valid_variables[reference] = variable
            except PydanticValidationError:
                continue

        return valid_variables

    @property
    def yaml_variables(self) -> dict[YamlVariableReferenceId, dict]:
        """
        Get the internal data for variables directly.
        Returns:

        """
        return self._get_yaml_dict_or_empty(EcalcYamlKeywords.variables)

    @property
    def facility_inputs(self) -> list[YamlFacilityModel]:
        facility_inputs: list[YamlFacilityModel] = []
        for facility_input in self._get_yaml_list_or_empty(EcalcYamlKeywords.facility_inputs):
            try:
                facility_inputs.append(TypeAdapter(YamlFacilityModel).validate_python(facility_input))
            except PydanticValidationError:
                pass

        return facility_inputs

    @property
    def models(self) -> list[YamlConsumerModel]:
        models: list[YamlConsumerModel] = []
        for model in self._get_yaml_list_or_empty(EcalcYamlKeywords.models):
            try:
                models.append(TypeAdapter(YamlConsumerModel).validate_python(model))
            except PydanticValidationError:
                pass

        return models

    @property
    def time_series(self) -> list[YamlTimeSeriesCollection]:
        """
        Get only valid time series, i.e. don't fail if one is invalid.
        """
        time_series: list[YamlTimeSeriesCollection] = []
        for time_series_data in self._get_yaml_list_or_empty(EcalcYamlKeywords.time_series):
            try:
                time_series.append(TypeAdapter(YamlTimeSeriesCollection).validate_python(time_series_data))
            except PydanticValidationError:
                pass

        return time_series

    @property
    def fuel_types(self):
        fuel_types = []
        for fuel_type in self._get_yaml_list_or_empty(EcalcYamlKeywords.fuel_types):
            try:
                fuel_types.append(TypeAdapter(YamlFuelType).validate_python(fuel_type))
            except PydanticValidationError:
                pass
        return fuel_types

    @property
    def installations(self) -> Iterable[YamlInstallation]:
        installations = []
        for installation in self._get_yaml_list_or_empty(EcalcYamlKeywords.installations):
            try:
                installations.append(TypeAdapter(YamlInstallation).validate_python(installation))
            except PydanticValidationError:
                pass
        return installations

    @property
    def inlet_streams(self) -> dict[str, YamlInletStream]:
        inlet_streams: dict[str, YamlInletStream] = {}
        raw = self._get_yaml_dict_or_empty(_INLET_STREAMS_KEY)
        adapter = TypeAdapter(YamlInletStream)

        for name, stream in raw.items():
            try:
                inlet_streams[name] = adapter.validate_python(stream)
            except PydanticValidationError:
                pass
        return inlet_streams

    @property
    def fluid_models(self) -> dict[str, YamlFluidModel]:
        fluid_models: dict[str, YamlFluidModel] = {}
        raw = self._get_yaml_dict_or_empty(_FLUID_MODELS_KEY)

        for name, fluid_model in raw.items():
            try:
                fluid_models[name] = TypeAdapter(YamlFluidModel).validate_python(fluid_model)
            except PydanticValidationError:
                pass
        return fluid_models

    @property
    def process_units(self) -> dict[str, YamlProcessUnit]:
        process_units: dict[str, YamlProcessUnit] = {}
        raw = self._get_yaml_dict_or_empty(_PROCESS_UNITS_KEY)

        for name, unit_data in raw.items():
            try:
                process_units[name] = TypeAdapter(YamlProcessUnit).validate_python(unit_data)
            except PydanticValidationError:
                pass
        return process_units

    @property
    def process_pipelines(self) -> dict[str, YamlProcessPipeline]:
        process_pipelines: dict[str, YamlProcessPipeline] = {}
        raw = self._get_yaml_dict_or_empty(_PROCESS_PIPELINES_KEY)

        for name, process_pipeline in raw.items():
            try:
                process_pipelines[name] = TypeAdapter(YamlProcessPipeline).validate_python(process_pipeline)
            except PydanticValidationError:
                pass

        return process_pipelines

    @property
    def process_simulations(self) -> list[YamlProcessSimulation]:
        process_simulations: list[YamlProcessSimulation] = []
        adapter = TypeAdapter(YamlProcessSimulation)
        for process_simulation in self._get_yaml_list_or_empty(_PROCESS_SIMULATIONS_KEY):
            try:
                process_simulations.append(adapter.validate_python(process_simulation))
            except PydanticValidationError:
                pass
        return process_simulations

    @property
    def ecalc_events(self) -> list[YamlEcalcEvent]:
        ecalc_events: list[YamlEcalcEvent] = []
        adapter = TypeAdapter(YamlEcalcEvent)
        for ecalc_event in self._get_yaml_list_or_empty(_ECALC_EVENTS_KEY):
            try:
                ecalc_events.append(adapter.validate_python(ecalc_event))
            except PydanticValidationError:
                pass
        return ecalc_events

    @property
    def process_events(self) -> list[YamlProcessEvent]:
        process_events: list[YamlProcessEvent] = []
        adapter = TypeAdapter(YamlProcessEvent)
        for process_event in self._get_yaml_list_or_empty(_PROCESS_EVENTS_KEY):
            try:
                process_events.append(adapter.validate_python(process_event))
            except PydanticValidationError:
                pass
        return process_events

    @property
    def pump_process_simulations(self) -> list[YamlPumpProcessSimulation]:
        pump_process_simulations: list[YamlPumpProcessSimulation] = []
        adapter = TypeAdapter(YamlPumpProcessSimulation)
        for pump_process_simulation in self._get_yaml_list_or_empty(_PUMP_PROCESS_SIMULATIONS_KEY):
            try:
                pump_process_simulations.append(adapter.validate_python(pump_process_simulation))
            except PydanticValidationError:
                pass
        return pump_process_simulations

    @property
    def start(self) -> datetime.datetime | None:
        start_value = self._internal_datamodel.get(EcalcYamlKeywords.start)
        return TypeAdapter(YamlDefaultDatetime).validate_python(start_value) if start_value is not None else None

    @property
    def end(self) -> datetime.datetime | None:
        end_value = self._internal_datamodel.get(EcalcYamlKeywords.end)
        return TypeAdapter(YamlDefaultDatetime).validate_python(end_value) if end_value is not None else None

    @property
    def dates(self) -> set[datetime.datetime]:
        """All dates in the yaml."""
        return set(find_date_keys_in_yaml(self._internal_datamodel))

    def validate(self, context: YamlModelValidationContext) -> Self:  # type: ignore[override]
        try:
            YamlAsset.model_validate(deepcopy(self._internal_datamodel), context=context)
            return self
        except PydanticValidationError as e:
            errors = []
            for error in custom_errors(e):
                yaml_path = self._get_yaml_path_from_pydantic_loc(error["loc"])
                file_context = self._get_closest_file_context(yaml_path.keys)
                errors.append(
                    ModelValidationError(
                        message=error["msg"],
                        location=Location.from_pydantic_loc(yaml_path.keys),
                        file_context=file_context,
                        data=None,
                    )
                )
            raise ModelValidationException(errors=errors) from e

    def _node_to_file_context(self, data: YamlDict) -> FileContext:
        return FileContext(
            name=self.name,
            start=FileMark(
                line_number=data.start_mark.line + 1,
                column=data.start_mark.column,
            ),
            end=FileMark(
                line_number=data.end_mark.line + 1,
                column=data.end_mark.column,
            ),
        )

    def _get_yaml_path_from_pydantic_loc(self, locs: tuple[str | int, ...]) -> YamlPath:
        """
        Filter extra pydantic locations.

        Pydantic adds extra locations for discriminators (specific TYPEs used in a union), tags ('single', 'temporal'
        for temporal model), and other validation functions when those fail.

        Args:
            locs:

        Returns:

        """
        # TODO: filter based on json_schema or pydantic field info definition?
        #   Tried implementing by filtering based on data, but that is at least not a good solution for missing_key error
        yaml_keys = YamlPath(())
        for loc in locs:
            if loc in ["single", "temporal"]:
                continue
            else:
                yaml_keys = yaml_keys.append(loc)

        return yaml_keys

    def _get_closest_file_context(
        self,
        yaml_path: Sequence[str | int | datetime.datetime],
        is_pydantic_loc: bool = False,
    ):
        current_data = self._internal_datamodel
        closest_file_context = self._node_to_file_context(current_data)
        for key in yaml_path:
            did_alter_data = False
            if isinstance(key, str) and isinstance(current_data, dict | YamlDict):
                key = key.upper()
                if key in current_data:
                    current_data = current_data[key]
                    did_alter_data = True
            elif isinstance(key, int) and isinstance(current_data, list | YamlList):
                try:
                    current_data = current_data[key]
                    did_alter_data = True
                except IndexError:
                    pass
            elif isinstance(key, datetime.datetime) and isinstance(current_data, dict | YamlDict):
                # Parse keys to datetime, find the matching value
                try:
                    current_data = next(
                        value for inner_key, value in current_data.items() if datetime_parser(inner_key) == key
                    )
                    did_alter_data = True
                except (PydanticValidationError, StopIteration):
                    # StopIteration if no matching datetime key, if the period has been altered by a late start,
                    #   i.e. model is defined from 2000, while START is 2010, then we will adjust the period for the
                    #   model, which means we won't find key in the yaml data.
                    # PydanticValidationError if unable to parse datetime
                    pass

            if did_alter_data and isinstance(current_data, YamlDict):
                closest_file_context = self._node_to_file_context(current_data)

            if not did_alter_data:
                if is_pydantic_loc:
                    # pydantic loc might contain extra keys such as discriminators (the specific TYPE used as discriminator), and tags ('temporal', 'single' for temporal model)
                    # Therefore, we continue past invalid keys in case we can find a more specific FileContext
                    continue
                else:
                    # Stop if we didn't find new data
                    return closest_file_context

        return closest_file_context

    def get_file_context(self, yaml_path: tuple[str | int | datetime.datetime, ...]) -> FileContext | None:
        return self._get_closest_file_context(yaml_path)


def find_date_keys_in_yaml(yaml_object: list | dict) -> list[datetime.datetime]:
    """The function will add any dates found in the yaml_object to the list named output.

    :param yaml_object: The content (or subset) of a yaml file
    :type yaml_object: Union[List,dict, CommentedMap]
    :return: The list with dates given as input to the function with any dates found in the yaml_object added to it
    :rtype:list[datetime.datetime]
    """

    def common_iterable(obj: list | dict) -> dict | Iterator[int]:
        """Helper function when iteration over something we beforehand don't know
        whether is adict,list or CommentedMap.

        :param obj: A subset of a nested YAML file to iterate over
        :type obj: Union[List,dict]
        :return: The object if the object is a dict, or the indices of the list if the object is a list
        :rtype: Union[Dict, Iterator[int]]
        """
        if isinstance(obj, dict):
            return obj
        else:
            return (idx for idx, value in enumerate(obj))

    output = []
    for index in common_iterable(yaml_object):
        if isinstance(index, datetime.date | datetime.datetime):
            index_to_datetime = convert_date_to_datetime(index)
            if index_to_datetime not in output:
                output.append(index_to_datetime)
        if isinstance(yaml_object[index], dict | list):
            output.extend(find_date_keys_in_yaml(yaml_object[index]))

    return output


def _find_file_references(node: dict | list | None) -> list[str]:
    """Recursively collect all string values under a `FILE` key in a parsed YAML datamodel.

    Structure-agnostic so we don't have to hard-code each section's nesting
    (e.g. CURVES inside COMPRESSOR_MODEL inside COMPRESSOR inside PROCESS_UNITS).
    Only string values are returned; anything else under FILE is ignored.
    """
    references: list[str] = []
    if isinstance(node, dict):
        for key, value in node.items():
            if key == EcalcYamlKeywords.file and isinstance(value, str):
                references.append(value)
            else:
                references.extend(_find_file_references(value))
    elif isinstance(node, list):
        for item in node:
            references.extend(_find_file_references(item))
    return references