upload ccurves

add support to upload custom curves

Author: robcalon

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "pyetm"
-version = "1.4.1"
+version = "1.4.2"
 
 description = "Python-ETM Connector"
 authors = [{name = "Rob Calon", email = "[email protected]"}]

src/pyetm/myc/model.py

@@ -6,7 +6,7 @@
 from dataclasses import dataclass
 from os import PathLike
 from pathlib import Path
-from typing import get_args, overload, Hashable, Literal, Sequence, TypedDict
+from typing import get_args, overload, Hashable, Literal, Iterable, Sequence, TypedDict
 
 import logging
 
@@ -376,7 +376,7 @@ def set_parameters(
         scenarios: ScenarioSlice | None = None,
         **kwargs
     ) -> None:
-        """can be done threaded as well?"""
+        """set parameters"""
 
         # collect scenarios
         scenarios = self.slice_cases(scenarios=scenarios)
@@ -392,25 +392,59 @@ def set_parameters(
         if errors:
             raise KeyError(f"unknown cases in dataframe: '{errors}'")
 
-        return self.pool.set_parameters(scenarios, parameters, **kwargs)
+        self.pool.set_parameters(scenarios, parameters, **kwargs)
+
+    def upload_custom_curves(
+        self,
+        ccurves: pd.Series | pd.DataFrame,
+        scenarios: ScenarioSlice | None = None,
+        **kwargs
+    ) -> None:
+        """upload custom curves"""
+        scenarios = self.slice_cases(scenarios=scenarios)
+        self.pool.upload_custom_curves(ccurves=ccurves, scenarios=scenarios, **kwargs)
+
+    def delete_custom_curves(
+        self,
+        keys: str | Iterable[str] | None = None,
+        scenarios: ScenarioSlice | None = None,
+        **kwargs
+    ) -> None:
+        """delete custom curves"""
+        scenarios = self.slice_cases(scenarios=scenarios)
+        self.pool.delete_custom_curves(keys=keys, scenarios=scenarios, **kwargs)
+
+    def set_custom_curves(
+        self,
+        ccurves: pd.Series | pd.DataFrame,
+        scenarios: ScenarioSlice | None = None,
+        **kwargs
+    ) -> None:
+        """set custom curves"""
+        scenarios = self.slice_cases(scenarios=scenarios)
+        self.pool.set_custom_curves(ccurves=ccurves, scenarios=scenarios, **kwargs)
 
     def get_gqueries(
         self,
-        gqueries: Sequence[str] | pd.Series | None = None,
+        gqueries: Iterable[str] | pd.Series | None = None,
         scenarios: ScenarioSlice | None = None,
         **kwargs
     ) -> pd.DataFrame:
         """get gqueries"""
 
-        # collect gqueries and scenarios
+        # collect gqueries
         gqueries = gqueries if gqueries else self.gqueries
+        if gqueries is None:
+            raise ValueError("no gqueries specified")
+
+        # collect scenarios
         scenarios = self.slice_cases(scenarios=scenarios)
 
         return self.pool.get_gqueries(scenarios=scenarios, gqueries=gqueries, **kwargs)
 
     def get_price_curves(
         self,
-        carriers: Carrier | Sequence[Carrier] | None = None,
+        carriers: Carrier | Iterable[Carrier] | None = None,
         scenarios: ScenarioSlice | None = None,
         **kwargs
     ) -> pd.DataFrame:

src/pyetm/myc/pool.py

@@ -230,15 +230,56 @@ def get_carrier_curves(
     #     with pool.get_client_from_session_id(scenario_id) as client:
     #         pass
 
+    @staticmethod
+    def upload_custom_curves(
+        pool: ClientPool,
+        scenario_id: int,
+        ccurves: pd.Series | pd.DataFrame,
+    ) -> pd.Series:
+        """upload custom curves"""
+
+        with pool.get_client_from_session_id(scenario_id) as client:
+            client.upload_custom_curves(ccurves=ccurves)
+
+        return pd.Series(name=scenario_id)
+
+    @staticmethod
+    def delete_custom_curves(
+        pool: ClientPool,
+        scenario_id: int,
+        keys: str | Iterable[str] | None = None,
+    ) -> pd.Series:
+        """delete custom curves"""
+
+        with pool.get_client_from_session_id(scenario_id) as client:
+            client.delete_custom_curves(keys=keys)
+
+        return pd.Series(name=scenario_id)
+
+    @staticmethod
+    def set_custom_curves(
+        pool: ClientPool,
+        scenario_id: int,
+        ccurves: pd.Series | pd.DataFrame,
+    ) -> pd.Series:
+        """set custom curves"""
+
+        with pool.get_client_from_session_id(scenario_id) as client:
+            client.set_custom_curves(ccurves=ccurves)
+
+        return pd.Series(name=scenario_id)
+
+
+
 class ClientPool:
     """pool of reusable clients"""
 
     def __init__(
         self,
         maxsize: int | None = None,
-        clients: Iterable[Client] | None = None,
+        clients: list[Client] | None = None,
         **kwargs
-    ):
+    ) -> None:
 
         if maxsize is None:
             maxsize = len(clients) if clients else 3
@@ -454,3 +495,45 @@ def get_carrier_curves(
             invert_sign_convention=invert_sign_convention,
             **kwargs,
         )
+
+    def upload_custom_curves(
+        self,
+        scenarios: Scenarios,
+        ccurves: pd.Series | pd.DataFrame,
+        **kwargs
+    ) -> None:
+        """upload custom curves"""
+        self.call_threaded(
+            func=self.tasks.upload_custom_curves,
+            scenarios=scenarios,
+            curves=ccurves,
+            **kwargs,
+        )
+
+    def delete_custom_curves(
+        self,
+        scenarios: Scenarios,
+        keys: str | Iterable[str] | None = None,
+        **kwargs
+    ) -> None:
+        """delete custom curves"""
+        self.call_threaded(
+            func=self.tasks.delete_custom_curves,
+            scenarios=scenarios,
+            keys=keys,
+            **kwargs,
+        )
+
+    def set_custom_curves(
+        self,
+        scenarios: Scenarios,
+        ccurves: pd.Series | pd.DataFrame,
+        **kwargs
+    ) -> None:
+        """set custom curves"""
+        self.call_threaded(
+            func=self.tasks.set_custom_curves,
+            scenarios=scenarios,
+            curves=ccurves,
+            **kwargs,
+        )