feat(simulators): Add support for running simulations with routine and model revision external IDs

CHANGELOG.md

@@ -68,6 +68,10 @@ As of 2025-08-29, changes are grouped as follows
 ### Changed
 - Attributes `run_time` and `simulation_time` are now automatically converted to timestamp format (when calling `to_pandas(...)`)
 
+## [Unreleased]
+### Added
+- [alpha] Support for running simulations using routine and model revision external IDs. The `SimulationRunWrite` class now supports two modes: running with `routine_external_id` only, or with both `routine_revision_external_id` and `model_revision_external_id`.
+
 ## [7.81.1] - 2025-08-20
 ### Fixed
 - [alpha] Breaking change: fixed naming inconsistencies in simulators module. Renamed `SimulatorRunList` to `SimulationRunList` and `SimulatorRunDataList` to `SimulationRunDataList`

cognite/client/_api/simulators/routines.py

@@ -228,7 +228,9 @@ def list(
 
     def run(
         self,
-        routine_external_id: str,
+        routine_external_id: str | None = None,
+        routine_revision_external_id: str | None = None,
+        model_revision_external_id: str | None = None,
         inputs: Sequence[SimulationInputOverride] | None = None,
         run_time: int | None = None,
         queue: bool | None = None,
@@ -238,10 +240,17 @@ def run(
     ) -> SimulationRun:
         """`Run a simulation <https://developer.cognite.com/api#tag/Simulation-Runs/operation/run_simulation_simulators_run_post>`_
 
-        Run a simulation for a given simulator routine.
+        Run a simulation for a given simulator routine. Supports two modes:
+        1. By routine external ID only
+        2. By routine revision external ID + model revision external ID
 
         Args:
-            routine_external_id (str): External id of the simulator routine to run
+            routine_external_id (str | None): External id of the simulator routine to run.
+                Cannot be specified together with routine_revision_external_id and model_revision_external_id.
+            routine_revision_external_id (str | None): External id of the simulator routine revision to run.
+                Must be specified together with model_revision_external_id.
+            model_revision_external_id (str | None): External id of the simulator model revision.
+                Must be specified together with routine_revision_external_id.
             inputs (Sequence[SimulationInputOverride] | None): List of input overrides
             run_time (int | None): Run time in milliseconds. Reference timestamp used for data pre-processing and data sampling.
             queue (bool | None): Queue the simulation run when connector is down.
@@ -253,17 +262,26 @@ def run(
             SimulationRun: Created simulation run
 
         Examples:
-            Create new simulation run:
+            Create new simulation run using routine external ID:
                 >>> from cognite.client import CogniteClient
                 >>> client = CogniteClient()
                 >>> run = client.simulators.routines.run(
                 ...     routine_external_id="routine1",
                 ...     log_severity="Debug"
                 ... )
+
+            Create new simulation run using routine and model revision external IDs:
+                >>> run = client.simulators.routines.run(
+                ...     routine_revision_external_id="routine_revision1",
+                ...     model_revision_external_id="model_revision1",
+                ...     log_severity="Debug"
+                ... )
         """
         self._warning.warn()
         run_object = SimulationRunWrite(
             routine_external_id=routine_external_id,
+            routine_revision_external_id=routine_revision_external_id,
+            model_revision_external_id=model_revision_external_id,
             inputs=list(inputs) if inputs is not None else None,
             run_time=run_time,
             queue=queue,

cognite/client/_api/simulators/runs.py

@@ -272,7 +272,7 @@ def create(self, items: SimulationRunWrite | Sequence[SimulationRunWrite]) -> Si
             SimulationRun | SimulationRunList: Created simulation run(s)
 
         Examples:
-            Create new simulation run:
+            Create new simulation run using routine external ID:
                 >>> from cognite.client import CogniteClient
                 >>> from cognite.client.data_classes.simulators.runs import SimulationRunWrite
                 >>> client = CogniteClient()
@@ -284,6 +284,17 @@ def create(self, items: SimulationRunWrite | Sequence[SimulationRunWrite]) -> Si
                 ...     ),
                 ... ]
                 >>> res = client.simulators.runs.create(run)
+
+            Create new simulation run using routine and model revision external IDs:
+                >>> run = [
+                ...     SimulationRunWrite(
+                ...         routine_revision_external_id="routine_revision1",
+                ...         model_revision_external_id="model_revision1",
+                ...         log_severity="Debug",
+                ...         run_type="external",
+                ...     ),
+                ... ]
+                >>> res = client.simulators.runs.create(run)
         """
         assert_type(items, "simulation_run", [SimulationRunWrite, Sequence])
 

cognite/client/data_classes/simulators/runs.py

@@ -87,20 +87,33 @@ class SimulationRunCore(WriteableCogniteResource["SimulationRunWrite"]):
     def __init__(
         self,
         run_type: str | None,
-        routine_external_id: str,
+        routine_external_id: str | None,
         run_time: int | None = None,
+        routine_revision_external_id: str | None = None,
+        model_revision_external_id: str | None = None,
     ) -> None:
         self.run_type = run_type
         self.run_time = run_time
         self.routine_external_id = routine_external_id
+        self.routine_revision_external_id = routine_revision_external_id
+        self.model_revision_external_id = model_revision_external_id
 
 
 class SimulationRunWrite(SimulationRunCore):
     """
     Request to run a simulator routine asynchronously.
 
+    This class supports two modes of running simulations:
+    1. By routine external ID only
+    2. By routine revision external ID + model revision external ID
+
     Args:
-        routine_external_id (str): External id of the associated simulator routine
+        routine_external_id (str | None): External id of the associated simulator routine.
+            Cannot be specified together with routine_revision_external_id and model_revision_external_id.
+        routine_revision_external_id (str | None): External id of the associated simulator routine revision.
+            Must be specified together with model_revision_external_id.
+        model_revision_external_id (str | None): External id of the associated simulator model revision.
+            Must be specified together with routine_revision_external_id.
         run_type (str | None): The type of the simulation run
         run_time (int | None): Run time in milliseconds. Reference timestamp used for data pre-processing and data sampling.
         queue (bool | None): Queue the simulation run when connector is down.
@@ -110,15 +123,34 @@ class SimulationRunWrite(SimulationRunCore):
 
     def __init__(
         self,
-        routine_external_id: str,
+        routine_external_id: str | None = None,
+        routine_revision_external_id: str | None = None,
+        model_revision_external_id: str | None = None,
         run_type: str | None = None,
         run_time: int | None = None,
         queue: bool | None = None,
         log_severity: str | None = None,
         inputs: list[SimulationInputOverride] | None = None,
     ) -> None:
+        # Validate that either routine_external_id OR (routine_revision_external_id + model_revision_external_id) is provided
+        if routine_external_id is not None:
+            if routine_revision_external_id is not None or model_revision_external_id is not None:
+                raise ValueError(
+                    "Cannot specify both 'routine_external_id' and revision-based parameters "
+                    "('routine_revision_external_id', 'model_revision_external_id'). "
+                    "Use either routine_external_id alone, or both routine_revision_external_id and model_revision_external_id."
+                )
+        else:
+            if not (routine_revision_external_id is not None and model_revision_external_id is not None):
+                raise ValueError(
+                    "Must specify either 'routine_external_id' alone, or both "
+                    "'routine_revision_external_id' and 'model_revision_external_id' together."
+                )
+
         super().__init__(
             routine_external_id=routine_external_id,
+            routine_revision_external_id=routine_revision_external_id,
+            model_revision_external_id=model_revision_external_id,
             run_type=run_type,
             run_time=run_time,
         )
@@ -131,7 +163,9 @@ def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None =
         inputs = resource.get("inputs", None)
         return cls(
             run_type=resource.get("runType"),
-            routine_external_id=resource["routineExternalId"],
+            routine_external_id=resource.get("routineExternalId"),
+            routine_revision_external_id=resource.get("routineRevisionExternalId"),
+            model_revision_external_id=resource.get("modelRevisionExternalId"),
             run_time=resource.get("runTime"),
             queue=resource.get("queue"),
             log_severity=resource.get("logSeverity"),
@@ -142,6 +176,16 @@ def dump(self, camel_case: bool = True) -> dict[str, Any]:
         output = super().dump(camel_case=camel_case)
         if self.inputs is not None:
             output["inputs"] = [input_.dump(camel_case=camel_case) for input_ in self.inputs]
+
+        # Remove fields based on the mode we're in
+        if self.routine_external_id is not None:
+            # Routine-only mode: remove revision fields that might be None
+            output.pop("routineRevisionExternalId", None)
+            output.pop("modelRevisionExternalId", None)
+        else:
+            # Revision mode: remove routine_external_id
+            output.pop("routineExternalId", None)
+
         return output
 
     def as_write(self) -> SimulationRunWrite:
@@ -227,6 +271,17 @@ def __init__(
         self._cognite_client = cast("CogniteClient", cognite_client)
 
     def as_write(self) -> SimulationRunWrite:
+        # Check if we have revision-based fields in the run
+        if hasattr(self, "routine_revision_external_id") and hasattr(self, "model_revision_external_id"):
+            if self.routine_revision_external_id and self.model_revision_external_id:
+                return SimulationRunWrite(
+                    routine_revision_external_id=self.routine_revision_external_id,
+                    model_revision_external_id=self.model_revision_external_id,
+                    run_type=self.run_type,
+                    run_time=self.run_time,
+                )
+
+        # Default to routine-based
         return SimulationRunWrite(
             routine_external_id=self.routine_external_id,
             run_type=self.run_type,

tests/tests_integration/test_api/test_simulators/test_routines.py

@@ -52,3 +52,27 @@ def test_sort(
         assert len(routines_asc) > 1
         for i in range(1, len(routines_asc)):
             assert routines_asc[i].created_time >= routines_asc[i - 1].created_time
+
+
+@pytest.mark.usefixtures(
+    "seed_resource_names",
+    "seed_simulator_routine_revisions",
+)
+class TestSimulatorRoutinesRunWithRevisions:
+    def test_run_with_revisions(self, cognite_client: CogniteClient, seed_resource_names) -> None:
+        """Test running a simulation using routine and model revision external IDs."""
+        routine_external_id = seed_resource_names["simulator_routine_external_id"]
+        routine_revision_external_id = f"{routine_external_id}_v1"
+        model_revision_external_id = seed_resource_names["simulator_model_revision_external_id"]
+
+        # Run simulation using revision external IDs
+        run = cognite_client.simulators.routines.run(
+            routine_revision_external_id=routine_revision_external_id,
+            model_revision_external_id=model_revision_external_id,
+            wait=False,  # Don't wait to avoid timeout in tests
+        )
+
+        assert run is not None
+        assert run.id is not None
+        assert run.routine_revision_external_id == routine_revision_external_id
+        assert run.model_revision_external_id == model_revision_external_id

tests/utils.py

@@ -483,6 +483,21 @@ def create_instance(self, resource_cls: type[T_Object], skip_defaulted_args: boo
                 keyword_arguments.pop("max_list_size", None)
         elif resource_cls is SimulatorRoutineStepArguments:
             keyword_arguments = {"data": {"reference_id": self._random_string(50), "arg2": self._random_string(50)}}
+        elif resource_cls.__name__ == "SimulationRunWrite":
+            # SimulationRunWrite requires either routine_external_id alone OR both routine_revision_external_id and model_revision_external_id
+            if self._random.choice([True, False]):
+                # Use routine_external_id only
+                keyword_arguments = {
+                    k: v
+                    for k, v in keyword_arguments.items()
+                    if k not in ["routine_revision_external_id", "model_revision_external_id"]
+                }
+                keyword_arguments["routine_external_id"] = self._random_string(50)
+            else:
+                # Use revision-based parameters
+                keyword_arguments = {k: v for k, v in keyword_arguments.items() if k != "routine_external_id"}
+                keyword_arguments["routine_revision_external_id"] = self._random_string(50)
+                keyword_arguments["model_revision_external_id"] = self._random_string(50)
 
         return resource_cls(*positional_arguments, **keyword_arguments)