9032957364

Committed 10 May 2024 01:25PM UTC coverage: 48.538% (+1.7%) from 46.79%

Build # 9032957364

Build Type

Pull #643

github

Specific Base 74d2ec

Committed by

web-flow

Commit Message

Merge aebf2c9af into ec2d8e4fe

Pull Request Pull Request #643: 8.19.0

Run Details

377 of 1074 new or added lines in 38 files covered. (35.1%)

977 existing lines in 19 files now uncovered.

3253 of 6702 relevant lines covered (48.54%)

0.97 hits per line

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

53.95

/iblrig/valve.py

import datetime
import warnings
from collections.abc import Sequence

import numpy as np
import scipy
from numpy.polynomial import Polynomial
from pydantic import 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, 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: PositiveFloat) -> PositiveFloat:
        return max((self._polynomial - volume_ul).roots())

    @validate_call
    def ms2ul(self, volume_ul: PositiveFloat | list[PositiveFloat]) -> PositiveFloat | np.ndarray:
        return self._polynomial(np.array(volume_ul))


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 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(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 Sequence	2✔
4
5	import numpy as np	2✔
6	import scipy	2✔
7	from numpy.polynomial import Polynomial	2✔
8	from pydantic import 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()	×
20	self.add_samples(open_times_ms, weights_g)	×
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)	×
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)	×
29	self._data = np.append(self._data, incoming)	×
30	self._data = np.sort(self._data)	×
31	self._update_fit()	×
32
33	def clear_data(self):	2✔
34	self._data = np.empty((0,), dtype=self._dtype)	×
35	self._update_fit()	×
36
37	@property	2✔
38	def open_times_ms(self) -> np.ndarray:	2✔
39	return self._data['open_times_ms']	×
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	×
48
49	def _update_fit(self) -> None:	2✔
50	if len(self._data) >= 2:	×
51	with warnings.catch_warnings():	×
52	warnings.simplefilter('ignore')	×
NEW 53	try:	×
NEW 54	c, _ = scipy.optimize.curve_fit(	×
55	self._fcn, self.open_times_ms, self.volumes_ul, bounds=([-np.inf, 0, 0], np.inf)
56	)
NEW 57	except RuntimeError:	×
NEW 58	c = [np.nan, np.nan, np.nan]	×
59	else:
60	c = [np.nan, np.nan, np.nan]	×
61	self._polynomial = Polynomial(coef=c)	×
62
63	@validate_call	2✔
64	def ul2ms(self, volume_ul: PositiveFloat) -> PositiveFloat:	2✔
65	return max((self._polynomial - volume_ul).roots())	×
66
67	@validate_call	2✔
68	def ms2ul(self, volume_ul: PositiveFloat \| list[PositiveFloat]) -> PositiveFloat \| np.ndarray:	2✔
69	return self._polynomial(np.array(volume_ul))	×
70
71
72	class Valve:	2✔
73	def __init__(self, settings: HardwareSettingsValve):	2✔
74	self._settings = settings	×
75	volumes_ul = settings.WATER_CALIBRATION_WEIGHT_PERDROP	×
76	weights_g = [volume / 1e3 for volume in volumes_ul]	×
77	self.values = ValveValues(settings.WATER_CALIBRATION_OPEN_TIMES, weights_g)	×
78
79	@property	2✔
80	def calibration_date(self) -> datetime.date:	2✔
81	return self._settings.WATER_CALIBRATION_DATE	×
82
83	@property	2✔
84	def calibration_range(self) -> list[float, float]:	2✔
85	return self._settings.WATER_CALIBRATION_RANGE	×
86
87	@property	2✔
88	def new_calibration_open_times(self) -> set[float]:	2✔
89	return set(np.linspace(self.calibration_range[0], self.calibration_range[1], self._settings.WATER_CALIBRATION_N))	×
90
91	@property	2✔
92	def free_reward_time(self) -> float:	2✔
93	return self._settings.FREE_REWARD_VOLUME_UL	×
94
95	@property	2✔
96	def settings(self) -> HardwareSettingsValve:	2✔
97	settings = self._settings	×
98	settings.WATER_CALIBRATION_OPEN_TIMES = self.values.open_times_ms	×
99	settings.WATER_CALIBRATION_WEIGHT_PERDROP = self.values.volumes_ul	×
100	return settings	×

int-brain-lab / iblrig / 9032957364

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