[CDF-22413] Simulators Part 1

cognite/client/_api/simulators/__init__.py

@@ -0,0 +1,3 @@
+from .simulators import SimulatorsAPI
+
+__all__ = ["SimulatorsAPI"]

cognite/client/_api/simulators/simulators.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+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
+from cognite.client.utils._experimental import FeaturePreviewWarning
+
+if TYPE_CHECKING:
+    from cognite.client import ClientConfig, CogniteClient
+
+
+class SimulatorsAPI(APIClient):
+    _RESOURCE_PATH = "/simulators"
+
+    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="beta", sdk_maturity="alpha", feature_name="Simulators")
+
+    def list(self, limit: int = DEFAULT_LIMIT_READ) -> SimulatorList:
+        """`Filter Simulators <https://api-docs.cognite.com/20230101-alpha/tag/Simulators/operation/filter_simulators_simulators_list_post>`_
+
+        List simulators
+
+        Args:
+            limit (int): The maximum number of simulators to return. Defaults to 25. Set to -1, float("inf") or None
+
+        Returns:
+            SimulatorList: List of simulators
+
+        Examples:
+
+            List simulators:
+
+                    >>> from cognite.client import CogniteClient
+                    >>> client = CogniteClient()
+                    >>> res = client.simulators.list()
+
+        """
+        self._warning.warn()
+        return self._list(
+            method="POST", limit=limit, resource_cls=Simulator, list_cls=SimulatorList, headers={"cdf-version": "beta"}
+        )

cognite/client/_cognite_client.py

@@ -22,6 +22,7 @@
 from cognite.client._api.raw import RawAPI
 from cognite.client._api.relationships import RelationshipsAPI
 from cognite.client._api.sequences import SequencesAPI
+from cognite.client._api.simulators import SimulatorsAPI
 from cognite.client._api.templates import TemplatesAPI
 from cognite.client._api.three_d import ThreeDAPI
 from cognite.client._api.time_series import TimeSeriesAPI
@@ -81,6 +82,7 @@ def __init__(self, config: ClientConfig | None = None) -> None:
         self.documents = DocumentsAPI(self._config, self._API_VERSION, self)
         self.workflows = WorkflowAPI(self._config, self._API_VERSION, self)
         self.units = UnitAPI(self._config, self._API_VERSION, self)
+        self.simulators = SimulatorsAPI(self._config, self._API_VERSION, self)
         # APIs just using base_url:
         self._api_client = APIClient(self._config, api_version=None, cognite_client=self)
 

cognite/client/data_classes/assets.py

@@ -890,7 +890,7 @@ def _count_subtree(xid: str, count: int = 0) -> int:
         counts.sort(key=lambda args: -args[-1])
         # The count for the fictitious "root of roots" is just len(assets), so we remove it:
         (count_dct := dict(counts)).pop(None, None)
-        return count_dct
+        return count_dct  # type: ignore[return-value]
 
     def _on_error(self, on_error: Literal["ignore", "warn", "raise"], message: str) -> None:
         if on_error == "warn":

cognite/client/data_classes/simulators/simulators.py

@@ -0,0 +1,205 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Any, Sequence
+
+from typing_extensions import Self
+
+from cognite.client.data_classes._base import (
+    CogniteObject,
+    CogniteResource,
+    CogniteResourceList,
+)
+from cognite.client.utils.useful_types import SequenceNotStr
+
+if TYPE_CHECKING:
+    from cognite.client import CogniteClient
+
+
+@dataclass
+class SimulatorUnitEntry(CogniteObject):
+    label: str
+    name: str
+
+    @classmethod
+    def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self:
+        return cls(
+            label=resource["label"],
+            name=resource["name"],
+        )
+
+
+@dataclass
+class SimulatorStepOption(CogniteObject):
+    label: str
+    value: str
+
+    @classmethod
+    def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self:
+        return cls(
+            label=resource["label"],
+            value=resource["value"],
+        )
+
+
+@dataclass
+class SimulatorModelType(CogniteObject):
+    name: str
+    key: str
+
+    @classmethod
+    def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self:
+        return cls(
+            name=resource["name"],
+            key=resource["key"],
+        )
+
+
+@dataclass
+class SimulatorQuantity(CogniteObject):
+    name: str
+    label: str
+    units: Sequence[SimulatorUnitEntry]
+
+    @classmethod
+    def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self:
+        return cls(
+            name=resource["name"],
+            label=resource["label"],
+            units=[SimulatorUnitEntry._load(unit_, cognite_client) for unit_ in resource["units"]],
+        )
+
+    def dump(self, camel_case: bool = True) -> dict[str, Any]:
+        output = super().dump(camel_case=camel_case)
+        output["units"] = [unit_.dump(camel_case=camel_case) for unit_ in self.units]
+
+        return output
+
+
+@dataclass
+class SimulatorStepField(CogniteObject):
+    name: str
+    label: str
+    info: str
+    options: Sequence[SimulatorStepOption] | None = None
+
+    @classmethod
+    def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self:
+        return cls(
+            name=resource["name"],
+            label=resource["label"],
+            info=resource["info"],
+            options=[SimulatorStepOption._load(option_, cognite_client) for option_ in resource["options"]]
+            if "options" in resource
+            else None,
+        )
+
+    def dump(self, camel_case: bool = True) -> dict[str, Any]:
+        output = super().dump(camel_case=camel_case)
+        if self.options is not None:
+            output["options"] = [option_.dump(camel_case=camel_case) for option_ in self.options]
+
+        return output
+
+
+@dataclass
+class SimulatorStep(CogniteObject):
+    step_type: str
+    fields: Sequence[SimulatorStepField]
+
+    @classmethod
+    def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self:
+        return cls(
+            step_type=resource["stepType"],
+            fields=[SimulatorStepField._load(field_, cognite_client) for field_ in resource["fields"]],
+        )
+
+    def dump(self, camel_case: bool = True) -> dict[str, Any]:
+        output = super().dump(camel_case=camel_case)
+        output["fields"] = [field_.dump(camel_case=camel_case) for field_ in self.fields]
+
+        return output
+
+
+class Simulator(CogniteResource):
+    """The simulator resource contains the definitions necessary for Cognite Data Fusion (CDF) to interact with a given simulator.
+
+    It serves as a central contract that allows APIs, UIs, and integrations (connectors) to utilize the same definitions
+    when dealing with a specific simulator.  Each simulator is uniquely identified and can be associated with various
+    file extension types, model types, step fields, and unit quantities. Simulators are essential for managing data
+    flows between CDF and external simulation tools, ensuring consistency and reliability in data handling.  ####
+    Limitations:  - A project can have a maximum of 100 simulators
+
+    This is the read/response format of the simulator.
+
+    Args:
+        id (int): A unique id of a simulator
+        external_id (str): External id of the simulator
+        name (str): Name of the simulator
+        file_extension_types (str | SequenceNotStr[str]): File extension types supported by the simulator
+        created_time (int): None
+        last_updated_time (int): None
+        model_types (SimulatorModelType | Sequence[SimulatorModelType] | None): Model types supported by the simulator
+        step_fields (SimulatorStep | Sequence[SimulatorStep] | None): Step types supported by the simulator when creating routines
+        unit_quantities (SimulatorQuantity | Sequence[SimulatorQuantity] | None): Quantities and their units supported by the simulator
+
+    """
+
+    def __init__(
+        self,
+        id: int,
+        external_id: str,
+        name: str,
+        file_extension_types: str | SequenceNotStr[str],
+        created_time: int,
+        last_updated_time: int,
+        model_types: SimulatorModelType | Sequence[SimulatorModelType] | None = None,
+        step_fields: SimulatorStep | Sequence[SimulatorStep] | None = None,
+        unit_quantities: SimulatorQuantity | Sequence[SimulatorQuantity] | None = None,
+    ) -> None:
+        self.external_id = external_id
+        self.name = name
+        self.file_extension_types = file_extension_types
+        self.model_types = model_types
+        self.step_fields = step_fields
+        self.unit_quantities = unit_quantities
+        self.id = id
+        self.created_time = created_time
+        self.last_updated_time = last_updated_time
+
+    @classmethod
+    def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self:
+        return cls(
+            id=resource["id"],
+            external_id=resource["externalId"],
+            name=resource["name"],
+            file_extension_types=resource["fileExtensionTypes"],
+            created_time=resource["createdTime"],
+            last_updated_time=resource["lastUpdatedTime"],
+            model_types=SimulatorModelType._load(resource["modelTypes"], cognite_client)
+            if "modelTypes" in resource
+            else None,
+            step_fields=SimulatorStep._load(resource["stepFields"], cognite_client)
+            if "stepFields" in resource
+            else None,
+            unit_quantities=SimulatorQuantity._load(resource["unitQuantities"], cognite_client)
+            if "unitQuantities" in resource
+            else None,
+        )
+
+    def dump(self, camel_case: bool = True) -> dict[str, Any]:
+        output = super().dump(camel_case=camel_case)
+        if isinstance(self.model_types, SimulatorModelType):
+            output["modelTypes" if camel_case else "model_types"] = self.model_types.dump(camel_case=camel_case)
+        if isinstance(self.step_fields, SimulatorStep):
+            output["stepFields" if camel_case else "step_fields"] = self.step_fields.dump(camel_case=camel_case)
+        if isinstance(self.unit_quantities, SimulatorQuantity):
+            output["unitQuantities" if camel_case else "unit_quantities"] = self.unit_quantities.dump(
+                camel_case=camel_case
+            )
+
+        return output
+
+
+class SimulatorList(CogniteResourceList[Simulator]):
+    _RESOURCE = Simulator

cognite/client/testing.py

@@ -39,6 +39,7 @@
 from cognite.client._api.raw import RawAPI, RawDatabasesAPI, RawRowsAPI, RawTablesAPI
 from cognite.client._api.relationships import RelationshipsAPI
 from cognite.client._api.sequences import SequencesAPI, SequencesDataAPI
+from cognite.client._api.simulators import SimulatorsAPI
 from cognite.client._api.synthetic_time_series import SyntheticDatapointsAPI
 from cognite.client._api.templates import (
     TemplateGroupsAPI,
@@ -138,6 +139,8 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
 
         self.relationships = MagicMock(spec_set=RelationshipsAPI)
 
+        self.simulators = MagicMock(spec=SimulatorsAPI)
+
         self.sequences = MagicMock(spec=SequencesAPI)
         self.sequences.data = MagicMock(spec_set=SequencesDataAPI)
 

cognite/client/utils/_concurrency.py

@@ -184,7 +184,7 @@ def uses_mainthread(cls) -> bool:
     @classmethod
     def get_executor(cls, max_workers: int) -> TaskExecutor:
         if cls.uses_threadpool():
-            return cls.get_thread_pool_executor(max_workers)
+            return cls.get_thread_pool_executor(max_workers)  # type: ignore[return-value]
         elif cls.uses_mainthread():
             return cls.get_mainthread_executor()
         raise RuntimeError(f"Invalid executor type '{cls.executor_type}'")

tests/tests_integration/test_api/test_simulators/test_simulators.py

@@ -0,0 +1,8 @@
+from cognite.client import CogniteClient
+
+
+class TestSimulators:
+    def test_list(self, cognite_client: CogniteClient) -> None:
+        simulators = cognite_client.simulators.list(limit=5)
+
+        assert len(simulators) > 0