fix(robot-server): Refactor hardware wrapper and other lifetime dependencies

api/src/opentrons/hardware_control/api.py

@@ -265,7 +265,7 @@ def _calculate_valid_attitude(self) -> DeckTransformState:
  return rb_cal.validate_attitude_deck_calibration(
  self._robot_calibration.deck_calibration)
 
- async def register_callback(self, cb):
+ def register_callback(self, cb):
  """ Allows the caller to register a callback, and returns a closure
  that can be used to unregister the provided callback
  """

robot-server/robot_server/app.py

@@ -1,4 +1,7 @@
 """Main FastAPI application."""
+
+
+from functools import partial
 import logging
 
 from opentrons import __version__
@@ -8,20 +11,36 @@
 from starlette.requests import Request
 from starlette.middleware.base import RequestResponseEndpoint
 
+from notify_server.clients import publisher as notify_server_publisher
+from notify_server.settings import Settings as NotifyServerSettings
+
+
 from .service.dependencies import (
  get_rpc_server,
  get_protocol_manager,
- get_hardware_wrapper,
  get_session_manager,
 )
 
 from .errors import exception_handlers
 from .router import router
 from .service import initialize_logging
 from . import constants
+from . import slow_initializing
+from . import hardware_initialization
 
-log = logging.getLogger(__name__)
 
+def _make_event_publisher() -> notify_server_publisher.Publisher:
+ """Create a notify-server event publisher instance."""
+ notify_server_settings = NotifyServerSettings()
+ event_publisher = notify_server_publisher.create(
+ notify_server_settings.publisher_address.connection_string()
+ )
+ # fixme(mm, 2021-08-09): Publisher has a .close() method that we must call, or
+ # we'll leak a socket.
+ return event_publisher
+
+
+log = logging.getLogger(__name__)
 
 app = FastAPI(
  title="Opentrons OT-2 HTTP API Spec",
@@ -60,7 +79,8 @@ async def on_startup() -> None:
  """Handle app startup."""
  initialize_logging()
  # Initialize api
- (await get_hardware_wrapper()).async_initialize()
+ factory = partial(hardware_initialization.initialize, _make_event_publisher())
+ app.state.hardware = slow_initializing.start_initializing(factory)
 
 
 @app.on_event("shutdown")

robot-server/robot_server/hardware_initialization.py

@@ -0,0 +1,62 @@
+"""Utilities for initializing the hardware interface."""
+
+
+import logging
+from pathlib import Path
+
+from opentrons import ThreadManager, initialize as initialize_api
+from opentrons.hardware_control.simulator_setup import load_simulator
+from opentrons.hardware_control.types import HardwareEvent, HardwareEventType
+from opentrons.util.helpers import utc_now
+
+from notify_server.clients.publisher import Publisher
+from notify_server.models import event, topics
+from notify_server.models.hardware_event import DoorStatePayload
+
+from .settings import get_settings
+
+
+log = logging.getLogger(__name__)
+
+
+async def initialize(event_publisher: Publisher) -> ThreadManager:
+ """Initialize the API."""
+ app_settings = get_settings()
+ thread_manager: ThreadManager
+ if app_settings.simulator_configuration_file_path:
+ log.info(
+ f"Loading simulator from "
+ f"{app_settings.simulator_configuration_file_path}"
+ )
+ # A path to a simulation configuration is defined. Let's use it.
+ thread_manager = ThreadManager(
+ load_simulator, Path(app_settings.simulator_configuration_file_path)
+ )
+ else:
+ # Create the hardware
+ thread_manager = await initialize_api()
+ await _init_event_watchers(thread_manager, event_publisher)
+ log.info("Opentrons API initialized")
+ return thread_manager
+
+
+async def _init_event_watchers(
+ thread_manager: ThreadManager,
+ event_publisher: Publisher,
+) -> None:
+ """Register the publisher callbacks with the hw thread manager."""
+ log.info("Starting hardware-event-notify publisher")
+
+ def publish_hardware_event(hw_event: HardwareEvent) -> None:
+ if hw_event.event == HardwareEventType.DOOR_SWITCH_CHANGE:
+ payload = DoorStatePayload(state=hw_event.new_state)
+ else:
+ return
+ topic = topics.RobotEventTopics.HARDWARE_EVENTS
+ publisher = "robot_server_event_publisher"
+ event_to_publish = event.Event(
+ createdOn=utc_now(), publisher=publisher, data=payload
+ )
+ event_publisher.send_nowait(topic, event_to_publish)
+
+ thread_manager.register_callback(publish_hardware_event)

robot-server/robot_server/service/dependencies.py

@@ -10,14 +10,11 @@
 
 from opentrons.hardware_control import ThreadManager, ThreadedAsyncLock
 
-from robot_server import constants, util, errors
-from robot_server.hardware_wrapper import HardwareWrapper
+from robot_server import app, constants, util, errors
 from robot_server.service.session.manager import SessionManager
 from robot_server.service.protocol.manager import ProtocolManager
 from robot_server.service.legacy.rpc import RPCServer
-
-from notify_server.clients import publisher
-from notify_server.settings import Settings as NotifyServerSettings
+from robot_server.slow_initializing import InitializationOngoingError, SlowInitializing
 
 
 class OutdatedApiVersionResponse(errors.ErrorDetails):
@@ -32,38 +29,30 @@ class OutdatedApiVersionResponse(errors.ErrorDetails):
  )
 
 
-@util.call_once
-async def get_event_publisher() -> publisher.Publisher:
- """A dependency creating a single notify-server event
- publisher instance."""
- notify_server_settings = NotifyServerSettings()
- event_publisher = publisher.create(
- notify_server_settings.publisher_address.connection_string()
- )
- return event_publisher
+async def get_app_state() -> State:
+ """Get the Starlette application's state from the framework.
 
+ See https://www.starlette.io/applications/#storing-state-on-the-app-instance
+ for more details.
+ """
 
-@util.call_once
-async def get_hardware_wrapper(
- event_publisher: publisher.Publisher = Depends(get_event_publisher),
-) -> HardwareWrapper:
- """Get the single HardwareWrapper instance."""
- return HardwareWrapper(event_publisher=event_publisher)
+ # Ideally, we would access the app state through request.app.state. However,
+ # this function might be depended upon by a WebSocket endpoint, and current FastAPI
+ # (v0.54.1) raises runtime errors when trying to resolve the built-in `request`
+ # dependency for WebSocket endpoints.
+ return app.app.state
 
 
-async def get_hardware(
- api_wrapper: HardwareWrapper = Depends(get_hardware_wrapper),
-) -> ThreadManager:
+async def get_hardware(state: State = Depends(get_app_state)) -> ThreadManager:
  """Hardware dependency"""
- hardware = api_wrapper.get_hardware()
-
- if hardware is None:
+ slow_initializing_hardware: SlowInitializing[ThreadManager] = state.hardware
+ try:
+ return slow_initializing_hardware.get_if_ready()
+ except InitializationOngoingError:
  raise HTTPException(
  status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
  detail="Robot motor controller is not ready.",
- )
-
- return hardware
+ ) from None
 
 
 @util.call_once
@@ -132,15 +121,6 @@ async def check_version_header(
  request.state.api_version = min(requested_version, constants.API_VERSION)
 
 
-def get_app_state(request: Request) -> State:
- """Get the Starlette application's state from the framework.
-
- See https://www.starlette.io/applications/#storing-state-on-the-app-instance
- for more details.
- """
- return request.app.state
-
-
 def get_unique_id() -> str:
  """Get a unique ID string to use as a resource identifier."""
  return str(uuid4())