water_valve

pydantic model aind_behavior_services.calibration.water_valve.CalibrationLogic[source]

Bases: AindBehaviorTaskLogicModel

Olfactometer operation control model that is used to run a calibration data acquisition workflow

Config:

  • extra: str = forbid

  • validate_assignment: bool = True

  • validate_defaults: bool = True

  • strict: bool = True

  • str_strip_whitespace: bool = True

Fields:
Validators:
field name: str = 'WaterValveCalibrationLogic'[source]
field task_parameters: CalibrationParameters [Required][source]
field version: Literal[TASK_LOGIC_VERSION] = '0.4.0'[source]
Validated by:

  • coerce_version

pydantic model aind_behavior_services.calibration.water_valve.CalibrationParameters[source]

Bases: TaskParameters

Config:

  • extra: str = allow

  • validate_assignment: bool = True

  • validate_defaults: bool = True

  • strict: bool = True

  • str_strip_whitespace: bool = True

Fields:
Validators:

field repeat_count: int = 200[source]

Number of times the valve opened per measure valve_open_time entry

Constraints:

  • ge = 1

field valve_open_interval: float = 0.2[source]

Time between two consecutive valve openings (s)

Constraints:

  • gt = 0

field valve_open_time: list[PositiveFloat] [Required][source]

An array with the times (s) the valve is open during calibration

Constraints:

  • min_length = 1

pydantic model aind_behavior_services.calibration.water_valve.CalibrationRig[source]

Bases: AindBehaviorRigModel

Fields:
Validators:
field version: Literal[RIG_VERSION] = '0.0.0'[source]
Validated by:
pydantic model aind_behavior_services.calibration.water_valve.Measurement[source]

Bases: BaseModel

Input for water valve calibration class

Fields:
field repeat_count: int [Required][source]

Number of times the valve opened.

Constraints:

  • ge = 0

field valve_open_interval: float [Required][source]

Time between two consecutive valve openings (s)

Constraints:

  • gt = 0

field valve_open_time: float [Required][source]

Valve open interval (s)

Constraints:

  • gt = 0

field water_weight: List[PositiveFloat] [Required][source]

Weight of water delivered (g)

Constraints:

  • min_length = 1

pydantic model aind_behavior_services.calibration.water_valve.WaterValveCalibration[source]

Bases: Calibration

Water valve calibration class

Fields:
field description: Literal['Calibration of the water valve delivery system'] = 'Calibration of the water valve delivery system'[source]
field device_name: str = 'WaterValve'[source]

Name of the device being calibrated

field input: WaterValveCalibrationInput [Required][source]
field output: WaterValveCalibrationOutput [Required][source]
pydantic model aind_behavior_services.calibration.water_valve.WaterValveCalibrationInput[source]

Bases: BaseModel

Fields:
field measurements: List[Measurement] = [][source]

List of measurements

calibrate_output(input: WaterValveCalibrationInput | None = None) → WaterValveCalibrationOutput[source]

Calibrate the water valve delivery system by populating the output field

pydantic model aind_behavior_services.calibration.water_valve.WaterValveCalibrationOutput[source]

Bases: BaseModel

Output for water valve calibration class

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

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

field offset: float [Required][source]

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

field r2: float | None = None[source]

R2 metric from the linear model.

Constraints:

  • ge = 0

  • le = 1

field slope: float [Required][source]

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

field valid_domain: List[PositiveFloat] | None = None[source]

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

Constraints:

  • min_length = 2

Example

import os

from aind_behavior_services.base import get_commit_hash
from aind_behavior_services.calibration import water_valve as wv
from aind_behavior_services.session import AindBehaviorSessionModel
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(),
)

calibration_logic = wv.CalibrationLogic(
    task_parameters=wv.CalibrationParameters(valve_open_time=_delta_times, valve_open_interval=0.5, repeat_count=200)
)

calibration_session = AindBehaviorSessionModel(
    root_path="C:\\Data",
    allow_dirty_repo=False,
    experiment="WaterValveCalibration",
    subject="WaterValve",
    experiment_version="WaterValveCalibration",
    commit_hash=get_commit_hash(),
    date=utcnow(),
)

rig = wv.CalibrationRig(rig_name="WaterValveRig")

seed_path = "local/water_valve_{suffix}.json"
os.makedirs(os.path.dirname(seed_path), exist_ok=True)

with open(seed_path.format(suffix="calibration_logic"), "w") as f:
    f.write(calibration_logic.model_dump_json(indent=3))
with open(seed_path.format(suffix="session"), "w") as f:
    f.write(calibration_session.model_dump_json(indent=3))
with open(seed_path.format(suffix="rig"), "w") as f:
    f.write(rig.model_dump_json(indent=3))