Merge upstream master into GhentAnalysis/master (0.3 release cycle)

.all-contributorsrc

@@ -175,6 +175,24 @@
       "contributions": [
         "code"
       ]
+    },
+    {
+      "login": "Lara813",
+      "name": "lamarkus",
+      "avatar_url": "https://avatars.githubusercontent.com/u/67378765?v=4",
+      "profile": "https://github.com/Lara813",
+      "contributions": [
+        "code"
+      ]
+    },
+    {
+      "login": "LennertGriesing",
+      "name": "Lennert Griesing",
+      "avatar_url": "https://avatars.githubusercontent.com/u/115483673?v=4",
+      "profile": "https://github.com/LennertGriesing",
+      "contributions": [
+        "code"
+      ]
     }
   ],
   "commitType": "docs"

README.md

@@ -154,6 +154,8 @@ For a better overview of the tasks that are triggered by the commands below, che
       <td align="center" valign="top" width="14.28%"><a href="https://github.com/aalvesan"><img src="https://avatars.githubusercontent.com/u/99343616?v=4?s=100" width="100px;" alt="Ana Andrade"/><br /><sub><b>Ana Andrade</b></sub></a><br /><a href="https://github.com/columnflow/columnflow/commits?author=aalvesan" title="Code">💻</a> <a href="https://github.com/columnflow/columnflow/pulls?q=is%3Apr+reviewed-by%3Aaalvesan" title="Reviewed Pull Requests">👀</a></td>
       <td align="center" valign="top" width="14.28%"><a href="https://github.com/philippgadow"><img src="https://avatars.githubusercontent.com/u/6804366?v=4?s=100" width="100px;" alt="philippgadow"/><br /><sub><b>philippgadow</b></sub></a><br /><a href="https://github.com/columnflow/columnflow/commits?author=philippgadow" title="Code">💻</a></td>
       <td align="center" valign="top" width="14.28%"><a href="https://github.com/LuSchaller"><img src="https://avatars.githubusercontent.com/u/30951523?v=4?s=100" width="100px;" alt="Lukas Schaller"/><br /><sub><b>Lukas Schaller</b></sub></a><br /><a href="https://github.com/columnflow/columnflow/commits?author=LuSchaller" title="Code">💻</a></td>
+      <td align="center" valign="top" width="14.28%"><a href="https://github.com/Lara813"><img src="https://avatars.githubusercontent.com/u/67378765?v=4?s=100" width="100px;" alt="lamarkus"/><br /><sub><b>lamarkus</b></sub></a><br /><a href="https://github.com/columnflow/columnflow/commits?author=Lara813" title="Code">💻</a></td>
+      <td align="center" valign="top" width="14.28%"><a href="https://github.com/LennertGriesing"><img src="https://avatars.githubusercontent.com/u/115483673?v=4?s=100" width="100px;" alt="Lennert Griesing"/><br /><sub><b>Lennert Griesing</b></sub></a><br /><a href="https://github.com/columnflow/columnflow/commits?author=LennertGriesing" title="Code">💻</a></td>
     </tr>
   </tbody>
 </table>

analysis_templates/cms_minimal/__cf_module_name__/columnflow_patches.py

@@ -4,8 +4,6 @@
 Collection of patches of underlying columnflow tasks.
 """
 
-import os
-
 import law
 from columnflow.util import memoize
 
@@ -17,19 +15,8 @@
 def patch_bundle_repo_exclude_files():
     from columnflow.tasks.framework.remote import BundleRepo
 
-    # get the relative path to CF_BASE
-    cf_rel = os.path.relpath(os.environ["CF_BASE"], os.environ["__cf_short_name_uc___BASE"])
-
-    # amend exclude files to start with the relative path to CF_BASE
-    exclude_files = [os.path.join(cf_rel, path) for path in BundleRepo.exclude_files]
-
-    # add additional files
-    exclude_files.extend([
-        "docs", "tests", "data", "assets", ".law", ".setups", ".data", ".github",
-    ])
-
-    # overwrite them
-    BundleRepo.exclude_files[:] = exclude_files
+    # add additional files to exclude
+    BundleRepo.exclude_files += ["docs"]
 
     logger.debug("patched exclude_files of cf.BundleRepo")
 

analysis_templates/cms_minimal/__cf_module_name__/config/analysis___cf_short_name_lc__.py

@@ -308,23 +308,20 @@
 cfg.add_variable(
     name="event",
     expression="event",
-    binning=(1, 0.0, 1.0e9),
+    binning=(100, 0.0, 1.0e9),
     x_title="Event number",
-    discrete_x=True,
 )
 cfg.add_variable(
     name="run",
     expression="run",
-    binning=(1, 100000.0, 500000.0),
+    binning=(100, 100000.0, 500000.0),
     x_title="Run number",
-    discrete_x=True,
 )
 cfg.add_variable(
     name="lumi",
     expression="luminosityBlock",
-    binning=(1, 0.0, 5000.0),
+    binning=(100, 0.0, 5000.0),
     x_title="Luminosity block",
-    discrete_x=True,
 )
 cfg.add_variable(
     name="n_jet",

analysis_templates/cms_minimal/law.cfg

@@ -209,6 +209,7 @@ cache_cleanup: $CF_WLCG_CACHE_CLEANUP
 cache_max_size: 15GB
 cache_global_lock: True
 cache_mtime_patience: -1
+rucio_report_access: True
 
 
 [wlcg_fs_global_redirector]
@@ -220,3 +221,4 @@ cache_cleanup: $CF_WLCG_CACHE_CLEANUP
 cache_max_size: 15GB
 cache_global_lock: True
 cache_mtime_patience: -1
+rucio_report_access: True

bin/cf_inspect

@@ -6,10 +6,10 @@ action () {
     local this_dir="$( cd "$( dirname "${this_file}" )" && pwd )"
 
     # check arguments
-    # [ "$#" -eq 0 ] && {
-    #     echo "ERROR: at least one file must be provided"
-    #     return 1
-    # }
+    [ "$#" -eq 0 ] && {
+        echo "ERROR: at least one file must be provided"
+        return 1
+    }
 
     # determine the sandbox to use
     local cf_inspect_sandbox="${CF_INSPECT_SANDBOX:-venv_columnar_dev}"

bin/cf_inspect.py

@@ -13,8 +13,8 @@
 import pickle
 
 import awkward as ak
-import numpy as np  # noqa
 
+from columnflow.columnar_util import update_ak_array, ChunkedIOHandler
 from columnflow.util import ipython_shell
 from columnflow.types import Any
 
@@ -35,11 +35,9 @@ def _load_parquet(fname: str, **kwargs) -> ak.Array:
 
 def _load_nano_root(fname: str, treepath: str | None = None, **kwargs) -> ak.Array:
     import uproot
-    import coffea.nanoevents
-
-    source = uproot.open(fname)
 
     # get the default treepath
+    source = uproot.open(fname)
     if treepath is None:
         for treepath in ["events", "Events"] + list(source.keys()):
             treepath = treepath.split(";", 1)[0]
@@ -48,24 +46,13 @@ def _load_nano_root(fname: str, treepath: str | None = None, **kwargs) -> ak.Arr
                 break
         else:
             raise ValueError(f"no default treepath determined in {fname}")
-    try:
-        return coffea.nanoevents.NanoEventsFactory.from_root(
-            source,
-            treepath=treepath,
-            delayed=False,
-            runtime_cache=None,
-            persistent_cache=None,
-        ).events()
-    except:
-        return uproot.open(fname)
-
-    return coffea.nanoevents.NanoEventsFactory.from_root(
-        source,
-        treepath=treepath,
-        mode="eager",
-        runtime_cache=None,
-        persistent_cache=None,
-    ).events()
+
+    # load the tree
+    tree = source[treepath]
+
+    # use the io handler to load arrays and attach nano schema
+    chunk_pos = ChunkedIOHandler.create_chunk_position(tree.num_entries, tree.num_entries, 0)
+    return ChunkedIOHandler.read_coffea_root(tree, chunk_pos)
 
 
 def load(fname: str, **kwargs) -> Any:
@@ -112,19 +99,57 @@ def list_content(data: Any) -> None:
     )
 
     ap.register("action", "help", argparse._HelpAction)
-    ap.add_argument("files", metavar="FILE", nargs="*", help="one or more supported files")
-    ap.add_argument("--events", "-e", action="store_true", help="assume files to contain event info")
-    ap.add_argument("--hists", "-h", action="store_true", help="assume files to contain histograms")
-    ap.add_argument("--treepath", "-t", type=str, help="name of the tree in ROOT files")
-    ap.add_argument("--list", "-l", action="store_true", help="list contents of the loaded file")
-    ap.add_argument("--help", action="help", help="show this help message and exit")
+    ap.add_argument(
+        "files",
+        metavar="FILE",
+        nargs="+",
+        help="one or more supported files",
+    )
+    ap.add_argument(
+        "--events",
+        "-e",
+        action="store_true",
+        help="assume files to contain event info",
+    )
+    ap.add_argument(
+        "--hists",
+        "-h",
+        action="store_true",
+        help="assume files to contain histograms",
+    )
+    ap.add_argument(
+        "--treepath",
+        "-t",
+        type=str,
+        help="name of the tree in ROOT files",
+    )
+    ap.add_argument(
+        "--list",
+        "-l",
+        action="store_true",
+        help="list contents of the loaded file",
+    )
+    ap.add_argument(
+        "--merge-columns",
+        "-m",
+        action="store_true",
+        help="merge columns of multiple files into one array",
+    )
+    ap.add_argument(
+        "--help",
+        action="help",
+        help="show this help message and exit",
+    )
 
     args = ap.parse_args()
 
     load_kwargs = {
         "treepath": args.treepath,
     }
     objects = [load(fname, **load_kwargs) for fname in args.files]
+
+    if len(objects) > 1 and args.merge_columns:
+        objects = [update_ak_array(objects[0], *objects[1:])]
     if len(objects) == 1:
         objects = objects[0]
     print("file content loaded into variable 'objects'")

columnflow/calibration/__init__.py

@@ -6,32 +6,46 @@
 
 from __future__ import annotations
 
+import copy
 import inspect
 
 import law
 
-from columnflow.util import DerivableMeta
 from columnflow.columnar_util import TaskArrayFunction
-from columnflow.types import Callable, Sequence, Any
+from columnflow.util import DerivableMeta, UNSET
+from columnflow.types import Callable, Sequence, Any, UNSET_TYPE
 
 
 class TaskArrayFunctionWithCalibratorRequirements(TaskArrayFunction):
 
     require_calibrators: Sequence[str] | set[str] | None = None
 
+    def __init__(self, *args, **kwargs):
+        if "require_calibrators" in kwargs or self.__class__.require_calibrators is None:
+            kwargs["require_calibrators"] = kwargs.get("require_calibrators") or []
+        elif isinstance(self.__class__.require_calibrators, (list, tuple)):
+            kwargs["require_calibrators"] = copy.copy(self.__class__.require_calibrators)
+
+        super().__init__(*args, **kwargs)
+
     def _req_calibrator(self, task: law.Task, calibrator: str) -> Any:
         # hook to customize how required calibrators are requested
         from columnflow.tasks.calibration import CalibrateEvents
         return CalibrateEvents.req_other_calibrator(task, calibrator=calibrator)
 
     def requires_func(self, task: law.Task, reqs: dict, **kwargs) -> None:
+        super().requires_func(task=task, reqs=reqs, **kwargs)
+
         # no requirements for workflows in pilot mode
         if callable(getattr(task, "is_workflow", None)) and task.is_workflow() and getattr(task, "pilot", False):
             return
 
         # add required calibrators when set
         if (calibs := self.require_calibrators):
-            reqs["required_calibrators"] = {calib: self._req_calibrator(task, calib) for calib in calibs}
+            reqs["required_calibrators"] = {
+                calib: self._req_calibrator(task, calib)
+                for calib in law.util.make_unique(calibs)
+            }
 
     def setup_func(
         self,
@@ -41,6 +55,8 @@ def setup_func(
         reader_targets: law.util.InsertableDict,
         **kwargs,
     ) -> None:
+        super().setup_func(task=task, reqs=reqs, inputs=inputs, reader_targets=reader_targets, **kwargs)
+
         if "required_calibrators" in inputs:
             for calib, inp in inputs["required_calibrators"].items():
                 reader_targets[f"required_calibrator_{calib}"] = inp["columns"]
@@ -62,9 +78,9 @@ def calibrator(
         cls,
         func: Callable | None = None,
         bases: tuple = (),
-        mc_only: bool = False,
-        data_only: bool = False,
-        require_calibrators: Sequence[str] | set[str] | None = None,
+        mc_only: bool | UNSET_TYPE = UNSET,
+        data_only: bool | UNSET_TYPE = UNSET,
+        require_calibrators: Sequence[str] | set[str] | None | UNSET_TYPE = UNSET,
         **kwargs,
     ) -> DerivableMeta | Callable:
         """
@@ -88,13 +104,13 @@ def calibrator(
         """
         def decorator(func: Callable) -> DerivableMeta:
             # create the class dict
-            cls_dict = {
-                **kwargs,
-                "call_func": func,
-                "mc_only": mc_only,
-                "data_only": data_only,
-                "require_calibrators": require_calibrators,
-            }
+            cls_dict = {**kwargs, "call_func": func}
+            if mc_only is not UNSET:
+                cls_dict["mc_only"] = mc_only
+            if data_only is not UNSET:
+                cls_dict["data_only"] = data_only
+            if require_calibrators is not UNSET:
+                cls_dict["require_calibrators"] = require_calibrators
 
             # get the module name
             frame = inspect.stack()[1]
@@ -113,8 +129,7 @@ def update_cls_dict(cls_name, cls_dict, get_attr):
                     raise Exception(f"calibrator {cls_name} received both mc_only and data_only")
                 if (mc_only or data_only) and cls_dict.get("skip_func"):
                     raise Exception(
-                        f"calibrator {cls_name} received custom skip_func, but either mc_only or "
-                        "data_only are set",
+                        f"calibrator {cls_name} received custom skip_func, but either mc_only or data_only are set",
                     )
 
                 if "skip_func" not in cls_dict:

columnflow/calibration/cms/egamma.py

@@ -97,7 +97,7 @@ def get_inputs(corrector, **additional_variables):
 
         # get scaled energy error
         smear = self.smear_syst_corrector.evaluate("smear", *get_inputs(self.smear_syst_corrector, pt=pt_scaled))
-        energy_err_scaled = (((coll.energyErr)**2 + (coll.energy * smear)**2) * scale)**0.5
+        energy_err_scaled = (((coll.energyErr)**2 + (coll.energy * smear)**2))**0.5 * scale
 
         # store columns
         events = set_ak_column_f32(events, f"{self.collection_name}.pt", pt_scaled)
@@ -127,7 +127,7 @@ def apply_smearing(syst):
             smear_factor = 1.0 + smear * rnd
             pt_smeared = coll.pt * smear_factor
             # get smeared energy error
-            energy_err_smeared = (((coll.energyErr)**2 + (coll.energy * smear)**2) * smear_factor)**0.5
+            energy_err_smeared = (((coll.energyErr)**2 + (coll.energy * smear)**2))**0.5 * smear_factor
             # return both
             return pt_smeared, energy_err_smeared
 
@@ -160,6 +160,8 @@ def apply_smearing(syst):
 
 @_egamma_scale_smear.init
 def _egamma_scale_smear_init(self: Calibrator, **kwargs) -> None:
+    super(_egamma_scale_smear, self).init_func(**kwargs)
+
     # store the config
     self.cfg = self.get_scale_smear_config()
 
@@ -182,6 +184,8 @@ def _egamma_scale_smear_init(self: Calibrator, **kwargs) -> None:
 
 @_egamma_scale_smear.requires
 def _egamma_scale_smear_requires(self, task: law.Task, reqs: dict[str, DotDict[str, Any]], **kwargs) -> None:
+    super(_egamma_scale_smear, self).requires_func(task=task, reqs=reqs, **kwargs)
+
     if "external_files" in reqs:
         return
 
@@ -198,6 +202,14 @@ def _egamma_scale_smear_setup(
     reader_targets: law.util.InsertableDict,
     **kwargs,
 ) -> None:
+    super(_egamma_scale_smear, self).setup_func(
+        task=task,
+        reqs=reqs,
+        inputs=inputs,
+        reader_targets=reader_targets,
+        **kwargs,
+    )
+
     # get and load the correction file
     corr_file = self.get_correction_file(reqs["external_files"].files)
     corr_set = load_correction_set(corr_file)