feat(ts-gen): add timeseries generation

antarest/core/tasks/model.py

@@ -23,8 +23,8 @@ class TaskType(str, Enum):
     ARCHIVE = "ARCHIVE"
     UNARCHIVE = "UNARCHIVE"
     SCAN = "SCAN"
-    WORKER_TASK = "WORKER_TASK"
     UPGRADE_STUDY = "UPGRADE_STUDY"
+    THERMAL_CLUSTER_SERIES_GENERATION = "THERMAL_CLUSTER_SERIES_GENERATION"
 
 
 class TaskStatus(Enum):

antarest/study/service.py

@@ -118,6 +118,9 @@
 from antarest.study.storage.study_upgrader import StudyUpgrader, check_versions_coherence, find_next_version
 from antarest.study.storage.utils import assert_permission, get_start_date, is_managed, remove_from_cache
 from antarest.study.storage.variantstudy.business.utils import transform_command_to_dto
+from antarest.study.storage.variantstudy.model.command.generate_thermal_cluster_timeseries import (
+    GenerateThermalClusterTimeSeries,
+)
 from antarest.study.storage.variantstudy.model.command.icommand import ICommand
 from antarest.study.storage.variantstudy.model.command.replace_matrix import ReplaceMatrix
 from antarest.study.storage.variantstudy.model.command.update_comments import UpdateComments
@@ -127,7 +130,6 @@
 from antarest.study.storage.variantstudy.model.model import CommandDTO
 from antarest.study.storage.variantstudy.variant_study_service import VariantStudyService
 from antarest.worker.archive_worker import ArchiveTaskArgs
-from antarest.worker.simulator_worker import GenerateTimeseriesTaskArgs
 
 logger = logging.getLogger(__name__)
 
@@ -149,6 +151,51 @@ def get_disk_usage(path: t.Union[str, Path]) -> int:
     return total_size
 
 
+class ThermalClusterTimeSeriesGeneratorTask:
+    """
+    Task to generate thermal clusters time series
+    """
+
+    def __init__(
+        self,
+        _study_id: str,
+        repository: StudyMetadataRepository,
+        storage_service: StudyStorageService,
+        event_bus: IEventBus,
+    ):
+        self._study_id = _study_id
+        self.repository = repository
+        self.storage_service = storage_service
+        self.event_bus = event_bus
+
+    def _generate_timeseries(self) -> None:
+        """Run the task (lock the database)."""
+        command_context = self.storage_service.variant_study_service.command_factory.command_context
+        command = GenerateThermalClusterTimeSeries(command_context=command_context)
+        with db():
+            study = self.repository.one(self._study_id)
+            file_study = self.storage_service.get_storage(study).get_raw(study)
+            execute_or_add_commands(study, file_study, [command], self.storage_service)
+            self.event_bus.push(
+                Event(
+                    type=EventType.STUDY_EDITED,
+                    payload=study.to_json_summary(),
+                    permissions=PermissionInfo.from_study(study),
+                )
+            )
+
+    def run_task(self, notifier: TaskUpdateNotifier) -> TaskResult:
+        msg = f"Generating thermal timeseries for study '{self._study_id}'"
+        notifier(msg)
+        self._generate_timeseries()
+        msg = f"Successfully generated thermal timeseries for study '{self._study_id}'"
+        notifier(msg)
+        return TaskResult(success=True, message=msg)
+
+    # Make `ThermalClusterTimeSeriesGeneratorTask` object callable
+    __call__ = run_task
+
+
 class StudyUpgraderTask:
     """
     Task to perform a study upgrade.
@@ -2385,19 +2432,32 @@ def unarchive_output_task(
 
         return task_id
 
-    def generate_timeseries(self, study: Study, params: RequestParameters) -> None:
-        self._assert_study_unarchived(study)
-        self.task_service.add_worker_task(
-            TaskType.WORKER_TASK,
-            "generate-timeseries",
-            GenerateTimeseriesTaskArgs(
-                study_id=study.id,
-                managed=is_managed(study),
-                study_path=str(study.path),
-                study_version=str(study.version),
-            ).dict(),
-            name=f"Generate timeseries for study {study.id}",
+    def generate_timeseries(self, study: Study, params: RequestParameters) -> str:
+        task_name = f"Generating thermal timeseries for study {study.name} ({study.id})"
+        study_tasks = self.task_service.list_tasks(
+            TaskListFilter(
+                ref_id=study.id,
+                type=[TaskType.THERMAL_CLUSTER_SERIES_GENERATION],
+                status=[TaskStatus.RUNNING, TaskStatus.PENDING],
+            ),
+            RequestParameters(user=DEFAULT_ADMIN_USER),
+        )
+        if len(study_tasks) > 0:
+            raise TaskAlreadyRunning()
+
+        thermal_cluster_timeseries_generation_task = ThermalClusterTimeSeriesGeneratorTask(
+            study.id,
+            repository=self.repository,
+            storage_service=self.storage_service,
+            event_bus=self.event_bus,
+        )
+
+        return self.task_service.add_task(
+            thermal_cluster_timeseries_generation_task,
+            task_name,
+            task_type=TaskType.THERMAL_CLUSTER_SERIES_GENERATION,
             ref_id=study.id,
+            custom_event_messages=None,
             request_params=params,
         )
 

antarest/study/storage/rawstudy/model/filesystem/matrix/matrix.py

@@ -29,6 +29,19 @@ class MatrixFrequency(str, Enum):
     HOURLY = "hourly"
 
 
+def dump_dataframe(df: pd.DataFrame, path: Path, float_format: Optional[str] = "%.6f") -> None:
+    if df.empty:
+        path.write_bytes(b"")
+    else:
+        df.to_csv(
+            path,
+            sep="\t",
+            header=False,
+            index=False,
+            float_format=float_format,
+        )
+
+
 class MatrixNode(LazyNode[Union[bytes, JSON], Union[bytes, JSON], JSON], ABC):
     def __init__(
         self,
@@ -142,13 +155,4 @@ def dump(
             self.config.path.write_bytes(data)
         else:
             df = pd.DataFrame(**data)
-            if df.empty:
-                self.config.path.write_bytes(b"")
-            else:
-                df.to_csv(
-                    self.config.path,
-                    sep="\t",
-                    header=False,
-                    index=False,
-                    float_format="%.6f",
-                )
+            dump_dataframe(df, self.config.path)

antarest/study/storage/variantstudy/business/command_reverter.py

@@ -16,6 +16,9 @@
 from antarest.study.storage.variantstudy.model.command.create_link import CreateLink
 from antarest.study.storage.variantstudy.model.command.create_renewables_cluster import CreateRenewablesCluster
 from antarest.study.storage.variantstudy.model.command.create_st_storage import CreateSTStorage
+from antarest.study.storage.variantstudy.model.command.generate_thermal_cluster_timeseries import (
+    GenerateThermalClusterTimeSeries,
+)
 from antarest.study.storage.variantstudy.model.command.icommand import ICommand
 from antarest.study.storage.variantstudy.model.command.remove_area import RemoveArea
 from antarest.study.storage.variantstudy.model.command.remove_binding_constraint import RemoveBindingConstraint
@@ -317,6 +320,12 @@ def _revert_update_district(
         extractor = base_command.get_command_extractor()
         return [extractor.generate_update_district(base, base_command.id)]
 
+    @staticmethod
+    def _revert_generate_thermal_cluster_timeseries(
+        base_command: GenerateThermalClusterTimeSeries, history: t.List["ICommand"], base: FileStudy
+    ) -> t.List[ICommand]:
+        raise NotImplementedError("The revert function for GenerateThermalClusterTimeSeries is not available")
+
     def revert(
         self,
         base_command: ICommand,

antarest/study/storage/variantstudy/command_factory.py

@@ -12,6 +12,9 @@
 from antarest.study.storage.variantstudy.model.command.create_link import CreateLink
 from antarest.study.storage.variantstudy.model.command.create_renewables_cluster import CreateRenewablesCluster
 from antarest.study.storage.variantstudy.model.command.create_st_storage import CreateSTStorage
+from antarest.study.storage.variantstudy.model.command.generate_thermal_cluster_timeseries import (
+    GenerateThermalClusterTimeSeries,
+)
 from antarest.study.storage.variantstudy.model.command.icommand import ICommand
 from antarest.study.storage.variantstudy.model.command.remove_area import RemoveArea
 from antarest.study.storage.variantstudy.model.command.remove_binding_constraint import RemoveBindingConstraint
@@ -54,6 +57,7 @@
     CommandName.UPDATE_DISTRICT.value: UpdateDistrict,
     CommandName.UPDATE_PLAYLIST.value: UpdatePlaylist,
     CommandName.UPDATE_SCENARIO_BUILDER.value: UpdateScenarioBuilder,
+    CommandName.GENERATE_THERMAL_CLUSTER_TIMESERIES.value: GenerateThermalClusterTimeSeries,
 }
 
 

antarest/study/storage/variantstudy/model/command/common.py

@@ -36,3 +36,4 @@ class CommandName(Enum):
     UPDATE_DISTRICT = "update_district"
     UPDATE_PLAYLIST = "update_playlist"
     UPDATE_SCENARIO_BUILDER = "update_scenario_builder"
+    GENERATE_THERMAL_CLUSTER_TIMESERIES = "generate_thermal_cluster_timeseries"

antarest/study/storage/variantstudy/model/command/generate_thermal_cluster_timeseries.py

@@ -0,0 +1,141 @@
+import logging
+import shutil
+import tempfile
+import typing as t
+from pathlib import Path
+
+import numpy as np
+import pandas as pd
+from antares.tsgen.duration_generator import ProbabilityLaw
+from antares.tsgen.random_generator import MersenneTwisterRNG
+from antares.tsgen.ts_generator import ThermalCluster, ThermalDataGenerator
+
+from antarest.study.storage.rawstudy.model.filesystem.config.model import Area, FileStudyTreeConfig
+from antarest.study.storage.rawstudy.model.filesystem.config.thermal import LocalTSGenerationBehavior
+from antarest.study.storage.rawstudy.model.filesystem.factory import FileStudy
+from antarest.study.storage.rawstudy.model.filesystem.matrix.matrix import dump_dataframe
+from antarest.study.storage.variantstudy.model.command.common import CommandName, CommandOutput
+from antarest.study.storage.variantstudy.model.command.icommand import ICommand, OutputTuple
+from antarest.study.storage.variantstudy.model.model import CommandDTO
+
+logger = logging.getLogger(__name__)
+
+
+MODULATION_CAPACITY_COLUMN = 2
+FO_RATE_COLUMN = 2
+PO_RATE_COLUMN = 3
+
+
+class GenerateThermalClusterTimeSeries(ICommand):
+    """
+    Command used to generate thermal cluster timeseries for an entire study
+    """
+
+    command_name = CommandName.GENERATE_THERMAL_CLUSTER_TIMESERIES
+    version = 1
+
+    def _apply_config(self, study_data: FileStudyTreeConfig) -> OutputTuple:
+        return CommandOutput(status=True, message="Nothing to do"), {}
+
+    def _apply(self, study_data: FileStudy) -> CommandOutput:
+        study_path = study_data.config.study_path
+        with tempfile.TemporaryDirectory(
+            suffix=".thermal_timeseries_gen.tmp", prefix="~", dir=study_path.parent
+        ) as path:
+            tmp_dir = Path(path)
+            try:
+                shutil.copytree(study_path / "input" / "thermal" / "series", tmp_dir, dirs_exist_ok=True)
+                self._build_timeseries(study_data, tmp_dir)
+            except Exception as e:
+                logger.error(f"Unhandled exception when trying to generate thermal timeseries: {e}", exc_info=True)
+                raise
+            else:
+                self._replace_safely_original_files(study_path, tmp_dir)
+                return CommandOutput(status=True, message="All time series were generated successfully")
+
+    def _build_timeseries(self, study_data: FileStudy, tmp_path: Path) -> None:
+        # 1- Get the seed and nb_years to generate
+        # NB: Default seed in IHM Legacy: 5489, default seed in web: 3005489.
+        general_data = study_data.tree.get(["settings", "generaldata"], depth=3)
+        thermal_seed = general_data["seeds - Mersenne Twister"]["seed-tsgen-thermal"]
+        nb_years = general_data["general"]["nbtimeseriesthermal"]
+        # 2 - Build the generator
+        rng = MersenneTwisterRNG(seed=thermal_seed)
+        generator = ThermalDataGenerator(rng=rng, days=365)
+        # 3- Loop through areas in alphabetical order
+        areas: t.Dict[str, Area] = study_data.config.areas
+        sorted_areas = {k: areas[k] for k in sorted(areas)}
+        for area_id, area in sorted_areas.items():
+            # 4- Loop through thermal clusters in alphabetical order
+            sorted_thermals = sorted(area.thermals, key=lambda x: x.id)
+            for thermal in sorted_thermals:
+                # 5 - Filters out clusters with no generation
+                if thermal.gen_ts == LocalTSGenerationBehavior.FORCE_NO_GENERATION:
+                    continue
+                # 6- Build the cluster
+                url = ["input", "thermal", "prepro", area_id, thermal.id.lower(), "modulation"]
+                matrix = study_data.tree.get_node(url)
+                matrix_df = matrix.parse(return_dataframe=True)  # type: ignore
+                modulation_capacity = matrix_df[MODULATION_CAPACITY_COLUMN].to_numpy()
+                url = ["input", "thermal", "prepro", area_id, thermal.id.lower(), "data"]
+                matrix = study_data.tree.get_node(url)
+                matrix_df = matrix.parse(return_dataframe=True)  # type: ignore
+                fo_duration, po_duration, fo_rate, po_rate, npo_min, npo_max = [
+                    np.array(matrix_df[i], dtype=float if i in [FO_RATE_COLUMN, PO_RATE_COLUMN] else int)
+                    for i in matrix_df.columns
+                ]
+                cluster = ThermalCluster(
+                    unit_count=thermal.unit_count,
+                    nominal_power=thermal.nominal_capacity,
+                    modulation=modulation_capacity,
+                    fo_law=ProbabilityLaw(thermal.law_forced.value.upper()),
+                    fo_volatility=thermal.volatility_forced,
+                    po_law=ProbabilityLaw(thermal.law_planned.value.upper()),
+                    po_volatility=thermal.volatility_planned,
+                    fo_duration=fo_duration,
+                    fo_rate=fo_rate,
+                    po_duration=po_duration,
+                    po_rate=po_rate,
+                    npo_min=npo_min,
+                    npo_max=npo_max,
+                )
+                # 7- Generate the time-series
+                results = generator.generate_time_series(cluster, nb_years)
+                generated_matrix = results.available_power
+                # 8- Write the matrix inside the input folder.
+                df = pd.DataFrame(data=generated_matrix, dtype=int)
+                target_path = self._build_matrix_path(tmp_path / area_id / thermal.id.lower())
+                dump_dataframe(df, target_path, None)
+
+    def to_dto(self) -> CommandDTO:
+        return CommandDTO(action=self.command_name.value, args={})
+
+    def match_signature(self) -> str:
+        return str(self.command_name.value)
+
+    def match(self, other: "ICommand", equal: bool = False) -> bool:
+        # Only used inside the cli app that no one uses I believe.
+        if not isinstance(other, GenerateThermalClusterTimeSeries):
+            return False
+        return True
+
+    def _create_diff(self, other: "ICommand") -> t.List["ICommand"]:
+        # Only used inside the cli app that no one uses I believe.
+        raise NotImplementedError()
+
+    def get_inner_matrices(self) -> t.List[str]:
+        # This is used to get used matrices and not remove them inside the garbage collector loop.
+        return []
+
+    @staticmethod
+    def _replace_safely_original_files(study_path: Path, tmp_path: Path) -> None:
+        original_path = study_path / "input" / "thermal" / "series"
+        shutil.rmtree(original_path)
+        tmp_path.rename(original_path)
+
+    @staticmethod
+    def _build_matrix_path(matrix_path: Path) -> Path:
+        real_path = matrix_path / "series.txt"
+        if not real_path.exists():
+            (matrix_path / "series.txt.link").rename(real_path)
+        return real_path

antarest/study/web/study_data_blueprint.py

@@ -1759,7 +1759,13 @@ def set_advanced_parameters(
     def generate_timeseries(
         uuid: str,
         current_user: JWTUser = Depends(auth.get_current_user),
-    ) -> t.Any:
+    ) -> str:
+        """
+        Generates time-series for thermal clusters and put them inside input data.
+
+        Args:
+        - `uuid`: The UUID of the study.
+        """
         logger.info(
             f"Generating timeseries for study {uuid}",
             extra={"user": current_user.id},

requirements.txt

@@ -1,5 +1,6 @@
 Antares-Launcher~=1.3.2
 antares-study-version~=1.0.3
+antares-timeseries-generation~=0.1.5
 
 alembic~=1.7.5
 asgi-ratelimit[redis]==0.7.0