Various fixes for instrument server problems (#409)

Addresses a few issues that have come up when running the web UI and instrument server.

* There is no longer a client so all references to ClientEnvironment are changed to Session
* Register processing job referred to a murfey_session_id which didn't exist
* em-spa-refine processing jobs were not made by the instrument server
* Copied the prometheus transferred file count function into the multigrid watcher as it was only in the client

Author: stephen-riggs

src/murfey/client/multigrid_control.py

@@ -223,6 +223,13 @@ def rsync_result(update: RSyncerUpdate):
                     self.rsync_processes[update.base_path].enqueue(update.file_path)
 
             self.rsync_processes[source].subscribe(rsync_result)
+            self.rsync_processes[source].subscribe(
+                partial(
+                    self._increment_transferred_files_prometheus,
+                    destination=destination,
+                    source=str(source),
+                )
+            )
             self.rsync_processes[source].subscribe(
                 partial(
                     self._increment_transferred_files,
@@ -399,6 +406,7 @@ def _start_dc(self, json, from_form: bool = False):
                     "em-spa-extract",
                     "em-spa-class2d",
                     "em-spa-class3d",
+                    "em-spa-refine",
                 ):
                     capture_post(
                         f"{str(self._environment.url.geturl())}/visits/{str(self._environment.visit)}/{self.session_id}/register_processing_job",
@@ -465,6 +473,33 @@ def _increment_file_count(
         }
         requests.post(url, json=data)
 
+    # Prometheus can handle higher traffic so update for every transferred file rather
+    # than batching as we do for the Murfey database updates in _increment_transferred_files
+    def _increment_transferred_files_prometheus(
+        self, update: RSyncerUpdate, source: str, destination: str
+    ):
+        if update.outcome is TransferResult.SUCCESS:
+            url = f"{str(self._environment.url.geturl())}/visits/{str(self._environment.visit)}/increment_rsync_transferred_files_prometheus"
+            data_files = (
+                [update]
+                if update.file_path.suffix in self._data_suffixes
+                and any(
+                    substring in update.file_path.name
+                    for substring in self._data_substrings
+                )
+                else []
+            )
+            data = {
+                "source": source,
+                "destination": destination,
+                "session_id": self.session_id,
+                "increment_count": 1,
+                "bytes": update.file_size,
+                "increment_data_count": len(data_files),
+                "data_bytes": sum(f.file_size for f in data_files),
+            }
+            requests.post(url, json=data)
+
     def _increment_transferred_files(
         self, updates: List[RSyncerUpdate], source: str, destination: str
     ):

src/murfey/server/__init__.py

@@ -171,15 +171,15 @@ def get_job_ids(tilt_series_id: int, appid: int) -> JobIDs:
             db.ProcessingJob,
             db.DataCollection,
             db.DataCollectionGroup,
-            db.ClientEnvironment,
+            db.Session,
         )
         .where(db.TiltSeries.id == tilt_series_id)
         .where(db.DataCollection.tag == db.TiltSeries.tag)
         .where(db.ProcessingJob.id == db.AutoProcProgram.pj_id)
         .where(db.AutoProcProgram.id == appid)
         .where(db.ProcessingJob.dc_id == db.DataCollection.id)
         .where(db.DataCollectionGroup.id == db.DataCollection.dcg_id)
-        .where(db.ClientEnvironment.session_id == db.TiltSeries.session_id)
+        .where(db.Session.id == db.TiltSeries.session_id)
     ).all()
     return JobIDs(
         dcgid=results[0][4].id,
@@ -1745,11 +1745,7 @@ def _register_3d_batch(message: dict, _db=murfey_db, demo: bool = False):
     other_options = dict(feedback_params)
 
     visit_name = (
-        _db.exec(
-            select(db.ClientEnvironment).where(
-                db.ClientEnvironment.session_id == message["session_id"]
-            )
-        )
+        _db.exec(select(db.Session).where(db.Session.id == message["session_id"]))
         .one()
         .visit
     )
@@ -2739,6 +2735,7 @@ def feedback_callback(header: dict, message: dict) -> None:
                 _transport_object.transport.ack(header)
             return None
         elif message["register"] == "processing_job":
+            murfey_session_id = message["session_id"]
             logger.info("registering processing job")
             assert isinstance(global_state["data_collection_ids"], dict)
             dc = murfey_db.exec(