Add support for simulator routines

CHANGELOG.md

@@ -19,6 +19,7 @@ Changes are grouped as follows
 
 ## [Unreleased]
 ### Added
+- Support for the `/simulators/routines` API endpoints.
 - Support for the `/simulators/models` and `/simulators/models/revisions` API endpoints.
 - Support for the `/simulators` and `/simulators/integration` API endpoints.
 

cognite/client/_api/simulators/__init__.py

@@ -5,6 +5,7 @@
 
 from cognite.client._api.simulators.integrations import SimulatorIntegrationsAPI
 from cognite.client._api.simulators.models import SimulatorModelsAPI
+from cognite.client._api.simulators.routines import SimulatorRoutinesAPI
 from cognite.client._api_client import APIClient
 from cognite.client._constants import DEFAULT_LIMIT_READ
 from cognite.client.data_classes.simulators.simulators import Simulator, SimulatorList
@@ -22,6 +23,7 @@ def __init__(self, config: ClientConfig, api_version: str | None, cognite_client
         super().__init__(config, api_version, cognite_client)
         self.integrations = SimulatorIntegrationsAPI(config, api_version, cognite_client)
         self.models = SimulatorModelsAPI(config, api_version, cognite_client)
+        self.routines = SimulatorRoutinesAPI(config, api_version, cognite_client)
         self._warning = FeaturePreviewWarning(
             api_maturity="General Availability", sdk_maturity="alpha", feature_name="Simulators"
         )

cognite/client/_api/simulators/models.py

@@ -7,9 +7,8 @@
 from cognite.client._api_client import APIClient
 from cognite.client._constants import DEFAULT_LIMIT_READ
 from cognite.client.data_classes._base import CogniteFilter
-from cognite.client.data_classes.simulators.filters import SimulatorModelsFilter
+from cognite.client.data_classes.simulators.filters import PropertySort, SimulatorModelsFilter
 from cognite.client.data_classes.simulators.models import (
-    CreatedTimeSort,
     SimulatorModel,
     SimulatorModelList,
     SimulatorModelUpdate,
@@ -41,14 +40,14 @@ def list(
         self,
         limit: int = DEFAULT_LIMIT_READ,
         filter: SimulatorModelsFilter | dict[str, Any] | None = None,
-        sort: CreatedTimeSort | None = None,
+        sort: PropertySort | None = None,
     ) -> SimulatorModelList:
         """`Filter simulator models <https://developer.cognite.com/api#tag/Simulator-Models/operation/filter_simulator_models_simulators_models_list_post>`_
         Retrieves a list of simulator models that match the given criteria
         Args:
             limit (int): Maximum number of results to return. Defaults to 25. Set to -1, float(“inf”) or None to return all items.
             filter (SimulatorModelsFilter | dict[str, Any] | None): Filter to apply.
-            sort (CreatedTimeSort | None): The criteria to sort by.
+            sort (PropertySort | None): The criteria to sort by.
         Returns:
             SimulatorModelList: List of simulator models
 
@@ -59,10 +58,11 @@ def list(
                 >>> res = client.simulators.models.list()
 
             Specify filter and sort order:
-                >>> from cognite.client.data_classes.simulators import SimulatorModelsFilter, CreatedTimeSort
+                >>> from cognite.client.data_classes.simulators import SimulatorModelsFilter
+                >>> from cognite.client.data_classes.simulators.filters import PropertySort
                 >>> res = client.simulators.models.list(
                 ...     filter=SimulatorModelsFilter(simulator_external_ids=["simulator_external_id"]),
-                ...     sort=CreatedTimeSort(order="asc")
+                ...     sort=PropertySort(order="asc")
                 ... )
 
         """
@@ -72,7 +72,7 @@ def list(
             limit=limit,
             resource_cls=SimulatorModel,
             list_cls=SimulatorModelList,
-            sort=[CreatedTimeSort.load(sort).dump()] if sort else None,
+            sort=[PropertySort.load(sort).dump()] if sort else None,
             filter=filter.dump(camel_case=True) if isinstance(filter, CogniteFilter) else filter,
         )
 

cognite/client/_api/simulators/models_revisions.py

@@ -6,9 +6,8 @@
 from cognite.client._api_client import APIClient
 from cognite.client._constants import DEFAULT_LIMIT_READ
 from cognite.client.data_classes._base import CogniteFilter
-from cognite.client.data_classes.simulators.filters import SimulatorModelRevisionsFilter
+from cognite.client.data_classes.simulators.filters import PropertySort, SimulatorModelRevisionsFilter
 from cognite.client.data_classes.simulators.models import (
-    CreatedTimeSort,
     SimulatorModelRevision,
     SimulatorModelRevisionList,
     SimulatorModelRevisionWrite,
@@ -36,14 +35,14 @@ def __init__(self, config: ClientConfig, api_version: str | None, cognite_client
     def list(
         self,
         limit: int = DEFAULT_LIMIT_READ,
-        sort: CreatedTimeSort | None = None,
+        sort: PropertySort | None = None,
         filter: SimulatorModelRevisionsFilter | dict[str, Any] | None = None,
     ) -> SimulatorModelRevisionList:
         """`Filter simulator model revisions <https://developer.cognite.com/api#tag/Simulator-Models/operation/filter_simulator_model_revisions_simulators_models_revisions_list_post>`_
         Retrieves a list of simulator model revisions that match the given criteria
         Args:
             limit (int): Maximum number of results to return. Defaults to 25. Set to -1, float(“inf”) or None to return all items.
-            sort (CreatedTimeSort | None): The criteria to sort by.
+            sort (PropertySort | None): The criteria to sort by.
             filter (SimulatorModelRevisionsFilter | dict[str, Any] | None): Filter to apply.
         Returns:
             SimulatorModelRevisionList: List of simulator model revisions
@@ -54,10 +53,11 @@ def list(
                 >>> res = client.simulators.models.revisions.list()
 
             Specify filter and sort order:
-                >>> from cognite.client.data_classes.simulators import SimulatorModelRevisionsFilter, CreatedTimeSort
+                >>> from cognite.client.data_classes.simulators import SimulatorModelRevisionsFilter
+                >>> from cognite.client.data_classes.simulators.filters import PropertySort
                 >>> res = client.simulators.models.revisions.list(
                 ...     filter=SimulatorModelRevisionsFilter(model_external_ids=["model_external_id"]),
-                ...     sort=CreatedTimeSort(order="asc")
+                ...     sort=PropertySort(order="asc")
                 ... )
         """
         self._warning.warn()
@@ -66,7 +66,7 @@ def list(
             limit=limit,
             resource_cls=SimulatorModelRevision,
             list_cls=SimulatorModelRevisionList,
-            sort=[CreatedTimeSort.load(sort).dump()] if sort else None,
+            sort=[PropertySort.load(sort).dump()] if sort else None,
             filter=filter.dump(camel_case=True) if isinstance(filter, CogniteFilter) else filter,
         )
 

cognite/client/_api/simulators/routines.py

@@ -0,0 +1,213 @@
+from __future__ import annotations
+
+from collections.abc import Iterator, Sequence
+from typing import TYPE_CHECKING, overload
+
+from cognite.client._api_client import APIClient
+from cognite.client._constants import DEFAULT_LIMIT_READ
+from cognite.client.data_classes.simulators.filters import PropertySort, SimulatorRoutinesFilter
+from cognite.client.data_classes.simulators.routines import (
+    SimulatorRoutine,
+    SimulatorRoutineList,
+    SimulatorRoutineWrite,
+)
+from cognite.client.utils._experimental import FeaturePreviewWarning
+from cognite.client.utils._identifier import IdentifierSequence
+from cognite.client.utils._validation import assert_type
+from cognite.client.utils.useful_types import SequenceNotStr
+
+if TYPE_CHECKING:
+    from cognite.client import ClientConfig, CogniteClient
+
+
+class SimulatorRoutinesAPI(APIClient):
+    _RESOURCE_PATH = "/simulators/routines"
+
+    def __init__(self, config: ClientConfig, api_version: str | None, cognite_client: CogniteClient) -> None:
+        super().__init__(config, api_version, cognite_client)
+        self._warning = FeaturePreviewWarning(
+            api_maturity="General Availability", sdk_maturity="alpha", feature_name="Simulators"
+        )
+
+    def __iter__(self) -> Iterator[SimulatorRoutine]:
+        """Iterate over simulator routines
+
+        Fetches simulator routines as they are iterated over, so you keep a limited number of simulator routines in memory.
+
+        Returns:
+            Iterator[SimulatorRoutine]: yields Simulator routines one by one.
+        """
+        return self()
+
+    @overload
+    def __call__(
+        self,
+        chunk_size: int,
+        model_external_ids: Sequence[str] | None = None,
+        simulator_integration_external_ids: Sequence[str] | None = None,
+        limit: int | None = None,
+    ) -> Iterator[SimulatorRoutineList]: ...
+
+    @overload
+    def __call__(
+        self,
+        chunk_size: None = None,
+        model_external_ids: Sequence[str] | None = None,
+        simulator_integration_external_ids: Sequence[str] | None = None,
+        limit: int | None = None,
+    ) -> Iterator[SimulatorRoutine]: ...
+
+    def __call__(
+        self,
+        chunk_size: int | None = None,
+        model_external_ids: Sequence[str] | None = None,
+        simulator_integration_external_ids: Sequence[str] | None = None,
+        limit: int | None = None,
+    ) -> Iterator[SimulatorRoutine] | Iterator[SimulatorRoutineList]:
+        """Iterate over simulator routines
+
+        Fetches simulator routines as they are iterated over, so you keep a limited number of simulator routines in memory.
+
+        Args:
+            chunk_size (int | None): Number of simulator routines to return in each chunk. Defaults to yielding one simulator routine a time.
+            model_external_ids (Sequence[str] | None): Filter on model external ids.
+            simulator_integration_external_ids (Sequence[str] | None): Filter on simulator integration external ids.
+            limit (int | None): Maximum number of simulator routines to return. Defaults to return all items.
+
+        Returns:
+            Iterator[SimulatorRoutine] | Iterator[SimulatorRoutineList]: yields SimulatorRoutine one by one if chunk is not specified, else SimulatorRoutineList objects.
+        """
+        routines_filter = SimulatorRoutinesFilter(
+            model_external_ids=model_external_ids,
+            simulator_integration_external_ids=simulator_integration_external_ids,
+        )
+        return self._list_generator(
+            list_cls=SimulatorRoutineList,
+            resource_cls=SimulatorRoutine,
+            method="POST",
+            filter=routines_filter.dump(),
+            chunk_size=chunk_size,
+            limit=limit,
+        )
+
+    @overload
+    def create(self, routine: Sequence[SimulatorRoutineWrite]) -> SimulatorRoutineList: ...
+
+    @overload
+    def create(self, routine: SimulatorRoutineWrite) -> SimulatorRoutineList: ...
+
+    def create(
+        self,
+        routine: SimulatorRoutineWrite | Sequence[SimulatorRoutineWrite],
+    ) -> SimulatorRoutine | SimulatorRoutineList:
+        """`Create simulator routine <https://developer.cognite.com/api#tag/Simulator-Routines/operation/create_simulator_routine_simulators_routines_post>`_
+        You can create an arbitrary number of simulator routines.
+        Args:
+            routine (SimulatorRoutineWrite | Sequence[SimulatorRoutineWrite]): Simulator routines to create.
+        Returns:
+            SimulatorRoutine | SimulatorRoutineList: Created simulator routine(s)
+        Examples:
+            Create new simulator routines:
+                >>> from cognite.client import CogniteClient
+                >>> from cognite.client.data_classes.simulators.routines import SimulatorRoutineWrite
+                >>> client = CogniteClient()
+                >>> routines = [
+                ...     SimulatorRoutineWrite(
+                ...         name="routine1",
+                ...         external_id="routine_ext_id",
+                ...         simulator_integration_external_id="integration_ext_id",
+                ...         model_external_id="model_ext_id",
+                ...     ),
+                ...     SimulatorRoutineWrite(
+                ...         name="routine2",
+                ...         external_id="routine_ext_id_2",
+                ...         simulator_integration_external_id="integration_ext_id_2",
+                ...         model_external_id="model_ext_id_2",
+                ...     )
+                ... ]
+                >>> res = client.simulators.routines.create(routines)
+        """
+        self._warning.warn()
+        assert_type(routine, "simulator_routines", [SimulatorRoutineWrite, Sequence])
+
+        return self._create_multiple(
+            list_cls=SimulatorRoutineList,
+            resource_cls=SimulatorRoutine,
+            items=routine,
+            input_resource_cls=SimulatorRoutineWrite,
+            resource_path=self._RESOURCE_PATH,
+        )
+
+    def delete(
+        self,
+        id: int | Sequence[int] | None = None,
+        external_ids: str | SequenceNotStr[str] | SequenceNotStr[str] | None = None,
+    ) -> None:
+        """`Delete one or more routines <https://developer.cognite.com/api#tag/Simulator-Routines/operation/delete_simulator_routine_simulators_routines_delete_post>`_
+
+        Args:
+            id (int | Sequence[int] | None): ids (or sequence of ids) for the routine(s) to delete.
+            external_ids (str | SequenceNotStr[str] | SequenceNotStr[str] | None): external ids (or sequence of external ids) for the routine(s) to delete.
+
+        Examples:
+
+            Delete routines by id or external id:
+
+                >>> from cognite.client import CogniteClient
+                >>> client = CogniteClient()
+                >>> client.simulators.routines.delete(id=[1,2,3], external_id="foo")
+        """
+        self._warning.warn()
+        self._delete_multiple(
+            identifiers=IdentifierSequence.load(ids=id, external_ids=external_ids),
+            wrap_ids=True,
+        )
+
+    def list(
+        self,
+        limit: int = DEFAULT_LIMIT_READ,
+        model_external_ids: Sequence[str] | None = None,
+        simulator_integration_external_ids: Sequence[str] | None = None,
+        sort: PropertySort | None = None,
+    ) -> SimulatorRoutineList:
+        """`Filter simulator routines <https://developer.cognite.com/api#tag/Simulator-Routines/operation/filter_simulator_routines_simulators_routines_list_post>`_
+
+        Retrieves a list of simulator routines that match the given criteria
+
+        Args:
+            limit (int): Maximum number of results to return. Defaults to 25. Set to -1, float(“inf”) or None to return all items.
+            model_external_ids (Sequence[str] | None): Filter on model external ids.
+            simulator_integration_external_ids (Sequence[str] | None): Filter on simulator integration external ids.
+            sort (PropertySort | None): The criteria to sort by.
+
+        Returns:
+            SimulatorRoutineList: List of simulator routines
+
+        Examples:
+            List simulator routines:
+                >>> from cognite.client import CogniteClient
+                >>> client = CogniteClient()
+                >>> res = client.simulators.routines.list()
+
+            Specify filter and sort order:
+                >>> from cognite.client.data_classes.simulators.filters import PropertySort
+                >>> res = client.simulators.routines.list(
+                ...     simulator_integration_external_ids=["integration_ext_id"],
+                ...     sort=PropertySort(order="asc")
+                ... )
+
+        """
+        routines_filter = SimulatorRoutinesFilter(
+            model_external_ids=model_external_ids,
+            simulator_integration_external_ids=simulator_integration_external_ids,
+        )
+        self._warning.warn()
+        return self._list(
+            limit=limit,
+            method="POST",
+            url_path="/simulators/routines/list",
+            resource_cls=SimulatorRoutine,
+            list_cls=SimulatorRoutineList,
+            sort=[PropertySort.load(sort).dump()] if sort else None,
+            filter=routines_filter.dump(),
+        )

cognite/client/data_classes/simulators/__init__.py

@@ -1,13 +1,12 @@
 from __future__ import annotations
 
 from cognite.client.data_classes.simulators.filters import (
+    PropertySort,
     SimulatorIntegrationFilter,
     SimulatorModelRevisionsFilter,
     SimulatorModelsFilter,
 )
 from cognite.client.data_classes.simulators.models import (
-    CreatedTimeSort,
-    PropertySort,
     SimulatorModel,
     SimulatorModelList,
     SimulatorModelRevision,
@@ -27,7 +26,7 @@
 )
 
 __all__ = [
-    "CreatedTimeSort",
+    "PropertySort",
     "PropertySort",
     "Simulator",
     "SimulatorIntegration",