Skip to content

Index

Pump control modules for the plant watering system.

This package defines the abstract base class for pump implementations and provides concrete implementations for supported hardware.

Included implementations
  • ad20p_1230e.CS_IO404_Based_AD20P_1230E: AD20P-1230E pump controlled via a CS-IO404 Modbus relay module.
Writing a custom pump module
  1. Create a new .py file in this package (e.g. my_pump.py).
  2. Define a class that inherits from Pump and a bus interface mixin from plant_controller.com_bus (I2CInterface or MODBUSInterface).
  3. Implement the pumping_callback method to activate the pump for the calculated duration and then record the event via self.db_save_function.
  4. The pump is instantiated by the Plant class. To use a custom pump, modify the Plant initialization or extend the config parsing.
Calibration

Pumps convert a dosage in milliliters to a pumping duration using linear calibration parameters (slope and offset). The relationship is::

time_seconds = slope * dosage_ml + offset

The calibration_parameters dict must contain 'slope' and 'offset' keys. Use the setup utility's calibration procedure to determine these.

Example minimal pump::

from plant_controller.pumps import Pump
from plant_controller.com_bus import MODBUSInterface, MODBUS
from plant_controller.datapoint import WateringEvent

class MyPump(Pump, MODBUSInterface):
    def __init__(self, bus, db_save_function, calibration_parameters,
                 **kwargs):
        super().__init__(bus, db_save_function, calibration_parameters)
        # ... hardware-specific setup ...

    async def pumping_callback(self, dosage: int):
        pump_time = self.calibration_parameters["slope"] * dosage \
                    + self.calibration_parameters["offset"]
        # ... activate pump for pump_time seconds ...
        await self.db_save_function(WateringEvent(dosage=dosage))

Pump

Bases: ABC

Abstract base class for all pump implementations.

Subclasses must also inherit from a bus interface mixin (I2CInterface or MODBUSInterface) so that the controller can determine which bus to pass during initialization.

Attributes:

Name Type Description
bus

The communication bus instance for hardware control.
db_save_function

Callable to persist WateringEvent datapoints.
calibration_parameters

Dict containing at minimum 'slope' (sec/ml) and 'offset' (sec) for dosage-to-time conversion.
Source code in pt/controller_3/src/plant_controller/pumps/__init__.py 
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
class Pump(ABC):
    """Abstract base class for all pump implementations.

    Subclasses must also inherit from a bus interface mixin
    (``I2CInterface`` or ``MODBUSInterface``) so that the controller
    can determine which bus to pass during initialization.

    Attributes:
        bus: The communication bus instance for hardware control.
        db_save_function: Callable to persist WateringEvent datapoints.
        calibration_parameters: Dict containing at minimum 'slope' (sec/ml)
            and 'offset' (sec) for dosage-to-time conversion.
    """

    def __init__(
        self,
        bus: Bus,
        db_save_function: Callable[[Datapoint | list[Datapoint]], None],
        calibration_parameters: dict[str, Any]
    ):
        """Initialize the pump.

        Args:
            bus: Communication bus instance (I2C or MODBUS).
            db_save_function: Function to save watering events to DB.
            calibration_parameters: Dict with 'slope' (sec/ml) and 'offset'
                (sec) for the linear dosage-to-time model.
        """
        self.bus = bus
        self.db_save_function = db_save_function
        self.calibration_parameters = calibration_parameters

    @abstractmethod
    async def pumping_callback(self, dosage: int):
        """Activate the pump to deliver the specified dosage.

        This method is called by the pump schedule and must block (await)
        until pumping is complete to ensure accurate dosage delivery.
        After pumping, implementations should record the event via
        ``await self.db_save_function(WateringEvent(dosage=dosage))``.

        Args:
            dosage: Amount to pump in milliliters.
        """
        pass

__init__(bus, db_save_function, calibration_parameters)

Initialize the pump.

Parameters:

Name Type Description Default
bus Bus

Communication bus instance (I2C or MODBUS).

 required
db_save_function Callable[[Datapoint | list[Datapoint]], None]

Function to save watering events to DB.

 required
calibration_parameters dict[str, Any]

Dict with 'slope' (sec/ml) and 'offset' (sec) for the linear dosage-to-time model.

 required
Source code in pt/controller_3/src/plant_controller/pumps/__init__.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def __init__(
    self,
    bus: Bus,
    db_save_function: Callable[[Datapoint | list[Datapoint]], None],
    calibration_parameters: dict[str, Any]
):
    """Initialize the pump.

    Args:
        bus: Communication bus instance (I2C or MODBUS).
        db_save_function: Function to save watering events to DB.
        calibration_parameters: Dict with 'slope' (sec/ml) and 'offset'
            (sec) for the linear dosage-to-time model.
    """
    self.bus = bus
    self.db_save_function = db_save_function
    self.calibration_parameters = calibration_parameters

pumping_callback(dosage) abstractmethod async

Activate the pump to deliver the specified dosage.

This method is called by the pump schedule and must block (await) until pumping is complete to ensure accurate dosage delivery. After pumping, implementations should record the event via await self.db_save_function(WateringEvent(dosage=dosage)).

Parameters:

Name Type Description Default
dosage int

Amount to pump in milliliters.

 required
Source code in pt/controller_3/src/plant_controller/pumps/__init__.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
@abstractmethod
async def pumping_callback(self, dosage: int):
    """Activate the pump to deliver the specified dosage.

    This method is called by the pump schedule and must block (await)
    until pumping is complete to ensure accurate dosage delivery.
    After pumping, implementations should record the event via
    ``await self.db_save_function(WateringEvent(dosage=dosage))``.

    Args:
        dosage: Amount to pump in milliliters.
    """
    pass