expand module emulation settings.

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

@@ -8,24 +8,21 @@
 from opentrons.drivers.mag_deck.driver import GCODE
 from opentrons.hardware_control.emulation.parser import Parser, Command
 from .abstract_emulator import AbstractEmulator
+from .settings import MagDeckSettings
 
 logger = logging.getLogger(__name__)
 
 
-SERIAL = "magnetic_emulator"
-MODEL = "mag_deck_v20"
-VERSION = "2.0.0"
-
-
 class MagDeckEmulator(AbstractEmulator):
     """Magdeck emulator"""
 
     height: float = 0
     position: float = 0
 
-    def __init__(self, parser: Parser) -> None:
-        self.reset()
+    def __init__(self, parser: Parser, settings: MagDeckSettings) -> None:
+        self._settings = settings
         self._parser = parser
+        self.reset()
 
     def handle(self, line: str) -> Optional[str]:
         """Handle a line"""
@@ -53,7 +50,11 @@ def _handle(self, command: Command) -> Optional[str]:
         elif command.gcode == GCODE.GET_CURRENT_POSITION:
             return f"Z:{self.position}"
         elif command.gcode == GCODE.DEVICE_INFO:
-            return f"serial:{SERIAL} model:{MODEL} version:{VERSION}"
+            return (
+                f"serial:{self._settings.serial_number} "
+                f"model:{self._settings.model} "
+                f"version:{self._settings.version}"
+            )
         elif command.gcode == GCODE.PROGRAMMING_MODE:
             pass
         return None

api/src/opentrons/hardware_control/emulation/module_server/helpers.py

@@ -60,6 +60,7 @@ async def handle_message(self, message: Message) -> None:
         Returns:
             None
         """
+
         def _next_index() -> int:
             index = self._hub_index
             self._hub_index += 1
@@ -79,7 +80,6 @@ def _next_index() -> int:
             await self._notify_method([], connections)
 
 
-
 async def wait_emulators(
     client: ModuleServerClient,
     modules: Sequence[ModuleType],

api/src/opentrons/hardware_control/emulation/scripts/run_module_emulator.py

@@ -17,9 +17,11 @@
 from opentrons.hardware_control.emulation.settings import Settings, ProxySettings
 
 emulator_builder: Final[Dict[str, Callable[[Settings], AbstractEmulator]]] = {
-    ModuleType.Magnetic.value: lambda s: MagDeckEmulator(Parser()),
-    ModuleType.Temperature.value: lambda s: TempDeckEmulator(Parser()),
-    ModuleType.Thermocycler.value: lambda s: ThermocyclerEmulator(Parser()),
+    ModuleType.Magnetic.value: lambda s: MagDeckEmulator(Parser(), s.magdeck),
+    ModuleType.Temperature.value: lambda s: TempDeckEmulator(Parser(), s.tempdeck),
+    ModuleType.Thermocycler.value: lambda s: ThermocyclerEmulator(
+        Parser(), s.thermocycler
+    ),
 }
 
 emulator_port: Final[Dict[str, Callable[[Settings], ProxySettings]]] = {

api/src/opentrons/hardware_control/emulation/settings.py

@@ -1,3 +1,4 @@
+from opentrons.hardware_control.emulation.util import TEMPERATURE_ROOM
 from pydantic import BaseSettings, BaseModel
 
 
@@ -17,7 +18,31 @@ class SmoothieSettings(BaseModel):
     port: int = 9996
 
 
-class ProxySettings(BaseSettings):
+class BaseModuleSettings(BaseModel):
+    serial_number: str
+    model: str
+    version: str
+
+
+class TemperatureModelSettings(BaseModel):
+    degrees_per_tick: float = 2.0
+    starting: float = float(TEMPERATURE_ROOM)
+
+
+class MagDeckSettings(BaseModuleSettings):
+    pass
+
+
+class TempDeckSettings(BaseModuleSettings):
+    temperature: TemperatureModelSettings
+
+
+class ThermocyclerSettings(BaseModuleSettings):
+    lid_temperature: TemperatureModelSettings
+    plate_temperature: TemperatureModelSettings
+
+
+class ProxySettings(BaseModel):
     """Settings for a proxy."""
 
     host: str = "0.0.0.0"
@@ -34,6 +59,22 @@ class ModuleServerSettings(BaseModel):
 
 class Settings(BaseSettings):
     smoothie: SmoothieSettings = SmoothieSettings()
+    magdeck: MagDeckSettings = MagDeckSettings(
+        serial_number="magnetic_emulator", model="mag_deck_v20", version="2.0.0"
+    )
+    tempdeck: TempDeckSettings = TempDeckSettings(
+        serial_number="temperature_emulator",
+        model="temp_deck_v20",
+        version="v2.0.1",
+        temperature=TemperatureModelSettings(starting=0.0),
+    )
+    thermocycler: ThermocyclerSettings = ThermocyclerSettings(
+        serial_number="thermocycler_emulator",
+        model="v02",
+        version="v1.1.0",
+        lid_temperature=TemperatureModelSettings(),
+        plate_temperature=TemperatureModelSettings(),
+    )
 
     host: str = "0.0.0.0"
 

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

@@ -9,6 +9,7 @@
 from opentrons.drivers.temp_deck.driver import GCODE
 from opentrons.hardware_control.emulation import util
 from opentrons.hardware_control.emulation.parser import Parser, Command
+from opentrons.hardware_control.emulation.settings import TempDeckSettings
 
 from .abstract_emulator import AbstractEmulator
 from .simulations import Temperature
@@ -17,17 +18,15 @@
 logger = logging.getLogger(__name__)
 
 
-SERIAL = "temperature_emulator"
-MODEL = "temp_deck_v20"
-VERSION = "v2.0.1"
-
-
 class TempDeckEmulator(AbstractEmulator):
     """TempDeck emulator"""
 
-    def __init__(self, parser: Parser) -> None:
-        self.reset()
+    _temperature: Temperature
+
+    def __init__(self, parser: Parser, settings: TempDeckSettings) -> None:
+        self._settings = settings
         self._parser = parser
+        self.reset()
 
     def handle(self, line: str) -> Optional[str]:
         """Handle a line"""
@@ -36,7 +35,10 @@ def handle(self, line: str) -> Optional[str]:
         return None if not joined else joined
 
     def reset(self):
-        self._temperature = Temperature(per_tick=0.25, current=0.0)
+        self._temperature = Temperature(
+            per_tick=self._settings.temperature.degrees_per_tick,
+            current=self._settings.temperature.starting,
+        )
 
     def _handle(self, command: Command) -> Optional[str]:
         """Handle a command."""
@@ -57,7 +59,11 @@ def _handle(self, command: Command) -> Optional[str]:
         elif command.gcode == GCODE.DISENGAGE:
             self._temperature.deactivate(util.TEMPERATURE_ROOM)
         elif command.gcode == GCODE.DEVICE_INFO:
-            return f"serial:{SERIAL} model:{MODEL} version:{VERSION}"
+            return (
+                f"serial:{self._settings.serial_number} "
+                f"model:{self._settings.model} "
+                f"version:{self._settings.version}"
+            )
         elif command.gcode == GCODE.PROGRAMMING_MODE:
             pass
         return None

api/src/opentrons/hardware_control/emulation/thermocycler.py

@@ -8,6 +8,7 @@
 from opentrons.drivers.thermocycler.driver import GCODE
 from opentrons.drivers.types import ThermocyclerLidStatus
 from opentrons.hardware_control.emulation.parser import Parser, Command
+from opentrons.hardware_control.emulation.settings import ThermocyclerSettings
 
 from .abstract_emulator import AbstractEmulator
 from .simulations import Temperature, TemperatureWithHold
@@ -16,11 +17,6 @@
 logger = logging.getLogger(__name__)
 
 
-SERIAL = "thermocycler_emulator"
-MODEL = "v02"
-VERSION = "v1.1.0"
-
-
 class ThermocyclerEmulator(AbstractEmulator):
     """Thermocycler emulator"""
 
@@ -30,9 +26,10 @@ class ThermocyclerEmulator(AbstractEmulator):
     plate_volume: util.OptionalValue[float]
     plate_ramp_rate: util.OptionalValue[float]
 
-    def __init__(self, parser: Parser) -> None:
-        self.reset()
+    def __init__(self, parser: Parser, settings: ThermocyclerSettings) -> None:
         self._parser = parser
+        self._settings = settings
+        self.reset()
 
     def handle(self, line: str) -> Optional[str]:
         """Handle a line"""
@@ -41,9 +38,13 @@ def handle(self, line: str) -> Optional[str]:
         return None if not joined else joined
 
     def reset(self):
-        self._lid_temperature = Temperature(per_tick=2, current=util.TEMPERATURE_ROOM)
+        self._lid_temperature = Temperature(
+            per_tick=self._settings.lid_temperature.degrees_per_tick,
+            current=self._settings.lid_temperature.starting,
+        )
         self._plate_temperature = TemperatureWithHold(
-            per_tick=2, current=util.TEMPERATURE_ROOM
+            per_tick=self._settings.plate_temperature.degrees_per_tick,
+            current=self._settings.plate_temperature.starting,
         )
         self.lid_status = ThermocyclerLidStatus.OPEN
         self.plate_volume = util.OptionalValue[float]()
@@ -115,7 +116,11 @@ def _handle(self, command: Command) -> Optional[str]:  # noqa: C901
         elif command.gcode == GCODE.DEACTIVATE_BLOCK:
             self._plate_temperature.deactivate(temperature=util.TEMPERATURE_ROOM)
         elif command.gcode == GCODE.DEVICE_INFO:
-            return f"serial:{SERIAL} model:{MODEL} version:{VERSION}"
+            return (
+                f"serial:{self._settings.serial_number} "
+                f"model:{self._settings.model} "
+                f"version:{self._settings.version}"
+            )
         return None
 
     @staticmethod

api/tests/opentrons/hardware_control/emulation/module_server/test_helpers.py

@@ -3,8 +3,10 @@
 import pytest
 from mock import AsyncMock
 from opentrons.drivers.rpi_drivers.types import USBPort
-from opentrons.hardware_control.emulation.module_server import helpers, \
-    ModuleServerClient
+from opentrons.hardware_control.emulation.module_server import (
+    helpers,
+    ModuleServerClient,
+)
 from opentrons.hardware_control.emulation.module_server import models
 from opentrons.hardware_control.modules import ModuleAtPort
 
@@ -45,18 +47,18 @@ def modules_at_port() -> List[ModuleAtPort]:
         ModuleAtPort(
             port=f"url{i}",
             name=f"module_type{i}",
-            usb_port=USBPort(name=f"identifier{i}", sub_names=[], hub=i),
+            usb_port=USBPort(name=f"identifier{i}", sub_names=[], hub=i + 1),
         )
         for i in range(5)
     ]
 
 
-async def test_handle_message_connected_empty(subject: helpers.ModuleListener, mock_callback: AsyncMock) -> None:
+async def test_handle_message_connected_empty(
+    subject: helpers.ModuleListener, mock_callback: AsyncMock
+) -> None:
     """It should call the call back with the correct modules to add."""
     message = models.Message(status="connected", connections=[])
-    await subject.handle_message(
-        message=message
-    )
+    await subject.handle_message(message=message)
     mock_callback.assert_called_once_with([], [])
 
 
@@ -68,9 +70,7 @@ async def test_handle_message_connected_one(
 ) -> None:
     """It should call the call back with the correct modules to add."""
     message = models.Message(status="connected", connections=connections[:1])
-    await subject.handle_message(
-        message=message
-    )
+    await subject.handle_message(message=message)
     mock_callback.assert_called_once_with(modules_at_port[:1], [])
 
 
@@ -82,18 +82,16 @@ async def test_handle_message_connected_many(
 ) -> None:
     """It should call the call back with the correct modules to add."""
     message = models.Message(status="connected", connections=connections)
-    await subject.handle_message(
-        message=message
-    )
+    await subject.handle_message(message=message)
     mock_callback.assert_called_once_with(modules_at_port, [])
 
 
-async def test_handle_message_disconnected_empty(subject: helpers.ModuleListener, mock_callback: AsyncMock) -> None:
+async def test_handle_message_disconnected_empty(
+    subject: helpers.ModuleListener, mock_callback: AsyncMock
+) -> None:
     """It should call the call back with the correct modules to remove."""
     message = models.Message(status="disconnected", connections=[])
-    await subject.handle_message(
-        message=message
-    )
+    await subject.handle_message(message=message)
     mock_callback.assert_called_once_with([], [])
 
 
@@ -105,9 +103,7 @@ async def test_handle_message_disconnected_one(
 ) -> None:
     """It should call the call back with the correct modules to remove."""
     message = models.Message(status="disconnected", connections=connections[:1])
-    await subject.handle_message(
-        message=message
-    )
+    await subject.handle_message(message=message)
     mock_callback.assert_called_once_with([], modules_at_port[:1])
 
 
@@ -119,18 +115,16 @@ async def test_handle_message_disconnected_many(
 ) -> None:
     """It should call the call back with the correct modules to remove."""
     message = models.Message(status="disconnected", connections=connections)
-    await subject.handle_message(
-        message=message
-    )
+    await subject.handle_message(message=message)
     mock_callback.assert_called_once_with([], modules_at_port)
 
 
-async def test_handle_message_dump_empty(subject: helpers.ModuleListener, mock_callback: AsyncMock) -> None:
+async def test_handle_message_dump_empty(
+    subject: helpers.ModuleListener, mock_callback: AsyncMock
+) -> None:
     """It should call the call back with the correct modules to load."""
     message = models.Message(status="dump", connections=[])
-    await subject.handle_message(
-        message=message
-    )
+    await subject.handle_message(message=message)
     mock_callback.assert_called_once_with([], [])
 
 
@@ -142,9 +136,7 @@ async def test_handle_message_dump_one(
 ) -> None:
     """It should call the call back with the correct modules to load."""
     message = models.Message(status="dump", connections=connections[:1])
-    await subject.handle_message(
-        message=message
-    )
+    await subject.handle_message(message=message)
     mock_callback.assert_called_once_with(modules_at_port[:1], [])
 
 
@@ -156,7 +148,5 @@ async def test_handle_message_dump_many(
 ) -> None:
     """It should call the call back with the correct modules to load."""
     message = models.Message(status="dump", connections=connections)
-    await subject.handle_message(
-        message=message
-    )
+    await subject.handle_message(message=message)
     mock_callback.assert_called_once_with(modules_at_port, [])