fix: migrate to pydantic v2.0

Fixes #33

psdm/base.py

@@ -5,10 +5,14 @@
 from __future__ import annotations
 
 import enum
+import json
 import pathlib
+import typing as t
 
 import pydantic
 
+T = t.TypeVar("T")
+
 
 class VoltageSystemType(enum.Enum):
     AC = "AC"
@@ -21,20 +25,25 @@ class CosphiDir(enum.Enum):
 
 
 class Base(pydantic.BaseModel):
-    class Config:
-        frozen = True
-        use_enum_values = True
+    model_config = {
+        "frozen": True,
+        "use_enum_values": True,
+    }
 
     @classmethod
     def from_file(cls, file_path: str | pathlib.Path) -> Base:
-        return cls.parse_file(file_path)
+        file_path = pathlib.Path(file_path)
+        with file_path.open("r", encoding="utf-8") as file_handle:
+            return cls.model_validate_json(file_handle.read())
 
     def to_json(self, file_path: str | pathlib.Path, indent: int = 2) -> None:
         file_path = pathlib.Path(file_path)
         file_path.parent.mkdir(parents=True, exist_ok=True)
         with file_path.open("w+", encoding="utf-8") as file_handle:
-            file_handle.write(self.json(indent=indent, sort_keys=True))
+            _json_data = self.model_dump_json()
+            _data = json.loads(_json_data)
+            json.dump(_data, file_handle, indent=indent, sort_keys=True)
 
     @classmethod
     def from_json(cls, json_str: str) -> Base:
-        return cls.parse_raw(json_str)
+        return cls.model_validate_json(json_str)

psdm/meta.py

@@ -5,6 +5,7 @@
 from __future__ import annotations
 
 import datetime
+import typing as t
 import uuid
 
 import pydantic
@@ -15,7 +16,7 @@
 
 
 class Meta(Base):
-    version = VERSION
+    version: t.ClassVar[str] = VERSION
     name: str
     date: datetime.date  # date of export
     id: uuid.UUID = pydantic.Field(default_factory=uuid.uuid4)  # noqa: A003

psdm/steadystate_case/active_power.py

@@ -7,22 +7,22 @@
 
 import pydantic
 
-from psdm.base import Base
+from psdm.topology.load import PowerBase
 from psdm.topology.load import validate_symmetry
 from psdm.topology.load import validate_total
 
 
-class ActivePower(Base):
+class ActivePower(PowerBase):
     value: float  # actual active power (three-phase)
     value_a: float  # actual active power (phase a)
     value_b: float  # actual active power (phase b)
     value_c: float  # actual active power (phase c)
     is_symmetrical: bool
 
-    @pydantic.root_validator(skip_on_failure=True)
-    def _validate_symmetry(cls, values: dict[str, float]) -> dict[str, float]:
-        return validate_symmetry(values)
+    @pydantic.model_validator(mode="after")  # type: ignore[arg-type]
+    def _validate_symmetry(cls, power: ActivePower) -> ActivePower:
+        return validate_symmetry(power)
 
-    @pydantic.root_validator(skip_on_failure=True)
-    def _validate_total(cls, values: dict[str, float]) -> dict[str, float]:
-        return validate_total(values)
+    @pydantic.model_validator(mode="after")  # type: ignore[arg-type]
+    def _validate_total(cls, power: ActivePower) -> ActivePower:
+        return validate_total(power)

psdm/steadystate_case/case.py

@@ -6,7 +6,6 @@
 
 from typing import TYPE_CHECKING
 
-import pydantic
 from loguru import logger
 
 from psdm.base import Base
@@ -21,9 +20,9 @@
 
 class Case(Base):
     meta: Meta
-    loads: pydantic.conlist(Load, unique_items=True)  # type: ignore[valid-type]
-    transformers: pydantic.conlist(Transformer, unique_items=True)  # type: ignore[valid-type]
-    external_grids: pydantic.conlist(ExternalGrid, unique_items=True)  # type: ignore[valid-type]
+    loads: list[Load]
+    transformers: list[Transformer]
+    external_grids: list[ExternalGrid]
 
     def is_valid_topology(self, topology: Topology) -> bool:
         logger.info("Verifying steadystate case ...")

psdm/steadystate_case/controller.py

@@ -6,7 +6,7 @@
 from __future__ import annotations
 
 import enum
-from typing import Annotated
+import typing as t
 
 import pydantic
 
@@ -37,38 +37,42 @@ class ControlledVoltageRef(enum.Enum):
     CA = "CA"
 
 
-class ControlQConst(Base):
+class ControlType(Base):
+    control_strategy: t.ClassVar[ControlStrategy]
+
+
+class ControlQConst(ControlType):
     # q-setpoint control mode
     q_set: float  # Setpoint of reactive power. Counted demand based.
 
-    control_strategy = ControlStrategy.Q_CONST
+    control_strategy: t.ClassVar[ControlStrategy] = ControlStrategy.Q_CONST
 
 
-class ControlUConst(Base):
+class ControlUConst(ControlType):
     # u-setpoint control mode
     u_set: float = pydantic.Field(ge=0)  # Setpoint of voltage.
     u_meas_ref: ControlledVoltageRef = ControlledVoltageRef.POS_SEQ  # voltage reference
 
-    control_strategy = ControlStrategy.U_CONST
+    control_strategy: t.ClassVar[ControlStrategy] = ControlStrategy.U_CONST
 
 
-class ControlTanphiConst(Base):
+class ControlTanphiConst(ControlType):
     # cos(phi) control mode
     cosphi_dir: CosphiDir
     cosphi: float = pydantic.Field(ge=0, le=1)  # cos(phi) for calculation of Q in relation to P.
 
-    control_strategy = ControlStrategy.TANPHI_CONST
+    control_strategy: t.ClassVar[ControlStrategy] = ControlStrategy.TANPHI_CONST
 
 
-class ControlCosphiConst(Base):
+class ControlCosphiConst(ControlType):
     # cos(phi) control mode
     cosphi_dir: CosphiDir
     cosphi: float = pydantic.Field(ge=0, le=1)  # cos(phi) for calculation of Q in relation to P.
 
-    control_strategy = ControlStrategy.COSPHI_CONST
+    control_strategy: t.ClassVar[ControlStrategy] = ControlStrategy.COSPHI_CONST
 
 
-class ControlCosphiP(Base):
+class ControlCosphiP(ControlType):
     # cos(phi(P)) control mode
     cosphi_ue: float = pydantic.Field(
         ge=0,
@@ -81,10 +85,10 @@ class ControlCosphiP(Base):
     p_threshold_ue: float = pydantic.Field(le=0)  # under excited: threshold for P.
     p_threshold_oe: float = pydantic.Field(le=0)  # over excited: threshold for P.
 
-    control_strategy = ControlStrategy.COSPHI_P
+    control_strategy: t.ClassVar[ControlStrategy] = ControlStrategy.COSPHI_P
 
 
-class ControlCosphiU(Base):
+class ControlCosphiU(ControlType):
     # cos(phi(U)) control mode
     cosphi_ue: float = pydantic.Field(
         ...,
@@ -99,10 +103,10 @@ class ControlCosphiU(Base):
     u_threshold_ue: float = pydantic.Field(..., ge=0)  # under excited: threshold for U.
     u_threshold_oe: float = pydantic.Field(..., ge=0)  # over excited: threshold for U.
 
-    control_strategy = ControlStrategy.COSPHI_U
+    control_strategy: t.ClassVar[ControlStrategy] = ControlStrategy.COSPHI_U
 
 
-class ControlQU(Base):
+class ControlQU(ControlType):
     # Q(U) characteristic control mode
     m_tg_2015: float = pydantic.Field(
         ...,
@@ -124,7 +128,7 @@ class ControlQU(Base):
     q_max_ue: float = pydantic.Field(..., ge=0)  # Under excited limit of Q: absolut value
     q_max_oe: float = pydantic.Field(..., ge=0)  # Over excited limit of Q: absolut value
 
-    control_strategy = ControlStrategy.Q_U
+    control_strategy: t.ClassVar[ControlStrategy] = ControlStrategy.Q_U
 
 
 def validate_pos(value: float | None) -> float | None:
@@ -134,32 +138,24 @@ def validate_pos(value: float | None) -> float | None:
     return value
 
 
-class ControlQP(Base):
+class ControlQP(ControlType):
     # Q(P) characteristic control mode
     q_p_characteristic_name: str
     q_max_ue: float | None = None  # Under excited limit of Q: absolut value
     q_max_oe: float | None = None  # Over excited limit of Q: absolut value
 
-    control_strategy = ControlStrategy.Q_P
-
-    validate_q_max_ue = pydantic.validator("q_max_ue", allow_reuse=True)(validate_pos)
-    validate_q_max_oe = pydantic.validator("q_max_oe", allow_reuse=True)(validate_pos)
+    control_strategy: t.ClassVar[ControlStrategy] = ControlStrategy.Q_P
 
+    @pydantic.field_validator("q_max_ue", mode="before")
+    def validate_q_max_ue(cls, v: float | None) -> float | None:
+        return validate_pos(v)
 
-ControlType = Annotated[
-    ControlQConst
-    | ControlUConst
-    | ControlTanphiConst
-    | ControlCosphiConst
-    | ControlCosphiP
-    | ControlCosphiU
-    | ControlQU
-    | ControlQP,
-    pydantic.Field(discriminator="control_strategy"),
-]
+    @pydantic.field_validator("q_max_oe", mode="before")
+    def validate_q_max_oe(cls, v: float | None) -> float | None:
+        return validate_pos(v)
 
 
-class Controller(Base):
+class Controller(ControlType):
     node_target: str  # the controlled node (which can be differ from node the load is connected to)
     control_type: ControlType | None = None
     external_controller_name: str | None = None  # if external controller is specified --> name

psdm/steadystate_case/reactive_power.py

@@ -6,24 +6,19 @@
 
 import pydantic
 
-from psdm.base import Base
 from psdm.steadystate_case.controller import Controller
+from psdm.topology.load import PowerBase
 from psdm.topology.load import validate_symmetry
 from psdm.topology.load import validate_total
 
 
-class ReactivePower(Base):
-    value: float  # actual reactive power (three-phase)
-    value_a: float  # actual reactive power (phase a)
-    value_b: float  # actual reactive power (phase b)
-    value_c: float  # actual reactive power (phase c)
-    is_symmetrical: bool
+class ReactivePower(PowerBase):
     controller: Controller | None = None
 
-    @pydantic.root_validator(skip_on_failure=True)
-    def _validate_symmetry(cls, values: dict[str, float]) -> dict[str, float]:
-        return validate_symmetry(values)
+    @pydantic.model_validator(mode="after")  # type: ignore[arg-type]
+    def _validate_symmetry(cls, power: ReactivePower) -> ReactivePower:
+        return validate_symmetry(power)
 
-    @pydantic.root_validator(skip_on_failure=True)
-    def _validate_total(cls, values: dict[str, float]) -> dict[str, float]:
-        return validate_total(values)
+    @pydantic.model_validator(mode="after")  # type: ignore[arg-type]
+    def _validate_total(cls, power: ReactivePower) -> ReactivePower:
+        return validate_total(power)

psdm/topology/branch.py

@@ -21,7 +21,7 @@ class Branch(Base):
     node_2: str
     name: str
     u_n: float  # nominal voltage of the branch connected nodes
-    i_r: float  # rated current of branch (thermal limit in continuous operation)
+    i_r: float | None  # rated current of branch (thermal limit in continuous operation)
     r1: float  # positive sequence values of PI-representation
     x1: float  # positive sequence values of PI-representation
     g1: float  # positive sequence values of PI-representation

psdm/topology/load.py

@@ -5,7 +5,7 @@
 from __future__ import annotations
 
 import enum
-from collections.abc import Sequence
+import typing as t
 
 import pydantic
 
@@ -78,31 +78,42 @@ class PowerType(enum.Enum):
 THRESHOLD = 0.51  # acceptable rounding error (0.5 W) + epsilon for calculation accuracy (0.01 W)
 
 
-def validate_total(values: dict[str, float]) -> dict[str, float]:
-    pow_total = values["value_a"] + values["value_b"] + values["value_c"]
-    diff = abs(values["value"] - pow_total)
+class PowerBase(Base):
+    value: float
+    value_a: float
+    value_b: float
+    value_c: float
+    is_symmetrical: bool
+
+
+T = t.TypeVar("T", bound=PowerBase)
+
+
+def validate_total(power: T) -> T:
+    pow_total = power.value_a + power.value_b + power.value_c
+    diff = abs(power.value - pow_total)
     if diff > THRESHOLD:
-        msg = f"Power mismatch: Total power should be {pow_total}, is {values['value']}."
+        msg = f"Power mismatch: Total power should be {pow_total}, is {power.value}."
         raise ValueError(msg)
 
-    return values
+    return power
 
 
-def validate_symmetry(values: dict[str, float]) -> dict[str, float]:
-    if values["value"] != 0:
-        if values["is_symmetrical"]:
-            if not (values["value_a"] == values["value_b"] == values["value_c"]):
+def validate_symmetry(power: T) -> T:
+    if power.value != 0:
+        if power.is_symmetrical:
+            if not (power.value_a == power.value_b == power.value_c):
                 msg = "Power mismatch: Three-phase power of load is not symmetrical."
                 raise ValueError(msg)
 
-        elif values["value_a"] == values["value_b"] == values["value_c"]:
+        elif power.value_a == power.value_b == power.value_c:
             msg = "Power mismatch: Three-phase power of load is symmetrical."
             raise ValueError(msg)
 
-    return values
+    return power
 
 
-class RatedPower(Base):
+class RatedPower(PowerBase):
     value: float = pydantic.Field(..., ge=0)  # rated power; base for p.u. calculation
     value_a: float = pydantic.Field(..., ge=0)  # rated power (phase a)
     value_b: float = pydantic.Field(..., ge=0)  # rated power (phase b)
@@ -114,19 +125,24 @@ class RatedPower(Base):
     power_type: PowerType
     is_symmetrical: bool
 
-    @pydantic.root_validator(skip_on_failure=True)
-    def _validate_symmetry(cls, values: dict[str, float]) -> dict[str, float]:
-        return validate_symmetry(values)
+    @pydantic.model_validator(mode="after")  # type: ignore[arg-type]
+    def _validate_symmetry(cls, power: RatedPower) -> RatedPower:
+        return validate_symmetry(power)
 
-    @pydantic.root_validator(skip_on_failure=True)
-    def _validate_total(cls, values: dict[str, float]) -> dict[str, float]:
-        return validate_total(values)
+    @pydantic.model_validator(mode="after")  # type: ignore[arg-type]
+    def _validate_total(cls, power: RatedPower) -> RatedPower:
+        return validate_total(power)
 
 
 class ConnectedPhases(Base):
-    phases_a: Sequence[Phase]
-    phases_b: Sequence[Phase]
-    phases_c: Sequence[Phase]
+    phases_a: tuple[Phase, Phase] | None
+    phases_b: tuple[Phase, Phase] | None
+    phases_c: tuple[Phase, Phase] | None
+
+    @pydantic.computed_field  # type: ignore[misc]
+    @property
+    def n_connected_phases(self) -> int:
+        return sum([getattr(self, f"phases_{idx}") is not None for idx in ["a", "b", "c"]])
 
 
 class Load(Base):  # including assets of type load and generator