Add pumps

Author: rickwierenga

docs/pylabrobot.pumps.rst

@@ -0,0 +1,24 @@
+.. currentmodule:: pylabrobot.pumps
+
+pylabrobot.pumps package
+========================
+
+This package contains APIs for working with pumps.
+
+.. autosummary::
+  :toctree: _autosummary
+  :nosignatures:
+  :recursive:
+
+    pylabrobot.pumps.pump.Pump
+
+
+Backends
+--------
+
+.. autosummary::
+  :toctree: _autosummary
+  :nosignatures:
+  :recursive:
+
+    pylabrobot.pumps.cole_parmer.masterflex.Masterflex

docs/pylabrobot.rst

@@ -10,5 +10,6 @@ Subpackages
     :maxdepth: 1
 
     pylabrobot.liquid_handling
+    pylabrobot.pumps
     pylabrobot.resources
     pylabrobot.utils

pylabrobot/pumps/__init__.py

@@ -0,0 +1,2 @@
+from .pump import Pump
+from .cole_parmer import Masterflex

pylabrobot/pumps/backend.py

@@ -0,0 +1,26 @@
+from abc import ABCMeta
+
+from pylabrobot.machine import MachineBackend
+
+
+class PumpBackend(MachineBackend, metaclass=ABCMeta):
+  """ Abstract base class for pump backends. """
+
+  def run_revolutions(self, num_revolutions: float):
+    """ Run for a given number of revolutions.
+
+    Args:
+      num_revolutions: number of revolutions to run.
+    """
+
+  def run_continuously(self, speed: float):
+    """ Run continuously at a given speed.
+
+    If speed is 0, the pump will be halted.
+
+    Args:
+      speed: speed in rpm.
+    """
+
+  def halt(self):
+    """ Halt the pump. """

pylabrobot/pumps/cole_parmer/__init__.py

@@ -0,0 +1 @@
+from .masterflex import Masterflex

pylabrobot/pumps/cole_parmer/masterflex.py

@@ -0,0 +1,56 @@
+from typing import Optional
+
+import serial # type: ignore
+
+from pylabrobot.pumps.backend import PumpBackend
+
+
+class Masterflex(PumpBackend):
+  """ Backend for the Cole Parmer Masterflex L/S 07551-20 pump
+
+  Documentation available at:
+    - https://pim-resources.coleparmer.com/instruction-manual/a-1299-1127b-en.pdf
+    - https://web.archive.org/web/20210924061132/https://pim-resources.coleparmer.com/
+      instruction-manual/a-1299-1127b-en.pdf
+  """
+
+  def __init__(self, com_port: str):
+    self.com_port = com_port
+    self.ser: Optional[serial.Serial] = None
+
+  async def setup(self):
+    self.ser = serial.Serial(
+      port=self.com_port,
+      baudrate=4800,
+      timeout=1,
+      parity=serial.PARITY_ODD,
+      stopbits=serial.STOPBITS_ONE,
+      bytesize=serial.SEVENBITS
+    )
+
+    self.ser.write(b"\x05") # Enquiry; ready to send.
+    self.send_command("P21")
+
+  def send_command(self, command: str):
+    assert self.ser is not None
+    command = "\x02" + command + "\x0d"
+    self.ser.write(command.encode())
+    return self.ser.read()
+
+  def run_revolutions(self, num_revolutions: float):
+    num_revolutions = round(num_revolutions, 2)
+    cmd = f"P21V{num_revolutions}G"
+    self.send_command(cmd)
+
+  def run_continuously(self, speed: float):
+    if speed == 0:
+      self.halt()
+      return
+
+    direction = "+" if speed > 0 else "-"
+    speed = int(abs(speed))
+    cmd = f"P21S{direction}{speed}G0"
+    self.send_command(cmd)
+
+  def halt(self):
+    self.send_command("P21H")

pylabrobot/pumps/pump.py

@@ -0,0 +1,27 @@
+from pylabrobot.machine import MachineFrontend
+
+from .backend import PumpBackend
+
+
+class Pump(MachineFrontend):
+  """ Frontend for a (peristaltic) pump. """
+
+  def __init__(self, backend: PumpBackend):
+    self.backend: PumpBackend = backend
+
+  def run_revolutions(self, num_revolutions: float):
+    self.backend.run_revolutions(num_revolutions=num_revolutions)
+
+  def run_continuously(self, speed: float):
+    """ Run continuously at a given speed.
+
+    If speed is 0, the pump will be halted.
+
+    Args:
+      speed: speed in rpm
+    """
+
+    self.backend.run_continuously(speed=speed)
+
+  def halt(self):
+    self.backend.halt()