Skip to content

Commit

Permalink
fix(api): fix thread manager definition (#12948)
Browse files Browse the repository at this point in the history
* attempted fix for repl calibration issue

* lint fixes

* attempted event loop fix
  • Loading branch information
pmoegenburg committed Jul 6, 2023
1 parent 84c0a0e commit 4635318
Showing 1 changed file with 7 additions and 8 deletions.
15 changes: 7 additions & 8 deletions api/src/opentrons/hardware_control/scripts/repl.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,8 @@
import asyncio
import logging
from logging.config import dictConfig
from opentrons.hardware_control.api import API
from opentrons.hardware_control.ot3api import OT3API

update_firmware = True
has_robot_server = True
Expand Down Expand Up @@ -54,7 +56,6 @@
find_axis_center,
gripper_pin_offsets_mean,
)
from opentrons.hardware_control import HardwareControlAPI # noqa: E402
from opentrons.hardware_control.thread_manager import ThreadManager # noqa: E402


Expand Down Expand Up @@ -85,11 +86,10 @@
}

if ff.enable_ot3_hardware_controller():
from opentrons.hardware_control.ot3api import OT3API

HCApi: Union[Type[OT3API], Type["API"]] = OT3API
HCApi: Union[Type[OT3API], Type[API]] = OT3API

def build_thread_manager() -> ThreadManager[Union["API", OT3API]]:
def build_thread_manager() -> ThreadManager[Union[API, OT3API]]:
return ThreadManager(
OT3API.build_hardware_controller,
use_usb_bus=ff.rear_panel_integration(),
Expand All @@ -99,14 +99,13 @@ def build_thread_manager() -> ThreadManager[Union["API", OT3API]]:
def wrap_async_util_fn(fn: Any, *bind_args: Any, **bind_kwargs: Any) -> Any:
@wraps(fn)
def synchronizer(*args: Any, **kwargs: Any) -> Any:
return asyncio.get_event_loop().run_until_complete(
return asyncio.new_event_loop().run_until_complete(
fn(*bind_args, *args, **bind_kwargs, **kwargs)
)

return synchronizer

else:
from opentrons.hardware_control.api import API

HCApi = API

Expand All @@ -125,7 +124,7 @@ def stop_server() -> None:
run(["systemctl", "stop", "opentrons-robot-server"])


def build_api() -> ThreadManager[HardwareControlAPI]:
def build_api() -> ThreadManager[Union[API, OT3API]]:
# NOTE: We are using StreamHandler so when the hw controller is
# being built we can log firmware update progress to stdout.
stream_handler = logging.StreamHandler()
Expand All @@ -146,7 +145,7 @@ async def _do_update() -> None:
return tm


def do_interact(api: ThreadManager[HardwareControlAPI]) -> None:
def do_interact(api: ThreadManager[Union[API, OT3API]]) -> None:
interact(
banner=(
"Hardware Control API REPL\nCall methods on api like "
Expand Down

0 comments on commit 4635318

Please sign in to comment.