Add some input validation

Signed-off-by: Sylvain Leclerc <[email protected]>

src/antares/tsgen/ts_generator.py

@@ -48,6 +48,84 @@ class ThermalCluster:
     po_law: ProbabilityLaw
     po_volatility: float
 
+    def __post_init__(self) -> None:
+        _check_cluster(self)
+
+
+def _check_1_dim(array: npt.NDArray, name: str) -> None:
+    if array.ndim != 1:
+        raise ValueError(f"{name} must be a 1 dimension array.")
+
+
+def _check_array(condition: npt.NDArray[np.bool_], message: str) -> None:
+    if condition.any():
+        raise ValueError(f"{message}: {condition.nonzero()[0].tolist()}")
+
+
+def _check_cluster(cluster: ThermalCluster) -> None:
+    if cluster.unit_count <= 0:
+        raise ValueError(
+            f"Unit count must be strictly positive, got {cluster.unit_count}."
+        )
+    if cluster.nominal_power <= 0:
+        raise ValueError(
+            f"Nominal power must be strictly positive, got {cluster.nominal_power}."
+        )
+    if cluster.fo_volatility < 0:
+        raise ValueError(
+            f"Forced outage volatility must be positive, got {cluster.unit_count}."
+        )
+    if cluster.po_volatility < 0:
+        raise ValueError(
+            f"Planned outage volatility must be positive, got {cluster.unit_count}."
+        )
+
+    _check_1_dim(cluster.fo_rate, "Forced outage failure rate")
+    _check_1_dim(cluster.fo_duration, "Forced outage duration")
+    _check_1_dim(cluster.po_rate, "Planned failure rate")
+    _check_1_dim(cluster.po_duration, "Planned outage duration")
+    _check_1_dim(cluster.npo_min, "Minimum count of planned outages")
+    _check_1_dim(cluster.npo_max, "Maximum count of planned outages")
+    _check_1_dim(cluster.modulation, "Hourly modulation")
+    if len(cluster.modulation) != 24:
+        raise ValueError("hourly modulation array must have 24 values.")
+
+    _check_array(
+        cluster.fo_rate < 0, "Forced failure rate is negative on following days"
+    )
+    _check_array(
+        cluster.fo_rate > 1, "Forced failure rate is greater than 1 on following days"
+    )
+    _check_array(
+        cluster.fo_duration < 0, "Forced outage duration is negative on following days"
+    )
+    _check_array(
+        cluster.po_rate < 0, "Planned failure rate is negative on following days"
+    )
+    _check_array(
+        cluster.po_rate > 1, "Planned failure rate is greater than 1 on following days"
+    )
+    _check_array(
+        cluster.po_duration < 0, "Planned outage duration is negative on following days"
+    )
+    _check_array(
+        cluster.modulation < 0, "Hourly modulation is negative on following hours"
+    )
+
+    lengths = {
+        len(a)
+        for a in [
+            cluster.fo_rate,
+            cluster.fo_duration,
+            cluster.po_rate,
+            cluster.po_duration,
+            cluster.npo_min,
+            cluster.npo_max,
+        ]
+    }
+    if len(lengths) != 1:
+        raise ValueError(f"Not all daily arrays have same size, got {lengths}")
+
 
 class OutputTimeseries:
     def __init__(self, ts_count: int, days: int) -> None:
@@ -96,6 +174,7 @@ def generate_time_series(
         """
         generation of multiple timeseries for a given thermal cluster
         """
+        _check_cluster(cluster)
 
         # TODO: Remove this log size limit, seems useless and error prone if very large durations
         log_size = 4000  # >= 5 * (max(df) + max(dp))

tests/test_unit.py

@@ -9,6 +9,7 @@
 # SPDX-License-Identifier: MPL-2.0
 #
 # This file is part of the Antares project.
+from typing import Any
 
 import numpy as np
 import numpy.testing as npt
@@ -18,6 +19,7 @@
     ProbabilityLaw,
     ThermalCluster,
     ThermalDataGenerator,
+    _check_cluster,
     _column_powers,
     _daily_to_hourly,
 )
@@ -64,7 +66,8 @@ def test_invalid_fo_rates(rng, base_cluster_365_days):
     cluster.fo_rate[10] = -0.1
 
     with pytest.raises(
-        ValueError, match="forced failure rate is negative on days \[10, 12\]"
+        ValueError,
+        match="Forced failure rate is negative on following days: \[10, 12\]",
     ):
         generator = ThermalDataGenerator(rng=rng, days=days)
         generator.generate_time_series(cluster, 1)
@@ -77,12 +80,72 @@ def test_invalid_po_rates(rng, base_cluster_365_days):
     cluster.po_rate[10] = -0.1
 
     with pytest.raises(
-        ValueError, match="planned failure rate is negative on days \[10, 12\]"
+        ValueError,
+        match="Planned failure rate is negative on following days: \[10, 12\]",
     ):
         generator = ThermalDataGenerator(rng=rng, days=days)
         generator.generate_time_series(cluster, 1)
 
 
+def valid_cluster() -> ThermalCluster:
+    days = 365
+    return ThermalCluster(
+        unit_count=10,
+        nominal_power=100,
+        modulation=np.ones(dtype=float, shape=24),
+        fo_law=ProbabilityLaw.UNIFORM,
+        fo_volatility=0,
+        po_law=ProbabilityLaw.UNIFORM,
+        po_volatility=0,
+        fo_duration=10 * np.ones(dtype=int, shape=days),
+        fo_rate=0.2 * np.ones(dtype=float, shape=days),
+        po_duration=10 * np.ones(dtype=int, shape=days),
+        po_rate=np.zeros(dtype=float, shape=days),
+        npo_min=np.zeros(dtype=int, shape=days),
+        npo_max=10 * np.ones(dtype=int, shape=days),
+    )
+
+
+def test_invalid_cluster():
+    cluster = valid_cluster()
+    _check_cluster(cluster)
+
+    cluster = valid_cluster()
+    with pytest.raises(ValueError):
+        cluster.nominal_power = -1
+        _check_cluster(cluster)
+
+    cluster = valid_cluster()
+    with pytest.raises(ValueError):
+        cluster.unit_count = -1
+        _check_cluster(cluster)
+
+    cluster = valid_cluster()
+    with pytest.raises(ValueError):
+        cluster.fo_duration[10] = -1
+        _check_cluster(cluster)
+
+    cluster = valid_cluster()
+    with pytest.raises(ValueError):
+        cluster.po_duration[10] = -1
+        _check_cluster(cluster)
+
+    cluster = valid_cluster()
+    with pytest.raises(ValueError):
+        cluster.modulation[10] = -1
+        _check_cluster(cluster)
+
+    cluster = valid_cluster()
+    with pytest.raises(ValueError):
+        cluster.modulation = np.ones(30)
+        _check_cluster(cluster)
+
+    cluster = valid_cluster()
+    with pytest.raises(ValueError):
+        cluster.fo_rate = cluster.fo_rate[:-2]
+        _check_cluster(cluster)
+
+
 def test_forced_outages(rng):
     days = 365
     cluster = ThermalCluster(