From 8bc9f27804857766727a25cd4230474f47b2f243 Mon Sep 17 00:00:00 2001 From: wahadameh Date: Wed, 30 Oct 2024 14:40:13 +0100 Subject: [PATCH] Including link unit testing and refactoring --- src/antares/tsgen/ts_generator.py | 77 +++++-- tests/test.py | 4 +- tests/test_ts_generator.py | 8 +- tests/test_unit.py | 352 +++++++++++++++++++++++------- 4 files changed, 338 insertions(+), 103 deletions(-) diff --git a/src/antares/tsgen/ts_generator.py b/src/antares/tsgen/ts_generator.py index 2bed74d..6f932b4 100644 --- a/src/antares/tsgen/ts_generator.py +++ b/src/antares/tsgen/ts_generator.py @@ -172,7 +172,7 @@ def _check_link_capacity(link_capacity: LinkCapacity) -> None: _check_1_dim(link_capacity.modulation_indirect, "Direct modulation") _check_1_dim(link_capacity.modulation_indirect, "Indirect hourly modulation") - if len(link_capacity.modulation_direct) != 8760 and len(link_capacity.modulation_indirect) != 8760: + if len(link_capacity.modulation_direct) != 8760 or len(link_capacity.modulation_indirect) != 8760: raise ValueError("hourly modulation array must have 8760 values.") _check_array(link_capacity.modulation_direct < 0, "Hourly direct modulation is negative on following hours") @@ -193,12 +193,13 @@ def _check_link_capacity(link_capacity: LinkCapacity) -> None: raise ValueError(f"Not all daily arrays have same size, got {lengths}") -# OutputTimeseries -> -class OutputTimeseries: +class OutageOutput: def __init__(self, ts_count: int, days: int) -> None: + # number_of_timeseries + self.ts_count = ts_count + # number of days + self.days = days self.available_units = np.zeros(shape=(days, ts_count), dtype=int) - # available power each hours - self.available_power = np.zeros((24 * days, ts_count), dtype=float) # number of pure planed, pure forced and mixed outage each day self.planned_outages = np.zeros((days, ts_count), dtype=int) self.forced_outages = np.zeros((days, ts_count), dtype=int) @@ -209,6 +210,24 @@ def __init__(self, ts_count: int, days: int) -> None: self.forced_outage_durations = np.zeros((days, ts_count), dtype=int) +class ClusterOutputTimeseries: + def __init__(self, outage_output: OutageOutput) -> None: + # output parameters + self.outage_output = outage_output + # available power each hours + self.available_power = np.zeros((24 * outage_output.days, outage_output.ts_count), dtype=float) + + +class LinkOutputTimeseries: + def __init__(self, outage_output: OutageOutput) -> None: + # output parameters + self.outage_output = outage_output + # direct available power each hours + self.direct_available_power = np.zeros((24 * outage_output.days, outage_output.ts_count), dtype=float) + # available power each hours + self.indirect_available_power = np.zeros((24 * outage_output.days, outage_output.ts_count), dtype=float) + + def _column_powers(column: FloatArray, width: int) -> npt.NDArray: """ Returns a matrix of given width where column[i] is the ith power of the input vector. @@ -333,7 +352,7 @@ def _combine_failure_rates(rates1: FloatArray, rates2: FloatArray) -> None: rates2[mask] *= (1 - rates1[mask]) / (1 - rates2[mask]) -class ThermalDataGenerator: +class TimeseriesGenerator: def __init__(self, rng: RNG = MersenneTwisterRNG(), days: int = 365) -> None: self.rng = rng self.days = days @@ -358,7 +377,7 @@ def _generate_outages( log_size: int, logp: ndarray[Any, dtype], number_of_timeseries: int, - output: OutputTimeseries, + output: OutageOutput, ) -> None: 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) @@ -387,9 +406,7 @@ def _generate_outages( pod_generator, ) - def generate_time_series_for_links( - self, link: LinkCapacity, number_of_timeseries: int - ) -> tuple[ndarray[Any, dtype[Any]], ndarray[Any, dtype[Any]]]: + def generate_time_series_for_links(self, link: LinkCapacity, number_of_timeseries: int) -> LinkOutputTimeseries: """ generation of multiple timeseries for a given link capacity """ @@ -407,22 +424,29 @@ def generate_time_series_for_links( # as a consequence, N + 2 time series will be computed # output that will be returned - output = OutputTimeseries(number_of_timeseries, self.days) + outage_params = OutageOutput(number_of_timeseries, self.days) + link_output = LinkOutputTimeseries(outage_params) - self._generate_outages(link.outage_gen_params, log, log_size, logp, number_of_timeseries, output) + self._generate_outages( + link.outage_gen_params, log, log_size, logp, number_of_timeseries, link_output.outage_output + ) - hourly_available_units = _daily_to_hourly(output.available_units) + hourly_available_units = _daily_to_hourly(link_output.outage_output.available_units) - direct_output = hourly_available_units * link.nominal_capacity * link.modulation_direct[:, np.newaxis] - indirect_output = hourly_available_units * link.nominal_capacity * link.modulation_indirect[:, np.newaxis] + link_output.direct_available_power = ( + hourly_available_units * link.nominal_capacity * link.modulation_direct[:, np.newaxis] + ) + link_output.indirect_available_power = ( + hourly_available_units * link.nominal_capacity * link.modulation_indirect[:, np.newaxis] + ) - return direct_output, indirect_output + return link_output def generate_time_series_for_clusters( self, cluster: ThermalCluster, number_of_timeseries: int, - ) -> OutputTimeseries: + ) -> ClusterOutputTimeseries: """ generation of multiple timeseries for a given thermal cluster """ @@ -444,15 +468,20 @@ def generate_time_series_for_clusters( # as a consequence, N + 2 time series will be computed # output that will be returned - output = OutputTimeseries(number_of_timeseries, self.days) + outage_output = OutageOutput(number_of_timeseries, self.days) + cluster_output = ClusterOutputTimeseries(outage_output) - self._generate_outages(cluster.outage_gen_params, log, log_size, logp, number_of_timeseries, output) + self._generate_outages( + cluster.outage_gen_params, log, log_size, logp, number_of_timeseries, cluster_output.outage_output + ) # - hourly_available_units = _daily_to_hourly(output.available_units) - output.available_power = hourly_available_units * cluster.nominal_power * cluster.modulation[:, np.newaxis] - np.round(output.available_power) - return output + hourly_available_units = _daily_to_hourly(cluster_output.outage_output.available_units) + cluster_output.available_power = ( + hourly_available_units * cluster.nominal_power * cluster.modulation[:, np.newaxis] + ) + np.round(cluster_output.available_power) + return cluster_output def output_generation( self, @@ -463,7 +492,7 @@ def output_generation( log_size: int, logp: ndarray[Any, dtype], number_of_timeseries: int, - output: OutputTimeseries, + output: OutageOutput, po_drawer: PlannedOutagesDrawer, pod_generator: DurationGenerator, ) -> None: diff --git a/tests/test.py b/tests/test.py index 58f0736..16f5cac 100644 --- a/tests/test.py +++ b/tests/test.py @@ -15,7 +15,7 @@ import pytest from antares.tsgen.cluster_import import import_thermal_cluster -from antares.tsgen.ts_generator import ThermalCluster, ThermalDataGenerator +from antares.tsgen.ts_generator import ThermalCluster, TimeseriesGenerator @pytest.fixture @@ -26,7 +26,7 @@ def cluster(data_directory) -> ThermalCluster: def test_cluster(cluster, output_directory): ts_nb = 1 - generator = ThermalDataGenerator() + generator = TimeseriesGenerator() results = generator.generate_time_series_for_clusters(cluster, ts_nb) tot_po = 0 diff --git a/tests/test_ts_generator.py b/tests/test_ts_generator.py index ea71b9e..13e0270 100644 --- a/tests/test_ts_generator.py +++ b/tests/test_ts_generator.py @@ -15,7 +15,7 @@ import pytest from antares.tsgen.cluster_import import import_thermal_cluster -from antares.tsgen.ts_generator import ThermalCluster, ThermalDataGenerator +from antares.tsgen.ts_generator import ThermalCluster, TimeseriesGenerator @pytest.fixture @@ -36,7 +36,7 @@ def cluster_high_por(data_directory) -> ThermalCluster: def test_one_unit_cluster(cluster_1, output_directory): ts_nb = 4 - generator = ThermalDataGenerator() + generator = TimeseriesGenerator() results = generator.generate_time_series_for_clusters(cluster_1, ts_nb) tot_po = 0 @@ -60,7 +60,7 @@ def test_one_unit_cluster(cluster_1, output_directory): def test_hundred_unit_cluster(cluster_100, output_directory): ts_nb = 50 - generator = ThermalDataGenerator() + generator = TimeseriesGenerator() results = generator.generate_time_series_for_clusters(cluster_100, ts_nb) tot_po = 0 @@ -109,7 +109,7 @@ def test_hundred_unit_cluster(cluster_100, output_directory): def test_max_po(cluster_high_por, output_directory): ts_nb = 4 - generator = ThermalDataGenerator() + generator = TimeseriesGenerator() results = generator.generate_time_series_for_clusters(cluster_high_por, ts_nb) # check the max PO diff --git a/tests/test_unit.py b/tests/test_unit.py index a82de47..d1af672 100644 --- a/tests/test_unit.py +++ b/tests/test_unit.py @@ -20,9 +20,10 @@ OutageGenerationParameters, ProbabilityLaw, ThermalCluster, - ThermalDataGenerator, + TimeseriesGenerator, _categorize_outages, _check_cluster, + _check_link_capacity, _column_powers, _daily_to_hourly, ) @@ -44,7 +45,6 @@ def test_elevate_to_power(): @pytest.fixture() def base_cluster_365_days(): - days = 365 outage_gen_params = valid_outage_params() return ThermalCluster( outage_gen_params, @@ -55,7 +55,6 @@ def base_cluster_365_days(): @pytest.fixture() def base_link_365_days(): - days = 365 outage_gen_params = valid_outage_params() return LinkCapacity( outage_gen_params, @@ -87,12 +86,9 @@ def test_outage_params_with_null_duration(rng): OutageGenerationParameters(**args) -def test_invalid_fo_rates(rng, base_cluster_365_days, base_link_365_days): +def test_link_invalid_fo_rates(rng, base_link_365_days): days = 365 - cluster = base_cluster_365_days link = base_link_365_days - cluster.outage_gen_params.fo_rate[12] = -0.2 - cluster.outage_gen_params.fo_rate[10] = -0.1 link.outage_gen_params.fo_rate[12] = -0.2 link.outage_gen_params.fo_rate[10] = -0.1 @@ -100,11 +96,25 @@ 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 = ThermalDataGenerator(rng=rng, days=days) + generator = TimeseriesGenerator(rng=rng, days=days) + generator.generate_time_series_for_links(link, 1) + + +def test_cluster_invalid_fo_rates(rng, base_cluster_365_days): + days = 365 + cluster = base_cluster_365_days + cluster.outage_gen_params.fo_rate[12] = -0.2 + cluster.outage_gen_params.fo_rate[10] = -0.1 + + with pytest.raises( + ValueError, + match="Forced failure rate is negative on following days: \[10, 12\]", + ): + generator = TimeseriesGenerator(rng=rng, days=days) generator.generate_time_series_for_clusters(cluster, 1) -def test_invalid_po_rates(rng, base_cluster_365_days): +def test_cluster_invalid_po_rates(rng, base_cluster_365_days): days = 365 cluster = base_cluster_365_days cluster.outage_gen_params.po_rate[12] = -0.2 @@ -114,12 +124,25 @@ def test_invalid_po_rates(rng, base_cluster_365_days): ValueError, match="Planned failure rate is negative on following days: \[10, 12\]", ): - generator = ThermalDataGenerator(rng=rng, days=days) + generator = TimeseriesGenerator(rng=rng, days=days) generator.generate_time_series_for_clusters(cluster, 1) -def valid_cluster() -> ThermalCluster: +def test_link_invalid_po_rates(rng, base_link_365_days): days = 365 + link = base_link_365_days + link.outage_gen_params.po_rate[12] = -0.2 + link.outage_gen_params.po_rate[10] = -0.1 + + with pytest.raises( + ValueError, + match="Planned failure rate is negative on following days: \[10, 12\]", + ): + generator = TimeseriesGenerator(rng=rng, days=days) + generator.generate_time_series_for_links(link, 1) + + +def valid_cluster() -> ThermalCluster: outage_gen_params = valid_outage_params() return ThermalCluster( outage_gen_params, @@ -128,6 +151,16 @@ def valid_cluster() -> ThermalCluster: ) +def valid_link() -> LinkCapacity: + outage_gen_params = valid_outage_params() + return LinkCapacity( + outage_gen_params, + nominal_capacity=100, + modulation_direct=np.ones(dtype=float, shape=8760), + modulation_indirect=np.ones(dtype=float, shape=8760), + ) + + def test_invalid_cluster(): cluster = valid_cluster() _check_cluster(cluster) @@ -168,6 +201,54 @@ def test_invalid_cluster(): _check_cluster(cluster) +def test_invalid_link(): + link = valid_link() + _check_link_capacity(link) + + link = valid_link() + with pytest.raises(ValueError): + link.nominal_capacity = -1 + _check_link_capacity(link) + + link = valid_link() + with pytest.raises(ValueError): + link.outage_gen_params.unit_count = -1 + _check_link_capacity(link) + + link = valid_link() + with pytest.raises(ValueError): + link.outage_gen_params.fo_duration[10] = -1 + _check_link_capacity(link) + + link = valid_link() + with pytest.raises(ValueError): + link.outage_gen_params.po_duration[10] = -1 + _check_link_capacity(link) + + link = valid_link() + with pytest.raises(ValueError): + link.modulation_direct[10] = -1 + link.modulation_indirect[10] = -1 + _check_link_capacity(link) + + link = valid_link() + with pytest.raises(ValueError): + link.modulation_direct = np.ones(30) + print(link.modulation_direct) + _check_link_capacity(link) + + link = valid_link() + with pytest.raises(ValueError): + link.modulation_indirect = np.ones(30) + print(link.modulation_direct) + _check_link_capacity(link) + + link = valid_link() + with pytest.raises(ValueError): + link.outage_gen_params.fo_rate = link.outage_gen_params.fo_rate[:-2] + _check_link_capacity(link) + + @pytest.mark.parametrize( "available_units,po_candidates,fo_candidates,expected", [ @@ -185,7 +266,7 @@ def test_distribute_outages(available_units, po_candidates, fo_candidates, expec assert outages == expected -def test_forced_outages(rng): +def test_cluster_forced_outages(rng): days = 365 # modifier valid_outage_params de facon à le paramétrer outage_gen_params = OutageGenerationParameters( @@ -206,18 +287,19 @@ def test_forced_outages(rng): nominal_power=100, modulation=np.ones(dtype=float, shape=8760), ) + cluster.modulation[12] = 0.5 - generator = ThermalDataGenerator(rng=rng, days=days) + generator = TimeseriesGenerator(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]) - npt.assert_equal(results.forced_outage_durations.T[0][:6], [0, 0, 0, 0, 10, 0]) + npt.assert_equal(results.outage_output.forced_outages.T[0][:6], [0, 0, 0, 0, 2, 0]) + npt.assert_equal(results.outage_output.forced_outage_durations.T[0][:6], [0, 0, 0, 0, 10, 0]) # No planned outage - npt.assert_equal(results.planned_outages.T[0], np.zeros(365)) - npt.assert_equal(results.planned_outage_durations.T[0], np.zeros(365)) + npt.assert_equal(results.outage_output.planned_outages.T[0], np.zeros(365)) + npt.assert_equal(results.outage_output.planned_outage_durations.T[0], np.zeros(365)) - npt.assert_equal(results.available_units.T[0][:5], [9, 9, 9, 9, 8]) + npt.assert_equal(results.outage_output.available_units.T[0][:5], [9, 9, 9, 9, 8]) # Check available power consistency with available units and modulation available_power = results.available_power.T assert available_power[0][0] == 900 @@ -225,32 +307,72 @@ def test_forced_outages(rng): assert available_power[0][4 * 24] == 800 -def test_planned_outages(rng): +def test_link_forced_outages(rng): days = 365 - outage_gen_params = valid_outage_params() - cluster = ThermalCluster( - outage_gen_params, - nominal_power=100, - modulation=np.ones(dtype=float, shape=8760), + # modifier valid_outage_params de facon à le paramétrer + outage_gen_params = OutageGenerationParameters( + unit_count=10, + 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), ) link = LinkCapacity( outage_gen_params, nominal_capacity=100, - modulation_indirect=np.ones(dtype=float, shape=8760), modulation_direct=np.ones(dtype=float, shape=8760), + modulation_indirect=np.ones(dtype=float, shape=8760), ) - cluster.modulation[12] = 0.5 - link.modulation_indirect[12] = 0.5 + link.modulation_direct[12] = 0.5 + link.modulation_indirect[12] = 0.5 + + generator = TimeseriesGenerator(rng=rng, days=days) + results = generator.generate_time_series_for_links(link, 1) + # 2 forced outages occur on day 5, with duration 10 + npt.assert_equal(results.outage_output.forced_outages.T[0][:6], [0, 0, 0, 0, 2, 0]) + npt.assert_equal(results.outage_output.forced_outage_durations.T[0][:6], [0, 0, 0, 0, 10, 0]) + # No planned outage + npt.assert_equal(results.outage_output.planned_outages.T[0], np.zeros(365)) + npt.assert_equal(results.outage_output.planned_outage_durations.T[0], np.zeros(365)) + + npt.assert_equal(results.outage_output.available_units.T[0][:5], [9, 9, 9, 9, 8]) + # Check available power consistency with available units and modulation + direct_available_power = results.direct_available_power.T + indirect_available_power = results.indirect_available_power.T + assert direct_available_power[0][0] == 900 + assert direct_available_power[0][12] == 450 # Modulation is 0.5 for hour 12 + assert direct_available_power[0][4 * 24] == 800 + assert indirect_available_power[0][0] == 900 + assert indirect_available_power[0][12] == 450 # Modulation is 0.5 for hour 12 + assert indirect_available_power[0][4 * 24] == 800 + + +def test_cluster_planned_outages(rng): + days = 365 + outage_gen_params = valid_outage_params() + cluster = ThermalCluster( + outage_gen_params, + nominal_power=100, + modulation=np.ones(dtype=float, shape=8760), + ) + + cluster.modulation[12] = 0.5 - generator = ThermalDataGenerator(rng=rng, days=days) + generator = TimeseriesGenerator(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)) + npt.assert_equal(results.outage_output.forced_outages.T[0], np.zeros(365)) + npt.assert_equal(results.outage_output.forced_outage_durations.T[0], np.zeros(365)) # No planned outage - npt.assert_equal(results.planned_outages.T[0][:6], [0, 0, 0, 0, 2, 0]) - npt.assert_equal(results.available_units.T[0][:5], [9, 9, 9, 9, 8]) + npt.assert_equal(results.outage_output.planned_outages.T[0][:6], [0, 0, 0, 0, 2, 0]) + npt.assert_equal(results.outage_output.available_units.T[0][:5], [9, 9, 9, 9, 8]) # Check available power consistency with available units and modulation available_power = results.available_power.T assert available_power[0][0] == 900 @@ -258,7 +380,39 @@ def test_planned_outages(rng): assert available_power[0][4 * 24] == 800 -def test_planned_outages_limitation(rng): +def test_link_planned_outages(rng): + days = 365 + outage_gen_params = valid_outage_params() + link = LinkCapacity( + outage_gen_params, + nominal_capacity=100, + modulation_direct=np.ones(dtype=float, shape=8760), + modulation_indirect=np.ones(dtype=float, shape=8760), + ) + + link.modulation_direct[12] = 0.5 + link.modulation_indirect[12] = 0.5 + + generator = TimeseriesGenerator(rng=rng, days=days) + results = generator.generate_time_series_for_links(link, 1) + # 0 forced outage + npt.assert_equal(results.outage_output.forced_outages.T[0], np.zeros(365)) + npt.assert_equal(results.outage_output.forced_outage_durations.T[0], np.zeros(365)) + # No planned outage + npt.assert_equal(results.outage_output.planned_outages.T[0][:6], [0, 0, 0, 0, 2, 0]) + npt.assert_equal(results.outage_output.available_units.T[0][:5], [9, 9, 9, 9, 8]) + # Check available power consistency with available units and modulation + direct_available_power = results.direct_available_power.T + indirect_available_power = results.indirect_available_power.T + assert direct_available_power[0][0] == 900 + assert direct_available_power[0][12] == 450 # Modulation is 0.5 for hour 12 + assert direct_available_power[0][4 * 24] == 800 + assert indirect_available_power[0][0] == 900 + assert indirect_available_power[0][12] == 450 # Modulation is 0.5 for hour 12 + assert indirect_available_power[0][4 * 24] == 800 + + +def test_cluster_planned_outages_limitation(rng): days = 365 # Maximum 1 planned outage at a time. outage_gen_params = OutageGenerationParameters( @@ -279,28 +433,64 @@ def test_planned_outages_limitation(rng): nominal_power=100, modulation=np.ones(dtype=float, shape=8760), ) + + generator = TimeseriesGenerator(rng=rng, days=days) + results = generator.generate_time_series_for_clusters(cluster, 1) + # No forced outage + npt.assert_equal(results.outage_output.forced_outages.T[0], np.zeros(365)) + npt.assert_equal(results.outage_output.forced_outage_durations.T[0], np.zeros(365)) + # Maximum one planned outage at a time + npt.assert_equal(results.outage_output.planned_outages.T[0][:6], [1, 0, 1, 0, 1, 0]) + npt.assert_equal(results.outage_output.planned_outage_durations.T[0][:6], [2, 0, 2, 0, 2, 0]) + npt.assert_equal(results.outage_output.available_units.T[0][:5], [9, 9, 9, 9, 9]) + # Check available power consistency with available units and modulation + available_power = results.available_power.T + assert available_power[0][0] == 900 + assert available_power[0][4 * 24] == 900 + + +def test_link_planned_outages_limitation(rng): + days = 365 + # Maximum 1 planned outage at a time. + outage_gen_params = OutageGenerationParameters( + unit_count=10, + 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=np.zeros(dtype=float, shape=days), + po_duration=2 * np.ones(dtype=int, shape=days), + po_rate=0.2 * np.ones(dtype=float, shape=days), + npo_min=np.zeros(dtype=int, shape=days), + npo_max=1 * np.ones(dtype=int, shape=days), + ) 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 = ThermalDataGenerator(rng=rng, days=days) - results = generator.generate_time_series_for_clusters(cluster, 1) + + generator = TimeseriesGenerator(rng=rng, days=days) + link_results = generator.generate_time_series_for_links(link, 1) # No 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)) + npt.assert_equal(link_results.outage_output.forced_outages.T[0], np.zeros(365)) + npt.assert_equal(link_results.outage_output.forced_outage_durations.T[0], np.zeros(365)) # Maximum one planned outage at a time - npt.assert_equal(results.planned_outages.T[0][:6], [1, 0, 1, 0, 1, 0]) - npt.assert_equal(results.planned_outage_durations.T[0][:6], [2, 0, 2, 0, 2, 0]) - npt.assert_equal(results.available_units.T[0][:5], [9, 9, 9, 9, 9]) + npt.assert_equal(link_results.outage_output.planned_outages.T[0][:6], [1, 0, 1, 0, 1, 0]) + npt.assert_equal(link_results.outage_output.planned_outage_durations.T[0][:6], [2, 0, 2, 0, 2, 0]) + npt.assert_equal(link_results.outage_output.available_units.T[0][:5], [9, 9, 9, 9, 9]) # Check available power consistency with available units and modulation - available_power = results.available_power.T - assert available_power[0][0] == 900 - assert available_power[0][4 * 24] == 900 + direct_available_power = link_results.direct_available_power.T + indirect_available_power = link_results.indirect_available_power.T + assert direct_available_power[0][0] == 900 + assert direct_available_power[0][4 * 24] == 900 + assert indirect_available_power[0][0] == 900 + assert indirect_available_power[0][4 * 24] == 900 -def test_planned_outages_min_limitation(rng): +def test_cluster_planned_outages_min_limitation(rng): days = 365 # Minimum 2 planned outages at a time outage_gen_params = OutageGenerationParameters( @@ -321,25 +511,59 @@ def test_planned_outages_min_limitation(rng): nominal_power=100, modulation=np.ones(dtype=float, shape=8760), ) + generator = TimeseriesGenerator(rng=rng, days=days) + results = generator.generate_time_series_for_clusters(cluster, 1) + # No forced outage + npt.assert_equal(results.outage_output.forced_outages.T[0], np.zeros(365)) + npt.assert_equal(results.outage_output.forced_outage_durations.T[0], np.zeros(365)) + # Maximum one planned outage at a time + npt.assert_equal(results.outage_output.planned_outages.T[0][:6], [0, 0, 1, 0, 0, 1]) + npt.assert_equal(results.outage_output.planned_outage_durations.T[0][:6], [0, 0, 10, 0, 0, 10]) + npt.assert_equal(results.outage_output.available_units.T[0][:5], [8, 8, 8, 8, 8]) + # Check available power consistency with available units and modulation + available_power = results.available_power.T + assert available_power[0][0] == 800 + assert available_power[0][4 * 24] == 800 + + +def test_link_planned_outages_min_limitation(rng): + days = 365 + # Minimum 2 planned outages at a time + outage_gen_params = OutageGenerationParameters( + unit_count=10, + 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=np.zeros(dtype=float, shape=days), + po_duration=10 * np.ones(dtype=int, shape=days), + po_rate=0.2 * np.ones(dtype=float, shape=days), + npo_min=2 * np.ones(dtype=int, shape=days), + npo_max=5 * np.ones(dtype=int, shape=days), + ) 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 = ThermalDataGenerator(rng=rng, days=days) - results = generator.generate_time_series_for_clusters(cluster, 1) + generator = TimeseriesGenerator(rng=rng, days=days) + link_results = generator.generate_time_series_for_links(link, 1) # No 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)) + npt.assert_equal(link_results.outage_output.forced_outages.T[0], np.zeros(365)) + npt.assert_equal(link_results.outage_output.forced_outage_durations.T[0], np.zeros(365)) # Maximum one planned outage at a time - npt.assert_equal(results.planned_outages.T[0][:6], [0, 0, 1, 0, 0, 1]) - npt.assert_equal(results.planned_outage_durations.T[0][:6], [0, 0, 10, 0, 0, 10]) - npt.assert_equal(results.available_units.T[0][:5], [8, 8, 8, 8, 8]) + npt.assert_equal(link_results.outage_output.planned_outages.T[0][:6], [0, 0, 1, 0, 0, 1]) + npt.assert_equal(link_results.outage_output.planned_outage_durations.T[0][:6], [0, 0, 10, 0, 0, 10]) + npt.assert_equal(link_results.outage_output.available_units.T[0][:5], [8, 8, 8, 8, 8]) # Check available power consistency with available units and modulation - available_power = results.available_power.T - assert available_power[0][0] == 800 - assert available_power[0][4 * 24] == 800 + direct_available_power = link_results.direct_available_power.T + indirect_available_power = link_results.indirect_available_power.T + assert direct_available_power[0][0] == 800 + assert direct_available_power[0][4 * 24] == 800 + assert indirect_available_power[0][0] == 800 + assert indirect_available_power[0][4 * 24] == 800 def test_with_long_fo_and_po_duration(data_directory): @@ -373,14 +597,8 @@ def test_with_long_fo_and_po_duration(data_directory): nominal_power=500, modulation=modulation_matrix, ) - link = LinkCapacity( - outage_gen_params, - nominal_capacity=500, - modulation_indirect=modulation_matrix, - modulation_direct=modulation_matrix, - ) rng = MersenneTwisterRNG(seed=3005489) - generator = ThermalDataGenerator(rng=rng, days=days) + generator = TimeseriesGenerator(rng=rng, days=days) results = generator.generate_time_series_for_clusters(cluster, 10) expected_matrix = np.loadtxt( @@ -404,15 +622,3 @@ def valid_outage_params() -> OutageGenerationParameters: npo_min=np.zeros(dtype=int, shape=days), npo_max=10 * np.ones(dtype=int, shape=days), ) - - -# def test_valid_outage_params(): - - -# def test_invalid_outage_params(): - - -# def test_valid_link_capacity(): - - -# def test_invalid_link_capacity():