pyfebiopt.optimize.reporting

Reporting helpers for optimization runs.

Classes

`Reporter`	Lightweight interface for emitting run lifecycle events.
`ReporterFactoryInput`	Inputs required to wire reporting transports.
`NullReporter`	No-op reporter used when monitoring/logging are disabled.
`CompositeReporter`	Fan-out reporter to keep console and monitor sinks independent.
`ConsoleReporter`	Console-only reporter that mirrors previous logging behavior.
`MonitorReporter`	Forward lifecycle events to the monitoring service when available.

Functions

build_reporter(→ Reporter)

Create and configure console/monitor reporters.

Module Contents

class pyfebiopt.optimize.reporting.Reporter

Bases: Protocol

Lightweight interface for emitting run lifecycle events.

run_started(phi0_vec: Array, theta0_vec: Array, bounds: pyfebiopt.optimize.parameters.BoundsPayload, optimizer_name: str, runner_jobs: int | None) → None: Called once before optimization begins.

record_iteration(index: int, phi_vec: Array, theta_vec: Array, cost: float, metrics: collections.abc.Mapping[str, Any], series: collections.abc.Mapping[str, dict[str, Any]], log_output: bool) → None: Called after each evaluation to log/monitor progress.

completed(phi_vec: Array | None, theta_opt: collections.abc.Mapping[str, float], optimizer_meta: collections.abc.Mapping[str, Any], metrics: collections.abc.Mapping[str, Any]) → None: Called when optimization finishes successfully.

failed(reason: str) → None: Called when optimization terminates with an error.

close() → None: Release any resources held by the reporter.

class pyfebiopt.optimize.reporting.ReporterFactoryInput

Inputs required to wire reporting transports.

logger: LoggerProtocol

parameter_space: pyfebiopt.optimize.parameters.ParameterSpace

options: pyfebiopt.optimize.options.EngineOptions

workspace: pyfebiopt.optimize.storage.StorageWorkspace

case_descriptions: list[collections.abc.Mapping[str, Any]]

runner_command: tuple[str, Ellipsis]

runner_env: collections.abc.Mapping[str, str] | None

optimizer_adapter: str

reparam_enabled: bool

pyfebiopt.optimize.reporting.build_reporter(config: ReporterFactoryInput) → Reporter

Create and configure console/monitor reporters.

Returns:: Reporter fan-out used by the engine.

class pyfebiopt.optimize.reporting.NullReporter

No-op reporter used when monitoring/logging are disabled.

run_started(_phi0_vec: Array, _theta0_vec: Array, _bounds: pyfebiopt.optimize.parameters.BoundsPayload, _optimizer_name: str, _runner_jobs: int | None) → None: Ignore run start event.

record_iteration(_index: int, _phi_vec: Array, _theta_vec: Array, _cost: float, _metrics: collections.abc.Mapping[str, Any], _series: collections.abc.Mapping[str, dict[str, Any]], _log_output: bool) → None: Ignore iteration event.

completed(_phi_vec: Array | None, _theta_opt: collections.abc.Mapping[str, float], _optimizer_meta: collections.abc.Mapping[str, Any], _metrics: collections.abc.Mapping[str, Any]) → None: Ignore completion event.

failed(_reason: str) → None: Ignore failure event.

close() → None: No resources to release.

class pyfebiopt.optimize.reporting.CompositeReporter(reporters: collections.abc.Iterable[Reporter], *, logger: LoggerProtocol | None = None)

Fan-out reporter to keep console and monitor sinks independent.

Collect multiple reporters and optionally log failures.

run_started(phi0_vec: Array, theta0_vec: Array, bounds: pyfebiopt.optimize.parameters.BoundsPayload, optimizer_name: str, runner_jobs: int | None) → None: Relay run start to all reporters.

record_iteration(index: int, phi_vec: Array, theta_vec: Array, cost: float, metrics: collections.abc.Mapping[str, Any], series: collections.abc.Mapping[str, dict[str, Any]], log_output: bool) → None: Relay iteration data to all reporters.

completed(phi_vec: Array | None, theta_opt: collections.abc.Mapping[str, float], optimizer_meta: collections.abc.Mapping[str, Any], metrics: collections.abc.Mapping[str, Any]) → None: Relay completion event.

failed(reason: str) → None: Relay failure event.

close() → None: Close all reporters.

class pyfebiopt.optimize.reporting.ConsoleReporter(logger: LoggerProtocol, parameter_space: pyfebiopt.optimize.parameters.ParameterSpace, case_descriptions: list[collections.abc.Mapping[str, Any]], workspace: pyfebiopt.optimize.storage.StorageWorkspace, *, reparam_enabled: bool, log_optimizer_space: bool = False)

Console-only reporter that mirrors previous logging behavior.

Prepare console reporter with configuration context.

logger

parameter_space

case_descriptions

workspace

reparam_enabled

log_optimizer_space = False

log_banner() → None: Print an ASCII banner on startup.

log_configuration(*, options: pyfebiopt.optimize.options.EngineOptions, runner_command: tuple[str, Ellipsis], runner_env: collections.abc.Mapping[str, str] | None, optimizer_adapter: str) → None: Log a structured summary of the optimization setup.

run_started(phi0_vec: Array, theta0_vec: Array, bounds: pyfebiopt.optimize.parameters.BoundsPayload, optimizer_name: str, runner_jobs: int | None) → None: Report initial configuration once optimization begins.

record_iteration(index: int, phi_vec: Array, theta_vec: Array, cost: float, metrics: collections.abc.Mapping[str, Any], series: collections.abc.Mapping[str, dict[str, Any]], log_output: bool) → None: Log iteration summary to the console.

completed(phi_vec: Array | None, theta_opt: collections.abc.Mapping[str, float], optimizer_meta: collections.abc.Mapping[str, Any], metrics: collections.abc.Mapping[str, Any]) → None: Log final parameter table and metric summary.

failed(reason: str) → None: Log a failure message.

close() → None: No resources to release for console logging.

class pyfebiopt.optimize.reporting.MonitorReporter(monitor_opts: pyfebiopt.optimize.options.MonitorOptions, parameter_space: pyfebiopt.optimize.parameters.ParameterSpace, workspace: pyfebiopt.optimize.storage.StorageWorkspace, case_descriptions: list[collections.abc.Mapping[str, Any]], logger: LoggerProtocol | None = None)

Forward lifecycle events to the monitoring service when available.

Initialize the monitoring client if enabled.

parameter_space

workspace

case_descriptions

logger

property enabled: bool: True when the monitoring client is active.

run_started(phi0_vec: Array, theta0_vec: Array, bounds: pyfebiopt.optimize.parameters.BoundsPayload, optimizer_name: str, runner_jobs: int | None) → None: Emit a run start event to the monitoring service.

record_iteration(index: int, phi_vec: Array, theta_vec: Array, cost: float, metrics: collections.abc.Mapping[str, Any], series: collections.abc.Mapping[str, dict[str, Any]], log_output: bool) → None: Send iteration payload to the monitoring service.

completed(phi_vec: Array | None, theta_opt: collections.abc.Mapping[str, float], optimizer_meta: collections.abc.Mapping[str, Any], metrics: collections.abc.Mapping[str, Any]) → None: Send completion summary to the monitoring service.

failed(reason: str) → None: Notify the monitoring service of a failure.

close() → None: Detach the monitoring client.