Base Evaluator

Unified evaluator: runs diagnostic-driven analyzers and reads s-file scalars.

The evaluator's job is to assemble a per-shot scalar dict for every shot in the current bin from two sources:

• analyzers: diagnostic-driven scan analyzers (run in-memory against the DataLogger frame). Each analyzer's ImageAnalyzerResult.scalars dict comes back with keys already namespaced via {metric_prefix}_{key}{metric_suffix} (defaults to the diagnostic's name for the prefix and empty for the suffix when unset). The namespacing is applied by ScanAnalysis's SingleDeviceScanAnalyzer._consume_result per #412; the analyzer side emits bare keys and this evaluator just forwards them through. So scalars["UC_TopView_x_fwhm"] is the convention regardless of which side of the contract was running.
• scalars: column names pulled directly from the current-bin DataFrame (i.e. raw s-file values like "U_Laser:Energy"). Used verbatim as keys.

Subclasses then implement either or both of:

• :meth:compute_objective — returns a float (the value Xopt optimizes)
• :meth:compute_observables — returns a dict of named auxiliary metrics (also returned to Xopt; required for BAX-style algorithms that don't have an objective)

Both hooks receive the same scalars dict and both can optionally override the per-shot list versions (:meth:compute_objective_from_shots, :meth:compute_observables_from_shots) for non-mean statistics or shot-level filtering.

Parameters:

Name	Type	Description	Default
analyzers	list of (str or dict)	Diagnostic stems or dict-form entries {diagnostic: X, ...overrides}. Each entry passes through :func:image_analysis.config.load_diagnostic and yields a scan analyzer built via :func:scan_analysis.config.create_scan_analyzer with use_injected_data=True.	None
scalars	list of str	Column names to pull from current_data_bin per shot. No analyzer is invoked for these; they're already in the DataLogger frame.	None
device_requirements	dict	Override the auto-generated requirements. The default is the union of per-analyzer blocks (each keyed on the GEECS device name).	None
scan_data_manager	injected at construction time		None
data_logger	injected at construction time		None

Attributes:

Name	Type	Description
diagnostics	list of DiagnosticAnalysisConfig	Resolved diagnostics from analyzers.
scan_analyzers	dict[str, ScanAnalyzer]	One ScanAnalyzer per diagnostic, keyed by GEECS device name.
scalar_keys	list of str	Column names that will be read from current_data_bin.
output_key	str or None	The key in the returned outputs dict that Xopt should treat as the objective. Defaults to "f". Set to None for BAX evaluators that emit only observables.
objective_tag	(str, optional)	Human-readable label written to log_entries (under Objective:<tag>). Xopt's actual objective key is hardcoded as "f"; this tag exists purely so the s-file row that gets written to disk carries a recognisable name. Defaults to the subclass name. Override via this kwarg from YAML, or via a class attribute on the subclass.

Methods:

Name	Description
get_current_data	Refresh current_data_bin and current_shot_numbers from data_logger.
filter_log_entries_by_bin	Return all log entries belonging to bin_num.
get_value	Refresh data, evaluate, normalise types, log, and return results.
__call__	Alias for :meth:get_value.
log_results_for_current_bin	Write results into every shot entry for the current bin.
compute_objective	Compute the scalar objective from mean-aggregated per-shot scalars.
compute_objective_from_shots	Compute the objective from a list of per-shot scalar dicts.
compute_observables	Return auxiliary scalar observables.
compute_observables_from_shots	Return auxiliary observables from the per-shot scalar list.

Source code in geecs_scanner/optimization/base_evaluator.py

def __init__(
    self,
    analyzers: Optional[List[Union[str, Dict[str, Any]]]] = None,
    scalars: Optional[List[str]] = None,
    objective_tag: Optional[str] = None,
    device_requirements: Optional[Dict[str, Any]] = None,
    scan_data_manager: Optional["ScanDataManager"] = None,
    data_logger: Optional["DataLogger"] = None,
):
    # --- Data sources ---------------------------------------------
    # Load each diagnostic into a typed config; build one scan
    # analyzer per diagnostic, keyed on the GEECS device name. Both
    # are stored so subclasses can introspect (e.g. for primary_device).
    self.diagnostics = []
    for entry in analyzers or []:
        name, overrides = _split_analyzer_entry(entry)
        self.diagnostics.append(load_diagnostic(name, overrides=overrides))

    self.scan_analyzers: Dict = {
        diag.name: create_scan_analyzer(diag, use_injected_data=True)
        for diag in self.diagnostics
    }

    # s-file scalar keys are plain column names from current_data_bin.
    # No analyzer needs to run for these.
    self.scalar_keys: List[str] = list(scalars or [])

    # --- Device requirements --------------------------------------
    # Auto-generate from analyzers if not supplied. s-file scalars
    # are assumed to already be in the DataLogger frame (configured
    # via the scanner's save-element layer), so they don't extend
    # device_requirements here.
    if device_requirements is None:
        devices = {
            diag.name: dict(_ANALYZER_DEVICE_REQUIREMENT_TEMPLATE)
            for diag in self.diagnostics
        }
        device_requirements = {"Devices": devices}
    self.device_requirements = device_requirements

    # --- Injected runtime context ---------------------------------
    self.scan_data_manager = scan_data_manager
    self.data_logger = data_logger
    self.scan_tag = (
        self.scan_data_manager.scan_paths.get_tag()
        if self.scan_data_manager is not None
        else None
    )

    # --- Per-evaluation state -------------------------------------
    self.bin_number: int = 0
    self.log_df = None  # pd.DataFrame
    self.current_data_bin = None  # filtered to current bin
    self.current_shot_numbers: Optional[List[int]] = None

    # objective_tag resolution order:
    # 1. ``objective_tag`` kwarg (typically set from YAML kwargs)
    # 2. class attribute (``objective_tag = "BeamSize"`` on the subclass)
    # 3. fall back to the subclass name
    if objective_tag is not None:
        self.objective_tag = objective_tag
    elif not self.objective_tag:
        self.objective_tag = self.__class__.__name__

Attributes

output_key class-attribute instance-attribute

output_key: Optional[str] = 'f'

objective_tag class-attribute instance-attribute

objective_tag: str = ''

diagnostics instance-attribute

diagnostics = []

scan_analyzers instance-attribute

scan_analyzers: Dict = {(name): (create_scan_analyzer(diag, use_injected_data=True)) for diag in (diagnostics)}

scalar_keys instance-attribute

scalar_keys: List[str] = list(scalars or [])

device_requirements instance-attribute

device_requirements = device_requirements

scan_data_manager instance-attribute

scan_data_manager = scan_data_manager

data_logger instance-attribute

data_logger = data_logger

scan_tag instance-attribute

scan_tag = get_tag() if scan_data_manager is not None else None

bin_number instance-attribute

bin_number: int = 0

log_df instance-attribute

log_df = None

current_data_bin instance-attribute

current_data_bin = None

current_shot_numbers instance-attribute

current_shot_numbers: Optional[List[int]] = None

primary_device property

primary_device: Optional[str]

GEECS device name of the first listed diagnostic, or None.

Convenience for subclasses that want to reference "the" device when there's just one analyzer (the common case).

Functions

get_current_data

get_current_data() -> None

Refresh current_data_bin and current_shot_numbers from data_logger.

Converts log_entries to a DataFrame (sorted by elapsed time so Shotnumber reflects acquisition order), then filters to the current bin.

Source code in geecs_scanner/optimization/base_evaluator.py

def get_current_data(self) -> None:
    """Refresh ``current_data_bin`` and ``current_shot_numbers`` from data_logger.

    Converts ``log_entries`` to a DataFrame (sorted by elapsed time so
    ``Shotnumber`` reflects acquisition order), then filters to the
    current bin.
    """
    import pandas as pd

    log_entries = self.data_logger.log_entries
    self.bin_number = self.data_logger.bin_num

    df = pd.DataFrame.from_dict(log_entries, orient="index")
    df = df.sort_values(by="Elapsed Time").reset_index(drop=True)
    df["Shotnumber"] = df.index + 1
    self.log_df = df

    bin_mask = df["Bin #"] == self.bin_number
    self.current_data_bin = df[bin_mask].copy()
    self.current_shot_numbers = df.loc[bin_mask, "Shotnumber"].tolist()

filter_log_entries_by_bin staticmethod

filter_log_entries_by_bin(log_entries: Dict[float, Dict[str, Any]], bin_num: int) -> List[Dict[str, Any]]

Return all log entries belonging to bin_num.

Source code in geecs_scanner/optimization/base_evaluator.py

@staticmethod
def filter_log_entries_by_bin(
    log_entries: Dict[float, Dict[str, Any]], bin_num: int
) -> List[Dict[str, Any]]:
    """Return all log entries belonging to *bin_num*."""
    return [
        entry for entry in log_entries.values() if entry.get("Bin #") == bin_num
    ]

get_value

get_value(input_data: Dict) -> Dict[str, float]

Refresh data, evaluate, normalise types, log, and return results.

Source code in geecs_scanner/optimization/base_evaluator.py

def get_value(self, input_data: Dict) -> Dict[str, float]:
    """Refresh data, evaluate, normalise types, log, and return results."""
    self.get_current_data()

    results = self._get_value(input_data=input_data)
    if not isinstance(results, dict):
        raise TypeError(
            f"{self.__class__.__name__}._get_value must return Dict[str, float]; "
            f"got {type(results)}"
        )

    results = {str(k): float(v) for k, v in results.items()}

    if self.output_key is not None and self.output_key not in results:
        raise KeyError(
            f"{self.__class__.__name__} requires objective key '{self.output_key}' "
            "in results, or set output_key = None for observables-only evaluators."
        )

    self.log_results_for_current_bin(results)
    return results

__call__

__call__(input_data: Dict) -> Dict

Alias for :meth:get_value.

Source code in geecs_scanner/optimization/base_evaluator.py

def __call__(self, input_data: Dict) -> Dict:
    """Alias for :meth:`get_value`."""
    return self.get_value(input_data)

log_results_for_current_bin

log_results_for_current_bin(results: Dict[str, float]) -> None

Write results into every shot entry for the current bin.

Source code in geecs_scanner/optimization/base_evaluator.py

def log_results_for_current_bin(self, results: Dict[str, float]) -> None:
    """Write *results* into every shot entry for the current bin."""
    if not self.current_shot_numbers:
        logger.warning("No shots found for current bin %s", self.bin_number)
        return
    for shot_num in self.current_shot_numbers:
        self._log_results_for_shot(shot_num, results)

compute_objective

compute_objective(scalars: Dict[str, float], bin_number: int) -> Optional[float]

Compute the scalar objective from mean-aggregated per-shot scalars.

Override this for simple evaluators where mean aggregation is enough. For full per-shot control (median, percentile, shot-level filtering), override :meth:compute_objective_from_shots instead.

scalars is a flat dict with analyzer outputs prefixed by device name ("UC_TopView_x_fwhm") and s-file columns as their natural names ("U_Laser:Energy").

Returns None by default — signals "this evaluator has no objective" (BAX mode). Subclasses with an objective must override this OR :meth:compute_objective_from_shots.

Source code in geecs_scanner/optimization/base_evaluator.py

def compute_objective(
    self, scalars: Dict[str, float], bin_number: int
) -> Optional[float]:
    """Compute the scalar objective from mean-aggregated per-shot scalars.

    Override this for simple evaluators where mean aggregation is enough.
    For full per-shot control (median, percentile, shot-level filtering),
    override :meth:`compute_objective_from_shots` instead.

    ``scalars`` is a flat dict with analyzer outputs prefixed by device
    name (``"UC_TopView_x_fwhm"``) and s-file columns as their natural
    names (``"U_Laser:Energy"``).

    Returns ``None`` by default — signals "this evaluator has no
    objective" (BAX mode). Subclasses with an objective must override
    this OR :meth:`compute_objective_from_shots`.
    """
    return None

compute_objective_from_shots

compute_objective_from_shots(scalars_list: List[Dict[str, float]], bin_number: int) -> Union[float, Dict[str, float], None]

Compute the objective from a list of per-shot scalar dicts.

The default mean-aggregates and delegates to :meth:compute_objective. Override for custom statistics::

def compute_objective_from_shots(self, scalars_list, bin_number):
    vals = [d["UC_TopView_x_fwhm"] for d in scalars_list]
    return float(np.median(vals))

Returns None to signal no objective (BAX mode), a float, or a dict that includes at least self.output_key plus any extras (e.g. f_noise) to pass through to Xopt.

Source code in geecs_scanner/optimization/base_evaluator.py

def compute_objective_from_shots(
    self,
    scalars_list: List[Dict[str, float]],
    bin_number: int,
) -> Union[float, Dict[str, float], None]:
    """Compute the objective from a list of per-shot scalar dicts.

    The default mean-aggregates and delegates to :meth:`compute_objective`.
    Override for custom statistics::

        def compute_objective_from_shots(self, scalars_list, bin_number):
            vals = [d["UC_TopView_x_fwhm"] for d in scalars_list]
            return float(np.median(vals))

    Returns ``None`` to signal no objective (BAX mode), a ``float``, or
    a dict that includes at least ``self.output_key`` plus any extras
    (e.g. ``f_noise``) to pass through to Xopt.
    """
    if not scalars_list:
        logger.warning(
            "compute_objective_from_shots received empty list for bin %s",
            bin_number,
        )
        return None
    aggregated = _mean_aggregate(scalars_list)
    return self.compute_objective(scalars=aggregated, bin_number=bin_number)

compute_observables

compute_observables(scalars: Dict[str, float], bin_number: int) -> Dict[str, float]

Return auxiliary scalar observables.

Override this for simple observables built from mean-aggregated per-shot scalars. For full per-shot control, override :meth:compute_observables_from_shots instead. Same scalars namespace as :meth:compute_objective. Default returns {}.

Required for BAX evaluators (they have no objective; observables are what Xopt models).

Source code in geecs_scanner/optimization/base_evaluator.py

def compute_observables(
    self, scalars: Dict[str, float], bin_number: int
) -> Dict[str, float]:
    """Return auxiliary scalar observables.

    Override this for simple observables built from mean-aggregated
    per-shot scalars. For full per-shot control, override
    :meth:`compute_observables_from_shots` instead. Same ``scalars``
    namespace as :meth:`compute_objective`. Default returns ``{}``.

    Required for BAX evaluators (they have no objective; observables are
    what Xopt models).
    """
    return {}

compute_observables_from_shots

compute_observables_from_shots(scalars_list: List[Dict[str, float]], bin_number: int) -> Dict[str, float]

Return auxiliary observables from the per-shot scalar list.

The default mean-aggregates and delegates to :meth:compute_observables. Override for per-shot statistics or shot-level filtering on observables — same shape as :meth:compute_objective_from_shots, the observables peer.

Source code in geecs_scanner/optimization/base_evaluator.py

def compute_observables_from_shots(
    self,
    scalars_list: List[Dict[str, float]],
    bin_number: int,
) -> Dict[str, float]:
    """Return auxiliary observables from the per-shot scalar list.

    The default mean-aggregates and delegates to
    :meth:`compute_observables`. Override for per-shot statistics or
    shot-level filtering on observables — same shape as
    :meth:`compute_objective_from_shots`, the observables peer.
    """
    if not scalars_list:
        return {}
    aggregated = _mean_aggregate(scalars_list)
    return self.compute_observables(scalars=aggregated, bin_number=bin_number)