Add logic for determining when a notification should be raised

d-j-hatton · d-j-hatton · commit cb6182195fcf · 2025-02-10T17:29:54.000Z
There are some hard coded parameters in here at the moment
diff --git a/pyproject.toml b/pyproject.toml
@@ -108,6 +108,7 @@ murfey = "murfey.client:run"
 "clem.register_lif_preprocessing_result" = "murfey.workflows.clem.register_preprocessing_results:register_lif_preprocessing_result"
 "clem.register_tiff_preprocessing_result" = "murfey.workflows.clem.register_preprocessing_results:register_tiff_preprocessing_result"
 "pato" = "murfey.workflows.notifications:notification_setup"
+"picked_particles" = "murfey.wrokflows.spa.picking:particles_picked"
 "spa.flush_spa_preprocess" = "murfey.workflows.spa.flush_spa_preprocess:flush_spa_preprocess"
 
 [tool.setuptools]
diff --git a/src/murfey/server/__init__.py b/src/murfey/server/__init__.py
@@ -573,298 +573,6 @@ def _get_spa_params(
     return relion_params, feedback_params
 
 
-def _register_picked_particles_use_diameter(
-    message: dict, _db=murfey_db, demo: bool = False
-):
-    """Received picked particles from the autopick service"""
-    # Add this message to the table of seen messages
-    params_to_forward = message.get("extraction_parameters")
-    assert isinstance(params_to_forward, dict)
-    pj_id = _pj_id(message["program_id"], _db)
-    ctf_params = db.CtfParameters(
-        pj_id=pj_id,
-        micrographs_file=params_to_forward["micrographs_file"],
-        extract_file=params_to_forward["extract_file"],
-        coord_list_file=params_to_forward["coord_list_file"],
-        ctf_image=params_to_forward["ctf_values"]["CtfImage"],
-        ctf_max_resolution=params_to_forward["ctf_values"]["CtfMaxResolution"],
-        ctf_figure_of_merit=params_to_forward["ctf_values"]["CtfFigureOfMerit"],
-        defocus_u=params_to_forward["ctf_values"]["DefocusU"],
-        defocus_v=params_to_forward["ctf_values"]["DefocusV"],
-        defocus_angle=params_to_forward["ctf_values"]["DefocusAngle"],
-    )
-    _db.add(ctf_params)
-    _db.commit()
-    _db.close()
-
-    picking_db_len = _db.exec(
-        select(func.count(db.ParticleSizes.id)).where(db.ParticleSizes.pj_id == pj_id)
-    ).one()
-    if picking_db_len > default_spa_parameters.nr_picks_before_diameter:
-        # If there are enough particles to get a diameter
-        instrument_name = (
-            _db.exec(select(db.Session).where(db.Session.id == message["session_id"]))
-            .one()
-            .instrument_name
-        )
-        machine_config = get_machine_config(instrument_name=instrument_name)[
-            instrument_name
-        ]
-        relion_params = _db.exec(
-            select(db.SPARelionParameters).where(db.SPARelionParameters.pj_id == pj_id)
-        ).one()
-        relion_options = dict(relion_params)
-        feedback_params = _db.exec(
-            select(db.SPAFeedbackParameters).where(
-                db.SPAFeedbackParameters.pj_id == pj_id
-            )
-        ).one()
-
-        particle_diameter = relion_params.particle_diameter
-
-        if feedback_params.picker_ispyb_id is None:
-            if demo or not _transport_object:
-                feedback_params.picker_ispyb_id = 1000
-            else:
-                assert feedback_params.picker_murfey_id is not None
-                feedback_params.picker_ispyb_id = _transport_object.do_buffer_lookup(
-                    message["program_id"], feedback_params.picker_murfey_id
-                )
-                if feedback_params.picker_ispyb_id is not None:
-                    _flush_class2d(message["session_id"], message["program_id"], _db)
-            _db.add(feedback_params)
-            _db.commit()
-            selection_stash = _db.exec(
-                select(db.SelectionStash).where(db.SelectionStash.pj_id == pj_id)
-            ).all()
-            for s in selection_stash:
-                _register_class_selection(
-                    {
-                        "session_id": s.session_id,
-                        "class_selection_score": s.class_selection_score or 0,
-                    },
-                    _db=_db,
-                    demo=demo,
-                )
-                _db.delete(s)
-                _db.commit()
-
-        if not particle_diameter:
-            # If the diameter has not been calculated then find it
-            picking_db = _db.exec(
-                select(db.ParticleSizes.particle_size).where(
-                    db.ParticleSizes.pj_id == pj_id
-                )
-            ).all()
-            particle_diameter = np.quantile(list(picking_db), 0.75)
-            relion_params.particle_diameter = particle_diameter
-            _db.add(relion_params)
-            _db.commit()
-
-            ctf_db = _db.exec(
-                select(db.CtfParameters).where(db.CtfParameters.pj_id == pj_id)
-            ).all()
-            for saved_message in ctf_db:
-                # Send on all saved messages to extraction
-                _db.expunge(saved_message)
-                zocalo_message: dict = {
-                    "parameters": {
-                        "micrographs_file": saved_message.micrographs_file,
-                        "coord_list_file": saved_message.coord_list_file,
-                        "output_file": saved_message.extract_file,
-                        "pixel_size": (
-                            relion_options["angpix"]
-                            * relion_options["motion_corr_binning"]
-                        ),
-                        "ctf_image": saved_message.ctf_image,
-                        "ctf_max_resolution": saved_message.ctf_max_resolution,
-                        "ctf_figure_of_merit": saved_message.ctf_figure_of_merit,
-                        "defocus_u": saved_message.defocus_u,
-                        "defocus_v": saved_message.defocus_v,
-                        "defocus_angle": saved_message.defocus_angle,
-                        "particle_diameter": particle_diameter,
-                        "downscale": relion_options["downscale"],
-                        "kv": relion_options["voltage"],
-                        "node_creator_queue": machine_config.node_creator_queue,
-                        "session_id": message["session_id"],
-                        "autoproc_program_id": _app_id(
-                            _pj_id(message["program_id"], _db, recipe="em-spa-extract"),
-                            _db,
-                        ),
-                        "batch_size": default_spa_parameters.batch_size_2d,
-                    },
-                    "recipes": ["em-spa-extract"],
-                }
-                if _transport_object:
-                    zocalo_message["parameters"][
-                        "feedback_queue"
-                    ] = _transport_object.feedback_queue
-                    _transport_object.send(
-                        "processing_recipe", zocalo_message, new_connection=True
-                    )
-        else:
-            # If the diameter is known then just send the new message
-            particle_diameter = relion_params.particle_diameter
-            zocalo_message = {
-                "parameters": {
-                    "micrographs_file": params_to_forward["micrographs_file"],
-                    "coord_list_file": params_to_forward["coord_list_file"],
-                    "output_file": params_to_forward["extract_file"],
-                    "pixel_size": (
-                        relion_options["angpix"] * relion_options["motion_corr_binning"]
-                    ),
-                    "ctf_image": params_to_forward["ctf_values"]["CtfImage"],
-                    "ctf_max_resolution": params_to_forward["ctf_values"][
-                        "CtfMaxResolution"
-                    ],
-                    "ctf_figure_of_merit": params_to_forward["ctf_values"][
-                        "CtfFigureOfMerit"
-                    ],
-                    "defocus_u": params_to_forward["ctf_values"]["DefocusU"],
-                    "defocus_v": params_to_forward["ctf_values"]["DefocusV"],
-                    "defocus_angle": params_to_forward["ctf_values"]["DefocusAngle"],
-                    "particle_diameter": particle_diameter,
-                    "downscale": relion_options["downscale"],
-                    "kv": relion_options["voltage"],
-                    "node_creator_queue": machine_config.node_creator_queue,
-                    "session_id": message["session_id"],
-                    "autoproc_program_id": _app_id(
-                        _pj_id(message["program_id"], _db, recipe="em-spa-extract"), _db
-                    ),
-                    "batch_size": default_spa_parameters.batch_size_2d,
-                },
-                "recipes": ["em-spa-extract"],
-            }
-            if _transport_object:
-                zocalo_message["parameters"][
-                    "feedback_queue"
-                ] = _transport_object.feedback_queue
-                _transport_object.send(
-                    "processing_recipe", zocalo_message, new_connection=True
-                )
-            if demo:
-                _register_incomplete_2d_batch(
-                    {
-                        "session_id": message["session_id"],
-                        "program_id": message["program_id"],
-                        "class2d_message": {
-                            "particles_file": "Select/job009/particles_split_1.star",
-                            "class2d_dir": "Class2D",
-                            "batch_size": 50000,
-                        },
-                    },
-                    _db=_db,
-                    demo=demo,
-                )
-
-    else:
-        # If not enough particles then save the new sizes
-        particle_list = message.get("particle_diameters")
-        assert isinstance(particle_list, list)
-        for particle in particle_list:
-            new_particle = db.ParticleSizes(pj_id=pj_id, particle_size=particle)
-            _db.add(new_particle)
-            _db.commit()
-    _db.close()
-
-
-def _register_picked_particles_use_boxsize(message: dict, _db=murfey_db):
-    """Received picked particles from the autopick service"""
-    # Add this message to the table of seen messages
-    params_to_forward = message.get("extraction_parameters")
-    assert isinstance(params_to_forward, dict)
-
-    instrument_name = (
-        _db.exec(select(db.Session).where(db.Session.id == message["session_id"]))
-        .one()
-        .instrument_name
-    )
-    machine_config = get_machine_config(instrument_name=instrument_name)[
-        instrument_name
-    ]
-    pj_id = _pj_id(message["program_id"], _db)
-    ctf_params = db.CtfParameters(
-        pj_id=pj_id,
-        micrographs_file=params_to_forward["micrographs_file"],
-        coord_list_file=params_to_forward["coord_list_file"],
-        extract_file=params_to_forward["extract_file"],
-        ctf_image=params_to_forward["ctf_values"]["CtfImage"],
-        ctf_max_resolution=params_to_forward["ctf_values"]["CtfMaxResolution"],
-        ctf_figure_of_merit=params_to_forward["ctf_values"]["CtfFigureOfMerit"],
-        defocus_u=params_to_forward["ctf_values"]["DefocusU"],
-        defocus_v=params_to_forward["ctf_values"]["DefocusV"],
-        defocus_angle=params_to_forward["ctf_values"]["DefocusAngle"],
-    )
-    _db.add(ctf_params)
-    _db.commit()
-    _db.close()
-
-    # Set particle diameter as zero and send box sizes
-    relion_params = _db.exec(
-        select(db.SPARelionParameters).where(db.SPARelionParameters.pj_id == pj_id)
-    ).one()
-    feedback_params = _db.exec(
-        select(db.SPAFeedbackParameters).where(db.SPAFeedbackParameters.pj_id == pj_id)
-    ).one()
-
-    if feedback_params.picker_ispyb_id is None and _transport_object:
-        assert feedback_params.picker_murfey_id is not None
-        feedback_params.picker_ispyb_id = _transport_object.do_buffer_lookup(
-            message["program_id"], feedback_params.picker_murfey_id
-        )
-        if feedback_params.picker_ispyb_id is not None:
-            _flush_class2d(message["session_id"], message["program_id"], _db)
-        _db.add(feedback_params)
-        _db.commit()
-        selection_stash = _db.exec(
-            select(db.SelectionStash).where(db.SelectionStash.pj_id == pj_id)
-        ).all()
-        for s in selection_stash:
-            _register_class_selection(
-                {
-                    "session_id": s.session_id,
-                    "class_selection_score": s.class_selection_score or 0,
-                },
-                _db=_db,
-            )
-            _db.delete(s)
-            _db.commit()
-
-    # Send the message to extraction with the box sizes
-    zocalo_message: dict = {
-        "parameters": {
-            "micrographs_file": params_to_forward["micrographs_file"],
-            "coord_list_file": params_to_forward["coord_list_file"],
-            "output_file": params_to_forward["extract_file"],
-            "pixel_size": relion_params.angpix * relion_params.motion_corr_binning,
-            "ctf_image": params_to_forward["ctf_values"]["CtfImage"],
-            "ctf_max_resolution": params_to_forward["ctf_values"]["CtfMaxResolution"],
-            "ctf_figure_of_merit": params_to_forward["ctf_values"]["CtfFigureOfMerit"],
-            "defocus_u": params_to_forward["ctf_values"]["DefocusU"],
-            "defocus_v": params_to_forward["ctf_values"]["DefocusV"],
-            "defocus_angle": params_to_forward["ctf_values"]["DefocusAngle"],
-            "particle_diameter": relion_params.particle_diameter,
-            "boxsize": relion_params.boxsize,
-            "small_boxsize": relion_params.small_boxsize,
-            "downscale": relion_params.downscale,
-            "kv": relion_params.voltage,
-            "node_creator_queue": machine_config.node_creator_queue,
-            "session_id": message["session_id"],
-            "autoproc_program_id": _app_id(
-                _pj_id(message["program_id"], _db, recipe="em-spa-extract"), _db
-            ),
-            "batch_size": default_spa_parameters.batch_size_2d,
-        },
-        "recipes": ["em-spa-extract"],
-    }
-    if _transport_object:
-        zocalo_message["parameters"][
-            "feedback_queue"
-        ] = _transport_object.feedback_queue
-        _transport_object.send("processing_recipe", zocalo_message, new_connection=True)
-    _db.close()
-
-
 def _release_2d_hold(message: dict, _db=murfey_db):
     relion_params, feedback_params = _get_spa_params(message["program_id"], _db)
     if not feedback_params.star_combination_job:
@@ -2841,31 +2549,6 @@ def feedback_callback(header: dict, message: dict) -> None:
             if _transport_object:
                 _transport_object.transport.ack(header)
             return None
-        elif message["register"] == "picked_particles":
-            movie = murfey_db.exec(
-                select(db.Movie).where(
-                    db.Movie.murfey_id == message["motion_correction_id"]
-                )
-            ).one()
-            movie.preprocessed = True
-            murfey_db.add(movie)
-            murfey_db.commit()
-            feedback_params = murfey_db.exec(
-                select(db.SPAFeedbackParameters).where(
-                    db.SPAFeedbackParameters.pj_id
-                    == _pj_id(message["program_id"], murfey_db)
-                )
-            ).one()
-            if feedback_params.estimate_particle_diameter:
-                _register_picked_particles_use_diameter(message)
-            else:
-                _register_picked_particles_use_boxsize(message)
-            prom.preprocessed_movies.labels(
-                processing_job=_pj_id(message["program_id"], murfey_db)
-            ).inc()
-            if _transport_object:
-                _transport_object.transport.ack(header)
-            return None
         elif message["register"] == "done_incomplete_2d_batch":
             _release_2d_hold(message)
             if _transport_object:
diff --git a/src/murfey/util/db.py b/src/murfey/util/db.py
@@ -395,6 +395,7 @@ class NotificationValue(SQLModel, table=True):  # type: ignore
         foreign_key="notificationparameter.id", primary_key=True
     )
     index: int
+    within_bounds: bool
     notification_parameter: Optional[DataCollectionGroup] = Relationship(
         back_populates="notification_values"
     )
diff --git a/src/murfey/workflows/notifications/__init__.py b/src/murfey/workflows/notifications/__init__.py
@@ -1,3 +1,4 @@
+import re
 from typing import Dict, Tuple
 
 from sqlmodel import Session, select
@@ -15,9 +16,10 @@ def notification_setup(
             parameter_name = k[:-3]
         else:
             continue
-        if parameter_name in parameters.keys():
+        snake_parameter_name = re.sub(r"(?<!^)(?=[A-Z])", "_", parameter_name).lower()
+        if snake_parameter_name in parameters.keys():
             continue
-        parameters[parameter_name] = (
+        parameters[snake_parameter_name] = (
             message.get(f"{parameter_name}Min", 0),
             message.get(f"{parameter_name}Max", 10000),
         )
diff --git a/src/murfey/workflows/spa/picking.py b/src/murfey/workflows/spa/picking.py

Original file line number	Diff line number	Diff line change
`@@ -395,6 +395,7 @@ class NotificationValue(SQLModel, table=True): # type: ignore`
`395`	`395`	`foreign_key="notificationparameter.id", primary_key=True`
`396`	`396`	`)`
`397`	`397`	`index: int`
	`398`	`+ within_bounds: bool`
`398`	`399`	`notification_parameter: Optional[DataCollectionGroup] = Relationship(`
`399`	`400`	`back_populates="notification_values"`
`400`	`401`	`)`