(Re)process runs from the GUI

damnit/backend/extract_data.py

@@ -41,7 +41,7 @@ def run_in_subprocess(args, **kwargs):
 
 def extract_in_subprocess(
         proposal, run, out_path, cluster=False, run_data=RunData.ALL, match=(),
-        python_exe=None, mock=False
+        variables=(), python_exe=None, mock=False
 ):
     if not python_exe:
         python_exe = sys.executable
@@ -52,8 +52,12 @@ def extract_in_subprocess(
         args.append('--cluster-job')
     if mock:
         args.append("--mock")
-    for m in match:
-        args.extend(['--match', m])
+    if variables:
+        for v in variables:
+            args.extend(['--var', v])
+    else:
+        for m in match:
+            args.extend(['--match', m])
 
     with TemporaryDirectory() as td:
         # Save a separate copy of the reduced data, so we can send an update
@@ -194,7 +198,7 @@ def update_db_vars(self):
         self.kafka_prd.flush()
 
     def extract_and_ingest(self, proposal, run, cluster=False,
-                           run_data=RunData.ALL, match=(), mock=False):
+                           run_data=RunData.ALL, match=(), variables=(), mock=False):
         if proposal is None:
             proposal = self.db.metameta['proposal']
 
@@ -206,7 +210,7 @@ def extract_and_ingest(self, proposal, run, cluster=False,
         python_exe = self.db.metameta.get('context_python', '')
         reduced_data = extract_in_subprocess(
             proposal, run, out_path, cluster=cluster, run_data=run_data,
-            match=match, python_exe=python_exe, mock=mock,
+            match=match, variables=variables, python_exe=python_exe, mock=mock,
         )
         log.info("Reduced data has %d fields", len(reduced_data))
         add_to_db(reduced_data, self.db, proposal, run)
@@ -248,6 +252,7 @@ def main(argv=None):
     # variables (confusing because all extraction now runs in cluster jobs)
     ap.add_argument('--cluster-job', action="store_true")
     ap.add_argument('--match', action="append", default=[])
+    ap.add_argument('--var',  action="append", default=[])
     ap.add_argument('--mock', action='store_true')
     ap.add_argument('--update-vars', action='store_true')
     args = ap.parse_args(argv)
@@ -272,6 +277,7 @@ def main(argv=None):
                             cluster=args.cluster_job,
                             run_data=RunData(args.run_data),
                             match=args.match,
+                            variables=args.var,
                             mock=args.mock)
 
 

damnit/backend/extraction_control.py

@@ -83,6 +83,7 @@ class ExtractionRequest:
     run_data: RunData
     cluster: bool = False
     match: tuple = ()
+    variables: tuple = ()   # Overrides match if present
     mock: bool = False
     update_vars: bool = True
 
@@ -94,8 +95,12 @@ def python_cmd(self):
         ]
         if self.cluster:
             cmd.append('--cluster-job')
-        for m in self.match:
-            cmd.extend(['--match', m])
+        if self.variables:
+            for v in self.variables:
+                cmd.extend(['--var', v])
+        else:
+            for m in self.match:
+                cmd.extend(['--match', m])
         if self.mock:
             cmd.append('--mock')
         if self.update_vars:

damnit/ctxsupport/ctxrunner.py

@@ -233,7 +233,7 @@ def vars_to_dict(self):
             for (name, v) in self.vars.items()
         }
 
-    def filter(self, run_data=RunData.ALL, cluster=True, name_matches=()):
+    def filter(self, run_data=RunData.ALL, cluster=True, name_matches=(), variables=()):
         new_vars = {}
         for name, var in self.vars.items():
 
@@ -247,9 +247,13 @@ def filter(self, run_data=RunData.ALL, cluster=True, name_matches=()):
             data_match = run_data == RunData.ALL or var.data == run_data
             # Skip data tagged cluster unless we're in a dedicated Slurm job
             cluster_match = cluster or not var.cluster
-            # Skip Variables that don't match the match list
-            name_match = (len(name_matches) == 0
-                          or any(m.lower() in title.lower() for m in name_matches))
+
+            if variables:  # --var: exact variable names (not titles)
+                name_match = name in variables
+            elif name_matches:  # --match: substring in variable titles
+                name_match = any(m.lower() in title.lower() for m in name_matches)
+            else:
+                name_match = True  # No --var or --match specification
 
             if data_match and cluster_match and name_match:
                 new_vars[name] = var
@@ -639,6 +643,7 @@ def main(argv=None):
     exec_ap.add_argument('--mock', action='store_true')
     exec_ap.add_argument('--cluster-job', action="store_true")
     exec_ap.add_argument('--match', action="append", default=[])
+    exec_ap.add_argument('--var', action="append", default=[])
     exec_ap.add_argument('--save', action='append', default=[])
     exec_ap.add_argument('--save-reduced', action='append', default=[])
 
@@ -673,7 +678,8 @@ def main(argv=None):
         ctx_whole = ContextFile.from_py_file(Path('context.py'))
         ctx_whole.check()
         ctx = ctx_whole.filter(
-            run_data=run_data, cluster=args.cluster_job, name_matches=args.match
+            run_data=run_data, cluster=args.cluster_job, name_matches=args.match,
+            variables=args.var,
         )
         log.info("Using %d variables (of %d) from context file %s",
              len(ctx.vars), len(ctx_whole.vars),

damnit/gui/main_window.py

@@ -25,8 +25,7 @@
 from ..api import DataType, RunVariables
 from ..backend import backend_is_running, initialize_and_start_backend
 from ..backend.db import BlobTypes, DamnitDB, MsgKind, ReducedData, db_path
-from ..backend.extract_data import get_context_file
-from ..backend.extraction_control import process_log_path
+from ..backend.extraction_control import process_log_path, ExtractionSubmitter
 from ..backend.user_variables import UserEditableVariable
 from ..definitions import UPDATE_BROKERS
 from ..util import StatusbarStylesheet, fix_data_for_plotting, icon_path
@@ -35,6 +34,7 @@
 from .open_dialog import OpenDBDialog
 from .new_context_dialog import NewContextFileDialog
 from .plot import Canvas, Plot
+from .process import ProcessingDialog
 from .table import DamnitTableModel, TableView, prettify_notation
 from .user_variables import AddUserVariableDialog
 from .web_viewer import PlotlyPlot, UrlSchemeHandler
@@ -340,6 +340,8 @@ def _create_menu_bar(self) -> None:
         self.action_export.setEnabled(False)
         self.context_dir_changed.connect(lambda _: self.action_export.setEnabled(True))
         self.action_export.triggered.connect(self.export_table)
+        self.action_process = QtWidgets.QAction("Reprocess runs", self)
+        self.action_process.triggered.connect(self.process_runs)
 
         action_adeqt = QtWidgets.QAction("Python console", self)
         action_adeqt.setShortcut("F12")
@@ -360,6 +362,7 @@ def _create_menu_bar(self) -> None:
         )
         fileMenu.addAction(action_autoconfigure)
         fileMenu.addAction(self.action_create_var)
+        fileMenu.addAction(self.action_process)
         fileMenu.addAction(self.action_export)
         fileMenu.addAction(action_adeqt)
         fileMenu.addAction(action_help)
@@ -668,6 +671,7 @@ def _create_view(self) -> None:
         self.table_view.doubleClicked.connect(self._inspect_data_proxy_idx)
         self.table_view.settings_changed.connect(self.save_settings)
         self.table_view.zulip_action.triggered.connect(self.export_selection_to_zulip)
+        self.table_view.process_action.triggered.connect(self.process_runs)
         self.table_view.log_view_requested.connect(self.show_run_logs)
 
         vertical_layout.addWidget(self.table_view)
@@ -895,6 +899,39 @@ def export_selection_to_zulip(self):
         df.replace(["None", '<NA>', 'nan'], '', inplace=True)
         self.zulip_messenger.send_table(df)
 
+    def process_runs(self):
+        sel_runs_by_prop = {}
+        for ix in self.table_view.selected_rows():
+            run_prop, run_num = self.table.row_to_proposal_run(ix.row())
+            sel_runs_by_prop.setdefault(run_prop, []).append(run_num)
+
+        if sel_runs_by_prop:
+            prop, sel_runs = max(sel_runs_by_prop.items(), key=lambda p: len(p[1]))
+            sel_runs.sort()
+        else:
+            prop = self.db.metameta.get("proposal", "")
+            sel_runs = []
+
+        var_ids_titles = zip(self.table.computed_columns(),
+                             self.table.computed_columns(by_title=True))
+
+        dlg = ProcessingDialog(str(prop), sel_runs, var_ids_titles, parent=self)
+        if dlg.exec() == QtWidgets.QDialog.Accepted:
+            submitter = ExtractionSubmitter(self.context_dir, self.db)
+
+            try:
+                reqs = dlg.extraction_requests()
+                for req in reqs:
+                    submitter.submit(req)
+            except Exception as e:
+                log.error("Error launching processing", exc_info=True)
+                self.show_status_message(f"Error launching processing: {e}",
+                                         10_000, stylesheet=StatusbarStylesheet.ERROR)
+            else:
+                self.show_status_message(
+                    f"Launched processing for {len(reqs)} runs", 10_000
+                )
+
     adeqt_window = None
 
     def show_adeqt(self):