Skip to content

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¤

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: 1200). :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 
22
23
24
25
26
27
28
29
30
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 
32
33
34
35
36
37
38
39
40
41
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. :param units: Units, eg 'ml/min'.

Source code in korobka/pumps/syringe_pump.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def infuse(self, rate: float | str, volume: str, 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.
    :param units: Units, eg 'ml/min'.
    """
    self.send_ascii_command(f"irate {rate} {units}")
    logger.debug("Syringe infuse rate set:%s", rate, units)
    self.send_ascii_command(f"tvolume {volume} ml")
    logger.debug("Syringe withdraw volume set:%s", volume)
    self.send_ascii_command("irun")
    self.wait_for_target()
wait_for_target(poll_interval=1.0, timeout_lim=600) ¤

Polls syringe and returns 'None' when syringe function is complete.

:param poll_interval: Poll interval. :param timeout_lim: Timeout.

Source code in korobka/pumps/syringe_pump.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def wait_for_target(self, poll_interval: float = 1.0, timeout_lim: int = 600) -> None:
    """Polls syringe and returns 'None' when syringe function is complete.

    :param poll_interval: Poll interval.
    :param timeout_lim: Timeout.
    """
    timeout = time.time() + timeout_lim
    valid_responses = [":", ">", "<", "*", "T*"]
    while True:
        response = self.send_ascii_command("status")
        if response in valid_responses:
            if response == "T*":
                logger.info("Syringe function complete")
                return None
            if response == "*":
                logger.error("Pump stalled")
                raise RuntimeError("Pump stalled")
            if response == "<":
                logger.debug("Pump withdrawing")
            if response == ">":
                logger.debug("Pump infusing")
            if response == ":":
                logger.debug("Pump idle")
        if response not in valid_responses:
            logger.debug("Invalid status update")
        if time.time() > timeout:
            logger.error("Pump timed-out")
            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. :param units: Units, eg 'ml/min'.

Source code in korobka/pumps/syringe_pump.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
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.
    :param units: Units, eg 'ml/min'.
    """
    self.send_ascii_command(f"wrate {rate} {units}")
    logger.debug("Syringe withdraw rate set:%s", rate, units)
    self.send_ascii_command(f"tvolume {volume} ml")
    logger.debug("Syringe withdraw volume set:%s", volume)
    self.send_ascii_command("wrun")
    self.wait_for_target()