diff --git a/cognite/client/_api/simulators/__init__.py b/cognite/client/_api/simulators/__init__.py new file mode 100644 index 0000000000..9d48db4f9f --- /dev/null +++ b/cognite/client/_api/simulators/__init__.py @@ -0,0 +1 @@ +from __future__ import annotations diff --git a/cognite/client/_api/simulators/simulation_runs.py b/cognite/client/_api/simulators/simulation_runs.py new file mode 100644 index 0000000000..805d43513f --- /dev/null +++ b/cognite/client/_api/simulators/simulation_runs.py @@ -0,0 +1,76 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +from cognite.client._api_client import APIClient +from cognite.client._constants import DEFAULT_LIMIT_READ +from cognite.client.data_classes.simulators.filters import SimulationRunsFilter +from cognite.client.data_classes.simulators.simulators import ( + CreatedTimeSort, + SimulationRun, + SimulationRunsList, + SimulationTimeSort, +) +from cognite.client.utils._experimental import FeaturePreviewWarning + +if TYPE_CHECKING: + from cognite.client import ClientConfig, CogniteClient + + +class SimulatorRunsAPI(APIClient): + _RESOURCE_PATH = "/simulators/runs" + + 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 list( + self, + limit: int = DEFAULT_LIMIT_READ, + filter: SimulationRunsFilter | dict[str, Any] | None = None, + sort: CreatedTimeSort | SimulationTimeSort | None = None, + ) -> SimulationRunsList: + """`Filter simulation runs `_ + + Retrieves a list of simulation runs that match the given criteria + + Args: + limit (int): The maximum number of simulation runs to return. + filter (SimulationRunsFilter | dict[str, Any] | None): Filter to apply. + sort (CreatedTimeSort | SimulationTimeSort | None): Sort simulation runs by created time or simulation time. + + Returns: + SimulationRunsList: List of simulation runs + + Examples: + + List simulation runs: + + >>> from cognite.client import CogniteClient + >>> client = CogniteClient() + >>> res = client.simulators.runs.list() + + """ + + sort_by = None + if isinstance(sort, CreatedTimeSort): + sort_by = [CreatedTimeSort.load(sort).dump()] + elif isinstance(sort, SimulationTimeSort): + sort_by = [SimulationTimeSort.load(sort).dump()] + + self._warning.warn() + return self._list( + method="POST", + limit=limit, + url_path="/simulators/runs/list", + resource_cls=SimulationRun, + list_cls=SimulationRunsList, + sort=sort_by, + filter=filter.dump() + if isinstance(filter, SimulationRunsFilter) + else filter + if isinstance(filter, dict) + else None, + ) diff --git a/cognite/client/_api/simulators/simulator_integrations.py b/cognite/client/_api/simulators/simulator_integrations.py new file mode 100644 index 0000000000..376c50100e --- /dev/null +++ b/cognite/client/_api/simulators/simulator_integrations.py @@ -0,0 +1,66 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +from cognite.client._api_client import APIClient +from cognite.client._constants import DEFAULT_LIMIT_READ +from cognite.client.data_classes.simulators.filters import SimulatorIntegrationFilter +from cognite.client.data_classes.simulators.simulators import ( + SimulatorIntegration, + SimulatorIntegrationList, +) +from cognite.client.utils._experimental import FeaturePreviewWarning + +if TYPE_CHECKING: + from cognite.client import CogniteClient + from cognite.client.config import ClientConfig + + +class SimulatorIntegrationsAPI(APIClient): + _RESOURCE_PATH = "/simulators/integrations" + + 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 list( + self, + limit: int = DEFAULT_LIMIT_READ, + filter: SimulatorIntegrationFilter | dict[str, Any] | None = None, + ) -> SimulatorIntegrationList: + """`Filter simulator integrations `_ + + Retrieves a list of simulator integrations that match the given criteria + + Args: + limit (int): The maximum number of simulator integrations to return. + filter (SimulatorIntegrationFilter | dict[str, Any] | None): Filter to apply. + + Returns: + SimulatorIntegrationList: List of simulator integrations + + Examples: + + List simulator integrations: + + >>> from cognite.client import CogniteClient + >>> client = CogniteClient() + >>> res = client.simulators.integrations.list() + + """ + + self._warning.warn() + return self._list( + method="POST", + limit=limit, + url_path="/simulators/integrations/list", + resource_cls=SimulatorIntegration, + list_cls=SimulatorIntegrationList, + filter=filter.dump() + if isinstance(filter, SimulatorIntegrationFilter) + else filter + if isinstance(filter, dict) + else None, + ) diff --git a/cognite/client/_api/simulators/simulator_models.py b/cognite/client/_api/simulators/simulator_models.py new file mode 100644 index 0000000000..5174ad8c6a --- /dev/null +++ b/cognite/client/_api/simulators/simulator_models.py @@ -0,0 +1,221 @@ +from __future__ import annotations + +from collections.abc import Sequence +from typing import TYPE_CHECKING, Any + +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 ( + CreatedTimeSort, + SimulatorModel, + SimulatorModelList, + SimulatorModelRevision, + SimulatorModelRevisionList, +) +from cognite.client.utils._experimental import FeaturePreviewWarning +from cognite.client.utils._identifier import IdentifierSequence +from cognite.client.utils.useful_types import SequenceNotStr + +if TYPE_CHECKING: + from cognite.client import ClientConfig, CogniteClient + + +class SimulatorModelRevisionsAPI(APIClient): + _RESOURCE_PATH = "/simulators/models/revisions" + + 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 list( + self, + limit: int = DEFAULT_LIMIT_READ, + sort: CreatedTimeSort | None = None, + filter: SimulatorModelRevisionsFilter | dict[str, Any] | None = None, + ) -> SimulatorModelRevisionList: + """`Filter simulator model revisions `_ + + Retrieves a list of simulator model revisions that match the given criteria + + Args: + limit (int): Maximum number of results to return. Defaults to 1000. Set to -1, float(“inf”) or None to return all items. + sort (CreatedTimeSort | None): Sort order for the results. + filter (SimulatorModelRevisionsFilter | dict[str, Any] | None): Filter to apply. + + Returns: + SimulatorModelRevisionList: List of simulator model revisions + + Examples: + + List simulator model revisions: + >>> from cognite.client import CogniteClient + >>> client = CogniteClient() + >>> res = client.simulators.models.revisions.list() + + """ + self._warning.warn() + return self._list( + method="POST", + limit=limit, + url_path="/simulators/models/revisions/list", + resource_cls=SimulatorModelRevision, + list_cls=SimulatorModelRevisionList, + sort=[CreatedTimeSort.load(sort).dump()] if sort else None, + filter=filter.dump() + if isinstance(filter, SimulatorModelRevisionsFilter) + else filter + if isinstance(filter, dict) + else None, + ) + + def retrieve(self, id: int | None = None, external_id: str | None = None) -> SimulatorModelRevision | None: + """`Retrieve simulator model revision `_ + + Retrieve a simulator model revision by ID or external ID + + Args: + id (int | None): The id of the simulator model revision. + external_id (str | None): The external id of the simulator model revision. + + Returns: + SimulatorModelRevision | None: Requested simulator model revision + + Examples: + + Get simulator model revision by id: + >>> from cognite.client import CogniteClient + >>> client = CogniteClient() + >>> res = client.simulators.models.revisions.retrieve(id=123) + + Get simulator model revision by external id: + >>> res = client.simulators.models.revisions.retrieve(external_id="abcdef") + + """ + identifiers = IdentifierSequence.load(ids=id, external_ids=external_id).as_singleton() + return self._retrieve_multiple( + list_cls=SimulatorModelRevisionList, + resource_cls=SimulatorModelRevision, + identifiers=identifiers, + resource_path="/simulators/models/revisions", + ) + + def retrieve_multiple( + self, + ids: Sequence[int] | None = None, + external_ids: SequenceNotStr[str] | None = None, + ) -> SimulatorModelRevisionList: + """`Retrieve simulator model revisions `_ + + Retrieve one or more simulator model revisions by IDs or external IDs. + + Args: + ids (Sequence[int] | None): IDs + external_ids (SequenceNotStr[str] | None): External IDs + ignore_unknown_ids (bool): Ignore IDs and external IDs that are not found rather than throw an exception. + + Returns: + SimulatorModelRevisionList: Requested simulator model revisions + + Examples: + + Get simulator model revisions by ids: + >>> from cognite.client import CogniteClient + >>> client = CogniteClient() + >>> res = client.simulators.models.revisions.retrieve_multiple(ids=[1, 2, 3]) + + Get simulator model revisions by external ids: + >>> res = client.simulators.models.revisions.retrieve_multiple(external_ids=["abc", "def"]) + """ + identifiers = IdentifierSequence.load(ids=ids, external_ids=external_ids) + return self._retrieve_multiple( + list_cls=SimulatorModelRevisionList, + resource_cls=SimulatorModelRevision, + identifiers=identifiers, + ) + + +class SimulatorModelsAPI(APIClient): + _RESOURCE_PATH = "/simulators/models" + + def __init__(self, config: ClientConfig, api_version: str | None, cognite_client: CogniteClient) -> None: + super().__init__(config, api_version, cognite_client) + self.revisions = SimulatorModelRevisionsAPI(config, api_version, cognite_client) + self._warning = FeaturePreviewWarning( + api_maturity="General Availability", sdk_maturity="alpha", feature_name="Simulators" + ) + + def list( + self, + limit: int = DEFAULT_LIMIT_READ, + filter: SimulatorModelsFilter | dict[str, Any] | None = None, + sort: CreatedTimeSort | None = None, + ) -> SimulatorModelList: + """`Filter simulator models `_ + + Retrieves a list of simulator models that match the given criteria + + Args: + limit (int): Maximum number of results to return. Defaults to 1000. 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. + + Returns: + SimulatorModelList: List of simulator models + + Examples: + + List simulator models: + + >>> from cognite.client import CogniteClient + >>> client = CogniteClient() + >>> res = client.simulators.models.list() + + """ + self._warning.warn() + return self._list( + method="POST", + limit=limit, + url_path="/simulators/models/list", + resource_cls=SimulatorModel, + list_cls=SimulatorModelList, + sort=[CreatedTimeSort.load(sort).dump()] if sort else None, + filter=filter.dump() + if isinstance(filter, SimulatorModelsFilter) + else filter + if isinstance(filter, dict) + else None, + ) + + def retrieve(self, id: int | None = None, external_id: str | None = None) -> SimulatorModel | None: + """`Retrieve simulator model `_ + + Retrieve a simulator model by ID or external ID + + Args: + id (int | None): The id of the simulator model. + external_id (str | None): The external id of the simulator model. + + Returns: + SimulatorModel | None: Requested simulator model + + Examples: + + Retrieve simulator model by id: + >>> from cognite.client import CogniteClient + >>> client = CogniteClient() + >>> res = client.simulators.models.retrieve(id=1) + + Retrieve simulator model by external id: + >>> res = client.simulators.models.retrieve(external_id="1") + + """ + identifiers = IdentifierSequence.load(ids=id, external_ids=external_id).as_singleton() + return self._retrieve_multiple( + list_cls=SimulatorModelList, + resource_cls=SimulatorModel, + identifiers=identifiers, + resource_path="/simulators/models", + ) diff --git a/cognite/client/_api/simulators/simulator_routines.py b/cognite/client/_api/simulators/simulator_routines.py new file mode 100644 index 0000000000..5d40998a40 --- /dev/null +++ b/cognite/client/_api/simulators/simulator_routines.py @@ -0,0 +1,194 @@ +from __future__ import annotations + +from collections.abc import Sequence +from typing import TYPE_CHECKING, Any + +from cognite.client._api_client import APIClient +from cognite.client._constants import DEFAULT_LIMIT_READ +from cognite.client.data_classes.simulators.filters import SimulatorRoutineRevisionsFilter, SimulatorRoutinesFilter +from cognite.client.data_classes.simulators.simulators import ( + CreatedTimeSort, + SimulatorRoutine, + SimulatorRoutineList, + SimulatorRoutineRevision, + SimulatorRoutineRevisionsList, +) +from cognite.client.utils._experimental import FeaturePreviewWarning +from cognite.client.utils._identifier import IdentifierSequence +from cognite.client.utils.useful_types import SequenceNotStr + +if TYPE_CHECKING: + from cognite.client import ClientConfig, CogniteClient + + +class SimulatorRoutineRevisionsAPI(APIClient): + _RESOURCE_PATH = "/simulators/routines/revisions" + + 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 list( + self, + limit: int = 20, # the maximum number of revisions to return is limited to 20 items. + sort: CreatedTimeSort | None = None, + filter: SimulatorRoutineRevisionsFilter | dict[str, Any] | None = None, + include_all_fields: bool = False, + ) -> SimulatorRoutineRevisionsList: + """`Filter simulator routine revisions `_ + + Retrieves a list of simulator routine revisions that match the given criteria. + + Args: + limit (int): Maximum number of results to return. Defaults to 10. Set to -1, float(“inf”) or None to return all items. + sort (CreatedTimeSort | None): The criteria to sort by. + filter (SimulatorRoutineRevisionsFilter | dict[str, Any] | None): Filter to apply. + include_all_fields (bool): If all fields should be included in the response. Defaults to false which does not include script, configuration.inputs and configuration.outputs in the response. + + Returns: + SimulatorRoutineRevisionsList: List of simulator routine revisions + + Examples: + + List simulator routine revisions: + >>> from cognite.client import CogniteClient + >>> client = CogniteClient() + >>> res = client.simulators.routines.revisions.list() + + """ + self._warning.warn() + return self._list( + method="POST", + limit=limit, + url_path="/simulators/routines/revisions/list", + resource_cls=SimulatorRoutineRevision, + list_cls=SimulatorRoutineRevisionsList, + filter=filter.dump() + if isinstance(filter, SimulatorRoutineRevisionsFilter) + else filter + if isinstance(filter, dict) + else None, + sort=[CreatedTimeSort.load(sort).dump()] if sort else None, + other_params={"includeAllFields": include_all_fields}, + ) + + def retrieve(self, id: int | None = None, external_id: str | None = None) -> SimulatorRoutineRevision | None: + """`Retrieve simulator routine revisions `_ + + Retrieve a simulator routine revision by ID or external ID + + Args: + id (int | None): The id of the simulator routine revision. + external_id (str | None): The external id of the simulator routine revision. + + Returns: + SimulatorRoutineRevision | None: Requested simulator routine revision + + Examples: + + Get simulator routine revision by id: + >>> from cognite.client import CogniteClient + >>> client = CogniteClient() + >>> res = client.simulators.routines.revisions.retrieve(ids=123) + + Get simulator routine revision by external id: + >>> res = client.simulators.routines.revisions.retrieve(external_ids="abcdef") + """ + identifiers = IdentifierSequence.load(ids=id, external_ids=external_id).as_singleton() + return self._retrieve_multiple( + resource_cls=SimulatorRoutineRevision, + list_cls=SimulatorRoutineRevisionsList, + identifiers=identifiers, + resource_path="/simulators/routines/revisions", + ) + + def retrieve_multiple( + self, + ids: Sequence[int] | None = None, + external_ids: SequenceNotStr[str] | None = None, + ignore_unknown_ids: bool = False, + ) -> SimulatorRoutineRevisionsList: + """`Retrieve simulator routine revisions `_ + + Retrieve one or more simulator routine revisions by IDs or external IDs + + Args: + ids (Sequence[int] | None): IDs + external_ids (SequenceNotStr[str] | None): External IDs + ignore_unknown_ids (bool): Ignore IDs and external IDs that are not found rather than throw an exception. + + Returns: + SimulatorRoutineRevisionsList: Requested simulator routine revisions + + Examples: + + Get simulator routine revisions by id: + >>> from cognite.client import CogniteClient + >>> client = CogniteClient() + >>> res = client.simulators.routines.revisions.retrieve_multiple(ids=[1, 2, 3]) + + Get simulator routine revisions by external id: + >>> res = client.simulators.routines.revisions.retrieve_multiple(external_ids=["abc", "def"]) + """ + identifiers = IdentifierSequence.load(ids=ids, external_ids=external_ids) + return self._retrieve_multiple( + list_cls=SimulatorRoutineRevisionsList, + resource_cls=SimulatorRoutineRevision, + identifiers=identifiers, + ignore_unknown_ids=ignore_unknown_ids, + ) + + +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.revisions = SimulatorRoutineRevisionsAPI(config, api_version, cognite_client) + self._warning = FeaturePreviewWarning( + api_maturity="General Availability", sdk_maturity="alpha", feature_name="Simulators" + ) + + def list( + self, + limit: int = DEFAULT_LIMIT_READ, + filter: SimulatorRoutinesFilter | dict[str, Any] | None = None, + sort: CreatedTimeSort | None = None, + ) -> SimulatorRoutineList: + """`Filter simulator routines `_ + + Retrieves a list of simulator routines that match the given criteria + + Args: + limit (int): Maximum number of results to return. Defaults to 1000. Set to -1, float(“inf”) or None to return all items. + filter (SimulatorRoutinesFilter | dict[str, Any] | None): The filter to narrow down simulator routines. + sort (CreatedTimeSort | 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() + + """ + self._warning.warn() + return self._list( + method="POST", + limit=limit, + url_path="/simulators/routines/list", + resource_cls=SimulatorRoutine, + list_cls=SimulatorRoutineList, + sort=[CreatedTimeSort.load(sort).dump()] if sort else None, + filter=filter.dump() + if isinstance(filter, SimulatorRoutinesFilter) + else filter + if isinstance(filter, dict) + else None, + ) diff --git a/cognite/client/_api/simulators/simulators.py b/cognite/client/_api/simulators/simulators.py new file mode 100644 index 0000000000..7c099524ab --- /dev/null +++ b/cognite/client/_api/simulators/simulators.py @@ -0,0 +1,56 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +from cognite.client._api.simulators.simulation_runs import SimulatorRunsAPI +from cognite.client._api.simulators.simulator_integrations import SimulatorIntegrationsAPI +from cognite.client._api.simulators.simulator_models import SimulatorModelsAPI +from cognite.client._api.simulators.simulator_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, +) +from cognite.client.utils._experimental import FeaturePreviewWarning + +if TYPE_CHECKING: + from cognite.client import CogniteClient + from cognite.client.config import ClientConfig + + +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.models = SimulatorModelsAPI(config, api_version, cognite_client) + self.runs = SimulatorRunsAPI(config, api_version, cognite_client) + self.integrations = SimulatorIntegrationsAPI(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" + ) + + def list(self, limit: int = DEFAULT_LIMIT_READ) -> SimulatorList: + """`Filter simulators `_ + + List simulators + + Args: + limit (int): Maximum number of results to return. Defaults to 1000. Set to -1, float(“inf”) or None to return all items. + + 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) diff --git a/cognite/client/_api_client.py b/cognite/client/_api_client.py index 34dca4b56c..f9f5c5354f 100644 --- a/cognite/client/_api_client.py +++ b/cognite/client/_api_client.py @@ -101,6 +101,7 @@ class APIClient: "geospatial/(compute|crs/byids|featuretypes/(byids|list))", "geospatial/featuretypes/[A-Za-z][A-Za-z0-9_]{0,31}/features/(aggregate|list|byids|search|search-streaming|[A-Za-z][A-Za-z0-9_]{0,255}/rasters/[A-Za-z][A-Za-z0-9_]{0,31})", "transformations/(filter|byids|jobs/byids|schedules/byids|query/run)", + "simulators/.*", "extpipes/(list|byids|runs/list)", "workflows/.*", "hostedextractors/.*", diff --git a/cognite/client/_cognite_client.py b/cognite/client/_cognite_client.py index b5c0f01b46..634963e9de 100644 --- a/cognite/client/_cognite_client.py +++ b/cognite/client/_cognite_client.py @@ -23,6 +23,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.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 @@ -83,6 +84,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) diff --git a/cognite/client/data_classes/simulators/__init__.py b/cognite/client/data_classes/simulators/__init__.py index 9d48db4f9f..16b1e6985f 100644 --- a/cognite/client/data_classes/simulators/__init__.py +++ b/cognite/client/data_classes/simulators/__init__.py @@ -1 +1,111 @@ from __future__ import annotations + +from cognite.client.data_classes.simulators.simulators import ( + SimulationRun, + SimulationRunCore, + SimulationRunsList, + SimulationValueUnitInput, + Simulator, + SimulatorCore, + SimulatorIntegration, + SimulatorIntegrationCore, + SimulatorIntegrationList, + SimulatorIntegrationWrite, + SimulatorIntegrationWriteList, + SimulatorList, + SimulatorModel, + SimulatorModelCore, + SimulatorModelList, + SimulatorModelRevision, + SimulatorModelRevisionCore, + SimulatorModelRevisionList, + SimulatorModelRevisionWrite, + SimulatorModelRevisionWriteList, + SimulatorModelType, + SimulatorModelWrite, + SimulatorModelWriteList, + SimulatorQuantity, + SimulatorRoutine, + SimulatorRoutineConfiguration, + SimulatorRoutineCore, + SimulatorRoutineDataSampling, + SimulatorRoutineInputConstant, + SimulatorRoutineInputTimeseries, + SimulatorRoutineList, + SimulatorRoutineLogicalCheckEnabled, + SimulatorRoutineOutput, + SimulatorRoutineRevision, + SimulatorRoutineRevisionCore, + SimulatorRoutineRevisionList, + SimulatorRoutineRevisionsList, + SimulatorRoutineRevisionWrite, + SimulatorRoutineRevisionWriteList, + SimulatorRoutineSchedule, + SimulatorRoutineStage, + SimulatorRoutineSteadyStateDetectionEnabled, + SimulatorRoutineStep, + SimulatorRoutineStepArguments, + SimulatorRoutineWrite, + SimulatorRoutineWriteList, + SimulatorStep, + SimulatorStepField, + SimulatorStepOption, + SimulatorUnitEntry, + SimulatorWrite, + SimulatorWriteList, +) + +__all__ = [ + "SimulationRun", + "SimulationRunCore", + "SimulationRunsList", + "SimulationValueUnitInput", + "Simulator", + "SimulatorCore", + "SimulatorIntegration", + "SimulatorIntegrationCore", + "SimulatorIntegrationList", + "SimulatorIntegrationWrite", + "SimulatorIntegrationWriteList", + "SimulatorList", + "SimulatorModel", + "SimulatorModelCore", + "SimulatorModelList", + "SimulatorModelRevision", + "SimulatorModelRevisionCore", + "SimulatorModelRevisionList", + "SimulatorModelRevisionWrite", + "SimulatorModelRevisionWriteList", + "SimulatorModelType", + "SimulatorModelWrite", + "SimulatorModelWriteList", + "SimulatorQuantity", + "SimulatorRoutine", + "SimulatorRoutineConfiguration", + "SimulatorRoutineCore", + "SimulatorRoutineDataSampling", + "SimulatorRoutineInputConstant", + "SimulatorRoutineInputTimeseries", + "SimulatorRoutineList", + "SimulatorRoutineLogicalCheckEnabled", + "SimulatorRoutineOutput", + "SimulatorRoutineRevision", + "SimulatorRoutineRevisionCore", + "SimulatorRoutineRevisionList", + "SimulatorRoutineRevisionWrite", + "SimulatorRoutineRevisionWriteList", + "SimulatorRoutineRevisionsList", + "SimulatorRoutineSchedule", + "SimulatorRoutineStage", + "SimulatorRoutineSteadyStateDetectionEnabled", + "SimulatorRoutineStep", + "SimulatorRoutineStepArguments", + "SimulatorRoutineWrite", + "SimulatorRoutineWriteList", + "SimulatorStep", + "SimulatorStepField", + "SimulatorStepOption", + "SimulatorUnitEntry", + "SimulatorWrite", + "SimulatorWriteList", +] diff --git a/cognite/client/data_classes/simulators/filters.py b/cognite/client/data_classes/simulators/filters.py new file mode 100644 index 0000000000..d5961acd9d --- /dev/null +++ b/cognite/client/data_classes/simulators/filters.py @@ -0,0 +1,88 @@ +from __future__ import annotations + +from collections.abc import Sequence +from typing import Any + +from cognite.client.data_classes._base import CogniteFilter + + +class SimulatorIntegrationFilter(CogniteFilter): + def __init__( + self, + simulator_external_ids: Sequence[str] | None = None, + active: bool | None = None, + ) -> None: + self.simulator_external_ids = simulator_external_ids + self.active = active + + def dump(self, camel_case: bool = True) -> dict[str, Any]: + return super().dump(camel_case=camel_case) + + +class SimulatorModelsFilter(CogniteFilter): + def __init__( + self, + simulator_external_ids: Sequence[str] | None = None, + ) -> None: + self.simulator_external_ids = simulator_external_ids + + def dump(self, camel_case: bool = True) -> dict[str, Any]: + return super().dump(camel_case=camel_case) + + +class SimulatorModelRevisionsFilter(CogniteFilter): + def __init__( + self, + model_external_ids: Sequence[str] | None = None, + all_versions: bool | None = None, + ) -> None: + self.model_external_ids = model_external_ids + self.all_versions = all_versions + + def dump(self, camel_case: bool = True) -> dict[str, Any]: + return super().dump(camel_case=camel_case) + + +class SimulatorRoutinesFilter(CogniteFilter): + def __init__( + self, + model_external_ids: Sequence[str] | None = None, + simulator_integration_external_ids: Sequence[str] | None = None, + ) -> None: + self.model_external_ids = model_external_ids + self.simulator_integration_external_ids = simulator_integration_external_ids + + def dump(self, camel_case: bool = True) -> dict[str, Any]: + return super().dump(camel_case=camel_case) + + +class SimulatorRoutineRevisionsFilter(CogniteFilter): + def __init__( + self, + routine_external_ids: Sequence[str] | None = None, + all_versions: bool | None = None, + model_external_ids: Sequence[str] | None = None, + simulator_integration_external_ids: Sequence[str] | None = None, + simulator_external_ids: Sequence[str] | None = None, + ) -> None: + self.model_external_ids = model_external_ids + self.all_versions = all_versions + self.routine_external_ids = routine_external_ids + self.simulator_integration_external_ids = simulator_integration_external_ids + self.simulator_external_ids = simulator_external_ids + + def dump(self, camel_case: bool = True) -> dict[str, Any]: + return super().dump(camel_case=camel_case) + + +class SimulationRunsFilter(CogniteFilter): + def __init__( + self, + model_external_ids: Sequence[str] | None = None, + all_versions: bool | None = None, + ) -> None: + self.model_external_ids = model_external_ids + self.all_versions = all_versions + + def dump(self, camel_case: bool = True) -> dict[str, Any]: + return super().dump(camel_case=camel_case) diff --git a/cognite/client/data_classes/simulators/simulators.py b/cognite/client/data_classes/simulators/simulators.py new file mode 100644 index 0000000000..a77f16f8e6 --- /dev/null +++ b/cognite/client/data_classes/simulators/simulators.py @@ -0,0 +1,1645 @@ +from __future__ import annotations + +from abc import ABC +from collections.abc import Sequence +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any, Literal + +from typing_extensions import Self + +from cognite.client.data_classes._base import ( + CogniteObject, + CogniteResource, + CogniteResourceList, + CogniteSort, + ExternalIDTransformerMixin, + IdTransformerMixin, + InternalIdTransformerMixin, + WriteableCogniteResource, + WriteableCogniteResourceList, +) +from cognite.client.utils.useful_types import SequenceNotStr + +if TYPE_CHECKING: + from cognite.client import CogniteClient + + +@dataclass +class SimulationValueUnitInput(CogniteObject): + name: str + quantity: str | None = None + + @classmethod + def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self: + return cls( + name=resource["name"], + quantity=resource.get("quantity"), + ) + + def dump(self, camel_case: bool = True) -> dict[str, Any]: + return super().dump(camel_case=camel_case) + + +@dataclass +class SimulatorRoutineInputTimeseries(CogniteObject): + name: str + reference_id: str + source_external_id: str + aggregate: str | None = None + save_timeseries_external_id: str | None = None + unit: SimulationValueUnitInput | None = None + + @classmethod + def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self: + return cls( + name=resource["name"], + reference_id=resource["referenceId"], + source_external_id=resource["sourceExternalId"], + aggregate=resource.get("aggregate"), + save_timeseries_external_id=resource.get("saveTimeseriesExternalId"), + unit=SimulationValueUnitInput._load(resource["unit"], cognite_client) if "unit" in resource else None, + ) + + def dump(self, camel_case: bool = True) -> dict[str, Any]: + output = super().dump(camel_case=camel_case) + if self.unit is not None: + output["unit"] = self.unit.dump(camel_case=camel_case) + + return output + + +@dataclass +class SimulatorRoutineInputConstant(CogniteObject): + name: str + reference_id: str + value: str + value_type: str + unit: SimulationValueUnitInput | None = None + save_timeseries_external_id: str | None = None + + @classmethod + def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self: + return cls( + name=resource["name"], + reference_id=resource["referenceId"], + value=resource["value"], + value_type=resource["valueType"], + unit=SimulationValueUnitInput._load(resource["unit"], cognite_client) if "unit" in resource else None, + save_timeseries_external_id=resource.get("saveTimeseriesExternalId"), + ) + + def dump(self, camel_case: bool = True) -> dict[str, Any]: + output = super().dump(camel_case=camel_case) + if self.unit is not None: + output["unit"] = self.unit.dump(camel_case=camel_case) + + return output + + +@dataclass +class SimulatorRoutineOutput(CogniteObject): + name: str + reference_id: str + value_type: str + unit: SimulationValueUnitInput | None = None + save_timeseries_external_id: str | None = None + + @classmethod + def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self: + return cls( + name=resource["name"], + reference_id=resource["referenceId"], + value_type=resource["valueType"], + unit=SimulationValueUnitInput._load(resource["unit"], cognite_client) if "unit" in resource else None, + save_timeseries_external_id=resource.get("saveTimeseriesExternalId"), + ) + + def dump(self, camel_case: bool = True) -> dict[str, Any]: + output = super().dump(camel_case=camel_case) + if self.unit is not None: + output["unit"] = self.unit.dump(camel_case=camel_case) + + return output + + +@dataclass +class SimulatorRoutineSchedule(CogniteObject): + enabled: bool = False + cron_expression: str | None = None + + def __init__(self, enabled: bool, cron_expression: str | None = None, **_: Any) -> None: + self.enabled = enabled + self.cron_expression = cron_expression + + @classmethod + def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> SimulatorRoutineSchedule: + return cls( + enabled=resource["enabled"], + cron_expression=resource.get("cronExpression"), + ) + + def dump(self, camel_case: bool = True) -> dict[str, Any]: + output = super().dump(camel_case=camel_case) + return output + + +@dataclass +class SimulatorRoutineDataSampling(CogniteObject): + enabled: bool = False + validation_window: int | None = None + sampling_window: int | None = None + granularity: str | None = None + + @classmethod + def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self: + 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) + + +@dataclass +class SimulatorRoutineLogicalCheckEnabled(CogniteObject): + enabled: bool = False + timeseries_external_id: str | None = None + aggregate: str | None = None + operator: str | None = None + value: float | None = None + + @classmethod + def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self: + 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) + + +@dataclass +class SimulatorRoutineSteadyStateDetectionEnabled(CogniteObject): + enabled: bool = False + timeseries_external_id: str | None = None + aggregate: str | None = None + min_section_size: int | None = None + var_threshold: float | None = None + slope_threshold: float | None = None + + @classmethod + def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self: + 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) + + +@dataclass +class SimulatorRoutineConfiguration(CogniteObject): + schedule: SimulatorRoutineSchedule + data_sampling: SimulatorRoutineDataSampling + logical_check: list[SimulatorRoutineLogicalCheckEnabled] + steady_state_detection: list[SimulatorRoutineSteadyStateDetectionEnabled] + inputs: list[SimulatorRoutineInputConstant | SimulatorRoutineInputTimeseries] + outputs: list[SimulatorRoutineOutput] + + @classmethod + def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self: + inputs = [] + outputs = [] + + if resource.get("inputs", None) is not None: + for input_ in resource["inputs"]: + if isinstance(input_, SimulatorRoutineInputConstant) or isinstance( + input_, SimulatorRoutineInputTimeseries + ): + inputs.append(input_) + + else: + if "value" in input_: + inputs.append(SimulatorRoutineInputConstant._load(input_, cognite_client)) + else: + inputs.append(SimulatorRoutineInputTimeseries._load(input_, cognite_client)) + + if resource.get("outputs", None) is not None: + for output_ in resource["outputs"]: + if isinstance(output_, SimulatorRoutineOutput): + outputs.append(output_) + + else: + outputs.append(SimulatorRoutineOutput._load(output_, cognite_client)) + + return cls( + schedule=SimulatorRoutineSchedule.load(resource["schedule"], cognite_client), + data_sampling=SimulatorRoutineDataSampling._load(resource["dataSampling"], cognite_client), + logical_check=[ + SimulatorRoutineLogicalCheckEnabled._load(check_, cognite_client) for check_ in resource["logicalCheck"] + ], + steady_state_detection=[ + SimulatorRoutineSteadyStateDetectionEnabled._load(detection_, cognite_client) + for detection_ in resource["steadyStateDetection"] + ], + inputs=inputs, + outputs=outputs, + ) + + def dump(self, camel_case: bool = True) -> dict[str, Any]: + output = super().dump(camel_case=camel_case) + output["schedule"] = self.schedule.dump(camel_case=camel_case) + output["dataSampling"] = self.data_sampling.dump(camel_case=camel_case) + output["logicalCheck"] = [check_.dump(camel_case=camel_case) for check_ in self.logical_check] + output["steadyStateDetection"] = [ + detection_.dump(camel_case=camel_case) for detection_ in self.steady_state_detection + ] + output["inputs"] = [input_.dump(camel_case=camel_case) for input_ in self.inputs] + output["outputs"] = [output_.dump(camel_case=camel_case) for output_ in self.outputs] + + return output + + +@dataclass +class SimulatorRoutineStepArguments(CogniteObject): + reference_id: str | None = None + object_name: str | None = None + object_property: str | None = None + + @classmethod + def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self: + 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) + + +@dataclass +class SimulatorRoutineStep(CogniteObject): + step_type: str + arguments: dict[str, Any] + + @classmethod + def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self: + return cls( + step_type=resource["stepType"], + arguments=resource["arguments"], + ) + + def dump(self, camel_case: bool = True) -> dict[str, Any]: + return super().dump(camel_case=camel_case) + + +@dataclass +class SimulatorRoutineStage(CogniteObject): + order: int + steps: list[SimulatorRoutineStep] + description: str | None + + @classmethod + def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self: + return cls( + order=resource["order"], + steps=[ + SimulatorRoutineStep._load(step_, cognite_client) if isinstance(step_, dict) else step_ + for step_ in resource["steps"] + ], + description=resource.get("description"), + ) + + def dump(self, camel_case: bool = True) -> dict[str, Any]: + output = super().dump(camel_case=camel_case) + output["steps"] = [step_.dump(camel_case=camel_case) for step_ in self.steps] + return output + + +@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) -> SimulatorModelType: + return cls( + name=resource["name"], + key=resource["key"], + ) + + @classmethod + def _load_list( + cls, resource: dict[str, Any] | list[dict[str, Any]], cognite_client: CogniteClient | None = None + ) -> SimulatorModelType | list[SimulatorModelType]: + if isinstance(resource, list): + return [cls._load(res, cognite_client) for res in resource] + + return cls._load(resource, cognite_client) + + +@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"]], + ) + + @classmethod + def _load_list( + cls, resource: dict[str, Any] | list[dict[str, Any]], cognite_client: CogniteClient | None = None + ) -> SimulatorQuantity | list[SimulatorQuantity]: + if isinstance(resource, list): + return [cls._load(res, cognite_client) for res in resource] + + return cls._load(resource, cognite_client) + + 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"]], + ) + + @classmethod + def _load_list( + cls, resource: dict[str, Any] | list[dict[str, Any]], cognite_client: CogniteClient | None = None + ) -> SimulatorStep | list[SimulatorStep]: + if isinstance(resource, list): + return [cls._load(res, cognite_client) for res in resource] + + return cls._load(resource, cognite_client) + + 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 SimulatorCore(WriteableCogniteResource["SimulatorWrite"], ABC): + """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. #### + + This is the read/response format of the simulator. + + Args: + 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 + 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, + external_id: str, + name: str, + file_extension_types: str | SequenceNotStr[str], + 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 + + @classmethod + def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self: + return cls( + external_id=resource["externalId"], + name=resource["name"], + file_extension_types=resource["fileExtensionTypes"], + model_types=SimulatorModelType._load_list(resource["modelTypes"], cognite_client) + if "modelTypes" in resource + else None, + step_fields=SimulatorStep._load_list(resource["stepFields"], cognite_client) + if "stepFields" in resource + else None, + unit_quantities=SimulatorQuantity._load_list(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 SimulatorWrite(SimulatorCore): + def __init__( + self, + external_id: str, + name: str, + file_extension_types: str | SequenceNotStr[str], + model_types: SimulatorModelType | Sequence[SimulatorModelType] | None = None, + step_fields: SimulatorStep | Sequence[SimulatorStep] | None = None, + unit_quantities: SimulatorQuantity | Sequence[SimulatorQuantity] | None = None, + ) -> None: + super().__init__( + external_id=external_id, + name=name, + file_extension_types=file_extension_types, + model_types=model_types, + step_fields=step_fields, + unit_quantities=unit_quantities, + ) + + def as_write(self) -> SimulatorWrite: + """Returns a writeable version of this resource""" + return self + + +class Simulator(SimulatorCore): + def __init__( + self, + external_id: str, + name: str, + file_extension_types: str | SequenceNotStr[str], + created_time: int | None = None, + last_updated_time: int | None = None, + id: int | None = None, + 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 + # 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 | None = id + self.created_time: int | None = created_time + self.last_updated_time: int | None = last_updated_time + + @classmethod + def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self: + load = super()._load(resource, cognite_client) + return cls( + external_id=load.external_id, + name=load.name, + file_extension_types=load.file_extension_types, + created_time=resource.get("createdTime"), + last_updated_time=resource.get("lastUpdatedTime"), + id=resource.get("id"), + model_types=load.model_types, + step_fields=load.step_fields, + unit_quantities=load.unit_quantities, + ) + + def as_write(self) -> SimulatorWrite: + """Returns a writeable version of this resource""" + return SimulatorWrite( + external_id=self.external_id, + name=self.name, + file_extension_types=self.file_extension_types, + model_types=self.model_types, + step_fields=self.step_fields, + unit_quantities=self.unit_quantities, + ) + + def __hash__(self) -> int: + return hash(self.external_id) + + +class SimulatorIntegrationCore(WriteableCogniteResource["SimulatorIntegrationWrite"], ABC): + """ + The simulator integration resource represents a simulator connector in Cognite Data Fusion (CDF). + It provides information about the configured connectors for a given simulator, including their status and additional + details such as dataset, name, license status, connector version, simulator version, and more. This resource is essential + for monitoring and managing the interactions between CDF and external simulators, ensuring proper data flow and integration. + + This is the read/response format of the simulator integration. + + Args: + + external_id (str): External id of the simulator integration + simulator_external_id (str): External id of the associated simulator + heartbeat (int): The interval in seconds between the last heartbeat and the current time + data_set_id (int): The id of the dataset associated with the simulator integration + connector_version (str): The version of the connector + license_status (str | None): The status of the license + simulator_version (str | None): The version of the simulator + license_last_checked_time (int | None): The time when the license was last checked + connector_status (str | None): The status of the connector + connector_status_updated_time (int | None): The time when the connector status was last updated + + """ + + def __init__( + self, + external_id: str, + simulator_external_id: str, + heartbeat: int, + data_set_id: int, + connector_version: str, + license_status: str | None = None, + simulator_version: str | None = None, + license_last_checked_time: int | None = None, + connector_status: str | None = None, + connector_status_updated_time: int | None = None, + ) -> None: + self.external_id = external_id + self.simulator_external_id = simulator_external_id + self.heartbeat = heartbeat + self.data_set_id = data_set_id + self.connector_version = connector_version + self.license_status = license_status + self.simulator_version = simulator_version + self.license_last_checked_time = license_last_checked_time + self.connector_status = connector_status + self.connector_status_updated_time = connector_status_updated_time + + @classmethod + def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self: + return cls( + external_id=resource["externalId"], + simulator_external_id=resource["simulatorExternalId"], + heartbeat=resource["heartbeat"], + data_set_id=resource["dataSetId"], + connector_version=resource["connectorVersion"], + license_status=resource.get("licenseStatus"), + simulator_version=resource.get("simulatorVersion"), + license_last_checked_time=resource.get("licenseLastCheckedTime"), + connector_status=resource.get("connectorStatus"), + connector_status_updated_time=resource.get("connectorStatusUpdatedTime"), + ) + + def dump(self, camel_case: bool = True) -> dict[str, Any]: + return super().dump(camel_case=camel_case) + + +class SimulatorIntegrationWrite(SimulatorIntegrationCore): + def __init__( + self, + external_id: str, + simulator_external_id: str, + heartbeat: int, + data_set_id: int, + connector_version: str, + license_status: str | None = None, + simulator_version: str | None = None, + license_last_checked_time: int | None = None, + connector_status: str | None = None, + connector_status_updated_time: int | None = None, + ) -> None: + super().__init__( + external_id=external_id, + simulator_external_id=simulator_external_id, + heartbeat=heartbeat, + data_set_id=data_set_id, + connector_version=connector_version, + license_status=license_status, + simulator_version=simulator_version, + license_last_checked_time=license_last_checked_time, + connector_status=connector_status, + connector_status_updated_time=connector_status_updated_time, + ) + + def as_write(self) -> SimulatorIntegrationWrite: + """Returns a writeable version of this resource""" + return self + + +class SimulatorIntegration(SimulatorIntegrationCore): + def __init__( + self, + external_id: str, + simulator_external_id: str, + heartbeat: int, + data_set_id: int, + connector_version: str, + license_status: str | None = None, + simulator_version: str | None = None, + license_last_checked_time: int | None = None, + connector_status: str | None = None, + connector_status_updated_time: int | None = None, + created_time: int | None = None, + last_updated_time: int | None = None, + id: int | None = None, + active: bool | None = None, + log_id: int | None = None, + ) -> None: + self.external_id = external_id + self.simulator_external_id = simulator_external_id + self.heartbeat = heartbeat + self.data_set_id = data_set_id + self.connector_version = connector_version + self.license_status = license_status + self.simulator_version = simulator_version + self.license_last_checked_time = license_last_checked_time + self.connector_status = connector_status + self.connector_status_updated_time = connector_status_updated_time + # 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 | None = id + self.created_time: int | None = created_time + self.last_updated_time: int | None = last_updated_time + self.active = active + self.log_id = log_id + + @classmethod + def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self: + load = super()._load(resource, cognite_client) + return cls( + external_id=load.external_id, + simulator_external_id=load.simulator_external_id, + heartbeat=load.heartbeat, + data_set_id=load.data_set_id, + connector_version=load.connector_version, + license_status=load.license_status, + simulator_version=load.simulator_version, + license_last_checked_time=load.license_last_checked_time, + connector_status=load.connector_status, + connector_status_updated_time=load.connector_status_updated_time, + created_time=resource.get("createdTime"), + last_updated_time=resource.get("lastUpdatedTime"), + id=resource.get("id"), + active=resource.get("active"), + log_id=resource.get("logId"), + ) + + def as_write(self) -> SimulatorIntegrationWrite: + """Returns a writeable version of this resource""" + return SimulatorIntegrationWrite( + external_id=self.external_id, + simulator_external_id=self.simulator_external_id, + heartbeat=self.heartbeat, + data_set_id=self.data_set_id, + connector_version=self.connector_version, + license_status=self.license_status, + simulator_version=self.simulator_version, + license_last_checked_time=self.license_last_checked_time, + connector_status=self.connector_status, + connector_status_updated_time=self.connector_status_updated_time, + ) + + def __hash__(self) -> int: + return hash(self.external_id) + + +class SimulatorModelRevisionCore(WriteableCogniteResource["SimulatorModelRevisionWrite"], ABC): + def __init__( + self, + external_id: str | None = None, + simulator_external_id: str | None = None, + model_external_id: str | None = None, + data_set_id: int | None = None, + file_id: int | None = None, + created_by_user_id: str | None = None, + status: str | None = None, + created_time: int | None = None, + last_updated_time: int | None = None, + version_number: int | None = None, + log_id: int | None = None, + description: str | None = None, + status_message: str | None = None, + ) -> None: + self.external_id = external_id + self.simulator_external_id = simulator_external_id + self.model_external_id = model_external_id + self.data_set_id = data_set_id + self.file_id = file_id + self.created_by_user_id = created_by_user_id + self.status = status + self.created_time = created_time + self.last_updated_time = last_updated_time + self.version_number = version_number + self.log_id = log_id + self.description = description + self.status_message = status_message + + @classmethod + def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self: + 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) + + +class SimulatorModelRevisionWrite(SimulatorModelRevisionCore): + def __init__( + self, + external_id: str | None = None, + model_external_id: str | None = None, + file_id: int | None = None, + description: str | None = None, + ) -> None: + super().__init__( + external_id=external_id, + model_external_id=model_external_id, + file_id=file_id, + description=description, + ) + + def as_write(self) -> SimulatorModelRevisionWrite: + """Returns a writeable version of this resource""" + return self + + @classmethod + def _load( + cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None + ) -> SimulatorModelRevisionWrite: + return cls( + external_id=resource.get("externalId"), + model_external_id=resource.get("modelExternalId"), + file_id=resource.get("fileId"), + description=resource.get("description"), + ) + + +class SimulatorModelRevision(SimulatorModelRevisionCore): + """ + + 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. + + Args: + id (int | None): No description. + external_id (str | None): External id of the simulator model revision + simulator_external_id (str | None): No description. + model_external_id (str | None): External id of the associated simulator model + data_set_id (int | None): The id of the dataset associated with the simulator model revision + file_id (int | None): The id of the file associated with the simulator model revision + created_by_user_id (str | None): The id of the user who created the simulator model revision + status (str | None): The status of the simulator model revision + created_time (int | None): The time when the simulator model revision was created + last_updated_time (int | None): The time when the simulator model revision was last updated + version_number (int | None): The version number of the simulator model revision + log_id (int | None): The id of the log associated with the simulator model revision + description (str | None): The description of the simulator model revision + status_message (str | None): The current status of the model revision + + """ + + def __init__( + self, + id: int | None = None, + external_id: str | None = None, + simulator_external_id: str | None = None, + model_external_id: str | None = None, + data_set_id: int | None = None, + file_id: int | None = None, + created_by_user_id: str | None = None, + status: str | None = None, + created_time: int | None = None, + last_updated_time: int | None = None, + version_number: int | None = None, + log_id: int | None = None, + description: str | None = None, + status_message: str | None = None, + ) -> None: + super().__init__( + external_id=external_id, + simulator_external_id=simulator_external_id, + model_external_id=model_external_id, + data_set_id=data_set_id, + file_id=file_id, + created_by_user_id=created_by_user_id, + status=status, + created_time=created_time, + last_updated_time=last_updated_time, + version_number=version_number, + log_id=log_id, + description=description, + status_message=status_message, + ) + # 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 | None = id + self.created_time: int | None = created_time + self.last_updated_time: int | None = last_updated_time + + @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) -> SimulatorModelRevisionWrite: + """Returns this SimulatorModelRevision in its writing version.""" + return SimulatorModelRevisionWrite( + external_id=self.external_id, + model_external_id=self.model_external_id, + file_id=self.file_id, + 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) + + +class SimulatorRoutineRevisionCore(WriteableCogniteResource["SimulatorRoutineRevisionWrite"], ABC): + def __init__( + self, + external_id: str, + simulator_external_id: str, + routine_external_id: str, + simulator_integration_external_id: str, + model_external_id: str, + data_set_id: int, + configuration: SimulatorRoutineConfiguration, + script: list[SimulatorRoutineStage], + ) -> None: + self.external_id = external_id + self.simulator_external_id = simulator_external_id + self.routine_external_id = routine_external_id + self.simulator_integration_external_id = simulator_integration_external_id + self.model_external_id = model_external_id + self.data_set_id = data_set_id + self.configuration = configuration + self.script = script + + @classmethod + def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self: + script = [] + + if resource.get("script", None) is not None: + script = [SimulatorRoutineStage._load(stage_, cognite_client) for stage_ in resource["script"]] + return cls( + external_id=resource["externalId"], + simulator_external_id=resource["simulatorExternalId"], + routine_external_id=resource["routineExternalId"], + simulator_integration_external_id=resource["simulatorIntegrationExternalId"], + model_external_id=resource["modelExternalId"], + data_set_id=resource["dataSetId"], + configuration=SimulatorRoutineConfiguration._load(resource["configuration"], cognite_client), + script=script, + ) + + def dump(self, camel_case: bool = True) -> dict[str, Any]: + output = super().dump(camel_case=camel_case) + output["configuration"] = self.configuration.dump(camel_case=camel_case) + output["script"] = [stage_.dump(camel_case=camel_case) for stage_ in self.script] + + return output + + +class SimulatorRoutineRevisionWrite(SimulatorRoutineRevisionCore): + def __init__( + self, + external_id: str, + simulator_external_id: str, + routine_external_id: str, + simulator_integration_external_id: str, + model_external_id: str, + data_set_id: int, + configuration: SimulatorRoutineConfiguration, + script: list[SimulatorRoutineStage], + ) -> None: + super().__init__( + external_id=external_id, + simulator_external_id=simulator_external_id, + routine_external_id=routine_external_id, + simulator_integration_external_id=simulator_integration_external_id, + model_external_id=model_external_id, + data_set_id=data_set_id, + configuration=configuration, + script=script, + ) + + def as_write(self) -> SimulatorRoutineRevisionWrite: + """Returns a writeable version of this resource""" + return self + + +class SimulatorRoutineRevision(SimulatorRoutineRevisionCore): + def __init__( + self, + external_id: str, + simulator_external_id: str, + routine_external_id: str, + simulator_integration_external_id: str, + model_external_id: str, + data_set_id: int, + configuration: SimulatorRoutineConfiguration, + script: list[SimulatorRoutineStage], + id: int | None = None, + created_by_user_id: str | None = None, + last_updated_time: int | None = None, + version_number: int | None = None, + created_time: int | None = None, + log_id: int | None = None, + ) -> None: + self.external_id = external_id + self.simulator_external_id = simulator_external_id + self.routine_external_id = routine_external_id + self.simulator_integration_external_id = simulator_integration_external_id + self.model_external_id = model_external_id + self.data_set_id = data_set_id + self.created_by_user_id = created_by_user_id + self.configuration = configuration + self.script = script + + # 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 | None = id + self.created_time: int | None = created_time + self.last_updated_time: int | None = last_updated_time + self.version_number = version_number + self.log_id = log_id + + @classmethod + def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self: + load = super()._load(resource, cognite_client) + return cls( + external_id=load.external_id, + simulator_external_id=load.simulator_external_id, + routine_external_id=load.routine_external_id, + simulator_integration_external_id=load.simulator_integration_external_id, + model_external_id=load.model_external_id, + data_set_id=load.data_set_id, + created_by_user_id=resource.get("createdByUserId"), + configuration=load.configuration, + script=load.script, + id=resource.get("id"), + created_time=resource.get("createdTime"), + last_updated_time=resource.get("lastUpdatedTime"), + version_number=resource.get("versionNumber"), + log_id=resource.get("logId"), + ) + + def as_write(self) -> SimulatorRoutineRevisionWrite: + """Returns a writeable version of this resource""" + return SimulatorRoutineRevisionWrite( + external_id=self.external_id, + simulator_external_id=self.simulator_external_id, + routine_external_id=self.routine_external_id, + simulator_integration_external_id=self.simulator_integration_external_id, + model_external_id=self.model_external_id, + data_set_id=self.data_set_id, + configuration=self.configuration, + script=self.script, + ) + + +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. + 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. + + 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 + type (str | None): The type key of the simulator model + description (str | None): The description of the simulator model + """ + + 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: + self.external_id = external_id + self.simulator_external_id = simulator_external_id + self.data_set_id = data_set_id + self.name = name + self.type = type + self.description = description + + @classmethod + def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self: + 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) + + +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( + external_id=resource.get("externalId"), + simulator_external_id=resource.get("simulatorExternalId"), + data_set_id=resource.get("dataSetId"), + name=resource.get("name"), + type=resource.get("type"), + 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. + + + 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: + load = super()._load(resource, cognite_client) + return cls( + external_id=load.external_id, + simulator_external_id=load.simulator_external_id, + data_set_id=load.data_set_id, + name=load.name, + id=resource.get("id"), + type=resource.get("type"), + description=load.description, + created_time=resource.get("createdTime"), + last_updated_time=resource.get("lastUpdatedTime"), + ) + + 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) + + +class SimulationRunCore(CogniteResource, ABC): + """""" + + """_summary_ + + Returns: + _type_: _description_ + """ + + def __init__( + self, + simulator_external_id: str | None = None, + simulator_integration_external_id: str | None = None, + model_external_id: str | None = None, + model_revision_external_id: str | None = None, + routine_external_id: str | None = None, + routine_revision_external_id: str | None = None, + run_time: int | None = None, + simulation_time: int | None = None, + status: str | None = None, + status_message: str | None = None, + data_set_id: int | None = None, + run_type: str | None = None, + user_id: str | None = None, + log_id: int | None = None, + ) -> None: + self.simulator_external_id = simulator_external_id + self.simulator_integration_external_id = simulator_integration_external_id + self.model_external_id = model_external_id + self.model_revision_external_id = model_revision_external_id + self.routine_external_id = routine_external_id + self.routine_revision_external_id = routine_revision_external_id + self.run_time = run_time + self.simulation_time = simulation_time + self.status = status + self.status_message = status_message + self.data_set_id = data_set_id + self.run_type = run_type + self.user_id = user_id + self.log_id = log_id + + @classmethod + def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self: + instance = super()._load(resource, cognite_client) + return instance + + def dump(self, camel_case: bool = True) -> dict[str, Any]: + return super().dump(camel_case) + + +class SimulationRun(SimulationRunCore): + """ + Every time a simulation routine executes, a simulation run object is created. + This object ensures that each execution of a routine is documented and traceable. + Each run has an associated simulation data resource, which stores the inputs and outputs of a + simulation run, capturing the values set into and read from the simulator model to ensure + the traceability and integrity of the simulation data. + + Simulation runs provide a historical record of the simulations performed, allowing users to analyze + and compare different runs, track changes over time, and make informed decisions based on the simulation results. + + This is the read/response format of a simulation run. + + Args: + simulator_external_id (str | None): External id of the associated simulator + simulator_integration_external_id (str | None): External id of the associated simulator integration + model_external_id (str | None): External id of the associated simulator model + model_revision_external_id (str | None): External id of the associated simulator model revision + routine_external_id (str | None): External id of the associated simulator routine + routine_revision_external_id (str | None): External id of the associated simulator routine revision + run_time (int | None): Run time in milliseconds. Reference timestamp used for data pre-processing and data sampling. + simulation_time (int | None): Simulation time in milliseconds. Timestamp when the input data was sampled. Used for indexing input and output time series. + status (str | None): The status of the simulation run + status_message (str | None): The status message of the simulation run + data_set_id (int | None): The id of the dataset associated with the simulation run + run_type (str | None): The type of the simulation run + user_id (str | None): The id of the user who executed the simulation run + log_id (int | None): The id of the log associated with the simulation run + id (int | None): A unique id of a simulation run + created_time (int | None): The number of milliseconds since epoch + last_updated_time (int | None): The number of milliseconds since epoch + + """ + + def __init__( + self, + simulator_external_id: str | None = None, + simulator_integration_external_id: str | None = None, + model_external_id: str | None = None, + model_revision_external_id: str | None = None, + routine_external_id: str | None = None, + routine_revision_external_id: str | None = None, + run_time: int | None = None, + simulation_time: int | None = None, + status: str | None = None, + status_message: str | None = None, + data_set_id: int | None = None, + run_type: str | None = None, + user_id: str | None = None, + log_id: int | None = None, + id: int | None = None, + created_time: int | None = None, + last_updated_time: int | None = None, + ) -> None: + super().__init__( + simulator_external_id=simulator_external_id, + simulator_integration_external_id=simulator_integration_external_id, + model_external_id=model_external_id, + model_revision_external_id=model_revision_external_id, + routine_external_id=routine_external_id, + routine_revision_external_id=routine_revision_external_id, + run_time=run_time, + simulation_time=simulation_time, + status=status, + status_message=status_message, + data_set_id=data_set_id, + run_type=run_type, + user_id=user_id, + log_id=log_id, + ) + # 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]. + # TODO: In the next major version we can make these properties required in the constructor + 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, cognite_client: CogniteClient | None = None) -> SimulationRun: + return super()._load(resource, cognite_client) + + def dump(self, camel_case: bool = True) -> dict[str, Any]: + return super().dump(camel_case=camel_case) + + def __hash__(self) -> int: + return hash(self.id) + + +class SimulatorRoutineCore(WriteableCogniteResource["SimulatorRoutineWrite"], ABC): + """ + The simulator routine resource defines instructions on interacting with a simulator model. A simulator routine includes: + + * Inputs (values set into the simulator model) + * Commands (actions to be performed by the simulator) + * Outputs (values read from the simulator model) + + Simulator routines can have multiple revisions, enabling users to track changes and evolve the routine over time. + Each model can have multiple routines, each performing different objectives such as calculating optimal + operation setpoints, forecasting production, benchmarking asset performance, and more. + + Each simulator routine can have a maximum of 10 revisions + + This is the read/response format of a simulator routine. + + Args: + external_id (str): External id of the simulator routine + simulator_external_id (str): External id of the associated simulator + model_external_id (str): External id of the associated simulator model + simulator_integration_external_id (str): External id of the associated simulator integration + name (str): The name of the simulator routine + data_set_id (int): The id of the dataset associated with the simulator routine + description (str | None): The description of the simulator routine + """ + + def __init__( + self, + external_id: str, + simulator_external_id: str, + model_external_id: str, + simulator_integration_external_id: str, + name: str, + data_set_id: int, + description: str | None = None, + ) -> None: + self.external_id = external_id + self.simulator_external_id = simulator_external_id + self.model_external_id = model_external_id + self.simulator_integration_external_id = simulator_integration_external_id + self.name = name + self.data_set_id = data_set_id + self.description = description + + @classmethod + def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self: + return cls( + external_id=resource["externalId"], + simulator_external_id=resource["simulatorExternalId"], + model_external_id=resource["modelExternalId"], + simulator_integration_external_id=resource["simulatorIntegrationExternalId"], + name=resource["name"], + data_set_id=resource["dataSetId"], + description=resource.get("description"), + ) + + def dump(self, camel_case: bool = True) -> dict[str, Any]: + return super().dump(camel_case=camel_case) + + +class SimulatorRoutineWrite(SimulatorRoutineCore): + def __init__( + self, + external_id: str, + simulator_external_id: str, + model_external_id: str, + simulator_integration_external_id: str, + name: str, + data_set_id: int, + description: str | None = None, + ) -> None: + super().__init__( + external_id=external_id, + simulator_external_id=simulator_external_id, + model_external_id=model_external_id, + simulator_integration_external_id=simulator_integration_external_id, + name=name, + data_set_id=data_set_id, + description=description, + ) + + def as_write(self) -> SimulatorRoutineWrite: + """Returns a writeable version of this resource""" + return self + + +class SimulatorRoutine(SimulatorRoutineCore): + def __init__( + self, + external_id: str, + simulator_external_id: str, + model_external_id: str, + simulator_integration_external_id: str, + name: str, + data_set_id: int, + description: str | None = None, + created_time: int | None = None, + last_updated_time: int | None = None, + id: int | None = None, + ) -> None: + self.external_id = external_id + self.simulator_external_id = simulator_external_id + self.model_external_id = model_external_id + self.simulator_integration_external_id = simulator_integration_external_id + self.name = name + self.data_set_id = data_set_id + self.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 | None = id + self.created_time: int | None = created_time + self.last_updated_time: int | None = last_updated_time + + @classmethod + def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self: + instance = cls( + external_id=resource["externalId"], + simulator_external_id=resource["simulatorExternalId"], + model_external_id=resource["modelExternalId"], + simulator_integration_external_id=resource["simulatorIntegrationExternalId"], + name=resource["name"], + data_set_id=resource["dataSetId"], + description=resource.get("description"), + created_time=resource.get("createdTime"), + last_updated_time=resource.get("lastUpdatedTime"), + id=resource.get("id"), + ) + return instance + + def as_write(self) -> SimulatorRoutineWrite: + """Returns a writeable version of this resource""" + return SimulatorRoutineWrite( + external_id=self.external_id, + simulator_external_id=self.simulator_external_id, + model_external_id=self.model_external_id, + simulator_integration_external_id=self.simulator_integration_external_id, + name=self.name, + data_set_id=self.data_set_id, + description=self.description, + ) + + def __hash__(self) -> int: + return hash(self.external_id) + + +class PropertySort(CogniteSort): + def dump(self, camel_case: bool = True) -> dict[str, Any]: + dumped = super().dump(camel_case=camel_case) + dumped["property"] = self.property + return dumped + + +class CreatedTimeSort(PropertySort): + def __init__( + self, + property: Literal["createdTime"] = "createdTime", + order: Literal["asc", "desc"] = "asc", + ): + super().__init__(property, order) + + +class SimulationTimeSort(PropertySort): + def __init__( + self, + property: Literal["simulationTime"] = "simulationTime", + order: Literal["asc", "desc"] = "asc", + ): + super().__init__(property, order) + + +class SimulatorRoutineRevisionWriteList(CogniteResourceList[SimulatorRoutineRevisionWrite], ExternalIDTransformerMixin): + _RESOURCE = SimulatorRoutineRevisionWrite + + +class SimulatorRoutineRevisionList( + WriteableCogniteResourceList[SimulatorRoutineRevisionWrite, SimulatorRoutineRevision], IdTransformerMixin +): + _RESOURCE = SimulatorRoutineRevision + + def as_write(self) -> SimulatorRoutineRevisionWriteList: + return SimulatorRoutineRevisionWriteList( + [a.as_write() for a in self.data], cognite_client=self._get_cognite_client() + ) + + +class SimulatorRoutineWriteList(CogniteResourceList[SimulatorRoutineWrite], ExternalIDTransformerMixin): + _RESOURCE = SimulatorRoutineWrite + + +class SimulatorRoutineList(WriteableCogniteResourceList[SimulatorRoutineWrite, SimulatorRoutine], IdTransformerMixin): + _RESOURCE = SimulatorRoutine + + def as_write(self) -> SimulatorRoutineWriteList: + return SimulatorRoutineWriteList([a.as_write() for a in self.data], cognite_client=self._get_cognite_client()) + + +class SimulatorRoutineRevisionsList( + WriteableCogniteResourceList[SimulatorRoutineRevisionWrite, SimulatorRoutineRevision], IdTransformerMixin +): + _RESOURCE = SimulatorRoutineRevision + + def as_write(self) -> SimulatorRoutineRevisionWriteList: + return SimulatorRoutineRevisionWriteList( + [a.as_write() for a in self.data], cognite_client=self._get_cognite_client() + ) + + +class SimulatorWriteList(CogniteResourceList[SimulatorWrite], ExternalIDTransformerMixin): + _RESOURCE = SimulatorWrite + + +class SimulatorList(WriteableCogniteResourceList[SimulatorWrite, Simulator], IdTransformerMixin): + _RESOURCE = Simulator + + def as_write(self) -> SimulatorWriteList: + return SimulatorWriteList([a.as_write() for a in self.data], cognite_client=self._get_cognite_client()) + + +class SimulatorIntegrationWriteList(CogniteResourceList[SimulatorIntegrationWrite], ExternalIDTransformerMixin): + _RESOURCE = SimulatorIntegrationWrite + + +class SimulatorIntegrationList( + WriteableCogniteResourceList[SimulatorIntegrationWrite, SimulatorIntegration], IdTransformerMixin +): + _RESOURCE = SimulatorIntegration + + def as_write(self) -> SimulatorIntegrationWriteList: + return SimulatorIntegrationWriteList( + [a.as_write() for a in self.data], cognite_client=self._get_cognite_client() + ) + + +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 SimulatorModelRevisionWriteList(CogniteResourceList[SimulatorModelRevisionWrite], ExternalIDTransformerMixin): + _RESOURCE = SimulatorModelRevisionWrite + + +class SimulatorModelRevisionList( + WriteableCogniteResourceList[SimulatorModelRevisionWrite, SimulatorModelRevision], IdTransformerMixin +): + _RESOURCE = SimulatorModelRevision + + def as_write(self) -> SimulatorModelRevisionWriteList: + return SimulatorModelRevisionWriteList( + [a.as_write() for a in self.data], cognite_client=self._get_cognite_client() + ) + + +class SimulationRunsList(CogniteResourceList[SimulationRun], InternalIdTransformerMixin): + _RESOURCE = SimulationRun diff --git a/cognite/client/testing.py b/cognite/client/testing.py index 2a7be50abd..74cabb499c 100644 --- a/cognite/client/testing.py +++ b/cognite/client/testing.py @@ -43,6 +43,11 @@ 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.simulation_runs import SimulatorRunsAPI +from cognite.client._api.simulators.simulator_integrations import SimulatorIntegrationsAPI +from cognite.client._api.simulators.simulator_models import SimulatorModelRevisionsAPI, SimulatorModelsAPI +from cognite.client._api.simulators.simulator_routines import SimulatorRoutineRevisionsAPI, SimulatorRoutinesAPI +from cognite.client._api.simulators.simulators import SimulatorsAPI from cognite.client._api.synthetic_time_series import SyntheticDatapointsAPI from cognite.client._api.templates import ( TemplateGroupsAPI, @@ -142,6 +147,14 @@ def __init__(self, *args: Any, **kwargs: Any) -> None: self.relationships = MagicMock(spec_set=RelationshipsAPI) + self.simulators = MagicMock(spec=SimulatorsAPI) + self.simulators.runs = MagicMock(spec_set=SimulatorRunsAPI) + self.simulators.integrations = MagicMock(spec_set=SimulatorIntegrationsAPI) + self.simulators.models = MagicMock(spec=SimulatorModelsAPI) + self.simulators.routines = MagicMock(spec=SimulatorRoutinesAPI) + self.simulators.models.revisions = MagicMock(spec_set=SimulatorModelRevisionsAPI) + self.simulators.routines.revisions = MagicMock(spec_set=SimulatorRoutineRevisionsAPI) + self.sequences = MagicMock(spec=SequencesAPI) self.sequences.data = MagicMock(spec_set=SequencesDataAPI) diff --git a/tests/tests_integration/test_api/test_simulators/seed/ShowerMixer.dwxmz b/tests/tests_integration/test_api/test_simulators/seed/ShowerMixer.dwxmz new file mode 100644 index 0000000000..b9c9348359 Binary files /dev/null and b/tests/tests_integration/test_api/test_simulators/seed/ShowerMixer.dwxmz differ diff --git a/tests/tests_integration/test_api/test_simulators/seed/data.py b/tests/tests_integration/test_api/test_simulators/seed/data.py new file mode 100644 index 0000000000..e3f18cde78 --- /dev/null +++ b/tests/tests_integration/test_api/test_simulators/seed/data.py @@ -0,0 +1,400 @@ +import time + +# data_set_id = 1521375514069 +# development_data_set_id = 97552494921583 +data_set_id = 97552494921583 + +resource_names = { + "simulator_external_id": "py_sdk_integration_tests", + "simulator_integration_external_id": "py_sdk_integration_tests_connector", + "simulator_model_external_id": "py_sdk_integration_tests_model", + "simulator_model_revision_external_id": "pysdk_model_revision", + "simulator_model_file_external_id": "ShowerMixer_simulator_model_file_3", + "simulator_routine_external_id": "pysdk_routine", + "simulator_routine_revision_external_id": "pysdk_routine_revision", + "simulator_test_data_set_id": data_set_id, +} + +simulator = { + "name": resource_names["simulator_external_id"], + "externalId": resource_names["simulator_external_id"], + "fileExtensionTypes": ["dwxmz"], + "modelTypes": [{"name": "Steady State", "key": "SteadyState"}], + "stepFields": [ + { + "stepType": "get/set", + "fields": [ + { + "name": "objectName", + "label": "Simulation Object Name", + "info": "Enter the name of the DWSIM object, i.e. Feed", + }, + { + "name": "objectProperty", + "label": "Simulation Object Property", + "info": "Enter the property of the DWSIM object, i.e. Temperature", + }, + ], + }, + { + "stepType": "command", + "fields": [ + { + "name": "command", + "label": "Command", + "info": "Select a command", + "options": [{"label": "Solve Flowsheet", "value": "Solve"}], + } + ], + }, + ], + "unitQuantities": [ + { + "name": "mass", + "label": "Mass", + "units": [{"label": "kg", "name": "kg"}, {"label": "g", "name": "g"}, {"label": "lb", "name": "lb"}], + }, + { + "name": "time", + "label": "Time", + "units": [{"label": "s", "name": "s"}, {"label": "min.", "name": "min."}, {"label": "h", "name": "h"}], + }, + { + "name": "accel", + "label": "Acceleration", + "units": [ + {"label": "m/s2", "name": "m/s2"}, + {"label": "cm/s2", "name": "cm/s2"}, + {"label": "ft/s2", "name": "ft/s2"}, + ], + }, + { + "name": "force", + "label": "Force", + "units": [ + {"label": "N", "name": "N"}, + {"label": "dyn", "name": "dyn"}, + {"label": "kgf", "name": "kgf"}, + {"label": "lbf", "name": "lbf"}, + ], + }, + { + "name": "volume", + "label": "Volume", + "units": [ + {"label": "m3", "name": "m3"}, + {"label": "cm3", "name": "cm3"}, + {"label": "L", "name": "L"}, + {"label": "ft3", "name": "ft3"}, + {"label": "bbl", "name": "bbl"}, + {"label": "gal[US]", "name": "gal[US]"}, + {"label": "gal[UK]", "name": "gal[UK]"}, + ], + }, + { + "name": "density", + "label": "Density", + "units": [ + {"label": "kg/m3", "name": "kg/m3"}, + {"label": "g/cm3", "name": "g/cm3"}, + {"label": "lbm/ft3", "name": "lbm/ft3"}, + ], + }, + { + "name": "diameter", + "label": "Diameter", + "units": [{"label": "mm", "name": "mm"}, {"label": "in", "name": "in"}], + }, + { + "name": "distance", + "label": "Distance", + "units": [{"label": "m", "name": "m"}, {"label": "ft", "name": "ft"}, {"label": "cm", "name": "cm"}], + }, + { + "name": "heatflow", + "label": "Heat Flow", + "units": [ + {"label": "kW", "name": "kW"}, + {"label": "kcal/h", "name": "kcal/h"}, + {"label": "BTU/h", "name": "BTU/h"}, + {"label": "BTU/s", "name": "BTU/s"}, + {"label": "cal/s", "name": "cal/s"}, + {"label": "HP", "name": "HP"}, + {"label": "kJ/h", "name": "kJ/h"}, + {"label": "kJ/d", "name": "kJ/d"}, + {"label": "MW", "name": "MW"}, + {"label": "W", "name": "W"}, + {"label": "BTU/d", "name": "BTU/d"}, + {"label": "MMBTU/d", "name": "MMBTU/d"}, + {"label": "MMBTU/h", "name": "MMBTU/h"}, + {"label": "kcal/s", "name": "kcal/s"}, + {"label": "kcal/h", "name": "kcal/h"}, + {"label": "kcal/d", "name": "kcal/d"}, + ], + }, + { + "name": "pressure", + "label": "Pressure", + "units": [ + {"label": "Pa", "name": "Pa"}, + {"label": "atm", "name": "atm"}, + {"label": "kgf/cm2", "name": "kgf/cm2"}, + {"label": "kgf/cm2g", "name": "kgf/cm2g"}, + {"label": "lbf/ft2", "name": "lbf/ft2"}, + {"label": "kPa", "name": "kPa"}, + {"label": "kPag", "name": "kPag"}, + {"label": "bar", "name": "bar"}, + {"label": "barg", "name": "barg"}, + {"label": "ftH2O", "name": "ftH2O"}, + {"label": "inH2O", "name": "inH2O"}, + {"label": "inHg", "name": "inHg"}, + {"label": "mbar", "name": "mbar"}, + {"label": "mH2O", "name": "mH2O"}, + {"label": "mmH2O", "name": "mmH2O"}, + {"label": "mmHg", "name": "mmHg"}, + {"label": "MPa", "name": "MPa"}, + {"label": "psi", "name": "psi"}, + {"label": "psig", "name": "psig"}, + ], + }, + { + "name": "velocity", + "label": "Velocity", + "units": [ + {"label": "m/s", "name": "m/s"}, + {"label": "cm/s", "name": "cm/s"}, + {"label": "mm/s", "name": "mm/s"}, + {"label": "km/h", "name": "km/h"}, + {"label": "ft/h", "name": "ft/h"}, + {"label": "ft/min", "name": "ft/min"}, + {"label": "ft/s", "name": "ft/s"}, + {"label": "in/s", "name": "in/s"}, + ], + }, + { + "name": "temperature", + "label": "Temperature", + "units": [ + {"label": "K", "name": "K"}, + {"label": "R", "name": "R"}, + {"label": "C", "name": "C"}, + {"label": "F", "name": "F"}, + ], + }, + { + "name": "volumetricFlow", + "label": "Volumetric Flow", + "units": [ + {"label": "m3/h", "name": "m3/h"}, + {"label": "cm3/s", "name": "cm3/s"}, + {"label": "L/h", "name": "L/h"}, + {"label": "L/min", "name": "L/min"}, + {"label": "L/s", "name": "L/s"}, + {"label": "ft3/h", "name": "ft3/h"}, + {"label": "ft3/min", "name": "ft3/min"}, + {"label": "ft3/s", "name": "ft3/s"}, + {"label": "gal[US]/h", "name": "gal[US]/h"}, + {"label": "gal[US]/min", "name": "gal[US]/min"}, + {"label": "gal[US]/s", "name": "gal[US]/s"}, + {"label": "gal[UK]/h", "name": "gal[UK]/h"}, + {"label": "gal[UK]/min", "name": "gal[UK]/min"}, + {"label": "gal[UK]/s", "name": "gal[UK]/s"}, + ], + }, + ], +} + + +simulator_integration = { + "externalId": resource_names["simulator_integration_external_id"], + "simulatorExternalId": resource_names["simulator_external_id"], + "heartbeat": int(time.time() * 1000), + "dataSetId": resource_names["simulator_test_data_set_id"], + "connectorVersion": "1.0.0", + "simulatorVersion": "1.0.0", + "licenseStatus": "AVAILABLE", + "licenseLastCheckedTime": 0, + "connectorStatus": "IDLE", + "connectorStatusUpdatedTime": 0, +} + + +simulator_model = { + "externalId": resource_names["simulator_model_external_id"], + "simulatorExternalId": resource_names["simulator_external_id"], + "name": "Test Simulator Model", + "description": "Test Simulator Model Desc", + "dataSetId": resource_names["simulator_test_data_set_id"], + "type": "SteadyState", +} + +simulator_model_revision = { + "externalId": resource_names["simulator_model_revision_external_id"], + "modelExternalId": resource_names["simulator_model_external_id"], + "description": "test sim model revision description", + "fileId": 00000000000000, +} + +simulator_routine = { + "externalId": resource_names["simulator_routine_external_id"], + "modelExternalId": resource_names["simulator_model_external_id"], + "simulatorIntegrationExternalId": resource_names["simulator_integration_external_id"], + "name": "Routine test", + "description": "test", +} + +simulator_routine_revision = { + "externalId": resource_names["simulator_routine_revision_external_id"], + "routineExternalId": resource_names["simulator_routine_external_id"], + "configuration": { + "schedule": {"enabled": True, "cronExpression": "*/10 * * * *"}, + "dataSampling": {"enabled": True, "validationWindow": None, "samplingWindow": 15, "granularity": 1}, + "logicalCheck": [], + "steadyStateDetection": [], + "inputs": [ + { + "name": "Cold Water Temperature - Constant", + "referenceId": "CWTC", + "value": 10, + "valueType": "DOUBLE", + "unit": {"name": "C", "quantity": "temperature"}, + "saveTimeseriesExternalId": "DWSIM-INPUT-constinput-CWTC-ShowerMixer-AI", + }, + { + "name": "Cold Water Pressure - Constant", + "referenceId": "CWPC", + "value": 3.6, + "valueType": "DOUBLE", + "unit": {"name": "bar", "quantity": "pressure"}, + "saveTimeseriesExternalId": "DWSIM-INPUT-constinput-CWPC-ShowerMixer", + }, + { + "name": "Cold Water Volumetric Flow - Constant", + "referenceId": "CWVFC", + "value": 0.37, + "valueType": "DOUBLE", + "unit": {"name": "m3/h", "quantity": "volumetricFlow"}, + "saveTimeseriesExternalId": "DWSIM-INPUT-constinput-CWVFC-ShowerMixer-AI", + }, + { + "name": "Hot Water Temperature - Constant", + "referenceId": "HWTC", + "value": 69, + "valueType": "DOUBLE", + "unit": {"name": "C", "quantity": "temperature"}, + "saveTimeseriesExternalId": "DWSIM-INPUT-constinput-HWTC-ShowerMixer-AI", + }, + { + "name": "Hot Water Pressure - Constant", + "referenceId": "HWPC", + "value": 2.8, + "valueType": "DOUBLE", + "unit": {"name": "bar", "quantity": "pressure"}, + "saveTimeseriesExternalId": "DWSIM-INPUT-constinput-HWPC-ShowerMixer-AI", + }, + { + "name": "Hot Water Volumetric Flow - Constant", + "referenceId": "HWVFC", + "value": 0.19, + "valueType": "DOUBLE", + "unit": {"name": "m3/h", "quantity": "volumetricFlow"}, + "saveTimeseriesExternalId": "DWSIM-INPUT-constinput-HWVFC-ShowerMixer-AI", + }, + ], + "outputs": [ + { + "name": "Shower Temperature", + "referenceId": "ST", + "unit": {"name": "C", "quantity": "temperature"}, + "valueType": "DOUBLE", + "saveTimeseriesExternalId": "DWSIM-OUTPUT-const_input-ST-ShowerMixer-AI", + }, + { + "name": "Shower Pressure", + "referenceId": "SP", + "unit": {"name": "bar", "quantity": "pressure"}, + "valueType": "DOUBLE", + "saveTimeseriesExternalId": "DWSIM-OUTPUT-const_input-SP-ShowerMixer-AI", + }, + { + "name": "Shower Volumetric Flow", + "referenceId": "SVF", + "unit": {"name": "m3/h", "quantity": "volumetricFlow"}, + "valueType": "DOUBLE", + "saveTimeseriesExternalId": "DWSIM-OUTPUT-const_input-SVF-ShowerMixer-AI", + }, + ], + }, + "script": [ + { + "order": 1, + "description": "Set Inputs", + "steps": [ + { + "order": 1, + "stepType": "Set", + "arguments": {"referenceId": "CWTC", "objectName": "Cold water", "objectProperty": "Temperature"}, + }, + { + "order": 2, + "stepType": "Set", + "arguments": {"referenceId": "CWPC", "objectName": "Cold water", "objectProperty": "Pressure"}, + }, + { + "order": 3, + "stepType": "Set", + "arguments": { + "referenceId": "CWVFC", + "objectName": "Cold water", + "objectProperty": "Volumetric Flow", + }, + }, + { + "order": 4, + "stepType": "Set", + "arguments": {"referenceId": "HWTC", "objectName": "Hot water", "objectProperty": "Temperature"}, + }, + { + "order": 5, + "stepType": "Set", + "arguments": {"referenceId": "HWPC", "objectName": "Hot water", "objectProperty": "Pressure"}, + }, + { + "order": 6, + "stepType": "Set", + "arguments": { + "referenceId": "HWVFC", + "objectName": "Hot water", + "objectProperty": "Volumetric Flow", + }, + }, + ], + }, + { + "order": 2, + "description": "Solve the flowsheet", + "steps": [{"order": 1, "stepType": "Command", "arguments": {"command": "Solve"}}], + }, + { + "order": 3, + "description": "Set simulation outputs", + "steps": [ + { + "order": 1, + "stepType": "Get", + "arguments": {"referenceId": "ST", "objectName": "Shower", "objectProperty": "Temperature"}, + }, + { + "order": 2, + "stepType": "Get", + "arguments": {"referenceId": "SP", "objectName": "Shower", "objectProperty": "Pressure"}, + }, + { + "order": 3, + "stepType": "Get", + "arguments": {"referenceId": "SVF", "objectName": "Shower", "objectProperty": "Volumetric Flow"}, + }, + ], + }, + ], +} diff --git a/tests/tests_integration/test_api/test_simulators/test_simulators.py b/tests/tests_integration/test_api/test_simulators/test_simulators.py new file mode 100644 index 0000000000..d9c2aa90d7 --- /dev/null +++ b/tests/tests_integration/test_api/test_simulators/test_simulators.py @@ -0,0 +1,290 @@ +import random +import time +import uuid + +import pytest + +from cognite.client import CogniteClient +from cognite.client.data_classes.files import FileMetadata +from cognite.client.data_classes.simulators.filters import ( + SimulatorIntegrationFilter, + SimulatorModelRevisionsFilter, + SimulatorModelsFilter, + SimulatorRoutineRevisionsFilter, + SimulatorRoutinesFilter, +) +from cognite.client.data_classes.simulators.simulators import CreatedTimeSort +from cognite.client.exceptions import CogniteAPIError +from tests.tests_integration.test_api.test_simulators.seed.data import ( + data_set_id, + resource_names, + simulator, + simulator_integration, + simulator_model, + simulator_model_revision, + simulator_routine, + simulator_routine_revision, +) + + +def truncated_uuid4(): + return str(uuid.uuid4())[:8] + + +model_unique_external_id = f"{resource_names['simulator_model_external_id']}_{truncated_uuid4()}" +model_revision_unique_external_id = f"{resource_names['simulator_model_revision_external_id']}_{truncated_uuid4()}" +simulator_routine_unique_external_id = f"{resource_names['simulator_routine_external_id']}_{truncated_uuid4()}" + + +@pytest.fixture(scope="class") +def seed_resource_names() -> dict[str, str]: + return resource_names + + +@pytest.fixture +def seed_file(cognite_client: CogniteClient, seed_resource_names) -> FileMetadata | None: + # check if file already exists + file = cognite_client.files.retrieve(external_id=seed_resource_names["simulator_model_file_external_id"]) + if (file is None) or (file is False): + file = cognite_client.files.upload( + path="tests/tests_integration/test_api/test_simulators/seed/ShowerMixer.dwxmz", + external_id=seed_resource_names["simulator_model_file_external_id"], + name="ShowerMixer.dwxmz", + data_set_id=data_set_id, + ) + yield file + + +@pytest.fixture +def seed_simulator(cognite_client: CogniteClient, seed_resource_names) -> None: + simulator_external_id = seed_resource_names["simulator_external_id"] + simulators = cognite_client.simulators.list() + simulator_exists = len(list(filter(lambda x: x.external_id == simulator_external_id, simulators))) > 0 + if not simulator_exists: + cognite_client.post( + f"/api/v1/projects/{cognite_client.config.project}/simulators", + json={"items": [simulator]}, + ) + + +@pytest.fixture +def seed_simulator_integration(cognite_client: CogniteClient, seed_simulator) -> None: + try: + simulator_integration["heartbeat"] = int(time.time() * 1000) + cognite_client.post( + f"/api/v1/projects/{cognite_client.config.project}/simulators/integrations", + json={"items": [simulator_integration]}, + ) + except CogniteAPIError: + simulator_integrations = cognite_client.simulators.integrations.list() + integration_id = next( + filter( + lambda x: x.external_id == simulator_integration["externalId"], + simulator_integrations, + ) + ).id + # update hearbeat instead + cognite_client.post( + f"/api/v1/projects/{cognite_client.config.project}/simulators/integrations/update", + json={"items": [{"id": integration_id, "update": {"heartbeat": {"set": int(time.time() * 1000)}}}]}, + ) + + +@pytest.fixture +def seed_simulator_models(cognite_client: CogniteClient, seed_simulator_integration) -> None: + models = cognite_client.simulators.models.list() + model_exists = len(list(filter(lambda x: x.external_id == model_unique_external_id, models))) > 0 + + if not model_exists: + cognite_client.post( + f"/api/v1/projects/{cognite_client.config.project}/simulators/models", + json={ + "items": [{**simulator_model, "externalId": model_unique_external_id}] + }, # Post actual simulator models here + ) + + +@pytest.fixture +def seed_simulator_model_revisions(cognite_client: CogniteClient, seed_simulator_models, seed_file) -> None: + model_revisions = cognite_client.simulators.models.revisions.list( + filter=SimulatorModelRevisionsFilter(model_external_ids=[model_unique_external_id]) + ) + model_revision_not_exists = ( + len(list(filter(lambda x: x.external_id == model_revision_unique_external_id, model_revisions))) == 0 + ) + + if model_revision_not_exists: + cognite_client.post( + f"/api/v1/projects/{cognite_client.config.project}/simulators/models/revisions", + json={ + "items": [ + { + **simulator_model_revision, + "fileId": seed_file.id, + "modelExternalId": model_unique_external_id, + "externalId": model_revision_unique_external_id, + } + ] + }, + ) + + +@pytest.fixture +def seed_simulator_routines(cognite_client: CogniteClient, seed_simulator_model_revisions) -> None: + routines = cognite_client.simulators.routines.list( + filter=SimulatorRoutinesFilter(model_external_ids=[model_unique_external_id]) + ) + routine_not_exists = ( + len(list(filter(lambda x: x.external_id == simulator_routine_unique_external_id, routines))) == 0 + ) + + if routine_not_exists: + cognite_client.post( + f"/api/v1/projects/{cognite_client.config.project}/simulators/routines", + json={ + "items": [ + { + **simulator_routine, + "modelExternalId": model_unique_external_id, + "externalId": simulator_routine_unique_external_id, + } + ] + }, + ) + + +@pytest.fixture +def seed_simulator_routine_revisions(cognite_client: CogniteClient, seed_simulator_routines) -> None: + revisions_all = cognite_client.simulators.routines.revisions.list( + filter=SimulatorRoutineRevisionsFilter( + routine_external_ids=[simulator_routine_unique_external_id], all_versions=True + ), + ) + + if len(revisions_all) == 0: + for _ in range(10): + revision = { + **simulator_routine_revision, + "externalId": f"{simulator_routine_revision['externalId']}_{truncated_uuid4()}", + "routineExternalId": simulator_routine_unique_external_id, + } + cognite_client.post( + f"/api/v1/projects/{cognite_client.config.project}/simulators/routines/revisions", + json={"items": [revision]}, + ) + + +@pytest.mark.usefixtures("seed_resource_names", "seed_simulator") +class TestSimulators: + def test_list_simulators(self, cognite_client: CogniteClient) -> None: + simulators = cognite_client.simulators.list(limit=5) + + assert len(simulators) > 0 + + +class TestSimulatorIntegrations: + @pytest.mark.usefixtures("seed_resource_names", "seed_simulator_integration") + def test_list_integrations(self, cognite_client: CogniteClient) -> None: + integrations = cognite_client.simulators.integrations.list(limit=5) + + assert len(integrations) > 0 + + def test_filter_integrations(self, cognite_client: CogniteClient, seed_resource_names) -> None: + all_integrations = cognite_client.simulators.integrations.list() + active_integrations = cognite_client.simulators.integrations.list( + filter=SimulatorIntegrationFilter(active=True) + ) + + filtered_integrations = cognite_client.simulators.integrations.list( + filter=SimulatorIntegrationFilter(simulator_external_ids=[seed_resource_names["simulator_external_id"]]) + ) + + assert len(all_integrations) > 0 + assert filtered_integrations[0].external_id == seed_resource_names["simulator_integration_external_id"] + # check time difference + assert filtered_integrations[0].active is True + + assert len(active_integrations) > 0 + assert len(filtered_integrations) > 0 + + +@pytest.mark.usefixtures( + "seed_resource_names", + "seed_simulator_model_revisions", +) +class TestSimulatorModels: + def test_list_models(self, cognite_client: CogniteClient, seed_resource_names) -> None: + models = cognite_client.simulators.models.list( + limit=5, filter=SimulatorModelsFilter(simulator_external_ids=[seed_resource_names["simulator_external_id"]]) + ) + assert len(models) > 0 + + def test_retrieve_model(self, cognite_client: CogniteClient, seed_resource_names) -> None: + model = cognite_client.simulators.models.retrieve(external_id=model_unique_external_id) + assert model is not None + assert model.external_id == model_unique_external_id + + def test_list_model_revisions(self, cognite_client: CogniteClient, seed_resource_names) -> None: + revisions = cognite_client.simulators.models.revisions.list( + limit=5, + filter=SimulatorModelRevisionsFilter(model_external_ids=[model_unique_external_id]), + ) + assert len(revisions) > 0 + + def test_retrieve_model_revision(self, cognite_client: CogniteClient, seed_resource_names) -> None: + model_revision = cognite_client.simulators.models.revisions.retrieve( + external_id=model_revision_unique_external_id + ) + assert model_revision is not None + assert model_revision.model_external_id == model_unique_external_id + + +@pytest.mark.usefixtures("seed_resource_names", "seed_simulator_routine_revisions") +class TestSimulatorRoutines: + def test_list_routines(self, cognite_client: CogniteClient) -> None: + routines = cognite_client.simulators.routines.list(limit=5) + assert len(routines) > 0 + + def test_list_and_filtering_routine_revisions(self, cognite_client: CogniteClient) -> None: + revisions_all = cognite_client.simulators.routines.revisions.list( + filter=SimulatorRoutineRevisionsFilter( + routine_external_ids=[simulator_routine_unique_external_id], all_versions=True + ), + ) + assert len(revisions_all) > 1 + revisions_filter = cognite_client.simulators.routines.revisions.list( + filter=SimulatorRoutineRevisionsFilter(model_external_ids=[model_unique_external_id], all_versions=True), + ) + assert len(revisions_filter) == 10 + revisions_sort_asc = cognite_client.simulators.routines.revisions.list( + sort=CreatedTimeSort(order="asc", property="createdTime"), + filter=SimulatorRoutineRevisionsFilter( + routine_external_ids=[simulator_routine_unique_external_id], all_versions=True + ), + ) + revisions_sort_desc = cognite_client.simulators.routines.revisions.list( + sort=CreatedTimeSort(order="desc", property="createdTime"), + filter=SimulatorRoutineRevisionsFilter( + routine_external_ids=[simulator_routine_unique_external_id], all_versions=True + ), + ) + assert revisions_sort_asc[0].external_id == revisions_sort_desc[-1].external_id + + def test_retrieve_routine_revision(self, cognite_client: CogniteClient) -> None: + revisions_all = cognite_client.simulators.routines.revisions.list( + filter=SimulatorRoutineRevisionsFilter( + routine_external_ids=[simulator_routine_unique_external_id], all_versions=True + ), + ) + + assert len(revisions_all) > 5 + + rev1 = revisions_all[random.randint(0, 4)] + rev2 = revisions_all[random.randint(0, 4)] + + rev1_retrieve = cognite_client.simulators.routines.revisions.retrieve(external_id=rev1.external_id) + assert rev1_retrieve is not None + assert rev1_retrieve.external_id == rev1.external_id + rev2_retrieve = cognite_client.simulators.routines.revisions.retrieve(external_id=rev2.external_id) + assert rev2_retrieve is not None + assert rev2_retrieve.external_id == rev2.external_id