Simulator resources creation

cognite/client/_api/simulators/simulator_models.py

@@ -1,18 +1,23 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any
+from collections.abc import Sequence
+from typing import TYPE_CHECKING, Any, overload
 
 from cognite.client._api_client import APIClient
 from cognite.client._constants import DEFAULT_LIMIT_READ
 from cognite.client.data_classes.simulators.filters import SimulatorModelRevisionsFilter, SimulatorModelsFilter
 from cognite.client.data_classes.simulators.simulators import (
     SimulatorModel,
+    SimulatorModelCore,
     SimulatorModelList,
     SimulatorModelRevision,
     SimulatorModelRevisionList,
+    SimulatorModelWrite,
 )
 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
@@ -160,3 +165,69 @@ def retrieve_revision(self, id: int | None = None, external_id: str | None = Non
             resource_path="/simulators/models/revisions",
             headers={"cdf-version": "beta"},
         )
+
+    @overload
+    def create_model(self, model: Sequence[SimulatorModel]) -> SimulatorModelList: ...
+
+    @overload
+    def create_model(self, model: SimulatorModel | SimulatorModelWrite) -> SimulatorModelList: ...
+
+    def create_model(
+        self, model: SimulatorModel | SimulatorModelWrite | Sequence[SimulatorModel] | Sequence[SimulatorModelWrite]
+    ) -> SimulatorModel | SimulatorModelList:
+        """`Create one or more simulator models. <https://api-docs.cognite.com/20230101-beta/tag/Simulator-Models>`_
+
+        You can create an arbitrary number of simulator models, and the SDK will split the request into multiple requests.
+
+        Args:
+            model (SimulatorModel | SimulatorModelWrite | Sequence[SimulatorModel] | Sequence[SimulatorModelWrite]): No description.
+
+        Returns:
+            SimulatorModel | SimulatorModelList: Created simulator model(s)
+
+        Examples:
+
+            Create new simulator models::
+
+                >>> from cognite.client import CogniteClient
+                >>> from cognite.client.data_classes import SimulatorModelWrite
+                >>> client = CogniteClient()
+                >>> models = [SimulatorModelWrite(name="model1"), SimulatorModelWrite(name="model2")]
+                >>> res = client.simulators.models.create(models)
+
+        """
+        assert_type(model, "simulator_model", [SimulatorModelCore, Sequence])
+
+        return self._create_multiple(
+            list_cls=SimulatorModelList,
+            resource_cls=SimulatorModel,
+            items=model,
+            input_resource_cls=SimulatorModelWrite,
+            resource_path="/simulators/models",
+            headers={"cdf-version": "beta"},
+        )
+
+    def delete_models(
+        self,
+        id: int | Sequence[int] | None = None,
+        external_id: str | SequenceNotStr[str] | None = None,
+    ) -> None:
+        """`Delete one or more models <https://api-docs.cognite.com/20230101-beta/tag/Simulator-Models>`_
+
+        Args:
+            id (int | Sequence[int] | None): Id or list of ids
+            external_id (str | SequenceNotStr[str] | None): External ID or list of external ids
+        Examples:
+
+            Delete models by id or external id::
+
+                >>> from cognite.client import CogniteClient
+                >>> client = CogniteClient()
+                >>> client.simulators.delete_models(id=[1,2,3], external_id="3")
+        """
+        self._delete_multiple(
+            identifiers=IdentifierSequence.load(ids=id, external_ids=external_id),
+            wrap_ids=True,
+            resource_path="/simulators/models",
+            headers={"cdf-version": "beta"},
+        )

cognite/client/data_classes/simulators/simulators.py

@@ -3,7 +3,7 @@
 from abc import ABC
 from collections.abc import Sequence
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, TypeVar
 
 from typing_extensions import Self
 
@@ -843,7 +843,7 @@ def dump(self, camel_case: bool = True) -> dict[str, Any]:
         return output
 
 
-class SimulatorModel(CogniteResource):
+class SimulatorModelCore(WriteableCogniteResource["SimulatorModelWrite"], ABC):
     """
     The simulator model resource represents an asset modeled in a simulator.
     This asset could range from a pump or well to a complete processing facility or refinery.
@@ -862,55 +862,158 @@ class SimulatorModel(CogniteResource):
     This is the read/response format of a simulator model.
 
     Args:
-        id (int): A unique id of a simulator model
-        external_id (str): External id of the simulator model
-        simulator_external_id (str): External id of the associated simulator
-        data_set_id (int): The id of the dataset associated with the simulator model
-        created_time (int): The time when the simulator model was created
-        last_updated_time (int): The time when the simulator model was last updated
-        name (str): The name of the simulator model
-        type_key (str | None): The type key of the simulator model
+        external_id (str | None): External id of the simulator model
+        simulator_external_id (str | None): External id of the associated simulator
+        data_set_id (int | None): The id of the dataset associated with the simulator model
+        name (str | None): The name of the simulator model
+        type (str | None): The type key of the simulator model
         description (str | None): The description of the simulator model
     """
 
     def __init__(
         self,
-        id: int,
-        external_id: str,
-        simulator_external_id: str,
-        data_set_id: int,
-        created_time: int,
-        last_updated_time: int,
-        name: str,
-        type_key: str | None = None,
+        external_id: str | None = None,
+        simulator_external_id: str | None = None,
+        data_set_id: int | None = None,
+        name: str | None = None,
+        type: str | None = None,
         description: str | None = None,
     ) -> None:
-        self.id = id
         self.external_id = external_id
         self.simulator_external_id = simulator_external_id
         self.data_set_id = data_set_id
-        self.created_time = created_time
-        self.last_updated_time = last_updated_time
         self.name = name
-        self.type_key = type_key
+        self.type = type
         self.description = description
 
     @classmethod
-    def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self:
+    def _load(
+        cls: type[T_SimulatorModel], resource: dict[str, Any], cognite_client: CogniteClient | None = None
+    ) -> T_SimulatorModel:
+        instance = super()._load(resource, cognite_client)
+        return instance
+
+    def dump(self, camel_case: bool = True) -> dict[str, Any]:
+        return super().dump(camel_case=camel_case)
+
+
+T_SimulatorModel = TypeVar("T_SimulatorModel", bound=SimulatorModelCore)
+
+
+class SimulatorModelWrite(SimulatorModelCore):
+    def __init__(
+        self,
+        external_id: str | None = None,
+        simulator_external_id: str | None = None,
+        data_set_id: int | None = None,
+        name: str | None = None,
+        type: str | None = None,
+        description: str | None = None,
+    ) -> None:
+        super().__init__(
+            external_id=external_id,
+            simulator_external_id=simulator_external_id,
+            data_set_id=data_set_id,
+            name=name,
+            type=type,
+            description=description,
+        )
+
+    @classmethod
+    def _load(cls, resource: dict, cognite_client: CogniteClient | None = None) -> SimulatorModelWrite:
         return cls(
-            id=resource["id"],
             external_id=resource["externalId"],
             simulator_external_id=resource["simulatorExternalId"],
-            name=resource["name"],
             data_set_id=resource["dataSetId"],
-            created_time=resource["createdTime"],
-            last_updated_time=resource["lastUpdatedTime"],
-            type_key=resource.get("typeKey"),
+            name=resource["name"],
+            type=resource.get("typeKey"),
             description=resource.get("description"),
         )
 
+    def as_write(self) -> SimulatorModelWrite:
+        """Returns self."""
+        return self
+
+
+class SimulatorModel(SimulatorModelCore):
+    """
+    The simulator model resource represents an asset modeled in a simulator.
+    This asset could range from a pump or well to a complete processing facility or refinery.
+    The simulator model is the root of its associated revisions, routines, runs, and results.
+    The dataset assigned to a model is inherited by its children. Deleting a model also deletes all its children, thereby
+    maintaining the integrity and hierarchy of the simulation data.
+
+    Simulator model revisions track changes and updates to a simulator model over time.
+    Each revision ensures that modifications to models are traceable and allows users to understand the evolution of a given model.
+
+    Limitations:
+        - A project can have a maximum of 1000 simulator models
+        - Each simulator model can have a maximum of 200 revisions
+
+
+    This is the read/response format of a simulator model.
+
+    Args:
+        external_id (str | None): External id of the simulator model
+        simulator_external_id (str | None): External id of the associated simulator
+        data_set_id (int | None): The id of the dataset associated with the simulator model
+        name (str | None): The name of the simulator model
+        id (int | None): A unique id of a simulator model
+        type (str | None): The type key of the simulator model
+        description (str | None): The description of the simulator model
+        created_time (int | None): The time when the simulator model was created
+        last_updated_time (int | None): The time when the simulator model was last updated
+    """
+
+    def __init__(
+        self,
+        external_id: str | None = None,
+        simulator_external_id: str | None = None,
+        data_set_id: int | None = None,
+        name: str | None = None,
+        id: int | None = None,
+        type: str | None = None,
+        description: str | None = None,
+        created_time: int | None = None,
+        last_updated_time: int | None = None,
+    ) -> None:
+        super().__init__(
+            external_id=external_id,
+            simulator_external_id=simulator_external_id,
+            data_set_id=data_set_id,
+            name=name,
+            type=type,
+            description=description,
+        )
+        # id/created_time/last_updated_time are required when using the class to read,
+        # but don't make sense passing in when creating a new object. So in order to make the typing
+        # correct here (i.e. int and not Optional[int]), we force the type to be int rather than
+        # Optional[int].
+        self.id: int = id  # type: ignore
+        self.created_time: int = created_time  # type: ignore
+        self.last_updated_time: int = last_updated_time  # type: ignore
+
+    @classmethod
+    def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self:
+        instance = super()._load(resource, cognite_client)
+        return instance
+
+    def as_write(self) -> SimulatorModelWrite:
+        """Returns this SimulatorModel in its writing version."""
+        return SimulatorModelWrite(
+            external_id=self.external_id,
+            simulator_external_id=self.simulator_external_id,
+            data_set_id=self.data_set_id,
+            name=self.name,
+            type=self.type,
+            description=self.description,
+        )
+
+    def __hash__(self) -> int:
+        return hash(self.external_id)
+
     def dump(self, camel_case: bool = True) -> dict[str, Any]:
-        return super().dump(camel_case=camel_case)
+        return super().dump(camel_case)
 
 
 class SimulationRun(CogniteResource):
@@ -1124,9 +1227,16 @@ def as_write(self) -> SimulatorIntegrationWriteList:
         )
 
 
-class SimulatorModelList(CogniteResourceList[SimulatorModel]):
+class SimulatorModelWriteList(CogniteResourceList[SimulatorModelWrite], ExternalIDTransformerMixin):
+    _RESOURCE = SimulatorModelWrite
+
+
+class SimulatorModelList(WriteableCogniteResourceList[SimulatorModelWrite, SimulatorModel], IdTransformerMixin):
     _RESOURCE = SimulatorModel
 
+    def as_write(self) -> SimulatorModelWriteList:
+        return SimulatorModelWriteList([a.as_write() for a in self.data], cognite_client=self._get_cognite_client())
+
 
 class SimulatorModelRevisionList(CogniteResourceList[SimulatorModelRevision]):
     _RESOURCE = SimulatorModelRevision

tests/tests_integration/test_api/test_simulators/test_simulators.py

@@ -1,7 +1,10 @@
+import datetime
+
 import pytest
 
 from cognite.client import CogniteClient
 from cognite.client.data_classes.simulators.filters import SimulatorIntegrationFilter
+from cognite.client.data_classes.simulators.simulators import SimulatorModel
 from tests.tests_integration.test_api.test_simulators.seed.data import (
     simulator,
     simulator_integration,
@@ -92,6 +95,8 @@ def test_filter_integrations(self, cognite_client: CogniteClient) -> None:
 
 
 class TestSimulatorModels:
+    TEST_DATA_SET_ID = 97552494921583
+
     def test_list_models(self, cognite_client: CogniteClient) -> None:
         models = cognite_client.simulators.models.list(limit=5)
         assert len(models) > 0
@@ -110,6 +115,24 @@ def test_retrieve_model_revision(self, cognite_client: CogniteClient) -> None:
         assert model is not None
         assert model.external_id == "Shower_mixer-1"
 
+    @pytest.fixture
+    def test_create_model(self, cognite_client: CogniteClient) -> None:
+        current_time = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
+        model_external_id = current_time
+        models_to_create = SimulatorModel(
+            name="model1",
+            simulator_external_id="DWSIM",
+            external_id=model_external_id,
+            data_set_id=self.TEST_DATA_SET_ID,
+            type="SteadyState",
+        )
+
+        models_created = cognite_client.simulators.create_model(models_to_create)
+        assert models_created is not None
+        assert models_created.external_id == model_external_id
+        # delete created model
+        cognite_client.simulators.delete_models(id=models_created.id)
+
 
 class TestSimulatorRoutines:
     def test_list_routines(self, cognite_client: CogniteClient) -> None: