10992769815

Committed 23 Sep 2024 10:50AM UTC coverage: 47.799% (+1.0%) from 46.79%

Build # 10992769815

Build Type

Pull #716

github

Specific Base 60cd00

Committed by

web-flow

Commit Message

Merge 73b6a53cb into a946a6ff9

Pull Request Pull Request #716: 8.24.1

Run Details

22 of 49 new or added lines in 9 files covered. (44.9%)

1015 existing lines in 22 files now uncovered.

4191 of 8768 relevant lines covered (47.8%)

0.96 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

83.33

/iblrig/valve.py

import datetime
import warnings
from collections.abc import Iterable, Sequence

import numpy as np
import scipy
from numpy.polynomial import Polynomial
from pydantic import NonNegativeFloat, PositiveFloat, validate_call

from iblrig.pydantic_definitions import HardwareSettingsValve


class ValveValues:
    _dtype = [('open_times_ms', float), ('weights_g', float)]
    _data: np.ndarray
    _polynomial: Polynomial

    def __init__(self, open_times_ms: Sequence[float], weights_g: Sequence[float]):
        self.clear_data()
        self.add_samples(open_times_ms, weights_g)

    @staticmethod
    def _fcn(x: np.ndarray, a: float, b: float, c: float) -> np.ndarray:
        return a + b * x + c * np.square(x)

    @validate_call
    def add_samples(self, open_times_ms: Sequence[PositiveFloat], weights_g: Sequence[PositiveFloat]):
        incoming = np.rec.fromarrays([open_times_ms, weights_g], dtype=self._dtype)
        self._data = np.append(self._data, incoming)
        self._data = np.sort(self._data)
        self._update_fit()

    def clear_data(self):
        self._data = np.empty((0,), dtype=self._dtype)
        self._update_fit()

    @property
    def open_times_ms(self) -> np.ndarray:
        return self._data['open_times_ms']

    @property
    def weights_g(self) -> np.ndarray:
        return self._data['weights_g']

    @property
    def volumes_ul(self) -> np.ndarray:
        return self._data['weights_g'] * 1e3

    def _update_fit(self) -> None:
        if len(self._data) >= 2:
            with warnings.catch_warnings():
                warnings.simplefilter('ignore')
                try:
                    c, _ = scipy.optimize.curve_fit(
                        self._fcn, self.open_times_ms, self.volumes_ul, p0=[0, 0, 0], bounds=([-np.inf, 0, 0], np.inf)
                    )
                except RuntimeError:
                    c = [np.nan, np.nan, np.nan]
        else:
            c = [np.nan, np.nan, np.nan]
        self._polynomial = Polynomial(coef=c)

    @validate_call
    def ul2ms(self, volume_ul: NonNegativeFloat | Iterable[NonNegativeFloat]) -> NonNegativeFloat | np.ndarray:
        if isinstance(volume_ul, Iterable):
            return np.array([self.ul2ms(v) for v in volume_ul])
        elif volume_ul == 0.0:
            return 0.0
        else:
            return max(np.append((self._polynomial - volume_ul).roots(), 0.0))

    @validate_call
    def ms2ul(self, time_ms: NonNegativeFloat | Iterable[NonNegativeFloat]) -> NonNegativeFloat | np.ndarray:
        if isinstance(time_ms, Iterable):
            return np.array([self.ms2ul(t) for t in time_ms])
        elif time_ms == 0.0:
            return 0.0
        else:
            return max(np.append(self._polynomial(time_ms), 0.0))


class Valve:
    def __init__(self, settings: HardwareSettingsValve):
        self._settings = settings
        volumes_ul = settings.WATER_CALIBRATION_WEIGHT_PERDROP
        weights_g = [volume / 1e3 for volume in volumes_ul]
        self.values = ValveValues(settings.WATER_CALIBRATION_OPEN_TIMES, weights_g)

    @property
    def calibration_date(self) -> datetime.date:
        return self._settings.WATER_CALIBRATION_DATE

    @property
    def is_calibrated(self) -> bool:
        return datetime.date.today() >= self.calibration_date

    @property
    def calibration_range(self) -> list[float, float]:
        return self._settings.WATER_CALIBRATION_RANGE

    @property
    def new_calibration_open_times(self) -> set[float]:
        return set(np.linspace(self.calibration_range[0], self.calibration_range[1], self._settings.WATER_CALIBRATION_N))

    @property
    def free_reward_time_sec(self) -> float:
        return self.values.ul2ms(self._settings.FREE_REWARD_VOLUME_UL) / 1000.0

    @property
    def free_reward_volume_ul(self) -> float:
        return self._settings.FREE_REWARD_VOLUME_UL

    @property
    def settings(self) -> HardwareSettingsValve:
        settings = self._settings
        settings.WATER_CALIBRATION_OPEN_TIMES = self.values.open_times_ms
        settings.WATER_CALIBRATION_WEIGHT_PERDROP = self.values.volumes_ul
        return settings

1	import datetime	2✔
2	import warnings	2✔
3	from collections.abc import Iterable, Sequence	2✔
4
5	import numpy as np	2✔
6	import scipy	2✔
7	from numpy.polynomial import Polynomial	2✔
8	from pydantic import NonNegativeFloat, PositiveFloat, validate_call	2✔
9
10	from iblrig.pydantic_definitions import HardwareSettingsValve	2✔
11
12
13	class ValveValues:	2✔
14	_dtype = [('open_times_ms', float), ('weights_g', float)]	2✔
15	_data: np.ndarray	2✔
16	_polynomial: Polynomial	2✔
17
18	def __init__(self, open_times_ms: Sequence[float], weights_g: Sequence[float]):	2✔
19	self.clear_data()	2✔
20	self.add_samples(open_times_ms, weights_g)	2✔
21
22	@staticmethod	2✔
23	def _fcn(x: np.ndarray, a: float, b: float, c: float) -> np.ndarray:	2✔
24	return a + b * x + c * np.square(x)	2✔
25
26	@validate_call	2✔
27	def add_samples(self, open_times_ms: Sequence[PositiveFloat], weights_g: Sequence[PositiveFloat]):	2✔
28	incoming = np.rec.fromarrays([open_times_ms, weights_g], dtype=self._dtype)	2✔
29	self._data = np.append(self._data, incoming)	2✔
30	self._data = np.sort(self._data)	2✔
31	self._update_fit()	2✔
32
33	def clear_data(self):	2✔
34	self._data = np.empty((0,), dtype=self._dtype)	2✔
35	self._update_fit()	2✔
36
37	@property	2✔
38	def open_times_ms(self) -> np.ndarray:	2✔
39	return self._data['open_times_ms']	2✔
40
41	@property	2✔
42	def weights_g(self) -> np.ndarray:	2✔
43	return self._data['weights_g']	×
44
45	@property	2✔
46	def volumes_ul(self) -> np.ndarray:	2✔
47	return self._data['weights_g'] * 1e3	2✔
48
49	def _update_fit(self) -> None:	2✔
50	if len(self._data) >= 2:	2✔
51	with warnings.catch_warnings():	2✔
52	warnings.simplefilter('ignore')	2✔
53	try:	2✔
54	c, _ = scipy.optimize.curve_fit(	2✔
55	self._fcn, self.open_times_ms, self.volumes_ul, p0=[0, 0, 0], bounds=([-np.inf, 0, 0], np.inf)
56	)
57	except RuntimeError:	×
58	c = [np.nan, np.nan, np.nan]	×
59	else:
60	c = [np.nan, np.nan, np.nan]	2✔
61	self._polynomial = Polynomial(coef=c)	2✔
62
63	@validate_call	2✔
64	def ul2ms(self, volume_ul: NonNegativeFloat \| Iterable[NonNegativeFloat]) -> NonNegativeFloat \| np.ndarray:	2✔
65	if isinstance(volume_ul, Iterable):	2✔
66	return np.array([self.ul2ms(v) for v in volume_ul])	2✔
67	elif volume_ul == 0.0:	2✔
68	return 0.0	2✔
69	else:
70	return max(np.append((self._polynomial - volume_ul).roots(), 0.0))	2✔
71
72	@validate_call	2✔
73	def ms2ul(self, time_ms: NonNegativeFloat \| Iterable[NonNegativeFloat]) -> NonNegativeFloat \| np.ndarray:	2✔
NEW 74	if isinstance(time_ms, Iterable):	×
NEW 75	return np.array([self.ms2ul(t) for t in time_ms])	×
NEW 76	elif time_ms == 0.0:	×
NEW 77	return 0.0	×
78	else:
NEW 79	return max(np.append(self._polynomial(time_ms), 0.0))	×
80
81
82	class Valve:	2✔
83	def __init__(self, settings: HardwareSettingsValve):	2✔
84	self._settings = settings	2✔
85	volumes_ul = settings.WATER_CALIBRATION_WEIGHT_PERDROP	2✔
86	weights_g = [volume / 1e3 for volume in volumes_ul]	2✔
87	self.values = ValveValues(settings.WATER_CALIBRATION_OPEN_TIMES, weights_g)	2✔
88
89	@property	2✔
90	def calibration_date(self) -> datetime.date:	2✔
91	return self._settings.WATER_CALIBRATION_DATE	2✔
92
93	@property	2✔
94	def is_calibrated(self) -> bool:	2✔
95	return datetime.date.today() >= self.calibration_date	2✔
96
97	@property	2✔
98	def calibration_range(self) -> list[float, float]:	2✔
99	return self._settings.WATER_CALIBRATION_RANGE	×
100
101	@property	2✔
102	def new_calibration_open_times(self) -> set[float]:	2✔
103	return set(np.linspace(self.calibration_range[0], self.calibration_range[1], self._settings.WATER_CALIBRATION_N))	×
104
105	@property	2✔
106	def free_reward_time_sec(self) -> float:	2✔
107	return self.values.ul2ms(self._settings.FREE_REWARD_VOLUME_UL) / 1000.0	2✔
108
109	@property	2✔
110	def free_reward_volume_ul(self) -> float:	2✔
111	return self._settings.FREE_REWARD_VOLUME_UL	×
112
113	@property	2✔
114	def settings(self) -> HardwareSettingsValve:	2✔
115	settings = self._settings	×
116	settings.WATER_CALIBRATION_OPEN_TIMES = self.values.open_times_ms	×
117	settings.WATER_CALIBRATION_WEIGHT_PERDROP = self.values.volumes_ul	×
118	return settings	×

int-brain-lab / iblrig / 10992769815

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous