diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md index d567848..c9427ba 100644 --- a/.github/copilot-instructions.md +++ b/.github/copilot-instructions.md @@ -52,13 +52,28 @@ histogram?* - The FLAF era is `era` (`--era`); `period` is refused by FLAF as a parameter, option or keyword and survives only as the AnaTuple column. -- `era` is FLAF's `EraParameter(max_len=1)`: a tuple holding one real era. It is read as - `self.era[0]` — `self.era` itself in a path, a label or a command is the tuple, not the name. - Requirements pass one real era as a tuple: `X.req(self, era=(e,), …)`. -- A meta-era (`era_groups:` in the card) never goes into `era`, because FLAF's `Setup` knows only - real eras: it goes into `meta_era`, and `era` holds one of its sub-eras as the `Setup` anchor. - `datacard_era` (`meta_era`, else `era[0]`) keys the outputs and store paths, and - `get_sub_eras()` lists the real eras whose histograms feed a card. +- Every task here takes several eras (`allow_multi_era = True` on `StatInferenceTask`, without + FLAF's era branches): `era` is the sorted tuple of `--era`, which must hold every real era + behind the configuration's cards (`card_eras()` raises naming the card and the missing + eras). An era's own settings are read through `self.setup_for(era)`; a setting the run reads + once through `self.common(getter, what)` or `statinference_settings()`, which raise when + the eras disagree. FLAF's `check_era_rename.py`, run in CI, rejects `self.era[...]` and an + `era=` keyword in `req()`; `self.global_params`, `self.setup`, `self.fs_*` + (except `fs_default`) and `self.datasets` are not checked statically and raise `EraScopeError` only + when run over several eras, which the unit tests do (three eras) — flag them in review. +- The tasks a run starts from that are not workflows (`ResonantLimitsTask`, the plot tasks, + `ResonantLimitsAndHistPlotTask`) check the eras against the cards when they are constructed + (`check_card_eras`), so the error stops `law run` before scheduling; the card workflows check + when they build their map, never in `__init__` (FLAF constructs them over single eras). +- `CreateDatacardsTask` and `PreprocessShapesTask` are one workflow each whose branches are the + cards (`eras:` of the configuration, in its order); `branch_data` is the card era, which may + be a group era (`era_groups:`), never in `era` because FLAF's `Setup` knows only real eras. + `get_sub_eras(card_era)` lists the real eras whose histograms feed a card; `MergedHists` + reads each with `hist_era` from that era's own `fs_HistTuple`. +- Outputs are keyed on the version and the card, the job state and the logs on the version and + the class (`store_parts()`, `remote_log_dir_target()`), never on `--era`, so a run over more + eras resumes the same jobs. One datacard configuration per version: the job file is named + after the branch numbers, i.e. the positions in `eras:`. ### Binning optimisation @@ -102,6 +117,6 @@ Verified 2026-08-27; re-check before relying on any of it. | Layout | `dc_make/` (datacard construction: `maker.py`, `create_datacards.py`, `process.py`, `model.py`, `uncertainty.py`, `binner.py`), `bin_opt/` (binning optimisation and limits), `common/` (`tools.py`, `param_parse.py`), `law/tasks.py`, `fit_val/`, `config/`, `legacy/`, `test/` (unit tests and their scratch analysis) | | Consumers | HH_bbtautau and HH_bbWW use this as a submodule; H_mumu does not | | Cards | analysis-side, e.g. `HH_bbWW/config/Datacards/CI_card.yaml` | -| Tests | `test/test_law_chain.py` (the law tasks on the real FLAF and law classes over a scratch analysis, `test/statinference_fixture.py`: per card the outputs, inputs and payload commands) and `test/test_shape_plots.py` (which histograms a card's shape plots show, through `input_categories`); no ROOT, Combine or dhi; run by `unit-tests` against a pinned FLAF commit | +| Tests | `test/test_law_chain.py` (the law tasks on the real FLAF and law classes over a scratch analysis, `test/statinference_fixture.py`: per card the outputs, inputs and payload commands), `test/test_card_eras.py` (the card workflows over several eras) and `test/test_shape_plots.py` (which histograms a card's shape plots show, through `input_categories`); no ROOT, Combine or dhi; run by `unit-tests` against a pinned FLAF commit | | Workflows | `formatting-check`, `unit-tests`, `repo-sanity-checks` (FLAF's sanity checks and FLAF's `run_tools/check_era_rename.py` for leftovers of the `period` → `era` rename), `trigger-flaf-integration`. Formatting is checked automatically — do not comment on it. A change to an assertion of `test/test_law_chain.py` (not to its helpers) changes what the chain is guaranteed to read, write or run — comment on it | | Integration test | Triggered by `@cms-flaf-bot please test`; the pipeline configuration lives in `cms-flaf/FLAF_ci` | diff --git a/.github/workflows/repo-sanity-checks.yaml b/.github/workflows/repo-sanity-checks.yaml index bac486d..2d881c9 100644 --- a/.github/workflows/repo-sanity-checks.yaml +++ b/.github/workflows/repo-sanity-checks.yaml @@ -26,7 +26,7 @@ jobs: repository: cms-flaf/FLAF # A fixed FLAF commit: the checker's rules change with FLAF, and this repository adopts # a new rule set by moving the pin, in a change that also fixes what the new rules find. - ref: e8e42edb2f1626a0875c29ceb5be0992942fb887 + ref: 4d9602cdbcdf6d09ba4566b4f83e3878e39975ca path: flaf_check sparse-checkout: run_tools/check_era_rename.py sparse-checkout-cone-mode: false diff --git a/.github/workflows/unit-tests.yaml b/.github/workflows/unit-tests.yaml index d9ca069..8f29e03 100644 --- a/.github/workflows/unit-tests.yaml +++ b/.github/workflows/unit-tests.yaml @@ -46,4 +46,5 @@ jobs: run: >- python -m pytest -q StatInference/test/test_law_chain.py + StatInference/test/test_card_eras.py StatInference/test/test_shape_plots.py diff --git a/README.md b/README.md index ff42a13..1c36cc0 100644 --- a/README.md +++ b/README.md @@ -101,13 +101,15 @@ rm -r output/binning_v2/tauTau_2018/workers `test/test_law_chain.py` checks the law tasks on the real FLAF and law classes over a scratch analysis (`test/statinference_fixture.py`): for each card, what it writes, which merged -histograms it reads and the commands it runs. `test/test_shape_plots.py` checks which -histograms the shape plots of a card show, for categories read through `input_categories`. +histograms it reads and the commands it runs. `test/test_card_eras.py` checks the card +workflows over several eras, and `test/test_shape_plots.py` which histograms the shape plots +of a card show, for categories read through `input_categories`. They need no ROOT, Combine or dhi. The `unit-tests` workflow runs them against a pinned FLAF commit; to run them locally, put a FLAF checkout at that commit (`ref:` in `.github/workflows/unit-tests.yaml`) next to this one and run from the directory holding both: ```bash pip install pyyaml numpy luigi==3.8.1 law==0.1.21 pytest -PYTHONPATH=$PWD python -m pytest -q StatInference/test/test_law_chain.py StatInference/test/test_shape_plots.py +PYTHONPATH=$PWD python -m pytest -q StatInference/test/test_law_chain.py \ + StatInference/test/test_card_eras.py StatInference/test/test_shape_plots.py ``` diff --git a/law/CreateDatacardsTask.py b/law/CreateDatacardsTask.py index 231c1d2..8ad6ab7 100644 --- a/law/CreateDatacardsTask.py +++ b/law/CreateDatacardsTask.py @@ -6,7 +6,11 @@ from string import Template from FLAF.RunKit.run_tools import ps_call -from FLAF.run_tools.law_customizations import HTCondorWorkflow, copy_param +from FLAF.run_tools.law_customizations import ( + EraDependentValueError, + HTCondorWorkflow, + copy_param, +) from StatInference.common.tools import CategoryNaming, InputCategories, importROOT @@ -15,76 +19,64 @@ class CreateDatacardsTask(StatInferenceTask, HTCondorWorkflow, law.LocalWorkflow): - """The datacards for one era, from the shapes the chain was pointed at. + """The datacards of the configuration, one branch per card (`eras:`, in its order). Runs dc_make/create_datacards.py on whatever `//.root` tree it is given -- the configuration's `preprocess:` step if it declares one, otherwise - the merged histograms directly -- and does no binning of its own. For a meta-era - (`meta_era` set, `era` one of its real sub-eras) it holds every sub-era's - histograms at once, which is what the summed shape behind each datacard bin is built - from, and why the stacked shape plots are drawn here. + the merged histograms directly -- and does no binning of its own. The card of a group + era holds every sub-era's histograms at once, which is what the summed shape behind + each datacard bin is built from, and why the stacked shape plots are drawn here. """ max_runtime = copy_param(HTCondorWorkflow.max_runtime, 2.0) n_cpus = copy_param(HTCondorWorkflow.n_cpus, 1) - # If set, build datacards for this meta-era instead of self.era[0]. self.era[0] must - # then be one of its real sub-eras -- it is only used to construct a valid FLAF Setup - # (Setup.getGlobal requires a real, known era; a meta-era name isn't one). - meta_era = luigi.Parameter(default="") - # Stacked plots of the shapes going into the cards. This task is where they belong: # it is the only one holding every sub-era's histograms at once, which is what the # merged shape (and hence each datacard bin) is built from. make_plots = luigi.BoolParameter(default=True) - def get_sub_eras(self): - """Real eras whose histograms feed into datacard_era: its constituent - sub-eras for a meta-era, otherwise just [self.era[0]].""" - return self.get_era_groups().get(self.datacard_era, [self.era[0]]) - - def input_hist_reqs(self): - """{key: task} for the shapes the cards are built from. + def input_hist_reqs(self, card_eras): + """{key: task} for the shapes the cards of `card_eras` are built from. - The configuration's `preprocess:` step if it declares one, otherwise the merged - histograms directly. Either way what arrives is "//.root", - so nothing below here knows which it got. + The configuration's `preprocess:` step if it declares one -- its workflow over the + same cards (law hands the --branches selection on), or from a branch its branch of + the same card -- otherwise the merged histograms directly. Either way what arrives + is "//.root", so nothing below here knows which it got. """ if self.preprocess_config(): - return { - "PreprocessShapes": PreprocessShapesTask.req( - self, meta_era=self.meta_era, branches=() - ) - } + return {"PreprocessShapes": PreprocessShapesTask.req(self)} return { f"MergedHists_{era}_{variable}": req for (era, variable), req in self.merged_hist_reqs( - self.get_sub_eras() + self.sub_eras_of(card_eras) ).items() } - def workflow_requires(self): + def stage_workflow_requires(self): # Merged with the base class's rather than replacing them: FLAF's HTCondorWorkflow # puts the software bundles a submitted job unpacks in there, and returning only # our own inputs drops them, so the jobs start without the bundle they need. - reqs = super().workflow_requires() - reqs.update(self.input_hist_reqs()) + reqs = super().stage_workflow_requires() + # the selected cards only (the map honours --branches) + reqs.update(self.input_hist_reqs(self.get_branch_map().values())) return reqs def requires(self): - return list(self.input_hist_reqs().values()) + return list(self.input_hist_reqs([self.branch_data]).values()) def create_branch_map(self): - return {0: None} + return dict(enumerate(self.card_eras())) def output(self): # fs_default. Note that combine cannot read these directly: # ResonantLimitsTask mirrors them back to datacards_dir() before handing them to # dhi -- see ResonantLimitsTask.stage_datacards. - return self.output_dir_target(self.version, "Datacards", self.datacard_era) + return self.output_dir_target(self.version, "Datacards", self.branch_data) def run(self): - statInf_entry = self.global_params["StatInference"] + card_era = self.branch_data + statInf_entry = self.statinference_settings() config = self.datacard_config_path() # ${ERA} in hist_bins names the era the cards are for, the same way the model's # input_file_pattern does. Each era has its own binning -- derived from its own @@ -93,7 +85,7 @@ def run(self): hist_bins = ( os.path.join( self.ana_path(), - Template(hist_bins_rel).safe_substitute(ERA=self.datacard_era), + Template(hist_bins_rel).safe_substitute(ERA=card_era), ) if hist_bins_rel else None @@ -107,12 +99,14 @@ def run(self): # PreprocessShapesTask already wrote the "//.root" # layout, so its output directory is the base input_file_pattern resolves # against -- nothing to stage. - reqs = self.input_hist_reqs()["PreprocessShapes"] + reqs = self.input_hist_reqs([card_era])["PreprocessShapes"] base_dir_local = stack.enter_context(reqs.output().localize("r")) else: targets = { key: req.output() - for key, req in self.merged_hist_reqs(self.get_sub_eras()).items() + for key, req in self.merged_hist_reqs( + self.get_sub_eras(card_era) + ).items() } self.check_inputs(targets) base_dir_local = stack.enter_context(self.stage_inputs(targets)) @@ -127,7 +121,7 @@ def run(self): "--config", config, "--eras", - self.datacard_era, + card_era, ] if hist_bins: cmd += ["--hist-bins", hist_bins] @@ -165,21 +159,33 @@ def _surviving_axis(base_dir): slices = node.get("slices") if isinstance(node, dict) else None return 0 if slices and "y_range" in slices[0] else 1 - def plot_variable(self, variable, axis=1): + def plot_variable(self, variable, sub_eras, axis=1): """The variable whose axis the shapes are binned along, for histograms.yaml. For 2D->1D rebinned input, the entry of the 2D variable's ``var_list: [x, y]`` - picked by `axis` (from _surviving_axis()); for 1D input, the variable itself. + picked by `axis` (from _surviving_axis()); for 1D input, the variable itself. The + plotted shapes are the sum over `sub_eras`, so their histograms.yaml must agree. """ - try: - import FLAF.Common.Setup as Setup - - hists = Setup.Setup(self.ana_path(), self.era[0], self.version).hists - var_list = hists[variable].get("var_list") - if var_list and len(var_list) > 1: - return var_list[axis] - except Exception as e: - print(f"Warning: no var_list for {variable} ({e}); plotting it as itself") + # None: the era's histograms.yaml has no entry for the variable + var_lists = {} + for era in sub_eras: + hists = self.setup_for(era).hists + try: + var_lists[era] = hists[variable].get("var_list") + except KeyError: + var_lists[era] = None + var_list = var_lists[sub_eras[0]] + if any(other != var_list for other in var_lists.values()): + raise EraDependentValueError( + f"{self.get_task_family()}: the var_list of {variable} in histograms.yaml " + f"differs between the eras summed in one plot ({var_lists})" + ) + if variable not in self.setup_for(sub_eras[0]).hists: + print( + f"Warning: no histograms.yaml entry for {variable}; plotting it as itself" + ) + elif var_list and len(var_list) > 1: + return var_list[axis] return variable @staticmethod @@ -298,6 +304,8 @@ def plot_rebinned_shapes(self, base_dir, config, output_dir): import glob cfg = self.get_config_data() + card_era = self.branch_data + sub_eras = self.get_sub_eras(card_era) plots_dir = os.path.join(output_dir, "plots") os.makedirs(plots_dir, exist_ok=True) plotter = os.path.join(self.ana_path(), "FLAF", "Analysis", "HistPlotter.py") @@ -309,7 +317,7 @@ def plot_rebinned_shapes(self, base_dir, config, output_dir): for variable in self.get_required_variables(): inputs = [ p - for era in self.get_sub_eras() + for era in sub_eras for p in glob.glob( os.path.join(base_dir, era, variable, f"{variable}.root") ) @@ -376,25 +384,28 @@ def plot_rebinned_shapes(self, base_dir, config, output_dir): "--globalConfig", os.path.join( self.ana_path(), - self.global_params["analysis_config_area"], + self.common( + lambda setup: setup.global_params["analysis_config_area"], + "analysis_config_area", + ), "global.yaml", ), # The surviving axis of the rebinned shapes, and already at its final # binning -- so no --rebin, which would coarsen it back to the histograms.yaml # grid and undo the whole point of the rebinning. "--var", - self.plot_variable(variable, self._surviving_axis(base_dir)), + self.plot_variable(variable, sub_eras, self._surviving_axis(base_dir)), # The plotted shapes are the sum over every sub-era, so the label has to # name the combination: HistPlotter reads config/plot/.yaml for the # luminosity, and a sub-era's file states that sub-era's luminosity alone. # --era stays a real era -- it builds the Setup, which does not know # meta-era names. "--year", - self.datacard_era, + card_era, "--ana_path", self.ana_path(), "--era", - self.era[0], + sub_eras[0], "--LAWrunVersion", self.version, "--wantSignals", diff --git a/law/MergedHists.py b/law/MergedHists.py index 24d8b20..d2a0d7f 100644 --- a/law/MergedHists.py +++ b/law/MergedHists.py @@ -18,13 +18,16 @@ class MergedHists(StatInferenceTask, law.ExternalTask): """ variable = luigi.Parameter(description="Hists_merged variable (directory) name") + # The era the file belongs to, one of `era`: a task over several eras requires the + # files of each, each from that era's own fs_HistTuple. + hist_era = luigi.Parameter(description="era of the merged histograms") def output(self): return self.remote_target( self.input_hists_version, "Hists_merged", - self.era[0], + self.hist_era, self.variable, f"{self.variable}.root", - fs=self.fs_HistTuple, + fs=self.setup_for(self.hist_era).get_fs("HistTuple"), ) diff --git a/law/PlotPullsAndImpactsTask.py b/law/PlotPullsAndImpactsTask.py index 771fac6..c9ef3ca 100644 --- a/law/PlotPullsAndImpactsTask.py +++ b/law/PlotPullsAndImpactsTask.py @@ -40,6 +40,8 @@ class PlotPullsAndImpactsTask(DhiPlotMixin, StatInferenceTask): # NO_STR leaves each of those at its own default. workflow = luigi.Parameter(default=law.parameter.NO_STR) + check_card_eras = True + def requires(self): # ResonantLimitsTask is what writes the combined cards this reads. return [ResonantLimitsTask.req(self)] diff --git a/law/PlotResonantLimitsTask.py b/law/PlotResonantLimitsTask.py index fd41998..d9553f8 100644 --- a/law/PlotResonantLimitsTask.py +++ b/law/PlotResonantLimitsTask.py @@ -43,6 +43,8 @@ class PlotResonantLimitsTask(DhiPlotMixin, StatInferenceTask): # NO_STR leaves each of those at its own default. workflow = luigi.Parameter(default=law.parameter.NO_STR) + check_card_eras = True + def requires(self): # ResonantLimitsTask is what mirrors the cards to datacards_dir(), which is what # the globs below resolve against -- the plots cannot be built before it has run. diff --git a/law/PreprocessShapesTask.py b/law/PreprocessShapesTask.py index f6337c0..39babbb 100644 --- a/law/PreprocessShapesTask.py +++ b/law/PreprocessShapesTask.py @@ -1,6 +1,5 @@ import contextlib import law -import luigi import os from string import Template @@ -27,43 +26,40 @@ class PreprocessShapesTask(StatInferenceTask, HTCondorWorkflow, law.LocalWorkflo is what the model's input_file_pattern resolves against and what the merged histograms already look like. That is the whole contract: the datacard step cannot tell whether it is reading preprocessed shapes or raw ones. + + One branch per card of the configuration, in the order of its `eras:`, as + CreateDatacardsTask, whose branch of a card requires the branch of the same card here. """ max_runtime = copy_param(HTCondorWorkflow.max_runtime, 4.0) n_cpus = copy_param(HTCondorWorkflow.n_cpus, 1) - # As on CreateDatacardsTask: preprocessing a meta-era needs its members, and - # self.era[0] must stay a real era because FLAF's Setup resolves it against config/. - meta_era = luigi.Parameter(default="") - - def get_sub_eras(self): - return self.get_era_groups().get(self.datacard_era, [self.era[0]]) - - def input_hist_reqs(self): + def input_hist_reqs(self, card_eras): return { f"MergedHists_{era}_{variable}": req for (era, variable), req in self.merged_hist_reqs( - self.get_sub_eras() + self.sub_eras_of(card_eras) ).items() } - def workflow_requires(self): + def stage_workflow_requires(self): # Merged with the base class's rather than replacing them: FLAF's HTCondorWorkflow # puts the software bundles a submitted job unpacks in there, and returning only # our own inputs drops them, so the jobs start without the bundle they need. - reqs = super().workflow_requires() - reqs.update(self.input_hist_reqs()) + reqs = super().stage_workflow_requires() + # the selected cards only (the map honours --branches) + reqs.update(self.input_hist_reqs(self.get_branch_map().values())) return reqs def requires(self): - return list(self.input_hist_reqs().values()) + return list(self.input_hist_reqs([self.branch_data]).values()) def create_branch_map(self): - return {0: None} + return dict(enumerate(self.card_eras())) def output(self): return self.output_dir_target( - self.version, "Hists_preprocessed", self.datacard_era + self.version, "Hists_preprocessed", self.branch_data ) def run(self): @@ -75,10 +71,13 @@ def run(self): "directly in that case and does not require it." ) script = os.path.join(self.ana_path(), cfg["script"]) + card_era = self.branch_data with contextlib.ExitStack() as stack: targets = { key: req.output() - for key, req in self.merged_hist_reqs(self.get_sub_eras()).items() + for key, req in self.merged_hist_reqs( + self.get_sub_eras(card_era) + ).items() } self.check_inputs(targets) base_dir_local = stack.enter_context(self.stage_inputs(targets)) @@ -92,7 +91,7 @@ def run(self): "--output", local_output.abspath, "--era", - self.datacard_era, + card_era, "--config", self.datacard_config_path(), ] @@ -101,7 +100,7 @@ def run(self): # against the analysis area, so a configuration can point at its own files # without absolute paths; every other argument is passed through untouched. for arg in cfg.get("args", []): - arg = Template(str(arg)).safe_substitute(ERA=self.datacard_era) + arg = Template(str(arg)).safe_substitute(ERA=card_era) if arg.startswith("config/"): arg = os.path.join(self.ana_path(), arg) cmd.append(arg) diff --git a/law/ResonantLimitsAndHistPlotTask.py b/law/ResonantLimitsAndHistPlotTask.py index eedb516..c83bc47 100644 --- a/law/ResonantLimitsAndHistPlotTask.py +++ b/law/ResonantLimitsAndHistPlotTask.py @@ -14,6 +14,11 @@ class ResonantLimitsAndHistPlotTask(StatInferenceTask): Depends on ResonantLimitsTask, not PlotResonantLimitsTask: this task is about getting the limits together with their input histograms, and the limit plots the configuration declares in `limit_plots` are asked for separately, by running PlotResonantLimitsTask. + + Its output is a marker written once everything below it ran, so a version it finished + is not looked at again: a later run of that version over more eras plots nothing new. + Run HistPlotTask or ResonantLimitsTask directly for that, or remove the marker first + (`--remove-output 0,a,y`). """ # Not a workflow itself. The parameter exists so that --workflow reaches the tasks @@ -21,42 +26,17 @@ class ResonantLimitsAndHistPlotTask(StatInferenceTask): # NO_STR leaves each of those at its own default. workflow = luigi.Parameter(default=law.parameter.NO_STR) - def get_plottable_eras(self): - """Real production eras, for FLAF's HistPlotTask. - - Deliberately not the same set as ResonantLimitsTask.get_eras(), which returns the - eras that carry their own datacards -- for a configuration with an `era_groups:` - entry that is the meta-era, and this is its members. The two are complements: - limits are set on the combination, while the input histograms can only be plotted - for eras that really exist, since HistPlotTask resolves `era` against - config//. - - So a group era is replaced by its members rather than dropped. Filtering it out - and stopping there left a configuration whose `eras:` names only the group -- the - mainline 2D one does -- with nothing to plot at all, and this task then reported - success having required no HistPlotTask. - """ - era_groups = self.get_era_groups() - eras = self.get_config_data().get("eras", [self.era[0]]) - plottable = [] - for e in eras: - for real_era in era_groups.get(e, [e]): - # A configuration may list a group and its members both; each era is - # plotted once. - if real_era not in plottable: - plottable.append(real_era) - return plottable + check_card_eras = True def requires(self): - reqs = [ResonantLimitsTask.req(self)] - for e in self.get_plottable_eras(): - # The histograms to plot are the ones the cards were built from, which live - # under hists_version -- req() would otherwise copy this task's own version - # and schedule a fresh production of them under it. - reqs.append( - HistPlotTask.req(self, era=(e,), version=self.input_hists_version) - ) - return reqs + # The histograms to plot are the ones the cards were built from, which live under + # hists_version -- req() would otherwise copy this task's own version and schedule + # a fresh production of them under it. One HistPlotTask over every era of --era, + # which holds every real era behind the cards (and may hold more, plotted too). + return [ + ResonantLimitsTask.req(self), + HistPlotTask.req(self, version=self.input_hists_version), + ] def output(self): return self.local_target("dummy.txt") diff --git a/law/ResonantLimitsTask.py b/law/ResonantLimitsTask.py index 633e220..5857b44 100644 --- a/law/ResonantLimitsTask.py +++ b/law/ResonantLimitsTask.py @@ -23,23 +23,20 @@ class ResonantLimitsTask(StatInferenceTask): # NO_STR leaves each of those at its own default. workflow = luigi.Parameter(default=law.parameter.NO_STR) + check_card_eras = True + def store_parts(self): return (self.version, self.__class__.__name__, "combined") - def get_eras(self): - """Every era the configuration lists: each gets its own cards and its own limit.""" - return self.get_all_eras() - - def _create_datacards_req(self, era, **kwargs): - era_groups = self.get_era_groups() - if era in era_groups: - return CreateDatacardsTask.req( - self, era=(era_groups[era][0],), meta_era=era, **kwargs - ) - return CreateDatacardsTask.req(self, era=(era,), **kwargs) - def requires(self): - return [self._create_datacards_req(e, branches=()) for e in self.get_eras()] + return CreateDatacardsTask.req(self, branches=()) + + def era_card_outputs(self): + """{card era: the output of the branch that writes that card}.""" + return { + task.branch_data: task.output() + for task in self.requires().get_branch_tasks().values() + } def output(self): return { @@ -77,12 +74,10 @@ def stage_datacards(self, era, remote_target): return local_dir def run(self): - eras = self.get_eras() era_cards = {} - for e in eras: - create_dc_br0 = self._create_datacards_req(e, branch=0, branches=()) - output_dir = self.stage_datacards(e, create_dc_br0.output()) + for e, card_output in self.era_card_outputs().items(): + output_dir = self.stage_datacards(e, card_output) era_cards[e] = glob.glob(os.path.join(output_dir, "*.txt")) masses = set() diff --git a/law/StatInferenceTask.py b/law/StatInferenceTask.py index 7d0ead7..555d8cf 100644 --- a/law/StatInferenceTask.py +++ b/law/StatInferenceTask.py @@ -6,7 +6,7 @@ import yaml from string import Template -from FLAF.run_tools.law_customizations import Task +from FLAF.run_tools.law_customizations import Task, era_csv class StatInferenceTask(Task): @@ -17,8 +17,26 @@ class StatInferenceTask(Task): ``StatInference.config``, not by global.yaml's own variable lists. This base class is the single place that file is read, so the four tasks below cannot disagree about which eras, masses or variables the analysis consists of. + + A task here may run over several eras (`--era A,B,...`), which must include every real + era behind the configuration's cards. The output paths are keyed on the version and the + card, the job state on the version and the class, never on `--era`: a card is the same + product whichever eras the run also covers. """ + allow_multi_era = True + # Check the eras of the run against the cards when the task is constructed. On for the + # tasks a run starts from that are not workflows, so the error stops `law run` before + # luigi schedules anything (a dependency that fails while scheduling leaves its pending + # parent, and a worker with keep_alive, waiting for ever). The card workflows check when + # they build their map: FLAF constructs them over single eras for their resource tokens. + check_card_eras = False + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + if self.check_card_eras: + self.card_eras() + # The Hists_merged tree to read. Separate from `version`, which names what this chain # *writes*: re-binning or re-fitting under a new version must not require the input # histograms to be reproduced (or copied) under that name. Not significant -- it @@ -49,11 +67,15 @@ def output_dir_target(self, *path): return law.LocalDirectoryTarget(os.path.join(*path), fs=fs) return self.remote_dir_target(*path) - def datacard_config_path(self): - return os.path.join( - self.ana_path(), self.global_params["StatInference"]["config"] + def statinference_settings(self): + """global.yaml's `StatInference` block, which every era of the task must share.""" + return self.common( + lambda setup: setup.global_params["StatInference"], "StatInference" ) + def datacard_config_path(self): + return os.path.join(self.ana_path(), self.statinference_settings()["config"]) + def get_config_data(self): # Cached per instance: requires()/workflow_requires() are re-entered many times # during graph construction and each call would otherwise re-parse the yaml. @@ -62,30 +84,19 @@ def get_config_data(self): self._config_data = yaml.safe_load(f) return self._config_data - @property - def datacard_era(self): - """The era this task's products belong to: its meta-era if it has one, else - `era`. Equal to `era` for every task that does not declare `meta_era`. - """ - return getattr(self, "meta_era", "") or self.era[0] - def store_parts(self): - """Keyed on datacard_era, not era. - - A meta-era instance is constructed with `era` set to one of its own sub-eras - -- FLAF's Setup only knows real eras -- so keying on `era` gave - CreateDatacardsTask(era=Run3_2022) and - CreateDatacardsTask(era=Run3_2022, meta_era=Run3_Early) the same store path. - The declared outputs never collided (they key on datacard_era already), but - law's HTCondor control files live here: htcondor_output_directory() is - local_path(), and the submission file is named only after the branches, which - for both is {0: None}. A run needing both -- any configuration listing a group - era and its members -- had the second workflow resume from the first's - submission data, poll jobs that were not its own, and finish incomplete. - - Identical to the old value for every task without a meta_era. + """Keyed on the version and the class, not on `--era`. + + The card workflows hold every card of the configuration as one branch each, so their + HTCondor job file (htcondor_output_directory() is local_path()) belongs to the + configuration, not to the eras of the run: a later run over more eras resumes the + same jobs. FLAF's default keys on the task's one era and refuses several. """ - return (self.version, self.__class__.__name__, self.datacard_era) + return (self.version, self.__class__.__name__) + + def remote_log_dir_target(self): + """Job logs of a version and class, for the same reason as store_parts().""" + return self.output_dir_target(self.version, "logs", self.__class__.__name__) def datacards_dir(self, era): """Local directory holding an era's datacards. @@ -115,7 +126,39 @@ def get_all_eras(self): different datasets, not a double count. Only combining them into one card would be -- see get_top_level_eras(). """ - return self.get_config_data().get("eras", [self.era[0]]) + eras = self.get_config_data().get("eras") + if not eras: + raise ValueError( + f"{self.datacard_config_path()} lists no `eras:`; the cards to build are " + "named there" + ) + return eras + + def get_sub_eras(self, card_era): + """The real eras whose histograms feed the card of `card_era`: the members of a + group era, else the era itself.""" + return self.get_era_groups().get(card_era, [card_era]) + + def card_eras(self): + """get_all_eras(), once every real era behind those cards is one of this task's + eras: a card is built from all of its sub-eras or not at all.""" + cards = self.get_all_eras() + missing = {} + for card in cards: + absent = [era for era in self.get_sub_eras(card) if era not in self.era] + if absent: + missing[card] = absent + if missing: + needed = sorted({era for absent in missing.values() for era in absent}) + raise ValueError( + f"{self.get_task_family()}(era={era_csv(self.era)}): the cards of " + f"{self.datacard_config_path()} need " + + "; ".join( + f"{card}: {', '.join(absent)}" for card, absent in missing.items() + ) + + f", which --era does not include; add {era_csv(needed)} to --era" + ) + return cards def get_top_level_eras(self): """The eras that may be *combined* with each other: group eras plus any standalone @@ -191,18 +234,24 @@ def get_required_variables(self): def merged_hist_reqs(self, eras): """{(era, variable): MergedHists} -- every merged input file for those eras. - Read by CreateDatacardsTask over the sub-eras of the era it builds cards for. + Read by the card tasks over the sub-eras of the cards they build. """ # Deferred: MergedHists subclasses this class, so importing it at module level # would be circular. from .MergedHists import MergedHists return { - (era, variable): MergedHists.req(self, era=(era,), variable=variable) + (era, variable): MergedHists.req(self, hist_era=era, variable=variable) for era in eras for variable in self.get_required_variables() } + def sub_eras_of(self, card_eras): + """The real eras behind the cards of `card_eras`, each once, in order.""" + return list( + dict.fromkeys(era for card in card_eras for era in self.get_sub_eras(card)) + ) + @contextlib.contextmanager def stage_inputs(self, targets): """Assemble the required merged histograms into the "// diff --git a/test/statinference_fixture.py b/test/statinference_fixture.py index 65e1d95..393ae4a 100644 --- a/test/statinference_fixture.py +++ b/test/statinference_fixture.py @@ -1,9 +1,10 @@ """A scratch analysis for the StatInference law tasks, on the real FLAF and law classes. -The analysis has three real eras (Run3_2022, Run3_2022EE, Run3_2023) and the datacard -configurations below. Run3_2022EE reads its merged histograms from a store of its own -(`fs_HistTuple` in config/Run3_2022EE/global.yaml), so an input built with another era's -Setup points to the wrong store and shows up in the paths. +The analysis has three real eras behind the cards (Run3_2022, Run3_2022EE, Run3_2023), one +more configured era (Run3_2023BPix) and the datacard configurations below. Run3_2022EE reads +its merged histograms from a store of its own (`fs_HistTuple` in +config/Run3_2022EE/global.yaml), so an input built with another era's Setup points to the +wrong store and shows up in the paths. The checkouts under test are found next to each other: `/StatInference` (this repository) and `/FLAF`, which is how the unit-test workflow lays them out. @@ -70,6 +71,8 @@ def _importable(name): ERA_C = "Run3_2023" GROUP = "Run3_Early" ERAS = (ERA_A, ERA_B, ERA_C) +#: a configured era behind no card +EXTRA_ERA = "Run3_2023BPix" #: the eras of every card: a group, one of its members and a standalone era CARD_ERAS = [GROUP, ERA_A, ERA_C] @@ -151,7 +154,7 @@ def card(preprocess): "config/Datacards/card.yaml": card(preprocess=False), "config/Datacards/card_preprocess.yaml": card(preprocess=True), } -for _era in ERAS: +for _era in ERAS + (EXTRA_ERA,): FILES[f"config/{_era}/processes.yaml"] = "data:\n datasets: [ Data ]\n" FILES[f"config/{_era}/datasets.yaml"] = "Data: {}\nTTto2L2Nu: {}\n" diff --git a/test/test_card_eras.py b/test/test_card_eras.py new file mode 100644 index 0000000..3146f90 --- /dev/null +++ b/test/test_card_eras.py @@ -0,0 +1,306 @@ +#!/usr/bin/env python3 +"""The card stages over several eras: one workflow whose branches are the cards. + +On the scratch analysis of statinference_fixture (cards Run3_Early = Run3_2022 + Run3_2022EE, +Run3_2022 and Run3_2023, in that order; Run3_2023BPix configured but behind no card): + +* `--era` must hold every real era behind the cards: over fewer, the tasks a run starts from + that are not workflows raise when constructed, a card workflow when its map is built, naming + the card and the missing eras, and luigi runs nothing; more eras give the same cards. A + --branches selection of cards reaches their inputs and their preprocessing. A configuration without `eras:` raises. +* Constructing a card workflow over one era does not raise: FLAF builds single-era + instances of a remote workflow for its luigi resources. +* The job state and the logs do not depend on `--era`: one job directory, one luigi + token and one log directory per version and class, the same over more eras. +* Branch i of CreateDatacardsTask requires branch i of PreprocessShapesTask, the same card; + every branch task carries all eras of the run, and each merged histogram is read with its + own era. +* ResonantLimitsTask requires the one card workflow over the run's eras and stages each card + from that card's branch; ResonantLimitsAndHistPlotTask requires one HistPlotTask over the + run's eras, and its marker is keyed on the version alone. +* A setting a run reads once must agree between its eras: the `StatInference` block, and + the var_list of a plotted variable between the eras summed in one plot. +""" + +import os +import shutil +import sys +import tempfile +import unittest +from unittest import mock + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +import statinference_fixture as fx # noqa: E402 + +import luigi # noqa: E402 + +VERSION = "v_cards" +HISTS_VERSION = "v_hists" +WITH_EXTRA = fx.ERAS + (fx.EXTRA_ERA,) + + +class CardTestCase(unittest.TestCase): + card_file = "config/Datacards/card.yaml" + #: extra files of the scratch analysis: {path: text} + extra_files = {} + + @classmethod + def setUpClass(cls): + cls.tmp = tempfile.mkdtemp(prefix="statinference_cards_") + cls.ana = fx.make_analysis(cls.tmp, cls.card_file) + for path, text in cls.extra_files.items(): + fx.write(os.path.join(cls.ana, path), text) + cls.env = mock.patch.dict( + os.environ, + { + "ANALYSIS_PATH": cls.ana, + "ANALYSIS_DATA_PATH": os.path.join(cls.tmp, "data"), + }, + ) + cls.env.start() + + @classmethod + def tearDownClass(cls): + cls.env.stop() + fx.clear_process_caches() + shutil.rmtree(cls.tmp, ignore_errors=True) + + def setUp(self): + fx.clear_process_caches() + + def task(self, cls, eras=fx.ERAS, **kwargs): + return cls(version=VERSION, era=eras, hists_version=HISTS_VERSION, **kwargs) + + def data(self, *path): + return os.path.join(self.tmp, "data", VERSION, *path) + + +class CardsNeedTheirEras(CardTestCase): + def test_fewer_eras_name_the_missing_ones(self): + for cls in (fx.si.CreateDatacardsTask, fx.si.PreprocessShapesTask): + with self.subTest(cls=cls.__name__): + workflow = self.task(cls, eras=(fx.ERA_A,)) + with self.assertRaises(ValueError) as raised: + workflow.get_branch_map() + message = str(raised.exception) + self.assertIn(f"{fx.GROUP}: {fx.ERA_B}", message) + self.assertIn(f"{fx.ERA_C}: {fx.ERA_C}", message) + self.assertIn(f"add {fx.ERA_B},{fx.ERA_C} to --era", message) + self.assertNotIn(f"{fx.ERA_A}: {fx.ERA_A}", message) + + def test_the_cards_are_the_configured_eras(self): + for eras in (fx.ERAS, WITH_EXTRA): + with self.subTest(eras=eras): + workflow = self.task(fx.si.CreateDatacardsTask, eras=eras) + self.assertEqual( + workflow.get_branch_map(), dict(enumerate(fx.CARD_ERAS)) + ) + + def test_a_run_starting_from_the_limits_stops_at_construction(self): + """The tasks a run starts from that are not workflows refuse too few eras when + they are constructed, before luigi schedules anything -- also when the limits + already exist, so that nothing below is asked to build the card map.""" + eras = (fx.ERA_A, fx.ERA_B) + for name in ( + "datacards/combined/combined_300.txt", + "ResonantLimitsTask/combined/limits.npz", + "ResonantLimitsTask/combined/era_limits/limits_Run3_2022.npz", + ): + fx.write(self.data(*name.replace("datacards", "Datacards").split("/")), "") + for cls in ( + fx.si.ResonantLimitsTask, + fx.si.ResonantLimitsAndHistPlotTask, + fx.si.PlotResonantLimitsTask, + fx.si.PlotPullsAndImpactsTask, + ): + with self.subTest(cls=cls.__name__): + with self.assertRaises(ValueError) as raised: + self.task(cls, eras=eras) + self.assertIn(f"add {fx.ERA_C} to --era", str(raised.exception)) + + def test_luigi_runs_nothing_over_too_few_eras(self): + """A card workflow started directly checks when luigi builds its map: over + --workflow local the error is a failed complete(), and luigi schedules nothing; + over htcondor luigi also asks the workflow for its process resources, which law's + remote proxy takes from the map, outside luigi's error handling, so the error ends + the run.""" + eras = (fx.ERA_A, fx.ERA_B) + local = self.task(fx.si.CreateDatacardsTask, eras=eras, workflow="local") + self.assertFalse( + luigi.build([local], local_scheduler=True, log_level="CRITICAL") + ) + fx.clear_process_caches() + remote = self.task(fx.si.CreateDatacardsTask, eras=eras, workflow="htcondor") + with self.assertRaises(ValueError) as raised: + luigi.build([remote], local_scheduler=True, log_level="CRITICAL") + self.assertIn(f"add {fx.ERA_C} to --era", str(raised.exception)) + self.assertFalse( + os.path.exists(os.path.join(self.tmp, "store", "default", VERSION)) + ) + + +class CardsWithoutEras(CardTestCase): + card_file = "config/Datacards/card_no_eras.yaml" + extra_files = { + "config/Datacards/card_no_eras.yaml": fx.card(preprocess=False).replace( + f"eras: [ {fx.GROUP}, {fx.ERA_A}, {fx.ERA_C} ]\n", "" + ) + } + + def test_a_configuration_without_eras_raises(self): + workflow = self.task(fx.si.CreateDatacardsTask) + with self.assertRaises(ValueError) as raised: + workflow.get_branch_map() + self.assertIn("card_no_eras.yaml lists no `eras:`", str(raised.exception)) + + +class StateWithoutEras(CardTestCase): + def test_one_workflow_over_one_era_can_be_built(self): + for cls in (fx.si.CreateDatacardsTask, fx.si.PreprocessShapesTask): + self.task(cls, eras=(fx.ERA_A,)) + + def test_job_state_and_logs_do_not_depend_on_the_eras(self): + for cls in (fx.si.CreateDatacardsTask, fx.si.PreprocessShapesTask): + with self.subTest(cls=cls.__name__): + directory = self.data(cls.__name__) + logs = os.path.join(self.tmp, "store", "default", VERSION, "logs") + for eras in (fx.ERAS, WITH_EXTRA): + workflow = self.task(cls, eras=eras, workflow="htcondor") + self.assertEqual( + workflow.htcondor_output_directory().abspath, directory + ) + self.assertEqual( + workflow.process_resources(), {f"flaf_jobs:{directory}": 1} + ) + self.assertEqual( + workflow.remote_log_dir_target().abspath, + os.path.join(logs, cls.__name__), + ) + + +class CardsPairAndCarryEveryEra(CardTestCase): + card_file = "config/Datacards/card_preprocess.yaml" + + def test_each_card_reads_the_preprocessed_shapes_of_that_card(self): + cards = self.task(fx.si.CreateDatacardsTask).get_branch_tasks() + outputs = set() + for branch, card in cards.items(): + (preprocess,) = card.requires() + self.assertIsInstance(preprocess, fx.si.PreprocessShapesTask) + self.assertEqual(preprocess.branch, branch) + self.assertEqual(preprocess.branch_data, card.branch_data) + outputs.add(preprocess.output().abspath) + self.assertEqual(len(outputs), len(fx.CARD_ERAS)) + + def test_the_workflow_requires_the_preprocess_workflow(self): + reqs = self.task(fx.si.CreateDatacardsTask).workflow_requires() + self.assertEqual(list(reqs), ["PreprocessShapes"]) + preprocess = reqs["PreprocessShapes"] + self.assertTrue(preprocess.is_workflow()) + self.assertEqual(preprocess.era, fx.ERAS) + + def test_every_branch_carries_every_era(self): + for cls in (fx.si.CreateDatacardsTask, fx.si.PreprocessShapesTask): + for card in self.task(cls).get_branch_tasks().values(): + self.assertEqual(card.era, fx.ERAS) + + def test_each_merged_histogram_is_read_with_its_own_era(self): + preprocess = self.task(fx.si.PreprocessShapesTask).get_branch_tasks() + for task in preprocess.values(): + for req in luigi.task.flatten(task.requires()): + self.assertEqual(req.era, fx.ERAS) + self.assertIn(req.hist_era, fx.SUB_ERAS[task.branch_data]) + self.assertIn(f"/Hists_merged/{req.hist_era}/", req.output().abspath) + + +class SelectedCards(CardTestCase): + card_file = "config/Datacards/card_preprocess.yaml" + + def test_a_selection_reaches_the_preprocessing(self): + workflow = self.task(fx.si.CreateDatacardsTask, branches=(1,)) + self.assertEqual(workflow.get_branch_map(), {1: fx.CARD_ERAS[1]}) + preprocess = workflow.workflow_requires()["PreprocessShapes"] + self.assertEqual(preprocess.get_branch_map(), {1: fx.CARD_ERAS[1]}) + reqs = luigi.task.flatten(preprocess.workflow_requires()) + self.assertEqual({r.hist_era for r in reqs}, set(fx.SUB_ERAS[fx.CARD_ERAS[1]])) + + +class SelectedCardsWithoutPreprocess(CardTestCase): + def test_a_selection_requires_its_cards_inputs_only(self): + workflow = self.task(fx.si.CreateDatacardsTask, branches=(0,)) + reqs = luigi.task.flatten(workflow.workflow_requires()) + self.assertEqual({r.hist_era for r in reqs}, set(fx.SUB_ERAS[fx.GROUP])) + + +class LimitsOverTheCards(CardTestCase): + def test_one_card_workflow_over_the_runs_eras(self): + limits = self.task(fx.si.ResonantLimitsTask, eras=WITH_EXTRA) + cards = limits.requires() + self.assertIsInstance(cards, fx.si.CreateDatacardsTask) + self.assertTrue(cards.is_workflow()) + self.assertEqual(cards.era, WITH_EXTRA) + self.assertEqual(cards.branches, ()) + outputs = limits.era_card_outputs() + self.assertEqual(list(outputs), fx.CARD_ERAS) + for card_era, output in outputs.items(): + self.assertEqual( + output.abspath, + os.path.join( + self.tmp, "store", "default", VERSION, "Datacards", card_era + ), + ) + + def test_one_histplot_over_the_runs_eras(self): + task = self.task(fx.si.ResonantLimitsAndHistPlotTask, eras=WITH_EXTRA) + (histplot,) = [ + r for r in task.requires() if isinstance(r, fx.analysis.HistPlotTask) + ] + self.assertEqual(histplot.era, WITH_EXTRA) + self.assertEqual(histplot.version, HISTS_VERSION) + for eras in (fx.ERAS, WITH_EXTRA): + marker = self.task(fx.si.ResonantLimitsAndHistPlotTask, eras=eras).output() + self.assertEqual( + marker.abspath, self.data("ResonantLimitsAndHistPlotTask", "dummy.txt") + ) + + +class SettingsAgreeBetweenEras(CardTestCase): + extra_files = { + f"config/{fx.ERA_C}/global.yaml": ( + "StatInference:\n config: config/Datacards/card.yaml\n" + " param_values: [ 500 ]\n" + ), + f"config/{fx.ERA_B}/plot/histograms.yaml": "".join( + f"{v}:\n var_list: [ HME, DNN_m{m} ]\n" + for v, m in zip(fx.VARIABLES, fx.MASSES) + ) + + "only_in_2022EE:\n var_list: [ a, b ]\n", + } + + def test_the_statinference_block_must_agree(self): + workflow = self.task(fx.si.CreateDatacardsTask) + with self.assertRaises(fx.lc.EraDependentValueError) as raised: + workflow.get_branch_map() + self.assertIn("StatInference differs between the eras", str(raised.exception)) + + def test_the_plotted_var_list_must_agree(self): + card = self.task(fx.si.CreateDatacardsTask, eras=(fx.ERA_A, fx.ERA_B), branch=0) + variable = fx.VARIABLES[0] + with self.assertRaises(fx.lc.EraDependentValueError) as raised: + card.plot_variable(variable, [fx.ERA_A, fx.ERA_B]) + self.assertIn(f"var_list of {variable}", str(raised.exception)) + self.assertEqual(card.plot_variable(variable, [fx.ERA_A]), "HME") + self.assertEqual( + card.plot_variable(variable, [fx.ERA_B]), f"DNN_m{fx.MASSES[0]}" + ) + self.assertEqual( + card.plot_variable("not_a_variable", [fx.ERA_A, fx.ERA_B]), "not_a_variable" + ) + # present in one of the summed eras only: they disagree as well + with self.assertRaises(fx.lc.EraDependentValueError): + card.plot_variable("only_in_2022EE", [fx.ERA_A, fx.ERA_B]) + + +if __name__ == "__main__": + unittest.main() diff --git a/test/test_law_chain.py b/test/test_law_chain.py index a142d2a..73c96ef 100644 --- a/test/test_law_chain.py +++ b/test/test_law_chain.py @@ -52,9 +52,9 @@ def root_params(): - """The parameters of the task the chain is run from. One era is enough: each card - chooses the eras it is built from.""" - return dict(version=VERSION, era=fx.ERA_A, hists_version=HISTS_VERSION) + """The parameters of the task the chain is run from, over every real era behind the + cards.""" + return dict(version=VERSION, era=fx.ERAS, hists_version=HISTS_VERSION) def limits_task(): @@ -66,22 +66,24 @@ def limits_and_histplot_task(): def card_workflows(limits): - """The CreateDatacardsTask workflows `limits` requires.""" + """The CreateDatacardsTask workflows `limits` requires: one, whose branches are the + cards.""" return luigi.task.flatten(limits.requires()) def cards_of(limits): """{card era: the task that writes that card}.""" return { - workflow.datacard_era: fx.si.CreateDatacardsTask.req(workflow, branch=0) + task.branch_data: task for workflow in card_workflows(limits) + for task in workflow.get_branch_tasks().values() } def preprocess_of(card): """The task that writes the preprocessed shapes `card` reads.""" - (workflow,) = card.requires() - return fx.si.PreprocessShapesTask.req(workflow, branch=0) + (preprocess,) = card.requires() + return preprocess def histplot_eras(task):