feature(api): Allow simulator to load multiple of a module

api/src/opentrons/drivers/heater_shaker/simulator.py

@@ -1,4 +1,4 @@
-from typing import Dict
+from typing import Dict, Optional
 from opentrons.util.async_helpers import ensure_yield
 from opentrons.drivers.heater_shaker.abstract import AbstractHeaterShakerDriver
 from opentrons.drivers.types import Temperature, RPM, HeaterShakerLabwareLatchStatus
@@ -7,12 +7,13 @@
 class SimulatingDriver(AbstractHeaterShakerDriver):
     DEFAULT_TEMP = 23
 
-    def __init__(self) -> None:
+    def __init__(self, serial_number: Optional[str] = None) -> None:
         self._labware_latch_state = HeaterShakerLabwareLatchStatus.IDLE_UNKNOWN
         self._current_temperature = self.DEFAULT_TEMP
         self._temperature = Temperature(current=self.DEFAULT_TEMP, target=None)
         self._rpm = RPM(current=0, target=None)
         self._homing_status = True
+        self._serial_number = serial_number
 
     @ensure_yield
     async def connect(self) -> None:
@@ -83,7 +84,7 @@ async def deactivate(self) -> None:
     @ensure_yield
     async def get_device_info(self) -> Dict[str, str]:
         return {
-            "serial": "dummySerialHS",
+            "serial": self._serial_number if self._serial_number else "dummySerialHS",
             "model": "dummyModelHS",
             "version": "dummyVersionHS",
         }

api/src/opentrons/drivers/mag_deck/simulator.py

@@ -11,9 +11,12 @@
 
 
 class SimulatingDriver(AbstractMagDeckDriver):
-    def __init__(self, sim_model: Optional[str] = None) -> None:
+    def __init__(
+        self, sim_model: Optional[str] = None, serial_number: Optional[str] = None
+    ) -> None:
         self._height = 0.0
         self._model = MAG_DECK_MODELS[sim_model] if sim_model else "mag_deck_v1.1"
+        self._serial_number = serial_number
 
     @ensure_yield
     async def probe_plate(self) -> None:
@@ -30,7 +33,7 @@ async def move(self, location: float) -> None:
     @ensure_yield
     async def get_device_info(self) -> Dict[str, str]:
         return {
-            "serial": "dummySerialMD",
+            "serial": self._serial_number if self._serial_number else "dummySerialMD",
             "model": self._model,
             "version": "dummyVersionMD",
         }

api/src/opentrons/drivers/rpi_drivers/interfaces.py

@@ -1,12 +1,15 @@
-from typing import List
+from typing import List, Union
 from typing_extensions import Protocol
 
-from opentrons.hardware_control.modules.types import ModuleAtPort
+from opentrons.hardware_control.modules.types import (
+    ModuleAtPort,
+    SimulatingModuleAtPort,
+)
 
 
 class USBDriverInterface(Protocol):
     def match_virtual_ports(
         self,
-        virtual_port: List[ModuleAtPort],
-    ) -> List[ModuleAtPort]:
+        virtual_port: Union[List[ModuleAtPort], List[SimulatingModuleAtPort]],
+    ) -> Union[List[ModuleAtPort], List[SimulatingModuleAtPort]]:
         ...

api/src/opentrons/drivers/rpi_drivers/usb.py

@@ -8,9 +8,12 @@
 import subprocess
 import re
 import os
-from typing import List
+from typing import List, Union
 
-from opentrons.hardware_control.modules.types import ModuleAtPort
+from opentrons.hardware_control.modules.types import (
+    ModuleAtPort,
+    SimulatingModuleAtPort,
+)
 from opentrons.hardware_control.types import BoardRevision
 
 from .interfaces import USBDriverInterface
@@ -79,8 +82,8 @@ def _read_usb_bus(self) -> List[USBPort]:
 
     def match_virtual_ports(
         self,
-        virtual_ports: List[ModuleAtPort],
-    ) -> List[ModuleAtPort]:
+        virtual_ports: Union[List[ModuleAtPort], List[SimulatingModuleAtPort]],
+    ) -> Union[List[ModuleAtPort], List[SimulatingModuleAtPort]]:
         """
         Match Virtual Ports
 
@@ -89,7 +92,7 @@ def match_virtual_ports(
         the physical usb port information.
         The virtual port path looks something like:
         dev/ot_module_[MODULE NAME]
-        :param virtual_ports: A list of ModuleAtPort objects
+        :param virtual_ports: A list of ModuleAtPort or SimulatingModuleAtPort objects
         that hold the name and virtual port of each module
         connected to the robot.
 

api/src/opentrons/drivers/rpi_drivers/usb_simulator.py

@@ -4,15 +4,19 @@
 A class to convert info from the usb bus into a
 more readable format.
 """
-from typing import List
+from typing import List, Union
 
-from opentrons.hardware_control.modules.types import ModuleAtPort
+from opentrons.hardware_control.modules.types import (
+    ModuleAtPort,
+    SimulatingModuleAtPort,
+)
 
 from .interfaces import USBDriverInterface
 
 
 class USBBusSimulator(USBDriverInterface):
     def match_virtual_ports(
-        self, virtual_port: List[ModuleAtPort]
-    ) -> List[ModuleAtPort]:
+        self,
+        virtual_port: Union[List[ModuleAtPort], List[SimulatingModuleAtPort]],
+    ) -> Union[List[ModuleAtPort], List[SimulatingModuleAtPort]]:
         return virtual_port

api/src/opentrons/drivers/temp_deck/simulator.py

@@ -11,10 +11,13 @@
 
 
 class SimulatingDriver(AbstractTempDeckDriver):
-    def __init__(self, sim_model: Optional[str] = None):
+    def __init__(
+        self, sim_model: Optional[str] = None, serial_number: Optional[str] = None
+    ):
         self._temp = Temperature(target=None, current=0)
         self._port: Optional[str] = None
         self._model = TEMP_DECK_MODELS[sim_model] if sim_model else "temp_deck_v1.1"
+        self._serial_number = serial_number
 
     @ensure_yield
     async def set_temperature(self, celsius: float) -> None:
@@ -48,7 +51,7 @@ async def enter_programming_mode(self) -> None:
     @ensure_yield
     async def get_device_info(self) -> Dict[str, str]:
         return {
-            "serial": "dummySerialTD",
+            "serial": self._serial_number if self._serial_number else "dummySerialTD",
             "model": self._model,
             "version": "dummyVersionTD",
         }

api/src/opentrons/drivers/thermocycler/simulator.py

@@ -10,14 +10,17 @@
 class SimulatingDriver(AbstractThermocyclerDriver):
     DEFAULT_TEMP = 23
 
-    def __init__(self, model: Optional[str] = None) -> None:
+    def __init__(
+        self, model: Optional[str] = None, serial_number: Optional[str] = None
+    ) -> None:
         self._ramp_rate: Optional[float] = None
         self._lid_status = ThermocyclerLidStatus.OPEN
         self._lid_temperature = Temperature(current=self.DEFAULT_TEMP, target=None)
         self._plate_temperature = PlateTemperature(
             current=self.DEFAULT_TEMP, target=None, hold=None
         )
         self._model = model if model else "thermocyclerModuleV1"
+        self._serial_number = serial_number
 
     def model(self) -> str:
         return self._model
@@ -103,7 +106,7 @@ async def deactivate_all(self) -> None:
     @ensure_yield
     async def get_device_info(self) -> Dict[str, str]:
         return {
-            "serial": "dummySerialTC",
+            "serial": self._serial_number if self._serial_number else "dummySerialTC",
             "model": "dummyModelTC",
             "version": "dummyVersionTC",
         }

api/src/opentrons/hardware_control/api.py

@@ -255,7 +255,7 @@ async def build_hardware_simulator(
         attached_instruments: Optional[
             Dict[top_types.Mount, Dict[str, Optional[str]]]
         ] = None,
-        attached_modules: Optional[List[str]] = None,
+        attached_modules: Optional[Dict[str, List[str]]] = None,
         config: Optional[Union[RobotConfig, OT3Config]] = None,
         loop: Optional[asyncio.AbstractEventLoop] = None,
         strict_attached_instruments: bool = True,
@@ -271,7 +271,7 @@ async def build_hardware_simulator(
             attached_instruments = {}
 
         if None is attached_modules:
-            attached_modules = []
+            attached_modules = {}
 
         checked_loop = use_or_initialize_loop(loop)
         if isinstance(config, RobotConfig):

api/src/opentrons/hardware_control/backends/ot3simulator.py

@@ -100,7 +100,7 @@ class OT3Simulator(FlexBackend):
     async def build(
         cls,
         attached_instruments: Dict[OT3Mount, Dict[str, Optional[str]]],
-        attached_modules: List[str],
+        attached_modules: Dict[str, List[str]],
         config: OT3Config,
         loop: asyncio.AbstractEventLoop,
         strict_attached_instruments: bool = True,
@@ -126,7 +126,7 @@ async def build(
     def __init__(
         self,
         attached_instruments: Dict[OT3Mount, Dict[str, Optional[str]]],
-        attached_modules: List[str],
+        attached_modules: Dict[str, List[str]],
         config: OT3Config,
         loop: asyncio.AbstractEventLoop,
         strict_attached_instruments: bool = True,
@@ -588,10 +588,16 @@ async def increase_z_l_hold_current(self) -> AsyncIterator[None]:
 
     @ensure_yield
     async def watch(self, loop: asyncio.AbstractEventLoop) -> None:
-        new_mods_at_ports = [
-            modules.ModuleAtPort(port=f"/dev/ot_module_sim_{mod}{str(idx)}", name=mod)
-            for idx, mod in enumerate(self._stubbed_attached_modules)
-        ]
+        new_mods_at_ports = []
+        for mod, serials in self._stubbed_attached_modules.items():
+            for serial in serials:
+                new_mods_at_ports.append(
+                    modules.SimulatingModuleAtPort(
+                        port=f"/dev/ot_module_sim_{mod}{str(serial)}",
+                        name=mod,
+                        serial_number=serial,
+                    )
+                )
         await self.module_controls.register_modules(new_mods_at_ports=new_mods_at_ports)
 
     @property

api/src/opentrons/hardware_control/backends/simulator.py

@@ -49,7 +49,7 @@ class Simulator:
     async def build(
         cls,
         attached_instruments: Dict[types.Mount, Dict[str, Optional[str]]],
-        attached_modules: List[str],
+        attached_modules: Dict[str, List[str]],
         config: RobotConfig,
         loop: asyncio.AbstractEventLoop,
         strict_attached_instruments: bool = True,
@@ -105,7 +105,7 @@ async def build(
     def __init__(
         self,
         attached_instruments: Dict[types.Mount, Dict[str, Optional[str]]],
-        attached_modules: List[str],
+        attached_modules: Dict[str, List[str]],
         config: RobotConfig,
         loop: asyncio.AbstractEventLoop,
         gpio_chardev: GPIODriverLike,
@@ -332,10 +332,16 @@ def set_active_current(self, axis_currents: Dict[Axis, float]) -> None:
 
     @ensure_yield
     async def watch(self) -> None:
-        new_mods_at_ports = [
-            modules.ModuleAtPort(port=f"/dev/ot_module_sim_{mod}{str(idx)}", name=mod)
-            for idx, mod in enumerate(self._stubbed_attached_modules)
-        ]
+        new_mods_at_ports = []
+        for mod, serials in self._stubbed_attached_modules.items():
+            for serial in serials:
+                new_mods_at_ports.append(
+                    modules.SimulatingModuleAtPort(
+                        port=f"/dev/ot_module_sim_{mod}{str(serial)}",
+                        name=mod,
+                        serial_number=serial,
+                    )
+                )
         await self.module_controls.register_modules(new_mods_at_ports=new_mods_at_ports)
 
     @contextmanager

api/src/opentrons/hardware_control/module_control.py

@@ -15,6 +15,8 @@
     save_module_calibration_offset,
 )
 from opentrons.hardware_control.modules.types import ModuleType
+from opentrons.hardware_control.modules import SimulatingModuleAtPort
+
 from opentrons.types import Point
 from .types import AionotifyEvent, BoardRevision, OT3Mount
 from . import modules
@@ -84,6 +86,7 @@ async def build_module(
         usb_port: types.USBPort,
         type: modules.ModuleType,
         sim_model: Optional[str] = None,
+        sim_serial_number: Optional[str] = None,
     ) -> modules.AbstractModule:
         return await modules.build(
             port=port,
@@ -93,6 +96,7 @@ async def build_module(
             hw_control_loop=self._api.loop,
             execution_manager=self._api._execution_manager,
             sim_model=sim_model,
+            sim_serial_number=sim_serial_number,
         )
 
     async def unregister_modules(
@@ -126,7 +130,9 @@ async def unregister_modules(
 
     async def register_modules(
         self,
-        new_mods_at_ports: Optional[List[modules.ModuleAtPort]] = None,
+        new_mods_at_ports: Optional[
+            Union[List[modules.ModuleAtPort], List[modules.SimulatingModuleAtPort]]
+        ] = None,
         removed_mods_at_ports: Optional[List[modules.ModuleAtPort]] = None,
     ) -> None:
         """
@@ -152,6 +158,9 @@ async def register_modules(
                 port=mod.port,
                 usb_port=mod.usb_port,
                 type=modules.MODULE_TYPE_BY_NAME[mod.name],
+                sim_serial_number=mod.serial_number
+                if isinstance(mod, SimulatingModuleAtPort)
+                else None,
             )
             self._available_modules.append(new_instance)
             log.info(

api/src/opentrons/hardware_control/modules/__init__.py

@@ -11,6 +11,7 @@
     BundledFirmware,
     UpdateError,
     ModuleAtPort,
+    SimulatingModuleAtPort,
     ModuleType,
     ModuleModel,
     TemperatureStatus,
@@ -33,6 +34,7 @@
     "BundledFirmware",
     "UpdateError",
     "ModuleAtPort",
+    "SimulatingModuleAtPort",
     "HeaterShaker",
     "ModuleType",
     "ModuleModel",

api/src/opentrons/hardware_control/modules/heater_shaker.py

@@ -49,6 +49,7 @@ async def build(
         poll_interval_seconds: Optional[float] = None,
         simulating: bool = False,
         sim_model: Optional[str] = None,
+        sim_serial_number: Optional[str] = None,
     ) -> "HeaterShaker":
         """
         Build a HeaterShaker
@@ -71,7 +72,7 @@ async def build(
             driver = await HeaterShakerDriver.create(port=port, loop=hw_control_loop)
             poll_interval_seconds = poll_interval_seconds or POLL_PERIOD
         else:
-            driver = SimulatingDriver()
+            driver = SimulatingDriver(serial_number=sim_serial_number)
             poll_interval_seconds = poll_interval_seconds or SIMULATING_POLL_PERIOD
 
         reader = HeaterShakerReader(driver=driver)

api/src/opentrons/hardware_control/modules/magdeck.py

@@ -53,13 +53,14 @@ async def build(
         poll_interval_seconds: Optional[float] = None,
         simulating: bool = False,
         sim_model: Optional[str] = None,
+        serial_number: Optional[str] = None,
     ) -> "MagDeck":
         """Factory function."""
         driver: AbstractMagDeckDriver
         if not simulating:
             driver = await MagDeckDriver.create(port=port, loop=hw_control_loop)
         else:
-            driver = SimulatingDriver(sim_model=sim_model)
+            driver = SimulatingDriver(sim_model=sim_model, serial_number=serial_number)
 
         mod = cls(
             port=port,

api/src/opentrons/hardware_control/modules/mod_abc.py

@@ -32,6 +32,7 @@ async def build(
         poll_interval_seconds: Optional[float] = None,
         simulating: bool = False,
         sim_model: Optional[str] = None,
+        serial_number: Optional[str] = None,
     ) -> "AbstractModule":
         """Modules should always be created using this factory.
 

api/src/opentrons/hardware_control/modules/tempdeck.py

@@ -39,6 +39,7 @@ async def build(
         poll_interval_seconds: Optional[float] = None,
         simulating: bool = False,
         sim_model: Optional[str] = None,
+        serial_number: Optional[str] = None,
     ) -> "TempDeck":
         """
         Build a TempDeck
@@ -60,7 +61,7 @@ async def build(
             driver = await TempDeckDriver.create(port=port, loop=hw_control_loop)
             poll_interval_seconds = poll_interval_seconds or TEMP_POLL_INTERVAL_SECS
         else:
-            driver = SimulatingDriver(sim_model=sim_model)
+            driver = SimulatingDriver(sim_model=sim_model, serial_number=serial_number)
             poll_interval_seconds = poll_interval_seconds or SIM_TEMP_POLL_INTERVAL_SECS
 
         reader = TempDeckReader(driver=driver)