water_valve

pydantic model aind_behavior_services.calibration.water_valve.Measurement

Bases: BaseModel

Input for water valve calibration class

Fields:

• repeat_count (int)

• valve_open_interval (float)

• valve_open_time (float)

• water_weight (List[float])

field repeat_count: int [Required]

Number of times the valve opened.

Constraints:

• ge = 0

field valve_open_interval: float [Required]

Time between two consecutive valve openings (s)

Constraints:

• gt = 0

field valve_open_time: float [Required]

Valve open interval (s)

Constraints:

• gt = 0

field water_weight: List[PositiveFloat] [Required]

Weight of water delivered (g)

Constraints:

• min_length = 1

pydantic model aind_behavior_services.calibration.water_valve.WaterValveCalibration

Bases: Calibration

Water valve calibration class

Fields:

• description (Literal['Calibration of the water valve delivery system'])

• device_name (str)

• input (aind_behavior_services.calibration.water_valve.WaterValveCalibrationInput)

• output (aind_behavior_services.calibration.water_valve.WaterValveCalibrationOutput)

field description: Literal['Calibration of the water valve delivery system'] = 'Calibration of the water valve delivery system'

field device_name: str = 'WaterValve'

Name of the device being calibrated

field input: WaterValveCalibrationInput [Required]

field output: WaterValveCalibrationOutput [Required]

pydantic model aind_behavior_services.calibration.water_valve.WaterValveCalibrationInput

Bases: BaseModel

Fields:

• measurements (List[aind_behavior_services.calibration.water_valve.Measurement])

field measurements: List[Measurement] = []

List of measurements

calibrate_output(input: WaterValveCalibrationInput | None = None) → WaterValveCalibrationOutput

Calibrate the water valve delivery system by populating the output field

pydantic model aind_behavior_services.calibration.water_valve.WaterValveCalibrationOutput

Bases: BaseModel

Output for water valve calibration class

Fields:

• interval_average (Dict[float, float] | None)

• offset (float)

• r2 (float | None)

• slope (float)

• valid_domain (List[float] | None)

field interval_average: Dict[PositiveFloat, PositiveFloat] | None = None

Dictionary keyed by measured valve interval and corresponding average single event volume.

field offset: float [Required]

Offset of the linear regression : Volume(g) = Slope(g/s) * time(s) + offset(g)

field r2: float | None = None

R2 metric from the linear model.

Constraints:

• ge = 0

• le = 1

field slope: float [Required]

Slope of the linear regression : Volume(g) = Slope(g/s) * time(s) + offset(g)

field valid_domain: List[PositiveFloat] | None = None

The optional time-intervals the calibration curve was calculated on.

Constraints:

• min_length = 2

Example

from aind_behavior_services.calibration import water_valve as wv
from aind_behavior_services.utils import utcnow


def linear_model(time, slope, offset):
    return slope * time + offset


_delta_times = [0.1, 0.2, 0.3, 0.4, 0.5]
_slope = 10.1
_offset = -0.3

_water_weights = [linear_model(x, _slope, _offset) for x in _delta_times]
_inputs = [
    wv.Measurement(valve_open_interval=0.5, valve_open_time=t[0], water_weight=[t[1]], repeat_count=1)
    for t in zip(_delta_times, _water_weights)
]


_outputs = wv.WaterValveCalibrationOutput(
    interval_average={interval: volume for interval, volume in zip(_delta_times, _water_weights)},
    slope=_slope,
    offset=_offset,
    r2=1.0,
    valid_domain=[value for value in _delta_times],
)

input = wv.WaterValveCalibrationInput(measurements=_inputs)

calibration = wv.WaterValveCalibration(
    input=input,
    output=input.calibrate_output(),
    device_name="WaterValve",
    date=utcnow(),
)