diff --git a/src/antares/tsgen/cluster_import.py b/src/antares/tsgen/cluster_import.py index 0771888..0b03ce0 100644 --- a/src/antares/tsgen/cluster_import.py +++ b/src/antares/tsgen/cluster_import.py @@ -14,7 +14,7 @@ import numpy as np -from .ts_generator import ProbabilityLaw, ThermalCluster, OutageGenerationParameters +from .ts_generator import OutageGenerationParameters, ProbabilityLaw, ThermalCluster def import_thermal_cluster(path: Path, days_per_year: int = 365) -> ThermalCluster: @@ -30,12 +30,12 @@ def import_thermal_cluster(path: Path, days_per_year: int = 365) -> ThermalClust fo_volatility=float(array[5][1]), po_law=law_dict[array[6][1]], po_volatility=float(array[7][1]), - fo_duration=array[8][1: days_per_year + 1].astype(int), - fo_rate=array[9][1: days_per_year + 1].astype(float), - po_duration=array[10][1: days_per_year + 1].astype(int), - po_rate=array[11][1: days_per_year + 1].astype(float), - npo_min=array[12][1: days_per_year + 1].astype(int), - npo_max=array[13][1: days_per_year + 1].astype(int), + fo_duration=array[8][1 : days_per_year + 1].astype(int), + fo_rate=array[9][1 : days_per_year + 1].astype(float), + po_duration=array[10][1 : days_per_year + 1].astype(int), + po_rate=array[11][1 : days_per_year + 1].astype(float), + npo_min=array[12][1 : days_per_year + 1].astype(int), + npo_max=array[13][1 : days_per_year + 1].astype(int), ) return ThermalCluster( outage_gen_params, diff --git a/src/antares/tsgen/duration_generator.py b/src/antares/tsgen/duration_generator.py index 4bc229b..9d67ba9 100644 --- a/src/antares/tsgen/duration_generator.py +++ b/src/antares/tsgen/duration_generator.py @@ -30,8 +30,7 @@ class DurationGenerator(ABC): """ @abstractmethod - def generate_duration(self, day: int) -> int: - ... + def generate_duration(self, day: int) -> int: ... class GeneratorWrapper(DurationGenerator): diff --git a/src/antares/tsgen/random_generator.py b/src/antares/tsgen/random_generator.py index 9bd3416..b705ed0 100644 --- a/src/antares/tsgen/random_generator.py +++ b/src/antares/tsgen/random_generator.py @@ -21,8 +21,7 @@ class RNG(ABC): """ @abstractmethod - def next(self) -> float: - ... + def next(self) -> float: ... class PythonRNG(ABC): diff --git a/src/antares/tsgen/ts_generator.py b/src/antares/tsgen/ts_generator.py index 0115641..189209e 100644 --- a/src/antares/tsgen/ts_generator.py +++ b/src/antares/tsgen/ts_generator.py @@ -11,11 +11,11 @@ # This file is part of the Antares project. from dataclasses import dataclass -from typing import Tuple, Any +from typing import Any, Tuple import numpy as np import numpy.typing as npt -from numpy import ndarray, dtype +from numpy import dtype, ndarray from antares.tsgen.duration_generator import ProbabilityLaw, make_duration_generator from antares.tsgen.random_generator import RNG, MersenneTwisterRNG @@ -55,7 +55,7 @@ def __post_init__(self) -> None: @dataclass class ThermalCluster: # available units of the cluster - #unit_count: int + # unit_count: int outage_gen_params: OutageGenerationParameters # nominal power nominal_power: float @@ -64,19 +64,19 @@ class ThermalCluster: # forced and planed outage parameters # indexed by day of the year - #fo_duration: IntArray - #fo_rate: FloatArray - #po_duration: IntArray - #po_rate: FloatArray - #npo_min: IntArray # number of planed outage min in a day - #npo_max: IntArray # number of planed outage max in a day + # fo_duration: IntArray + # fo_rate: FloatArray + # po_duration: IntArray + # po_rate: FloatArray + # npo_min: IntArray # number of planed outage min in a day + # npo_max: IntArray # number of planed outage max in a day # forced and planed outage probability law and volatility # volatility characterizes the distance from the expect at which the value drawn can be - #fo_law: ProbabilityLaw - #fo_volatility: float - #po_law: ProbabilityLaw - #po_volatility: float + # fo_law: ProbabilityLaw + # fo_volatility: float + # po_law: ProbabilityLaw + # po_volatility: float def __post_init__(self) -> None: _check_cluster(self) @@ -84,13 +84,13 @@ def __post_init__(self) -> None: @dataclass class LinkCapacity: - #outage generation parameters + # outage generation parameters outage_gen_params: OutageGenerationParameters - #nominal capacity + # nominal capacity nominal_capacity: float - #direct / indirect modulation of the nominal capacity + # direct / indirect modulation of the nominal capacity modulation_direct: FloatArray modulation_indirect: FloatArray @@ -192,7 +192,8 @@ def _check_link_capacity(link_capacity: LinkCapacity): if len(lengths) != 1: raise ValueError(f"Not all daily arrays have same size, got {lengths}") -#OutputTimeseries -> + +# OutputTimeseries -> class OutputTimeseries: def __init__(self, ts_count: int, days: int) -> None: self.available_units = np.zeros(shape=(days, ts_count), dtype=int) @@ -337,12 +338,7 @@ def __init__(self, rng: RNG = MersenneTwisterRNG(), days: int = 365) -> None: self.rng = rng self.days = days - def _compare_apparent_PO( - self, - current_available_units: int, - po_candidates: int, - stock: int - ): + def _compare_apparent_PO(self, current_available_units: int, po_candidates: int, stock: int): candidate = po_candidates + stock if 0 <= candidate <= current_available_units: po_candidates = candidate @@ -356,13 +352,7 @@ def _compare_apparent_PO( return po_candidates, stock def _generate_outages( - self, - outage_gen_params: OutageGenerationParameters, - log, - log_size, - logp, - number_of_timeseries, - output + self, outage_gen_params: OutageGenerationParameters, log, log_size, logp, number_of_timeseries, output ): daily_fo_rate = _compute_failure_rates(outage_gen_params.fo_rate, outage_gen_params.fo_duration) daily_po_rate = _compute_failure_rates(outage_gen_params.po_rate, outage_gen_params.po_duration) @@ -371,19 +361,28 @@ def _generate_outages( fo_drawer = ForcedOutagesDrawer(self.rng, outage_gen_params.unit_count, daily_fo_rate) po_drawer = PlannedOutagesDrawer(self.rng, outage_gen_params.unit_count, daily_po_rate) - fod_generator = make_duration_generator(self.rng, outage_gen_params.fo_law, - outage_gen_params.fo_volatility, outage_gen_params.fo_duration) - pod_generator = make_duration_generator(self.rng, outage_gen_params.po_law, - outage_gen_params.po_volatility, outage_gen_params.po_duration) - - self.output_generation(outage_gen_params, fo_drawer, fod_generator, - log, log_size, logp, - number_of_timeseries, output, po_drawer, pod_generator) + fod_generator = make_duration_generator( + self.rng, outage_gen_params.fo_law, outage_gen_params.fo_volatility, outage_gen_params.fo_duration + ) + pod_generator = make_duration_generator( + self.rng, outage_gen_params.po_law, outage_gen_params.po_volatility, outage_gen_params.po_duration + ) + + self.output_generation( + outage_gen_params, + fo_drawer, + fod_generator, + log, + log_size, + logp, + number_of_timeseries, + output, + po_drawer, + pod_generator, + ) def generate_time_series_for_links( - self, - link: LinkCapacity, - number_of_timeseries: int + self, link: LinkCapacity, number_of_timeseries: int ) -> tuple[ndarray[Any, dtype[Any]], ndarray[Any, dtype[Any]]]: """ generation of multiple timeseries for a given link capacity @@ -414,9 +413,9 @@ def generate_time_series_for_links( return direct_output, indirect_output def generate_time_series_for_clusters( - self, - cluster: ThermalCluster, - number_of_timeseries: int, + self, + cluster: ThermalCluster, + number_of_timeseries: int, ) -> OutputTimeseries: """ generation of multiple timeseries for a given thermal cluster @@ -434,8 +433,6 @@ def generate_time_series_for_clusters( # failure rate means the probability to enter in outage each day # its value is given by: OR / [OR + OD * (1 - OR)] - - # --- calculation --- # the two first generated time series will be dropped, necessary to make system stable and physically coherent # as a consequence, N + 2 time series will be computed @@ -451,9 +448,19 @@ def generate_time_series_for_clusters( np.round(output.available_power) return output - def output_generation(self, outage_gen_params, fo_drawer, fod_generator, log, log_size, logp, number_of_timeseries, - output, - po_drawer, pod_generator): + def output_generation( + self, + outage_gen_params, + fo_drawer, + fod_generator, + log, + log_size, + logp, + number_of_timeseries, + output, + po_drawer, + pod_generator, + ): # dates now = 0 # current number of PO and AU (avlaible units) diff --git a/tests/test_unit.py b/tests/test_unit.py index 3f86425..f1988ee 100644 --- a/tests/test_unit.py +++ b/tests/test_unit.py @@ -15,19 +15,29 @@ import pytest from antares.tsgen.random_generator import MersenneTwisterRNG -import antares.tsgen.ts_generator +from antares.tsgen.ts_generator import ( + LinkCapacity, + OutageGenerationParameters, + ProbabilityLaw, + ThermalCluster, + ThermalDataGenerator, + _categorize_outages, + _check_cluster, + _column_powers, + _daily_to_hourly, +) def test_daily_to_hourly(): daily = np.array([[1, 2]]) - hourly = antares.tsgen.ts_generator._daily_to_hourly(daily) + hourly = _daily_to_hourly(daily) expected = [[1, 2]] * 24 npt.assert_equal(hourly, expected) def test_elevate_to_power(): input = np.array([1, 0.5, 0.1]) - powers = antares.tsgen.ts_generator._column_powers(input, 3) + powers = _column_powers(input, 3) expected = np.array([[1, 1, 1], [1, 0.5, 0.25], [1, 0.1, 0.01]]) npt.assert_almost_equal(powers, expected, decimal=3) @@ -36,31 +46,33 @@ def test_elevate_to_power(): def base_cluster_365_days(): days = 365 outage_gen_params = valid_outage_params() - return antares.tsgen.ts_generator.ThermalCluster( + return ThermalCluster( outage_gen_params, nominal_power=100, modulation=np.ones(dtype=float, shape=8760), ) + @pytest.fixture() def base_link_365_days(): days = 365 outage_gen_params = valid_outage_params() - return antares.tsgen.ts_generator.LinkCapacity( + return LinkCapacity( outage_gen_params, nominal_capacity=100, modulation_indirect=np.ones(dtype=float, shape=8760), modulation_direct=np.ones(dtype=float, shape=8760), ) + # cluster -> outage gen params def test_outage_params_with_null_duration(rng): days = 365 args = { "unit_count": 10, - "fo_law": antares.tsgen.ts_generator.ProbabilityLaw.UNIFORM, + "fo_law": ProbabilityLaw.UNIFORM, "fo_volatility": 0, - "po_law": antares.tsgen.ts_generator.ProbabilityLaw.UNIFORM, + "po_law": ProbabilityLaw.UNIFORM, "po_volatility": 0, "fo_duration": 10 * np.ones(dtype=int, shape=days), "fo_rate": 0.2 * np.zeros(dtype=float, shape=days), @@ -72,7 +84,7 @@ def test_outage_params_with_null_duration(rng): for duration_type in ["po_duration", "fo_duration"]: args[duration_type] = 10 * np.zeros(dtype=int, shape=days) with pytest.raises(ValueError, match="outage duration is null or negative on following days"): - antares.tsgen.ts_generator.OutageGenerationParameters(**args) + OutageGenerationParameters(**args) def test_invalid_fo_rates(rng, base_cluster_365_days, base_link_365_days): @@ -88,7 +100,7 @@ def test_invalid_fo_rates(rng, base_cluster_365_days, base_link_365_days): ValueError, match="Forced failure rate is negative on following days: \[10, 12\]", ): - generator = antares.tsgen.ts_generator.ThermalDataGenerator(rng=rng, days=days) + generator = ThermalDataGenerator(rng=rng, days=days) generator.generate_time_series_for_clusters(cluster, 1) @@ -102,14 +114,14 @@ def test_invalid_po_rates(rng, base_cluster_365_days): ValueError, match="Planned failure rate is negative on following days: \[10, 12\]", ): - generator = antares.tsgen.ts_generator.ThermalDataGenerator(rng=rng, days=days) + generator = ThermalDataGenerator(rng=rng, days=days) generator.generate_time_series_for_clusters(cluster, 1) -def valid_cluster() -> antares.tsgen.ts_generator.ThermalCluster: +def valid_cluster() -> ThermalCluster: days = 365 outage_gen_params = valid_outage_params() - return antares.tsgen.ts_generator.ThermalCluster( + return ThermalCluster( outage_gen_params, nominal_power=100, modulation=np.ones(dtype=float, shape=8760), @@ -118,42 +130,42 @@ def valid_cluster() -> antares.tsgen.ts_generator.ThermalCluster: def test_invalid_cluster(): cluster = valid_cluster() - antares.tsgen.ts_generator._check_cluster(cluster) + _check_cluster(cluster) cluster = valid_cluster() with pytest.raises(ValueError): cluster.nominal_power = -1 - antares.tsgen.ts_generator._check_cluster(cluster) + _check_cluster(cluster) cluster = valid_cluster() with pytest.raises(ValueError): cluster.outage_gen_params.unit_count = -1 - antares.tsgen.ts_generator._check_cluster(cluster) + _check_cluster(cluster) cluster = valid_cluster() with pytest.raises(ValueError): cluster.outage_gen_params.fo_duration[10] = -1 - antares.tsgen.ts_generator._check_cluster(cluster) + _check_cluster(cluster) cluster = valid_cluster() with pytest.raises(ValueError): cluster.outage_gen_params.po_duration[10] = -1 - antares.tsgen.ts_generator._check_cluster(cluster) + _check_cluster(cluster) cluster = valid_cluster() with pytest.raises(ValueError): cluster.modulation[10] = -1 - antares.tsgen.ts_generator._check_cluster(cluster) + _check_cluster(cluster) cluster = valid_cluster() with pytest.raises(ValueError): cluster.modulation = np.ones(30) - antares.tsgen.ts_generator._check_cluster(cluster) + _check_cluster(cluster) cluster = valid_cluster() with pytest.raises(ValueError): cluster.outage_gen_params.fo_rate = cluster.outage_gen_params.fo_rate[:-2] - antares.tsgen.ts_generator._check_cluster(cluster) + _check_cluster(cluster) @pytest.mark.parametrize( @@ -169,20 +181,20 @@ def test_invalid_cluster(): ], ) def test_distribute_outages(available_units, po_candidates, fo_candidates, expected): - outages = antares.tsgen.ts_generator._categorize_outages(available_units, po_candidates, fo_candidates) + outages = _categorize_outages(available_units, po_candidates, fo_candidates) assert outages == expected def test_forced_outages(rng): days = 365 - #modifier valid_outage_params de facon à le paramétrer + # modifier valid_outage_params de facon à le paramétrer outage_gen_params = valid_outage_params() - cluster = antares.tsgen.ts_generator.ThermalCluster( + cluster = ThermalCluster( outage_gen_params, nominal_power=100, modulation=np.ones(dtype=float, shape=8760), ) - link_capacity=antares.tsgen.ts_generator.LinkCapacity( + link_capacity = LinkCapacity( outage_gen_params, nominal_capacity=100, modulation_direct=np.ones(dtype=float, shape=8760), @@ -190,7 +202,7 @@ def test_forced_outages(rng): ) cluster.modulation[12] = 0.5 - generator = antares.tsgen.ts_generator.ThermalDataGenerator(rng=rng, days=days) + generator = ThermalDataGenerator(rng=rng, days=days) results = generator.generate_time_series_for_clusters(cluster, 1) # 2 forced outages occur on day 5, with duration 10 npt.assert_equal(results.forced_outages.T[0][:6], [0, 0, 0, 0, 2, 0]) @@ -210,12 +222,12 @@ def test_forced_outages(rng): def test_planned_outages(rng): days = 365 outage_gen_params = valid_outage_params() - cluster = antares.tsgen.ts_generator.ThermalCluster( + cluster = ThermalCluster( outage_gen_params, nominal_power=100, modulation=np.ones(dtype=float, shape=8760), ) - link=antares.tsgen.ts_generator.LinkCapacity( + link = LinkCapacity( outage_gen_params, nominal_capacity=100, modulation_indirect=np.ones(dtype=float, shape=8760), @@ -225,8 +237,8 @@ def test_planned_outages(rng): link.modulation_indirect[12] = 0.5 link.modulation_direct[12] = 0.5 - generator = antares.tsgen.ts_generator.ThermalDataGenerator(rng=rng, days=days) - results = generator.generate_time_series_for_clusters(cluster, 1) + generator = ThermalDataGenerator(rng=rng, days=days) + results = generator.generate_time_series_for_clusters(cluster, 1) # 0 forced outage npt.assert_equal(results.forced_outages.T[0], np.zeros(365)) npt.assert_equal(results.forced_outage_durations.T[0], np.zeros(365)) @@ -244,18 +256,18 @@ def test_planned_outages_limitation(rng): days = 365 # Maximum 1 planned outage at a time. outage_gen_params = valid_outage_params() - cluster = antares.tsgen.ts_generator.ThermalCluster( + cluster = ThermalCluster( outage_gen_params, nominal_power=100, modulation=np.ones(dtype=float, shape=8760), ) - link=antares.tsgen.ts_generator.LinkCapacity( + link = LinkCapacity( outage_gen_params, nominal_capacity=100, modulation_direct=np.ones(dtype=float, shape=8760), modulation_indirect=np.ones(dtype=float, shape=8760), ) - generator = antares.tsgen.ts_generator.ThermalDataGenerator(rng=rng, days=days) + generator = ThermalDataGenerator(rng=rng, days=days) results = generator.generate_time_series_for_clusters(cluster, 1) # No forced outage npt.assert_equal(results.forced_outages.T[0], np.zeros(365)) @@ -274,18 +286,18 @@ def test_planned_outages_min_limitation(rng): days = 365 # Minimum 2 planned outages at a time outage_gen_params = valid_outage_params() - cluster = antares.tsgen.ts_generator.ThermalCluster( + cluster = ThermalCluster( outage_gen_params, nominal_power=100, modulation=np.ones(dtype=float, shape=8760), ) - link=antares.tsgen.ts_generator.LinkCapacity( + link = LinkCapacity( outage_gen_params, nominal_capacity=100, modulation_direct=np.ones(dtype=float, shape=8760), modulation_indirect=np.ones(dtype=float, shape=8760), ) - generator = antares.tsgen.ts_generator.ThermalDataGenerator(rng=rng, days=days) + generator = ThermalDataGenerator(rng=rng, days=days) results = generator.generate_time_series_for_clusters(cluster, 1) # No forced outage npt.assert_equal(results.forced_outages.T[0], np.zeros(365)) @@ -314,19 +326,19 @@ def test_with_long_fo_and_po_duration(data_directory): fo_rate[:31] = 0.1 po_rate[:31] = 0.02 outage_gen_params = valid_outage_params() - cluster = antares.tsgen.ts_generator.ThermalCluster( + cluster = ThermalCluster( outage_gen_params, nominal_power=500, modulation=modulation_matrix, ) - link=antares.tsgen.ts_generator.LinkCapacity( + link = LinkCapacity( outage_gen_params, nominal_capacity=500, modulation_indirect=modulation_matrix, modulation_direct=modulation_matrix, ) rng = MersenneTwisterRNG(seed=3005489) - generator = antares.tsgen.ts_generator.ThermalDataGenerator(rng=rng, days=days) + generator = ThermalDataGenerator(rng=rng, days=days) results = generator.generate_time_series_for_clusters(cluster, 10) expected_matrix = np.loadtxt( @@ -334,13 +346,14 @@ def test_with_long_fo_and_po_duration(data_directory): ) assert np.array_equal(results.available_power, expected_matrix) -def valid_outage_params() -> antares.tsgen.ts_generator.OutageGenerationParameters: + +def valid_outage_params() -> OutageGenerationParameters: days = 365 - return antares.tsgen.ts_generator.OutageGenerationParameters( + return OutageGenerationParameters( unit_count=10, - fo_law=antares.tsgen.ts_generator.ProbabilityLaw.UNIFORM, + fo_law=ProbabilityLaw.UNIFORM, fo_volatility=0, - po_law=antares.tsgen.ts_generator.ProbabilityLaw.UNIFORM, + po_law=ProbabilityLaw.UNIFORM, po_volatility=0, fo_duration=10 * np.ones(dtype=int, shape=days), fo_rate=np.zeros(dtype=float, shape=days), @@ -350,13 +363,14 @@ def valid_outage_params() -> antares.tsgen.ts_generator.OutageGenerationParamete npo_max=10 * np.ones(dtype=int, shape=days), ) -#def test_valid_outage_params(): + +# def test_valid_outage_params(): -#def test_invalid_outage_params(): +# def test_invalid_outage_params(): -#def test_valid_link_capacity(): +# def test_valid_link_capacity(): -#def test_invalid_link_capacity(): +# def test_invalid_link_capacity():