diff --git a/g-code-testing/g_code_parsing/g_code_engine.py b/g-code-testing/g_code_parsing/g_code_engine.py index d33b3134e954..36c323cb617d 100644 --- a/g-code-testing/g_code_parsing/g_code_engine.py +++ b/g-code-testing/g_code_parsing/g_code_engine.py @@ -1,20 +1,11 @@ import os import sys -import threading import asyncio -import time from multiprocessing import Process from typing import Generator, Callable from collections import namedtuple -from opentrons.hardware_control.emulation.magdeck import MagDeckEmulator -from opentrons.hardware_control.emulation.parser import Parser -from opentrons.hardware_control.emulation.run_emulator import \ - run_emulator_client from opentrons.hardware_control.emulation.settings import Settings -from opentrons.hardware_control.emulation.tempdeck import TempDeckEmulator -from opentrons.hardware_control.emulation.thermocycler import \ - ThermocyclerEmulator from opentrons.hardware_control.emulation.types import ModuleType from opentrons.protocols.parse import parse from opentrons.protocols.execution import execute @@ -69,7 +60,9 @@ def _get_loop() -> asyncio.AbstractEventLoop: @staticmethod def _set_env_vars(settings: Settings) -> None: """Set URLs of where to find modules and config for smoothie""" - os.environ["OT_MAGNETIC_EMULATOR_URI"] = GCodeEngine.URI_TEMPLATE % settings.magdeck_proxy.driver_port + os.environ["OT_MAGNETIC_EMULATOR_URI"] = ( + GCodeEngine.URI_TEMPLATE % settings.magdeck_proxy.driver_port + ) os.environ["OT_THERMOCYCLER_EMULATOR_URI"] = ( GCodeEngine.URI_TEMPLATE % settings.thermocycler_proxy.driver_port ) @@ -83,14 +76,18 @@ def _start_emulation_app(emulator_settings: Settings) -> Process: modules = [ModuleType.Magnetic, ModuleType.Temperature, ModuleType.Thermocycler] def runit(): - asyncio.run(run_app.run(emulator_settings, modules=[m.value for m in modules])) + asyncio.run( + run_app.run(emulator_settings, modules=[m.value for m in modules]) + ) proc = Process(target=runit) proc.daemon = True proc.start() async def _wait_ready() -> None: - c = await module_server.ModuleServerClient.connect(host="localhost", port=emulator_settings.module_server.port) + c = await module_server.ModuleServerClient.connect( + host="localhost", port=emulator_settings.module_server.port + ) await module_server.wait_emulators(client=c, modules=modules, timeout=5) c.close() @@ -152,7 +149,9 @@ def run_http(self, executable: Callable): """ app_process = self._start_emulation_app(emulator_settings=self._config) with GCodeWatcher(emulator_settings=self._config) as watcher: - asyncio.run(executable(hardware=self._emulate_hardware(settings=self._config))) + asyncio.run( + executable(hardware=self._emulate_hardware(settings=self._config)) + ) yield GCodeProgram.from_g_code_watcher(watcher) app_process.terminate() app_process.join() diff --git a/g-code-testing/tests/g_code_parsing/test_g_code_program.py b/g-code-testing/tests/g_code_parsing/test_g_code_program.py index 64f5804d8e02..b4f54a1aa9a6 100644 --- a/g-code-testing/tests/g_code_parsing/test_g_code_program.py +++ b/g-code-testing/tests/g_code_parsing/test_g_code_program.py @@ -1,4 +1,4 @@ -from typing import Generator, Iterator +from typing import Iterator import pytest from opentrons.hardware_control.emulation.settings import Settings