Machine settings debug logging and knobs (#105)

* Debug messages
* Extract no knobs/all knobs/default knobs
* pytimber warnings

CHANGELOG.md

@@ -1,5 +1,13 @@
 # pylhc Changelog
 
+## Version 0.7.4
+
+Changes in Machine Settings Info
+ - Default behaviour for no knobs given: extract None.
+ - Old behaviour of extracting all restored by giving `knobs = ["all"]`
+ - Option `["default"]` available for default knobs.
+ - Additional debug logging
+
 ## Version 0.7.3 
 
 Fixes:

pylhc/__init__.py

@@ -10,7 +10,7 @@
 __title__ = "pylhc"
 __description__ = "An accelerator physics script collection for the OMC team at CERN."
 __url__ = "https://github.com/pylhc/pylhc"
-__version__ = "0.7.3"
+__version__ = "0.7.4"
 __author__ = "pylhc"
 __author_email__ = "[email protected]"
 __license__ = "MIT"

pylhc/data_extract/lsa.py

@@ -59,12 +59,17 @@ def find_knob_names(self, accelerator: str = "lhc", regexp: str = "") -> list:
         Returns:
             Sorted list of knob names.
         """
+        LOG.debug(f"Getting knobs for {accelerator}.")
         req = self._ParametersRequestBuilder()
         req.setAccelerator(self._getAccelerator(accelerator))
         req.setParameterTypeName("KNOB")
         lst = self._parameterService.findParameters(req.build())
-        reg = re.compile(regexp, re.IGNORECASE)
-        return sorted(filter(reg.search, [pp.getName() for pp in lst]))
+        LOG.debug(f"{len(lst)} Knobs extracted for {accelerator}.")
+        if regexp:
+            LOG.debug(f"Selecting Knobs containing expression: {regexp}")
+            reg = re.compile(regexp, re.IGNORECASE)
+            return sorted(filter(reg.search, [pp.getName() for pp in lst]))
+        return sorted(pp.getName() for pp in lst)
 
     def find_existing_knobs(self, knobs: List[str]) -> List[str]:
         """
@@ -80,6 +85,7 @@ def find_existing_knobs(self, knobs: List[str]) -> List[str]:
         Returns:
             A list of the knob names that actually exist.
         """
+        LOG.debug(f"Checking if the following knobs exist: {knobs}")
         dont_exist = [k for k in knobs if self._getParameter(k) is None]
         if len(dont_exist):
             LOG.warning(f"The following knobs do not exist and will be filtered: {dont_exist}.")
@@ -134,19 +140,22 @@ def find_beamprocess_history(
         """
         cts = self.findUserContextMappingHistory(t_start.timestamp(), t_end.timestamp(), accelerator=accelerator)
 
-        db = pytimber.LoggingDB(source=source)
+        db = pytimber.LoggingDB(source=source, loglevel=logging.WARNING)
         fillnts, fillnv = try_to_acquire_data(
             db.get, "HX:FILLN", t_start.timestamp(), t_end.timestamp()
         )["HX:FILLN"]
 
         if not len(fillnv):
             raise ValueError(f"No beamprocesses for {accelerator} ({source}) found between {t_start} - {t_end}.")
 
+        LOG.debug(f"{len(fillnts)} fills aqcuired.")
+        # map beam-processes to fills
         fills = {}
         for ts, name in zip(cts.timestamp, cts.name):
             idx = fillnts.searchsorted(ts) - 1
             filln = int(fillnv[idx])
             fills.setdefault(filln, []).insert(0, (ts, name))
+        LOG.debug(f"Beamprocess History extracted.")
         return fills
 
     def get_trim_history(
@@ -169,6 +178,7 @@ def get_trim_history(
         Returns:
             Dictionary of trims and their data (as TrimTuples, i.e. NamedTuple of lists of time and data).
         """
+        LOG.debug("Extracting Trim history.")
         if knobs is None or len(knobs) == 0:
             knobs = self.find_knob_names(accelerator)
         else:
@@ -182,13 +192,15 @@ def get_trim_history(
         if end_time is not None:
             end_time = end_time.timestamp() 
 
+        LOG.debug(f"Getting trims for {len(knobs)} knobs.")
         try:
             trims = self.getTrims(parameter=knobs, beamprocess=beamprocess, start=start_time, end=end_time)
         except jpype.java.lang.NullPointerException as e:
             # In the past this happened, when a knob was not defined, but
             # this should have been caught by the filter_existing_knobs above
             raise ValueError(f"Something went wrong when extracting trims for the knobs: {knobs}") from e
 
+        LOG.debug(f"{len(trims)} trims extracted.")
         trims_not_found = [k for k in knobs if k not in trims.keys()]
         if len(trims_not_found):
             LOG.warning(f"The following knobs were not found in '{beamprocess}': {trims_not_found}")
@@ -205,9 +217,10 @@ def get_beamprocess_info(self, beamprocess: Union[str, object]) -> Dict:
             Dictionary with context info.
         """
         if isinstance(beamprocess, str):
+            LOG.debug(f"Extracting Beamprocess {beamprocess} from ContextService")
             beamprocess = self._contextService.findStandAloneBeamProcess(beamprocess)
         bp_dict = _beamprocess_to_dict(beamprocess)
-        LOG.debug(str(bp_dict))
+        LOG.debug(f"Beamprocess details: {str(bp_dict)}")
         return bp_dict
 
     def find_active_beamprocess_at_time(
@@ -285,11 +298,13 @@ def get_knob_circuits(self, knob_name: str, optics: str) -> tfs.TfsDataFrame:
     def get_madx_name_from_circuit(self, circuit: str):
         """Returns the ``MAD-X`` Strength Name (Circuit/Knob) from the given circuit name."""
         logical_name = circuit.split("/")[0]
-        slist = jpype.java.util.Collections.singletonList(
+        slist = jpype.java.util.Collections.singletonList(  # python lists did not work (jdilly)
             logical_name
-        )  # python lists did not work (jdilly)
-        madx_name = self._deviceService.findMadStrengthNamesByLogicalNames(slist)  # returns a map
-        return madx_name[logical_name]
+        )
+        madx_name_map = self._deviceService.findMadStrengthNamesByLogicalNames(slist)  # returns a map
+        madx_name = madx_name_map[logical_name]
+        LOG.debug(f"Name conversion: {circuit} -> {logical_name} -> {madx_name}")
+        return madx_name
 
 
 # Single Instance LSAClient ####################################################

pylhc/machine_settings_info.py

@@ -53,6 +53,7 @@
 from omc3.utils.time_tools import AccDatetime, AcceleratorDatetime
 from pathlib import Path
 from typing import Tuple, Iterable, Dict, Union
+from omc3.knob_extractor import name2lsa, KNOB_CATEGORIES
 
 from pylhc.constants import machine_settings_info as const
 from pylhc.data_extract.lsa import COL_NAME as LSA_COLUMN_NAME, LSA
@@ -90,7 +91,12 @@ def _get_params() -> dict:
             default=None,
             nargs="+",
             type=str,
-            help="List of knobnames."),
+            help="List of knobnames. "
+                 "If `None` (or omitted) no knobs will be extracted. "
+                 "If it is just the string ``'all'``, "
+                 "all knobs will be extracted (can be slow)."
+                 "Use the string ``'default'`` the main knobs of interest."
+        ),
         accel=dict(
             default='lhc',
             type=str,
@@ -138,6 +144,12 @@ def get_info(opt) -> Dict[str, object]:
         - **knobs** *(str)*:
 
             List of knobnames.
+            If `None` (or omitted) no knobs will be extracted.
+            If it is just the string ``'all'``,
+            all knobs will be extracted (can be slow).
+            Use the string ``'default'`` the main knobs of interest.
+            If this is called from python, the strings need
+            to be put as single items into a list.
 
             default: ``None``
 
@@ -198,13 +210,21 @@ def get_info(opt) -> Dict[str, object]:
     except ValueError as e:
         LOG.error(str(e))
     else:
-        trim_histories = LSA.get_trim_history(beamprocess_info.Object, opt.knobs, start_time=acc_start_time, end_time=acc_time, accelerator=opt.accel)
-        trims = _get_last_trim(trim_histories)
+        if opt.knobs is not None:
+            if len(opt.knobs) == 1 and opt.knobs[0].lower() == 'all':
+                opt.knobs = []  # will extract all knobs in get_trim_history
+            if len(opt.knobs) == 1 and opt.knobs[0].lower() == 'default':
+                opt.knobs = [name2lsa(knob) for category in KNOB_CATEGORIES.values()
+                             for knob in category]
+
+            trim_histories = LSA.get_trim_history(beamprocess_info.Object, opt.knobs, start_time=acc_start_time, end_time=acc_time, accelerator=opt.accel)
+            trims = _get_last_trim(trim_histories)
+
         if opt.knob_definitions:
-            if opt.knobs:
-                knob_definitions = _get_knob_definitions(opt.knobs, optics_info.Name)
+            if trim_histories:  # this now works with 'all'/`default`. Might create a lot of files
+                knob_definitions = _get_knob_definitions(trim_histories.keys(), optics_info.Name)
             else:
-                LOG.error("Writing out knob definitions requires providing a list of knobs.")
+                LOG.error("Writing out knob definitions requested, but no knobs extracted.")
 
     if opt.log:
         log_summary(acc_time, beamprocess_info, optics_info, trims)
@@ -384,6 +404,10 @@ def _get_beamprocess(acc_time: AccDatetime, accel: str, source: str) -> DotDict:
         raise ValueError(f"In fill {fill_no} the {str(e)}") from e
     bp_info = LSA.get_beamprocess_info(beamprocess)
     bp_info.update({"Fill": fill_no, "StartTime": start_time})
+    LOG.debug(
+        f"Beamprocess {bp_info['Name']} in fill {fill_no}"
+        f" extracted at time {start_time}."
+    )
     return DotDict(bp_info)
 
 
@@ -396,6 +420,7 @@ def _get_beamprocess_start(beamprocesses: Iterable[Tuple[float, str]], acc_time:
     ts = acc_time.timestamp()
     for time, name in sorted(beamprocesses, key=lambda x: x[0], reverse=True):
         if time <= ts and name == bp_name:
+            LOG.debug(f"Found start for beamprocess '{bp_name}' at timestamp {time}.")
             return acc_time.__class__.from_timestamp(time)
     raise ValueError(
         f"Beamprocess '{bp_name}' was not found."
@@ -407,8 +432,10 @@ def _get_beamprocess_start(beamprocesses: Iterable[Tuple[float, str]], acc_time:
 
 def _get_optics(acc_time: AccDatetime, beamprocess: str, bp_start: AccDatetime) -> DotDict:
     """Get the info about the active optics at ``acc_time``."""
+    LOG.debug(f"Getting optics for {beamprocess} at time {acc_time.utc_string}")
     optics_table = LSA.getOpticTable(beamprocess)
     optics, start_time = _get_last_optics(optics_table, beamprocess, bp_start, acc_time)
+    LOG.debug(f"Optics {optics} extracted at time {start_time.utc_string}.")
     return DotDict({"Name": optics, "StartTime": start_time})
 
 
@@ -431,6 +458,7 @@ def _get_last_optics(
 
 def _get_knob_definitions(knobs: list, optics: str):
     """Get knob definitions."""
+    LOG.debug(f"Extracting knob definitions for {len(knobs)} in optics '{optics}'.")
     defs = {}
     for knob in knobs:
         try:
@@ -449,6 +477,7 @@ def _get_last_trim(trims: dict) -> dict:
     Returns:
         Dictionary of knob names and their values.
     """
+    LOG.debug("Extracting last trim from found trim histories.")
     trim_dict = {trim: trims[trim].data[-1] for trim in trims.keys()}  # return last set value
     for trim, value in trim_dict.items():
         try: