lots of typing fixes

Author: CamDavidsonPilon

core/pioreactor/actions/leader/backup_database.py

@@ -50,7 +50,7 @@ def _local_available_space(path: str) -> int:
 
 def count_writes_occurring() -> int:
     with local_intermittent_storage("mqtt_to_db_streaming") as c:
-        return c.get("inserts_in_last_60s", 0)
+        return int(c.get("inserts_in_last_60s", 0))  # type: ignore[call-overload]
 
 
 def backup_database(output_file: str, force: bool = False, backup_to_workers: int = 0) -> None:

core/pioreactor/actions/led_intensity.py

@@ -44,7 +44,7 @@ def change_leds_intensities_temporarily(
     old_state = {}
     try:
         with local_intermittent_storage("leds") as cache:
-            old_state = {c: float(cache.get(c, 0.0)) for c in desired_state.keys()}
+            old_state = {c: float(cache.get(c, 0.0)) for c in desired_state.keys()}  # type: ignore[arg-type]
         if not led_intensity(desired_state, **kwargs):
             raise ValueError("Unable to update LED.")
 
@@ -86,15 +86,17 @@ def _update_current_state(
     with local_intermittent_storage("leds") as led_cache:
         # rehydrate old cache
         old_state: LEDsToIntensityMapping = {
-            channel: led_cache.get(str(channel), 0.0) for channel in ALL_LED_CHANNELS
+            channel: float(led_cache.get(str(channel), 0.0))  # type: ignore[arg-type]
+            for channel in ALL_LED_CHANNELS
         }
 
         # update cache
         for channel, intensity in state.items():
             led_cache[channel] = intensity
 
         new_state: LEDsToIntensityMapping = {
-            channel: led_cache.get(str(channel), 0.0) for channel in ALL_LED_CHANNELS
+            channel: float(led_cache.get(str(channel), 0.0))  # type: ignore[arg-type]
+            for channel in ALL_LED_CHANNELS
         }
 
         return new_state, old_state

core/pioreactor/actions/od_blank.py

@@ -4,6 +4,7 @@
 from json import dumps
 from json import loads
 from typing import Any
+from typing import cast
 from typing import Iterator
 from typing import Optional
 
@@ -113,7 +114,7 @@ def delete_od_blank(unit: pt.Unit | None = None, experiment: pt.Experiment | Non
             retain=True,
         )
 
-        means = loads(cache[experiment])
+        means = loads(cast(str | bytes | bytearray, cache[experiment]))
         for channel, mean in means.items():
             pubsub.publish(
                 f"pioreactor/{unit}/{experiment}/{action_name}/mean/{channel}",

core/pioreactor/actions/self_test.py

@@ -132,7 +132,9 @@ def test_REF_is_in_correct_position(
     reference_channel = cast(PdChannel, next((k for k, v in pd_channels.items() if v == REF_keyword), None))
     assert reference_channel is not None, "REF required for this self-test"
 
-    signal_channels = [channel for channel, angle in pd_channels.items() if angle != REF_keyword]
+    signal_channels: list[PdChannel] = [
+        cast(PdChannel, channel) for channel, angle in pd_channels.items() if angle != REF_keyword
+    ]
 
     test_channels = {str(channel): "90" for channel, angle in pd_channels.items()}
 
@@ -467,6 +469,7 @@ def test_PD_is_near_0_volts_for_blank(
     signal_pd_channels = {k: v for k, v in pd_channels.items() if v != REF_keyword}
 
     for channel, angle in signal_pd_channels.items():
+        channel = cast(PdChannel, channel)
         assert angle in ["90", "45", "135"], f"Angle {angle} not valid for this test."
 
         signals = []

core/pioreactor/automations/base.py

@@ -148,7 +148,7 @@ def latest_growth_rate(self) -> float:
                 "readings are too stale (over 5 minutes old) - are `od_reading` and `growth_rate_calculating` running?"
             )
 
-        return cast(float, self._latest_growth_rate)
+        return self._latest_growth_rate
 
     @property
     def latest_normalized_od(self) -> float:
@@ -169,7 +169,7 @@ def latest_normalized_od(self) -> float:
                 "readings are too stale (over 5 minutes old) - are `od_reading` and `growth_rate_calculating` running?"
             )
 
-        return cast(float, self._latest_normalized_od)
+        return self._latest_normalized_od
 
     @property
     def latest_od(self) -> dict[pt.PdChannel, float]:
@@ -211,7 +211,7 @@ def latest_od_fused(self) -> float:
                 f"fused readings are too stale (over 5 minutes old) - is `od_reading` running?. Last reading occurred at {self.latest_od_fused_at}."
             )
 
-        return cast(float, self._latest_od_fused)
+        return self._latest_od_fused
 
     def latest_biomass_value(
         self,

core/pioreactor/background_jobs/base.py

@@ -8,6 +8,7 @@
 from os import getpid
 from time import sleep
 from time import time
+from types import FunctionType
 from typing import Self
 
 from msgspec.json import decode as loads
@@ -271,7 +272,7 @@ def wrapped_init(self: t.Any, *args: t.Any, **kwargs: t.Any) -> None:
                     pass
                 raise
 
-        cls.__init__ = wrapped_init
+        setattr(cls, "__init__", t.cast(FunctionType, wrapped_init))
 
     def __init__(self, unit: pt.Unit, experiment: pt.Experiment, source: str = "app") -> None:
         if self.job_name in DISALLOWED_JOB_NAMES:
@@ -368,6 +369,10 @@ def __post__init__(self) -> None:
     def job_key(self) -> str:
         return f"{self.job_name}/{self.job_id}"
 
+    @staticmethod
+    def _ignore_future_sighups(*args: t.Any) -> None:
+        signal.signal(signal.SIGHUP, signal.SIG_IGN)
+
     def start_passive_listeners(self) -> None:
         # overwrite this to in subclasses to subscribe to topics in MQTT
         # using this handles reconnects correctly.
@@ -738,7 +743,7 @@ def exit_gracefully(reason: int | str, *args: object) -> None:
                     [
                         exit_gracefully,
                         # add a "ignore all future SIGUPs" onto the top of the stack.
-                        lambda *args: signal.signal(signal.SIGHUP, signal.SIG_IGN),
+                        self._ignore_future_sighups,
                     ],
                 )
             except AttributeError:

core/pioreactor/background_jobs/dosing_automation.py

@@ -557,10 +557,10 @@ def _init_volume_throughput(self) -> None:
         )
 
         with local_persistent_storage("alt_media_throughput") as cache:
-            self.alt_media_throughput = cache.get(self.experiment, 0.0)
+            self.alt_media_throughput = float(cache.get(self.experiment, 0.0))  # type: ignore[arg-type]
 
         with local_persistent_storage("media_throughput") as cache:
-            self.media_throughput = cache.get(self.experiment, 0.0)
+            self.media_throughput = float(cache.get(self.experiment, 0.0))  # type: ignore[arg-type]
 
         return
 

core/pioreactor/background_jobs/growth_rate_calculating.py

@@ -321,31 +321,31 @@ def _get_od_blank_from_cache(self) -> dict[pt.PdChannel, float]:
             result = cache.get(self.experiment)
 
         if result is not None:
-            od_blanks = result
+            od_blanks = cast(bytes | str, result)
             return loads(od_blanks)
         else:
             return defaultdict(lambda: 0.0)
 
     def _get_growth_rate_from_cache(self) -> float:
         with local_persistent_storage("growth_rate") as cache:
-            return cache.get(self.experiment, 0.0)
+            return float(cast(float | int, cache.get(self.experiment, 0.0)))
 
     def _get_filtered_od_from_cache(self) -> float:
         with local_persistent_storage("od_filtered") as cache:
-            return cache.get(self.experiment, 1.0)
+            return float(cast(float | int, cache.get(self.experiment, 1.0)))
 
     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]:
         with local_persistent_storage("od_normalization_mean") as cache:
-            result = cache[self.experiment]
+            result = cast(bytes | str, cache[self.experiment])
             return loads(result)
 
     def _get_od_variances_from_cache(self) -> dict[pt.PdChannel, float]:
         with local_persistent_storage("od_normalization_variance") as cache:
-            result = cache[self.experiment]
+            result = cast(bytes | str, cache[self.experiment])
             return loads(result)
 
     @staticmethod
@@ -402,7 +402,9 @@ def _update_state_from_observation(
         updated_state_, covariance_ = self.ekf.update(
             list(scaled_observations.values()), dt, self._recent_dilution
         )
-        latest_od_filtered, latest_growth_rate = float(updated_state_[0]), float(updated_state_[1])
+        updated_state = cast(Any, updated_state_)
+        covariance = cast(Any, covariance_)
+        latest_od_filtered, latest_growth_rate = float(updated_state[0]), float(updated_state[1])
 
         if self._obs_since_last_dose is not None and self._obs_required_to_reset is not None:
             self._obs_since_last_dose += 1
@@ -422,8 +424,8 @@ def _update_state_from_observation(
         )
 
         kf_outputs = structs.KalmanFilterOutput(
-            state=self.ekf.state_.tolist(),
-            covariance_matrix=covariance_.tolist(),
+            state=cast(Any, self.ekf.state_).tolist(),
+            covariance_matrix=covariance.tolist(),
             timestamp=timestamp,
         )
 
@@ -539,6 +541,7 @@ def click_growth_rate_calculating(ctx: click.Context, ignore_cache: bool) -> Non
         experiment = whoami.get_assigned_experiment_name(unit)
 
         use_fused_od = _should_use_fused_od(unit)
+        od_stream: MqttODSource | MqttODFusedSource
         if use_fused_od:
             od_stream = MqttODFusedSource(unit=unit, experiment=experiment, skip_first=5)
         else:

core/pioreactor/background_jobs/leader/mqtt_to_db_streaming.py

@@ -140,7 +140,7 @@ def callback(message: pt.MQTTMessage) -> None:
                 values_placeholder = ", ".join(":" + c for c in new_row.keys())
                 SQL = f"""INSERT INTO {table} ({cols_placeholder}) VALUES ({values_placeholder})"""
                 try:
-                    self.sqliteworker.execute(SQL, new_row)
+                    self.sqliteworker.execute(SQL, cast(object, new_row))  # type: ignore[arg-type]
                 except Exception as e:
                     self.logger.warning(e)
                     self.logger.debug(f"SQL that caused error: `{SQL}`")

core/pioreactor/background_jobs/monitor.py

@@ -5,6 +5,7 @@
 from threading import Thread
 from time import sleep
 from typing import Callable
+from typing import cast
 from typing import Optional
 
 import click
@@ -47,8 +48,10 @@
 
 
 class classproperty(property):
-    def __get__(self, obj: object, objtype: type[object] | None = None) -> object:
-        return self.fget(objtype)
+    def __get__(self, obj: object | None, objtype: type[object] | None = None) -> object:  # type: ignore[override]
+        assert self.fget is not None
+        owner = objtype if objtype is not None else type(obj)
+        return self.fget(owner)
 
 
 class Monitor(LongRunningBackgroundJob):
@@ -119,7 +122,7 @@ def pretty_version(info: tuple | None) -> str:
 
         # previously I had pioreactor_version and model name here, but this starts before the webserver is online, and it
         # would crash this job.
-        self.versions = {
+        self.versions: dict[str, str | None] = {
             "app": pretty_version(version.software_version_info),
             "hat": pretty_version(version.hardware_version_info),
             "firmware": pretty_version(version.get_firmware_version()),
@@ -373,7 +376,7 @@ def check_heater_pcb_temperature(self) -> None:
 
         observed_tmp = tmp_driver.get_temperature()
 
-        if observed_tmp >= self.MAX_TEMP_TO_SHUTDOWN:
+        if observed_tmp >= cast(float, self.MAX_TEMP_TO_SHUTDOWN):
             # something is wrong - temperature_automation should have detected this, but didn't, so it must have failed / incorrectly cleaned up.
             # we're going to just shutdown to be safe.
             self.logger.error(
@@ -382,9 +385,9 @@ def check_heater_pcb_temperature(self) -> None:
 
             subprocess.call("sudo shutdown now --poweroff", shell=True)
 
-        elif observed_tmp >= self.MAX_TEMP_TO_SHUTDOWN_IF_NO_TEMP_AUTOMATION and not utils.is_pio_job_running(
-            "temperature_automation"
-        ):
+        elif observed_tmp >= cast(
+            float, self.MAX_TEMP_TO_SHUTDOWN_IF_NO_TEMP_AUTOMATION
+        ) and not utils.is_pio_job_running("temperature_automation"):
             # errant PWM?
             # false positive: small chance this is in an incubator?
 
@@ -431,7 +434,7 @@ def check_for_mqtt_connection_to_leader(self) -> None:
     def check_for_last_backup(self) -> None:
         with utils.local_persistent_storage("database_backups") as cache:
             if cache.get("latest_backup_timestamp"):
-                latest_backup_at = to_datetime(cache["latest_backup_timestamp"])
+                latest_backup_at = to_datetime(cast(str, cache["latest_backup_timestamp"]))
 
                 if (current_utc_datetime() - latest_backup_at).days > 30:
                     self.logger.warning(