Source code for ogstools.materiallib.core.phase
# Copyright (c) 2012-2025, OpenGeoSys Community (http://www.opengeosys.org)
#            Distributed under a Modified BSD License.
#            See accompanying file LICENSE.txt or
#            http://www.opengeosys.org/project/license
#
import logging
from typing import Any
from ogstools.materiallib.schema.process_schema import PROCESS_SCHEMAS
from .component import Component
from .components import Components
from .material import Material
from .property import MaterialProperty
logger = logging.getLogger(__name__)
[docs]
class Phase:
[docs]
    def __init__(
        self,
        phase_type: str,
        gas_material: Material | None = None,
        liquid_material: Material | None = None,
        solid_material: Material | None = None,
        process: str = "",
    ):
        self.type = phase_type
        self.process = process
        self.gas_material = gas_material
        self.liquid_material = liquid_material
        self.solid_material = solid_material
        schema = PROCESS_SCHEMAS.get(process)
        if not schema:
            msg = f"No process schema found for '{process}'."
            raise ValueError(msg)
        self.schema: dict[str, Any] = schema
        # Check for material consistency
        match phase_type:
            case "AqueousLiquid":
                if liquid_material is None:
                    raise ValueError(
                        ("AqueousLiquid requires liquid_material.",)[0]
                    )
                if solid_material is not None:
                    raise ValueError(
                        ("AqueousLiquid must not have solid_material.",)[0]
                    )
            case "Gas":
                if gas_material is None:
                    raise ValueError(("Gas requires gas_material.",)[0])
                if solid_material is not None:
                    raise ValueError(("Gas must not have solid_material.",)[0])
            case "Solid":
                if solid_material is None:
                    raise ValueError(("Solid requires solid_material.",)[0])
                if gas_material is not None or liquid_material is not None:
                    raise ValueError(
                        (
                            "Solid must not have gas_material or liquid_material.",
                        )[0]
                    )
        self.properties: list[MaterialProperty] = []
        self.components: list[Component] = []
        if not self.schema:
            msg = f"No process schema found for '{process}'."
            raise ValueError(msg)
        self._load_phase_properties()
        if (
            gas_material is not None
            and liquid_material is not None
            and any(
                "components" in p
                for p in self.schema.get("phases", [])
                if p.get("type") == self.type
            )
        ):
            self._load_components(gas_material, liquid_material) 
    def _load_phase_properties(self) -> None:
        logger.debug("Loading properties for phase type: %s", self.type)
        assert self.schema is not None
        phase_def = next(
            (
                p
                for p in self.schema.get("phases", [])
                if p.get("type") == self.type
            ),
            None,
        )
        if phase_def is None:
            msg = f"No phase definition found for type '{self.type}'"
            raise ValueError(msg)
        logger.debug("Found phase definition for %s", self.type)
        required = set(phase_def.get("properties", []))
        logger.debug("Required properties: %s", required)
        source = {
            "AqueousLiquid": self.liquid_material,
            "Gas": self.gas_material,
            "Solid": self.solid_material,
        }.get(self.type)
        if source is None:
            msg = f"Don't know how to load properties for phase type '{self.type}'"
            raise ValueError(msg)
        logger.debug("Source material: %s", source.name)
        self.properties = [
            prop
            for prop in source.properties
            if prop.name in required
            and (
                prop.extra.get("scope") == "phase" or "scope" not in prop.extra
            )
        ]
        loaded = {prop.name for prop in self.properties}
        missing = required - loaded
        if missing:
            msg = f"Missing required properties for phase type '{self.type}', material '{source.name}': {missing}"
            raise ValueError(msg)
        logger.debug(
            "Loaded %s properties for phase type '%s'",
            len(self.properties),
            self.type,
        )
        logger.debug(self.properties)
    def _load_components(
        self,
        gas_material: Material,
        liquid_material: Material,
        Diffusion_coefficient: float | None = None,
    ) -> None:
        self.components_obj = Components(
            phase_type=self.type,
            gas_component=gas_material,
            liquid_component=liquid_material,
            process=self.process,
            Diffusion_coefficient=Diffusion_coefficient,
        )
        self.components = [
            self.components_obj.gas_component_obj,
            self.components_obj.liquid_component_obj,
        ]
[docs]
    def add_property(self, prop: MaterialProperty) -> None:
        self.properties.append(prop) 
[docs]
    def add_component(self, component: Component) -> None:
        self.components.append(component) 
[docs]
    def validate(self) -> bool:
        self._validate_phase_exists()
        self._validate_required_properties()
        self._validate_extra_properties()
        self._validate_components()
        return True 
    def _validate_phase_exists(self) -> None:
        if not any(p["type"] == self.type for p in self.schema["phases"]):
            msg = f"Phase '{self.type}' is not defined for this process."
            raise ValueError(msg)
    def _validate_required_properties(self) -> None:
        required = self._required_properties()
        found = self._found_property_names()
        missing = [p for p in required if p not in found]
        if missing:
            msg = (
                f"Phase '{self.type}' is missing required properties: {missing}"
            )
            raise ValueError(msg)
    def _validate_extra_properties(self) -> None:
        required = self._required_properties()
        found = self._found_property_names()
        extra = [p for p in found if p not in required]
        if extra:
            msg = f"Phase '{self.type}' has unknown/unsupported properties: {extra}"
            raise ValueError(msg)
    def _validate_components(self) -> None:
        for component in self.components:
            if not component.validate():
                msg = f"Component '{component.name}' in phase '{self.type}' is invalid."
                raise ValueError(msg)
    def _required_properties(self) -> list[str]:
        for p in self.schema["phases"]:
            if p["type"] == self.type:
                return p.get("properties", [])
        return []
    def _found_property_names(self) -> list[str]:
        return [p.name for p in self.properties]
    # -----------------------
    # Representation
    # -----------------------
    def __repr__(self) -> str:
        lines = [f"<Phase '{self.type}'>"]
        if self.properties:
            lines.append(f"  ├─ {len(self.properties)} properties:")
            for prop in self.properties:
                lines.append("  │   " + repr(prop))
        else:
            lines.append("  ├─ no properties")
        if self.components:
            lines.append(f"  ├─ {len(self.components)} component groups:")
            for comp in self.components:
                for line in repr(comp).splitlines():
                    lines.append("  │   " + line)
        else:
            lines.append("  └─ no components")
        return "\n".join(lines)