Remove the logic for skipping existing processing

src/murfey/client/multigrid_control.py

@@ -40,7 +40,6 @@ class MultigridController:
     finalising: bool = False
     dormant: bool = False
     multigrid_watcher_active: bool = True
-    processing_enabled: bool = True
     do_transfer: bool = True
     dummy_dc: bool = False
     force_mdoc_metadata: bool = True
@@ -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 {}
 
@@ -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,
@@ -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 = "",
@@ -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(partial(self._start_dc, from_form=True))
             self.analysers[source].start()
             if transfer:
                 self.rsync_processes[source].subscribe(self.analysers[source].enqueue)
@@ -502,21 +492,6 @@ 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):
         if self.dummy_dc:
             return

src/murfey/client/watchdir_multigrid.py

@@ -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)
@@ -30,7 +29,6 @@ def __init__(
         )
         # Toggleable settings
         self._analyse = True
-        self._skip_existing_processing = skip_existing_processing
         self._stopping = False
 
     def start(self):
@@ -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)
@@ -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"]:
@@ -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)

src/murfey/instrument_server/api.py

@@ -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, {}),
@@ -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(

src/murfey/server/api/instrument.py

@@ -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()
                     },

src/murfey/util/instrument_models.py

@@ -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

src/murfey/util/models.py

@@ -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] = []