syringe_pump

korobka.pumps.syringe_pump

Module for working with KDScientific Legato 180 syringe pump using RS485 Protocol.

Uses the BaseSerial class for serial communication.

Classes

PumpStatus

Bases: NamedTuple

Parsed fields of a Legato status reply.

Attributes

moving property

True while the motor is turning.

stalled property

True if the pump reports a stall.

target_reached property

True once the target volume has been delivered.

volume_ml property

Delivered volume in millilitres.

SyringePump(port, baudrate=115200, timeout=1)

Bases: BaseSerial

Handles direct communication with the syringe pump over RS485.

:param port: Serial port for the pump. :param baudrate: Communication baud rate (default: 115200). :param timeout: Serial read timeout (default: 1s).

Initializes the pump communication using BaseSerial.

:param port: Serial port for the pump. :param baudrate: Communication baud rate (default: 115200). :param timeout: Serial read timeout (default: 1s).

Source code in korobka/pumps/syringe_pump.py

def __init__(self, port: str, baudrate: int = 115200, timeout: int = 1):
    """Initializes the pump communication using BaseSerial.

    :param port: Serial port for the pump.
    :param baudrate: Communication baud rate (default: 115200).
    :param timeout: Serial read timeout (default: 1s).
    """
    super().__init__(port=port, baudrate=baudrate, parity="E", stopbits=1, timeout=timeout)
    logger.info("SyringePump initialized on port %s", port)

Methods:

config_syringe(diameter, volume)

Configures syringe properties.

:param diameter: Diameter of the syringe. :param volume: Volume of the syringe.

Source code in korobka/pumps/syringe_pump.py

def config_syringe(self, diameter: float, volume: float) -> None:
    """Configures syringe properties.

    :param diameter: Diameter of the syringe.
    :param volume: Volume of the syringe.
    """
    self.send_ascii_command(f"diameter {diameter}")
    logger.info("Config diameter %s", diameter)
    self.send_ascii_command(f"svolume {volume} ml")
    logger.info("Config volume: %s", volume)

infuse(rate, volume, units='ml/min')

Operates the pump to infuse a given volume at a given rate.

:param rate: Infuse rate. :param volume: Infuse volume in ml. :param units: Units, eg 'ml/min'.

Source code in korobka/pumps/syringe_pump.py

def infuse(self, rate: float | str, volume: float, units: str = "ml/min") -> None:
    """Operates the pump to infuse a given volume at a given rate.

    :param rate: Infuse rate.
    :param volume: Infuse volume in ml.
    :param units: Units, eg 'ml/min'.
    """
    self.send_ascii_command(f"irate {rate} {units}")
    logger.debug("Syringe infuse rate set: %s %s", rate, units)
    self.send_ascii_command(f"tvolume {volume} ml")
    logger.debug("Syringe infuse volume set: %s ml", volume)
    self._clear_counters()
    self.send_ascii_command("irun")
    self.wait_for_target()

read_status()

Queries the pump and returns its parsed status.

:return: Parsed status fields. :raises ValueError: If no status line can be found in the reply.

Source code in korobka/pumps/syringe_pump.py

def read_status(self) -> PumpStatus:
    """Queries the pump and returns its parsed status.

    :return: Parsed status fields.
    :raises ValueError: If no status line can be found in the reply.
    """
    reply = self.send_ascii_command("status")

    matches = list(_STATUS_RE.finditer(reply))
    if not matches:
        raise ValueError(f"Unparsable status reply: {reply!r}")

    rate, elapsed, volume, flags = matches[-1].groups()

    # The prompt trails the data line. If it has not arrived inside the read window
    # the last line is the data line itself, which the length test rejects.
    lines = [line.strip() for line in reply.splitlines() if line.strip()]
    prompt = lines[-1] if len(lines[-1]) <= _MAX_PROMPT_LEN else ""

    return PumpStatus(
        rate_fl_s=int(rate),
        elapsed_ms=int(elapsed),
        volume_fl=int(volume),
        flags=flags,
        prompt=prompt,
    )

stop()

Stops the pump immediately.

Source code in korobka/pumps/syringe_pump.py

def stop(self) -> None:
    """Stops the pump immediately."""
    self.send_ascii_command("stop")
    logger.info("Pump stopped")

wait_for_target(poll_interval=0.2, timeout_lim=600)

Polls the pump until the target volume is reached.

:param poll_interval: Pause between polls. The serial read itself already costs roughly 1.5 s btw. :param timeout_lim: Timeout. :raises RuntimeError: If the pump stalls or status becomes unreadable. :raises TimeoutError: If the target is not reached within timeout_lim.

Source code in korobka/pumps/syringe_pump.py

def wait_for_target(self, poll_interval: float = 0.2, timeout_lim: int = 600) -> None:
    """Polls the pump until the target volume is reached.

    :param poll_interval: Pause between polls. The serial read itself already costs
        roughly 1.5 s btw.
    :param timeout_lim: Timeout.
    :raises RuntimeError: If the pump stalls or status becomes unreadable.
    :raises TimeoutError: If the target is not reached within timeout_lim.
    """
    deadline = time.time() + timeout_lim
    armed = False
    glitches = 0

    while True:
        try:
            status = self.read_status()
        except (ValueError, RuntimeError) as exc:
            glitches += 1
            logger.warning("Status read failed (%d/%d): %s", glitches, _MAX_STATUS_GLITCHES, exc)
            if glitches >= _MAX_STATUS_GLITCHES:
                self._safe_stop()
                raise
            time.sleep(poll_interval)
            continue
        glitches = 0

        if status.stalled:
            logger.error("Pump stalled after %.4f ml", status.volume_ml)
            raise RuntimeError("Pump stalled")

        # The target-reached flag survives the end of a run, so a stale "T" from the
        # previous transfer would otherwise satisfy this loop before the motor moves.
        # _clear_counters zeroes elapsed_ms beforehand, making it a start-of-run gate.
        armed = armed or status.moving or status.elapsed_ms > 0

        if armed and status.target_reached:
            logger.info(
                "Target reached: %.4f ml in %.2f s",
                status.volume_ml,
                status.elapsed_ms / 1000,
            )
            return

        if time.time() > deadline:
            logger.error("Pump timed out after %.4f ml", status.volume_ml)
            self._safe_stop()
            raise TimeoutError("Timeout")

        time.sleep(poll_interval)

withdraw(rate, volume, units='ml/min')

Operates the pump to withdraw a given volume at a given rate.

For most cases rate can be set to 'max'.

:param rate: Withdraw rate. :param volume: Withdraw volume in ml. :param units: Units, eg 'ml/min'.

Source code in korobka/pumps/syringe_pump.py

def withdraw(self, rate: float | str, volume: float, units: str = "ml/min") -> None:
    """Operates the pump to withdraw a given volume at a given rate.

    For most cases rate can be set to 'max'.

    :param rate: Withdraw rate.
    :param volume: Withdraw volume in ml.
    :param units: Units, eg 'ml/min'.
    """
    self.send_ascii_command(f"wrate {rate} {units}")
    logger.debug("Syringe withdraw rate set: %s %s", rate, units)
    self.send_ascii_command(f"tvolume {volume} ml")
    logger.debug("Syringe withdraw volume set: %s ml", volume)
    self._clear_counters()
    self.send_ascii_command("wrun")
    self.wait_for_target()