wip

Author: CamDavidsonPilon

core/pioreactor/background_jobs/growth_rate_calculating.py

@@ -41,6 +41,7 @@
 from threading import Thread
 from time import sleep
 from typing import Generator
+from typing import Iterator
 
 import click
 from msgspec.json import decode as loads
@@ -55,6 +56,7 @@
 from pioreactor.config import config
 from pioreactor.utils import local_persistent_storage
 from pioreactor.utils.streaming import DosingObservationSource
+from pioreactor.utils.streaming import IteratorBackedStream
 from pioreactor.utils.streaming import merge_historical_streams
 from pioreactor.utils.streaming import merge_live_streams
 from pioreactor.utils.streaming import MqttDosingSource
@@ -108,13 +110,13 @@ def __init__(
         self._recent_dilution = False
 
     def initialize_extended_kalman_filter(
-        self, od_std: float, rate_std: float, obs_std: float, od_stream: ODObservationSource
+        self, od_std: float, rate_std: float, obs_std: float, od_iter: Iterator[structs.ODReadings]
     ) -> CultureGrowthEKF:
         import numpy as np
 
         self.logger.debug(f"{od_std=}, {rate_std=}, {obs_std=}")
 
-        initial_nOD, initial_growth_rate = self.get_initial_values(od_stream)
+        initial_nOD, initial_growth_rate = self.get_initial_values(od_iter)
 
         initial_state = np.array([initial_nOD, initial_growth_rate])
         self.logger.debug(f"Initial state: {repr(initial_state)}")
@@ -240,13 +242,13 @@ def _compute_and_cache_od_statistics(
 
         return means, variances
 
-    def get_initial_values(self, od_stream: ODObservationSource) -> tuple[float, float]:
+    def get_initial_values(self, od_iter: Iterator[structs.ODReadings]) -> tuple[float, float]:
         if self.ignore_cache:
             initial_growth_rate = 0.0
-            initial_nOD = self.get_filtered_od_from_stream(od_stream)
+            initial_nOD = self.get_filtered_od_from_iterator(od_iter)
         else:
             initial_growth_rate = self.get_growth_rate_from_cache()
-            initial_nOD = self.get_filtered_od_from_cache_or_stream(od_stream)
+            initial_nOD = self.get_filtered_od_from_cache_or_iterator(od_iter)
         return (initial_nOD, initial_growth_rate)
 
     def get_precomputed_values(
@@ -304,16 +306,16 @@ def get_growth_rate_from_cache(self) -> float:
         with local_persistent_storage("growth_rate") as cache:
             return cache.get(self.cache_key, 0.0)
 
-    def get_filtered_od_from_cache_or_stream(self, od_stream: ODObservationSource) -> float:
+    def get_filtered_od_from_cache_or_iterator(self, od_iter: Iterator[structs.ODReadings]) -> float:
         with local_persistent_storage("od_filtered") as cache:
             value = cache.get(self.cache_key)
         if value:
             return value
         else:
-            return self.get_filtered_od_from_stream(od_stream)
+            return self.get_filtered_od_from_iterator(od_iter)
 
-    def get_filtered_od_from_stream(self, od_stream: ODObservationSource) -> float:
-        scaled_od_readings = self.scale_raw_observations(next(iter(od_stream)))
+    def get_filtered_od_from_iterator(self, od_iter: Iterator[structs.ODReadings]) -> float:
+        scaled_od_readings = self.scale_raw_observations(next(od_iter))
         return mean(scaled_od_readings[channel] for channel in scaled_od_readings.keys())
 
     def get_od_normalization_from_cache(self) -> dict[pt.PdChannel, float]:
@@ -454,18 +456,20 @@ def process_until_disconnected_or_exhausted(
         self.logger.debug(f"od_blank={dict(self.od_blank)}")
 
         # create kalman filter
+        od_iter = iter(od_stream)
         self.ekf = self.initialize_extended_kalman_filter(
             od_std=config.getfloat("growth_rate_kalman", "od_std"),
             rate_std=config.getfloat("growth_rate_kalman", "rate_std"),
             obs_std=config.getfloat("growth_rate_kalman", "obs_std"),
-            od_stream=od_stream,
+            od_iter=od_iter,
         )
 
         # how should we merge streams?
         if od_stream.is_live and dosing_stream.is_live:
-            merged_streams = merge_live_streams(od_stream, dosing_stream, stop_event=self.stopping_event)
+            od_iter_stream = IteratorBackedStream(od_iter, od_stream.set_stop_event)
+            merged_streams = merge_live_streams(od_iter_stream, dosing_stream, stop_event=self.stopping_event)
         elif not od_stream.is_live and not dosing_stream.is_live:
-            merged_streams = merge_historical_streams(od_stream, dosing_stream, key=lambda t: t.timestamp)
+            merged_streams = merge_historical_streams(od_iter, dosing_stream, key=lambda t: t.timestamp)
         else:
             raise ValueError("Both streams must be live or both must be historical.")
 

core/pioreactor/background_jobs/od_reading.py

@@ -980,12 +980,11 @@ def __call__(self, raw_od_readings: structs.ODReadings) -> structs.ODFused | Non
             od_fused_value = compute_fused_od(self.estimator, fused_inputs)
             if self._should_warn_about_bounds(od_fused_value):
                 self.logger.warning(
-                    "Fused OD estimate hit estimator bounds: estimator=%s min_logc=%s max_logc=%s od_fused=%s ref_normalization=%s",
+                    "Fused OD estimate hit estimator bounds: estimator=%s min_logc=%s max_logc=%s od_fused=%s",
                     self.estimator.estimator_name,
                     self.estimator.min_logc,
                     self.estimator.max_logc,
                     od_fused_value,
-                    config.get("od_reading.config", "ref_normalization", fallback="classic"),
                 )
         except Exception as e:
             self.logger.debug(f"Failed to compute fused OD: {e}", exc_info=True)

core/pioreactor/calibrations/protocols/stirring_dc_based.py

@@ -168,6 +168,8 @@ def start_dc_based_session(
     min_dc: float | None = None,
     max_dc: float | None = None,
 ) -> CalibrationSession:
+    if any(is_pio_job_running(["stirring"])):
+        raise ValueError("Stirring must be off before starting.")
     session_id = str(uuid.uuid4())
     now = utc_iso_timestamp()
     return CalibrationSession(

core/pioreactor/pubsub.py

@@ -47,8 +47,12 @@ def loop_stop(self) -> MQTTErrorCode:
             return MQTTErrorCode.MQTT_ERR_INVAL
         self._thread_terminate = True
         # Wake the network loop (select) so it can observe _thread_terminate promptly.
-        # Paho uses a sockpair to interrupt the loop; writing a byte is enough and avoids closing fds.
-        self._reset_sockets(sockpair_only=True)
+        # Avoid closing sockpair fds while the loop thread might be blocked on recv.
+        if self._sockpairW is not None:
+            try:
+                self._sockpairW.send(b"x")
+            except OSError:
+                pass
 
         if threading.current_thread() != thread:
             thread.join()

core/pioreactor/utils/streaming.py

@@ -256,6 +256,20 @@ def __iter__(self):
 S = ODObservationSource | DosingObservationSource
 
 
+class IteratorBackedStream(ODObservationSource):
+    is_live = True
+
+    def __init__(self, iterator: Iterator[ODReadings], stop_event_setter: Callable[[Event], None]) -> None:
+        self._iterator = iterator
+        self._stop_event_setter = stop_event_setter
+
+    def set_stop_event(self, ev: Event) -> None:
+        self._stop_event_setter(ev)
+
+    def __iter__(self) -> Iterator[ODReadings]:
+        return self._iterator
+
+
 def merge_live_streams(
     *iterables: S,
     stop_event: Event | None = None,

core/tests/test_growth_rate_calculating.py

@@ -241,7 +241,25 @@ def test_restart(self) -> None:
                         timestamp="2010-01-01T12:00:35.000Z",
                     ),
                 )
-                assert wait_for(lambda: float(calc1.processor.ekf.state_[-1]) != 0, timeout=3.0)
+                publish(
+                    f"pioreactor/{unit}/{experiment}/od_reading/ods",
+                    create_encoded_od_raw_batched(
+                        ["1", "2"],
+                        [1.155, 0.935],
+                        ["90", "135"],
+                        timestamp="2010-01-01T12:00:35.000Z",
+                    ),
+                )
+                publish(
+                    f"pioreactor/{unit}/{experiment}/od_reading/ods",
+                    create_encoded_od_raw_batched(
+                        ["1", "2"],
+                        [1.156, 0.936],
+                        ["90", "135"],
+                        timestamp="2010-01-01T12:00:35.000Z",
+                    ),
+                )
+                assert wait_for(lambda: float(calc1.processor.ekf.state_[-1]) != 0, timeout=10.0)
 
             with GrowthRateCalculator(unit=unit, experiment=experiment) as calc2:
                 od_stream, dosing_stream = create_od_stream_from_mqtt(