add Machine class

Author: rickwierenga

pylabrobot/liquid_handling/backends/backend.py

@@ -3,6 +3,7 @@
 from abc import ABCMeta, abstractmethod
 from typing import List, Type, Optional
 
+from pylabrobot.machine import MachineBackend
 from pylabrobot.resources import Resource
 from pylabrobot.liquid_handling.standard import (
   Pickup,
@@ -17,7 +18,7 @@
 )
 
 
-class LiquidHandlerBackend(object, metaclass=ABCMeta):
+class LiquidHandlerBackend(MachineBackend, metaclass=ABCMeta):
   """
   Abstract base class for liquid handling robot backends.
 
@@ -28,15 +29,6 @@ class LiquidHandlerBackend(object, metaclass=ABCMeta):
     setup_finished: Whether the backend has been set up.
   """
 
-  def __init__(self):
-    self.setup_finished = False
-
-  async def setup(self):
-    self.setup_finished = True
-
-  async def stop(self):
-    self.setup_finished = False
-
   async def assigned_resource_callback(self, resource: Resource):
     """ Called when a new resource was assigned to the robot.
 

pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py

@@ -180,7 +180,6 @@ def __init__(self):
     self.commands = []
 
   async def setup(self) -> None:
-    self.setup_finished = True
     self._num_channels = 8
     self.iswap_installed = True
     self.core96_head_installed = True

pylabrobot/liquid_handling/backends/saver_backend.py

@@ -19,6 +19,9 @@ async def setup(self):
     await super().setup()
     self.commands_received = []
 
+  async def stop(self):
+    await super().stop()
+
   async def send_command(self, command: str, data: Dict[str, Any]):
     self.commands_received.append({"command": command, "data": data})
 

pylabrobot/liquid_handling/backends/serializing_backend.py

@@ -45,9 +45,11 @@ async def send_command(
 
   async def setup(self):
     await self.send_command(command="setup")
+    await super().setup()
 
   async def stop(self):
     await self.send_command(command="stop")
+    await super().stop()
 
   async def assigned_resource_callback(self, resource: Resource):
     await self.send_command(command="resource_assigned", data={"resource": resource.serialize(),
@@ -172,10 +174,7 @@ class Command(TypedDict):
 
   async def setup(self):
     self.sent_commands: List[SerializingSavingBackend.Command] = []
-    self.setup_finished = True
-
-  async def stop(self):
-    self.setup_finished = False
+    await super().setup()
 
   async def send_command(self, command: str, data: Optional[Dict[str, Any]] = None):
     self.sent_commands.append({"command": command, "data": data})

pylabrobot/liquid_handling/liquid_handler.py

@@ -3,7 +3,6 @@
 from __future__ import annotations
 
 import asyncio
-import functools
 import inspect
 import json
 import logging
@@ -13,6 +12,7 @@
 from typing import Any, Callable, Dict, Union, Optional, List, Sequence, Set
 import warnings
 
+from pylabrobot.machine import MachineFrontend, need_setup_finished
 from pylabrobot.liquid_handling.strictness import Strictness, get_strictness
 from pylabrobot.plate_reading import PlateReader
 from pylabrobot.resources import (
@@ -52,33 +52,13 @@
 logger = logging.getLogger("pylabrobot")
 
 
-def need_setup_finished(func: Callable): # pylint: disable=no-self-argument
-  """ Decorator for methods that require the liquid handler to be set up.
-
-  Checked by verifying `self.setup_finished` is `True`.
-
-  Raises:
-    RuntimeError: If the liquid handler is not set up.
-  """
-
-  @functools.wraps(func)
-  async def wrapper(self, *args, **kwargs):
-    if not self.setup_finished:
-      raise RuntimeError("The setup has not finished. See `LiquidHandler.setup`.")
-    await func(self, *args, **kwargs) # pylint: disable=not-callable
-  return wrapper
-
-
-class LiquidHandler:
+class LiquidHandler(MachineFrontend):
   """
   Front end for liquid handlers.
 
   This class is the front end for liquid handlers; it provides a high-level interface for
   interacting with liquid handlers. In the background, this class uses the low-level backend (
   defined in `pyhamilton.liquid_handling.backends`) to communicate with the liquid handler.
-
-  Attributes:
-    setup_finished: Whether the liquid handler has been setup.
   """
 
   def __init__(self, backend: LiquidHandlerBackend, deck: Deck):
@@ -89,8 +69,9 @@ def __init__(self, backend: LiquidHandlerBackend, deck: Deck):
       deck: Deck to use.
     """
 
-    self.backend = backend
-    self.setup_finished = False
+    super().__init__(backend=backend)
+
+    self.backend: LiquidHandlerBackend = backend
     self._picked_up_tips96: Optional[TipRack] = None # TODO: replace with tracker.
 
     self.deck = deck
@@ -99,21 +80,23 @@ def __init__(self, backend: LiquidHandlerBackend, deck: Deck):
 
     self.head: Dict[int, TipTracker] = {}
 
+
   async def setup(self):
     """ Prepare the robot for use. """
 
     if self.setup_finished:
       raise RuntimeError("The setup has already finished. See `LiquidHandler.stop`.")
 
     await self.backend.setup()
-    self.setup_finished = True
 
     self.head = {c: TipTracker() for c in range(self.backend.num_channels)}
 
     self.resource_assigned_callback(self.deck)
     for resource in self.deck.children:
       self.resource_assigned_callback(resource)
 
+    super().setup()
+
   def update_head_state(self, state: Dict[int, Optional[Tip]]):
     """ Update the state of the liquid handler head.
 
@@ -139,10 +122,6 @@ def clear_head_state(self):
 
     self.update_head_state({c: None for c in self.head.keys()})
 
-  async def stop(self):
-    await self.backend.stop()
-    self.setup_finished = False
-
   def _run_async_in_thread(self, func, *args, **kwargs):
     def callback(*args, **kwargs):
       loop = asyncio.new_event_loop()

pylabrobot/machine.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+import functools
+from typing import Callable
+
+def need_setup_finished(func: Callable):
+  """ Decorator for methods that require the liquid handler to be set up.
+
+  Checked by verifying `self.setup_finished` is `True`.
+
+  Raises:
+    RuntimeError: If the liquid handler is not set up.
+  """
+
+  @functools.wraps(func)
+  async def wrapper(self: MachineFrontend, *args, **kwargs):
+    if not self.setup_finished:
+      raise RuntimeError("The setup has not finished. See `setup`.")
+    await func(self, *args, **kwargs)
+  return wrapper
+
+
+class MachineBackend(ABC):
+  """ Abstract class for machine backends. """
+
+  @abstractmethod
+  async def setup(self):
+    pass
+
+  @abstractmethod
+  async def stop(self):
+    pass
+
+
+class MachineFrontend(ABC):
+  """ Abstract class for machine frontends. """
+
+  @abstractmethod
+  def __init__(self, backend: MachineBackend):
+    self.backend = backend
+    self._setup_finished = False
+
+  @property
+  def setup_finished(self) -> bool:
+    return self._setup_finished
+
+  async def setup(self):
+    self.backend.setup()
+
+  @need_setup_finished
+  async def stop(self):
+    self.backend.stop()

pylabrobot/plate_reading/backend.py

@@ -1,16 +1,18 @@
 from __future__ import annotations
 
-from abc import ABC, abstractmethod
+from abc import ABCMeta, abstractmethod
 import sys
 from typing import List, Optional, Type
 
+from pylabrobot.machine import MachineBackend
+
 if sys.version_info >= (3, 8):
   from typing import Literal
 else:
   from typing_extensions import Literal
 
 
-class PlateReaderBackend(ABC):
+class PlateReaderBackend(MachineBackend, metaclass=ABCMeta):
   """ An abstract class for a plate reader. Plate readers are devices that can read luminescence,
   absorbance, or fluorescence from a plate. """
 

pylabrobot/plate_reading/plate_reader.py

@@ -1,7 +1,7 @@
-import functools
 import sys
-from typing import Callable, List, cast
+from typing import List, cast
 
+from pylabrobot.machine import MachineFrontend, need_setup_finished
 from pylabrobot.resources import Coordinate, Resource, Plate
 from pylabrobot.plate_reading.backend import PlateReaderBackend
 
@@ -15,26 +15,7 @@ class NoPlateError(Exception):
   pass
 
 
-# copied from LiquidHandler.py, maybe we need a shared base class?
-
-def need_setup_finished(func: Callable): # pylint: disable=no-self-argument
-  """ Decorator for methods that require the plate reader to be set up.
-
-  Checked by verifying `self.setup_finished` is `True`.
-
-  Raises:
-    RuntimeError: If the liquid handler is not set up.
-  """
-
-  @functools.wraps(func)
-  async def wrapper(self, *args, **kwargs):
-    if not self.setup_finished:
-      raise RuntimeError("The setup has not finished. See `PlateReader.setup`.")
-    await func(self, *args, **kwargs) # pylint: disable=not-callable
-  return wrapper
-
-
-class PlateReader(Resource):
+class PlateReader(Resource, MachineFrontend):
   """ The front end for plate readers. Plate readers are devices that can read luminescence,
   absorbance, or fluorescence from a plate.
 
@@ -61,9 +42,9 @@ class PlateReader(Resource):
   """
 
   def __init__(self, name: str, backend: PlateReaderBackend) -> None:
-    super().__init__(name=name, size_x=0, size_y=0, size_z=0, category="plate_reader")
-    self.backend = backend
-    self.setup_finished = False
+    MachineFrontend.__init__(self, backend=backend)
+    self.backend: PlateReaderBackend = backend
+    Resource.__init__(self, name=name, size_x=0, size_y=0, size_z=0, category="plate_reader")
 
   def assign_child_resource(self, resource):
     if len(self.children) >= 1:
@@ -77,14 +58,6 @@ def get_plate(self) -> Plate:
       raise NoPlateError("There is no plate in the plate reader.")
     return cast(Plate, self.children[0])
 
-  async def setup(self) -> None:
-    await self.backend.setup()
-    self.setup_finished = True
-
-  async def stop(self) -> None:
-    await self.backend.stop()
-    self.setup_finished = False
-
   async def open(self) -> None:
     await self.backend.open()