Merge branch 'release/25.5.22'

* release/25.5.22: (43 commits)
  bump version
  remove this check. It's flakey + users can make modifications more easily afterwards now
  more test
  need to restart mqtt to db since we are changing a struct defintion
  run mqtt publish in another thread so we aren't blocking the main checks
  new pump logging
  adding support for negative dosing events
  adding support for negative dosing events
  adding echo about moving ui foldeR
  changelog and test
  I don't need some of these pragma for my use case
  fix more tests
  try this
  improvements to creating mqtt clients; adding logic to handle if the BJ subclass's __init__ fails - we gracefully clean up
  try this
  try this
  revert and simply
  this fixes the db is lcoked errors
  no error letting you know what happened
  lets see the consequence of changing the isolation level
  ...

Author: CamDavidsonPilon

.github/workflows/ci.yaml

@@ -70,7 +70,7 @@ jobs:
 
       - name: Run tests
         run:  |
-          pytest pioreactor/tests/ -vv --timeout 600 --random-order --durations 15
+          pytest pioreactor/tests/ -vv --timeout 240 --random-order --durations 15 --log-level DEBUG
           pytest pioreactor/tests/test_automation_imports.py
         env:
           TESTING: 1

CHANGELOG.md

@@ -1,3 +1,34 @@
+### 25.5.22
+
+#### Enhancements
+ - new _System logs_ under _Inventory_ to track logs happening outside of experiments in your cluster.
+  - Better organization of logs in the UI. System logs, like calibrations, worker additions, etc. won't show up on the Overview page.
+ - Exported data zips have folders for each dataset requested.
+ - Improvements to the Kalman filter. For users using the growth-rate model with media dosing, you should see improvements to your growth-rate time series. We recommend the following configuration:
+ ```
+[growth_rate_kalman]
+# obs_std ↑ smooths growth rate, rate_std ↑ more responsive growth rate
+obs_std=1.5
+od_std=0.0025
+rate_std=0.25
+ ```
+  **Note: the acceleration term is removed**
+ - Added the column `hours_since_experiment_created` to dataset exports that details hours since experiment was created.
+ - A running pump now fires off an incremental dosing event every N seconds (N=0.5 currently) to tell the software its progress. Previously, we would fire off a single event that represented the total amount moved. This is most noticeable when watching the vial volume change over time (looks more accurate over a short period).
+ - When a pump runs, it _first_ fires off a dosing_event, which stores information about how much liquid is moved. However, if the pump is stopped early, there was no correction issued to the amount of liquid actually moved. Now, when a pump is stopped early, a _negative_ volume is sent s.t. the delta between the initial amount and new amount is equal to the dosed amount (so when you sum up the volume changes, you get the actual change, as expected).
+ - Performance optimizations
+ - New image installs only:
+   - updated base OS to the latest 25-05-06 Raspberry Pi OS. The big change is using Linux kernel 6.12.
+
+#### Bug fixes
+ - fixed stir bar not spinning on Pioreactor page (UI) in some cases
+ - alert user if their OD reading is constant before starting the growth-rate calculator, which would break things.
+ - alert user if their software is installed in a non-standard location. If so, try `pio uninstall pioreactor -y`.
+ - Added a warning if the OD calibration is invalid (ex: a constant line)
+ - Fix for Raspberry Pi 5 using upstream Adafruit libraries.
+
+
+
 ### 25.5.1
 
 #### Enhancements

config.dev.ini

@@ -149,10 +149,9 @@ ws_protocol=ws
 use_tls=0
 
 [growth_rate_kalman]
-acc_std=0.0008
 obs_std=1.5
-od_std=0.005
-rate_std=0.1
+od_std=0.0025
+rate_std=0.25
 
 
 [dosing_automation.config]
@@ -165,7 +164,5 @@ max_subdose=1.0
 [growth_rate_calculating.config]
 # these next two parameters control the length and magnitude
 # of the variance shift that our Kalman filter performs after a dosing event
-ekf_variance_shift_post_dosing_minutes=0.40
-ekf_variance_shift_post_dosing_factor=2500
 ekf_outlier_std_threshold=3.0
 samples_for_od_statistics=35

pioreactor/actions/leader/experiment_profile.py

@@ -550,6 +550,7 @@ def _callable() -> None:
                 "level": options.level,
                 "source": parent_job.job_key,
                 "task": parent_job.job_key,
+                "_souce": "app",
             }
             post_into_leader(
                 f"/api/workers/{unit}/experiments/{experiment}/logs", json=body

pioreactor/actions/leader/export_experiment_data.py

@@ -26,11 +26,24 @@ def source_exists(cursor, table_name_to_check: str) -> bool:
     return cursor.execute(query, (table_name_to_check,)).fetchone() is not None
 
 
-def generate_timestamp_to_localtimestamp_clause(timestamp_columns) -> str:
+def generate_timestamp_to_localtimestamp_clause(timestamp_columns: list[str]) -> str:
     if not timestamp_columns:
         return ""
 
-    clause = ",".join([f"datetime({c}, 'localtime') as {c}_localtime" for c in timestamp_columns])
+    clause = ",".join([f"datetime(T.{c}, 'localtime') as {c}_localtime" for c in timestamp_columns])
+
+    return clause
+
+
+def generate_timestamp_to_relative_time_clause(default_order_by: str) -> str:
+    if not default_order_by:
+        return ""
+
+    START_TIME = "E.created_at"
+
+    clause = (
+        f"(unixepoch({default_order_by}) - unixepoch({START_TIME}))/3600.0 as hours_since_experiment_created"
+    )
 
     return clause
 
@@ -73,7 +86,7 @@ def create_experiment_clause(
         existing_placeholders = existing_placeholders | {
             f"experiment{i}": experiment for i, experiment in enumerate(experiments)
         }
-        return f"experiment IN ({quoted_experiments})", existing_placeholders
+        return f"T.experiment IN ({quoted_experiments})", existing_placeholders
 
 
 def create_timespan_clause(
@@ -82,15 +95,15 @@ def create_timespan_clause(
     if start_time is not None and end_time is not None:
         existing_placeholders["start_time"] = start_time
         existing_placeholders["end_time"] = end_time
-        return f"{time_column} >= :start_time AND {time_column} <= :end_time", existing_placeholders
+        return f"T.{time_column} >= :start_time AND {time_column} <= :end_time", existing_placeholders
 
     elif start_time is not None:
         existing_placeholders["start_time"] = start_time
-        return f"{time_column} >= :start_time", existing_placeholders
+        return f"T.{time_column} >= :start_time", existing_placeholders
 
     elif end_time is not None:
         existing_placeholders["end_time"] = end_time
-        return f"{time_column} <= :end_time", existing_placeholders
+        return f"T.{time_column} <= :end_time", existing_placeholders
     else:
         raise ValueError
 
@@ -101,20 +114,24 @@ def create_sql_query(
     existing_placeholders: dict[str, str],
     where_clauses: list[str] | None = None,
     order_by_col: str | None = None,
+    has_experiment: bool = False,
 ) -> tuple[str, dict[str, str]]:
     """
     Constructs an SQL query with SELECT, FROM, WHERE, and ORDER BY clauses.
     """
     # Base SELECT and FROM clause
-    query = f"SELECT {', '.join(selects)} FROM ({table_or_subquery})"
+    query = f"SELECT {', '.join(selects)} FROM ({table_or_subquery}) T"
+
+    if has_experiment:
+        query += " JOIN experiments E ON E.experiment = T.experiment"
 
     # Add WHERE clause if provided
     if where_clauses:
         query += f" WHERE {' AND '.join(where_clauses)}"
 
     # Add ORDER BY clause if provided
     if order_by_col:
-        query += f' ORDER BY "{order_by_col}"'
+        query += f' ORDER BY "T.{order_by_col}"'
 
     return query, existing_placeholders
 
@@ -143,7 +160,7 @@ def export_experiment_data(
         click.echo("At least one dataset name must be provided.")
         sys.exit(1)
 
-    logger = create_logger("export_experiment_data")
+    logger = create_logger("export_experiment_data", experiment="$experiment")
     logger.info(
         f"Starting export of dataset{'s' if len(dataset_names) > 1 else ''}: {', '.join(dataset_names)}."
     )
@@ -159,8 +176,6 @@ def export_experiment_data(
             "BASE64", 1, decode_base64
         )  # TODO: until next OS release which implements a native sqlite3 base64 function
 
-        con.set_trace_callback(logger.debug)
-
         cursor = con.cursor()
         cursor.executescript(
             """
@@ -171,6 +186,7 @@ def export_experiment_data(
             PRAGMA cache_size = -4000;
         """
         )
+        con.set_trace_callback(logger.debug)
 
         for dataset_name in dataset_names:
             try:
@@ -185,15 +201,15 @@ def export_experiment_data(
 
             _partition_by_unit = dataset.has_unit and (partition_by_unit or dataset.always_partition_by_unit)
             _partition_by_experiment = dataset.has_experiment and partition_by_experiment
-            filenames: list[str] = []
+            path_to_files: list[Path] = []
             placeholders: dict[str, str] = {}
 
             order_by_col = dataset.default_order_by
             table_or_subquery = dataset.table or dataset.query
             assert table_or_subquery is not None
 
             where_clauses: list[str] = []
-            selects = ["*"]
+            selects = ["T.*"]
 
             if dataset.timestamp_columns:
                 selects.append(generate_timestamp_to_localtimestamp_clause(dataset.timestamp_columns))
@@ -202,6 +218,9 @@ def export_experiment_data(
                 experiment_clause, placeholders = create_experiment_clause(experiments, placeholders)
                 where_clauses.append(experiment_clause)
 
+            if dataset.has_experiment and dataset.default_order_by:
+                selects.append(generate_timestamp_to_relative_time_clause(dataset.default_order_by))
+
             if dataset.timestamp_columns and (start_time or end_time):
                 assert dataset.default_order_by is not None
                 timespan_clause, placeholders = create_timespan_clause(
@@ -210,7 +229,7 @@ def export_experiment_data(
                 where_clauses.append(timespan_clause)
 
             query, placeholders = create_sql_query(
-                selects, table_or_subquery, placeholders, where_clauses, order_by_col
+                selects, table_or_subquery, placeholders, where_clauses, order_by_col, dataset.has_experiment
             )
             cursor.execute(query, placeholders)
 
@@ -243,15 +262,17 @@ def export_experiment_data(
                     )
 
                     if rows_partition not in parition_to_writer_map:
+                        # create a new csv writer for this partition since it doesn't exist yet
                         filename = f"{dataset_name}-" + "-".join(rows_partition) + f"-{time}.csv"
                         filename = filename.replace(" ", "_")
-                        filenames.append(filename)
-                        path_to_file = Path(Path(output).parent / filename)
+                        path_to_file = Path(output).parent / filename
                         parition_to_writer_map[rows_partition] = csv.writer(
                             stack.enter_context(open(path_to_file, "w")), delimiter=","
                         )
                         parition_to_writer_map[rows_partition].writerow(headers)
 
+                        path_to_files.append(path_to_file)
+
                     parition_to_writer_map[rows_partition].writerow(row)
 
                     if count % 10_000 == 0:
@@ -261,10 +282,10 @@ def export_experiment_data(
             if count == 0:
                 logger.warning(f"No data present in {dataset_name}. Check database?")
 
-            for filename in filenames:
-                path_to_file = Path(Path(output).parent / filename)
-                zf.write(path_to_file, arcname=filename)
-                Path(path_to_file).unlink()
+            zf.mkdir(dataset_name)
+            for path_to_file in path_to_files:
+                zf.write(path_to_file, arcname=f"{dataset_name}/{path_to_file.name}")
+                path_to_file.unlink()
 
     logger.info("Finished export.")
     return