Skip to content

Remove the logic for skipping existing processing #735

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
stephen-riggs wants to merge 3 commits into
base: main
Choose a base branch
from
+72 −138
Open

Remove the logic for skipping existing processing #735

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
e3fc2aa
Remove the old tui force metadata code
stephen-riggs Jan 27, 2026
3b35bcd
No more skip existing processing and first loop logic
stephen-riggs Jan 27, 2026
809e5a3
Remove redundant form logic
stephen-riggs Feb 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 2 additions & 10 deletions src/murfey/client/analyser.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -308,11 +308,7 @@ def _analyse(self):
dc_metadata["acquisition_software"] = (
self._context._acquisition_software
)
self.notify(
{
"form": dc_metadata,
}
)
self.notify(dc_metadata)

# If a file with a CLEM context is identified, immediately post it
elif isinstance(self._context, CLEMContext):
Expand Down Expand Up @@ -366,11 +362,7 @@ def _analyse(self):
dc_metadata["acquisition_software"] = (
self._context._acquisition_software
)
self.notify(
{
"form": dc_metadata,
}
)
self.notify(dc_metadata)
elif isinstance(
self._context,
(
Expand Down
167 changes: 66 additions & 101 deletions src/murfey/client/multigrid_control.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,6 @@ class MultigridController:
finalising: bool = False
dormant: bool = False
multigrid_watcher_active: bool = True
processing_enabled: bool = True
Copy link
Contributor

@tieneupin tieneupin Feb 2, 2026 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just to confirm, by removing processing_enabled, are we moving towards relying entirely on the API calls from the backend server to determine whether the Analyser is activated for a given session?

Copy link
Contributor Author

@stephen-riggs stephen-riggs Feb 2, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The processing_enabled flag didn't really do what it says, the analyse one if the thing which determines whether the Analyser is activated

Copy link
Contributor

@tieneupin tieneupin Feb 2, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Got you. So, the MultigridController will have no attribute related to whether the Analyser should be triggered, but these will be directly set when setting up the RSyncers via the messages sent from the backend?

do_transfer: bool = True
dummy_dc: bool = False
force_mdoc_metadata: bool = True
Expand Down Expand Up @@ -86,8 +85,6 @@ def __post_init__(self):
for ds in val.values()
for s in ds
]
self._data_collection_form_complete = False
self._register_dc: bool | None = None
self.rsync_processes = self.rsync_processes or {}
self.analysers = self.analysers or {}

Expand Down Expand Up @@ -260,7 +257,6 @@ def _start_rsyncer_multigrid(
self._start_rsyncer(
source,
destination,
force_metadata=self.processing_enabled,
analyse=analyse,
remove_files=remove_files,
tag=tag,
Expand Down Expand Up @@ -324,7 +320,6 @@ def _start_rsyncer(
source: Path,
destination: str,
visit_path: str = "",
force_metadata: bool = False,
analyse: bool = True,
remove_files: bool = False,
tag: str = "",
Expand Down Expand Up @@ -455,12 +450,7 @@ def rsync_result(update: RSyncerUpdate):
force_mdoc_metadata=self.force_mdoc_metadata,
limited=limited,
)
if force_metadata:
self.analysers[source].subscribe(
partial(self._start_dc, from_form=True)
)
else:
self.analysers[source].subscribe(self._data_collection_form)
self.analysers[source].subscribe(self._start_dc)
self.analysers[source].start()
if transfer:
self.rsync_processes[source].subscribe(self.analysers[source].enqueue)
Expand Down Expand Up @@ -502,32 +492,13 @@ def _rsync_update_converter(p: Path) -> None:
)
self._environment.watchers[source].start()

def _data_collection_form(self, response: dict):
log.info("data collection form ready")
if self._data_collection_form_complete:
return
if self._register_dc and response.get("form"):
self._form_values = {k: str(v) for k, v in response.get("form", {}).items()}
log.info(
f"gain reference is set to {self._form_values.get('gain_ref')}, {self._environment.gain_ref}"
)
if self._form_values.get("gain_ref") in (None, "None"):
self._form_values["gain_ref"] = self._environment.gain_ref
self._data_collection_form_complete = True
elif self._register_dc is None:
self._data_collection_form_complete = True

def _start_dc(self, metadata_json, from_form: bool = False):
def _start_dc(self, metadata_json):
if self.dummy_dc:
return
# for multigrid the analyser sends the message straight to _start_dc by-passing user input
# it is then necessary to extract the data from the message
if from_form:
metadata_json = metadata_json.get("form", {})
# Safely convert all entries into strings, but leave None as-is
metadata_json = {
k: str(v) if v is not None else None for k, v in metadata_json.items()
}
# Safely convert all entries into strings, but leave None as-is
metadata_json = {
k: str(v) if v is not None else None for k, v in metadata_json.items()
}
self._environment.dose_per_frame = metadata_json.get("dose_per_frame")
self._environment.gain_ref = metadata_json.get("gain_ref")
self._environment.symmetry = metadata_json.get("symmetry")
Expand Down Expand Up @@ -601,82 +572,76 @@ def _start_dc(self, metadata_json, from_form: bool = False):
environment=self._environment,
token=self.token,
)
if from_form:
data = {
"voltage": metadata_json["voltage"],
"pixel_size_on_image": metadata_json["pixel_size_on_image"],
"experiment_type": metadata_json["experiment_type"],
"image_size_x": metadata_json["image_size_x"],
"image_size_y": metadata_json["image_size_y"],
"file_extension": metadata_json["file_extension"],
"acquisition_software": metadata_json["acquisition_software"],
"image_directory": str(
self._environment.default_destinations[source]
),
"tag": str(source),
"source": str(source),
"magnification": metadata_json["magnification"],
"total_exposed_dose": metadata_json.get("total_exposed_dose"),
"c2aperture": metadata_json.get("c2aperture"),
"exposure_time": metadata_json.get("exposure_time"),
"slit_width": metadata_json.get("slit_width"),
"phase_plate": metadata_json.get("phase_plate", False),
}
data = {
"voltage": metadata_json["voltage"],
"pixel_size_on_image": metadata_json["pixel_size_on_image"],
"experiment_type": metadata_json["experiment_type"],
"image_size_x": metadata_json["image_size_x"],
"image_size_y": metadata_json["image_size_y"],
"file_extension": metadata_json["file_extension"],
"acquisition_software": metadata_json["acquisition_software"],
"image_directory": str(self._environment.default_destinations[source]),
"tag": str(source),
"source": str(source),
"magnification": metadata_json["magnification"],
"total_exposed_dose": metadata_json.get("total_exposed_dose"),
"c2aperture": metadata_json.get("c2aperture"),
"exposure_time": metadata_json.get("exposure_time"),
"slit_width": metadata_json.get("slit_width"),
"phase_plate": metadata_json.get("phase_plate", False),
}
capture_post(
base_url=str(self._environment.url.geturl()),
router_name="workflow.router",
function_name="start_dc",
token=self.token,
visit_name=self._environment.visit,
session_id=self.session_id,
data=data,
)
for recipe in (
"em-spa-preprocess",
"em-spa-extract",
"em-spa-class2d",
"em-spa-class3d",
"em-spa-refine",
):
capture_post(
base_url=str(self._environment.url.geturl()),
router_name="workflow.router",
function_name="start_dc",
function_name="register_proc",
token=self.token,
visit_name=self._environment.visit,
session_id=self.session_id,
data=data,
)
for recipe in (
"em-spa-preprocess",
"em-spa-extract",
"em-spa-class2d",
"em-spa-class3d",
"em-spa-refine",
):
capture_post(
base_url=str(self._environment.url.geturl()),
router_name="workflow.router",
function_name="register_proc",
token=self.token,
visit_name=self._environment.visit,
session_id=self.session_id,
data={
"tag": str(source),
"source": str(source),
"recipe": recipe,
},
)
log.info(f"Posting SPA processing parameters: {metadata_json}")
response = capture_post(
base_url=str(self._environment.url.geturl()),
router_name="workflow.spa_router",
function_name="register_spa_proc_params",
token=self.token,
session_id=self.session_id,
data={
**{
k: None if v == "None" else v
for k, v in metadata_json.items()
},
"tag": str(source),
"source": str(source),
"recipe": recipe,
},
)
if response and not str(response.status_code).startswith("2"):
log.warning(f"{response.reason}")
capture_post(
base_url=str(self._environment.url.geturl()),
router_name="workflow.spa_router",
function_name="flush_spa_processing",
token=self.token,
visit_name=self._environment.visit,
session_id=self.session_id,
data={"tag": str(source)},
)
log.info(f"Posting SPA processing parameters: {metadata_json}")
response = capture_post(
base_url=str(self._environment.url.geturl()),
router_name="workflow.spa_router",
function_name="register_spa_proc_params",
token=self.token,
session_id=self.session_id,
data={
**{k: None if v == "None" else v for k, v in metadata_json.items()},
"tag": str(source),
},
)
if response and not str(response.status_code).startswith("2"):
log.warning(f"{response.reason}")
capture_post(
base_url=str(self._environment.url.geturl()),
router_name="workflow.spa_router",
function_name="flush_spa_processing",
token=self.token,
visit_name=self._environment.visit,
session_id=self.session_id,
data={"tag": str(source)},
)

def _increment_file_count(
self, observed_files: List[Path], source: str, destination: str
Expand Down
26 changes: 4 additions & 22 deletions src/murfey/client/watchdir_multigrid.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,6 @@ def __init__(
self,
path: str | os.PathLike,
machine_config: dict,
skip_existing_processing: bool = False,
):
super().__init__()
self._basepath = Path(path)
Expand All @@ -30,7 +29,6 @@ def __init__(
)
# Toggleable settings
self._analyse = True
self._skip_existing_processing = skip_existing_processing
self._stopping = False

def start(self):
Expand Down Expand Up @@ -61,21 +59,14 @@ def _handle_metadata(self, directory: Path, extra_directory: str):
)
self._seen_dirs.append(directory)

def _handle_fractions(self, directory: Path, first_loop: bool):
def _handle_fractions(self, directory: Path):
processing_started = False
for d02 in directory.glob("Images-Disc*"):
if d02 not in self._seen_dirs:
# If 'skip_existing_processing' is set, do not process for
# any data directories found on the first loop.
# This allows you to avoid triggering processing again if Murfey is restarted
self.notify(
d02,
remove_files=True,
analyse=(
not (first_loop and self._skip_existing_processing)
if self._analyse
else False
),
analyse=self._analyse,
tag="fractions",
)
self._seen_dirs.append(d02)
Expand All @@ -88,17 +79,12 @@ def _handle_fractions(self, directory: Path, first_loop: bool):
):
self.notify(
directory,
analyse=(
not (first_loop and self._skip_existing_processing)
if self._analyse
else False
),
analyse=self._analyse,
tag="fractions",
)
self._seen_dirs.append(directory)

def _process(self):
first_loop = True
while not self._stopping:
for d in self._basepath.glob("*"):
if d.name in self._machine_config["create_directories"]:
Expand Down Expand Up @@ -133,18 +119,14 @@ def _process(self):
self._handle_fractions(
sample.parent.parent.parent
/ f"{sample.parent.name}_{sample.name}",
first_loop,
)

else:
if d.is_dir() and d not in self._seen_dirs:
self._handle_metadata(
d, extra_directory=f"metadata_{d.name}"
)
self._handle_fractions(d.parent.parent / d.name, first_loop)

if first_loop:
first_loop = False
self._handle_fractions(d.parent.parent / d.name)
time.sleep(15)

self.notify(final=True)
2 changes: 0 additions & 2 deletions src/murfey/instrument_server/api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -174,7 +174,6 @@ def setup_multigrid_watcher(
session_id,
murfey_url=_get_murfey_url(),
do_transfer=True,
processing_enabled=not watcher_spec.skip_existing_processing,
_machine_config=machine_config,
token=tokens.get(session_id, "token"),
data_collection_parameters=data_collection_parameters.get(label, {}),
Expand All @@ -190,7 +189,6 @@ def setup_multigrid_watcher(
watchers[session_id] = MultigridDirWatcher(
watcher_spec.source,
machine_config,
skip_existing_processing=watcher_spec.skip_existing_processing,
)
watchers[session_id].subscribe(
partial(
Expand Down
1 change: 0 additions & 1 deletion src/murfey/server/api/instrument.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -154,7 +154,6 @@ async def setup_multigrid_watcher(
"visit": visit,
"label": visit,
"instrument_name": instrument_name,
"skip_existing_processing": watcher_spec.skip_existing_processing,
"destination_overrides": {
str(k): v for k, v in watcher_spec.destination_overrides.items()
},
Expand Down
1 change: 0 additions & 1 deletion src/murfey/util/instrument_models.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,6 @@ class MultigridWatcherSpec(BaseModel):
label: str
visit: str
instrument_name: str
skip_existing_processing: bool = False
destination_overrides: Dict[Path, str] = {}
rsync_restarts: List[str] = []
visit_end_time: Optional[datetime] = None
1 change: 0 additions & 1 deletion src/murfey/util/models.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -203,7 +203,6 @@ class BatchPositionParameters(BaseModel):

class MultigridWatcherSetup(BaseModel):
source: Path
skip_existing_processing: bool = False
destination_overrides: Dict[Path, str] = {}
rsync_restarts: List[str] = []

Expand Down