pyfebiopt.optimize.reporting ============================ .. py:module:: pyfebiopt.optimize.reporting .. autoapi-nested-parse:: Reporting helpers for optimization runs. Classes ------- .. autoapisummary:: pyfebiopt.optimize.reporting.Reporter pyfebiopt.optimize.reporting.ReporterFactoryInput pyfebiopt.optimize.reporting.NullReporter pyfebiopt.optimize.reporting.CompositeReporter pyfebiopt.optimize.reporting.ConsoleReporter pyfebiopt.optimize.reporting.MonitorReporter Functions --------- .. autoapisummary:: pyfebiopt.optimize.reporting.build_reporter Module Contents --------------- .. py:class:: Reporter Bases: :py:obj:`Protocol` Lightweight interface for emitting run lifecycle events. .. py:method:: 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. .. py:method:: 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. .. py:method:: 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. .. py:method:: failed(reason: str) -> None Called when optimization terminates with an error. .. py:method:: close() -> None Release any resources held by the reporter. .. py:class:: ReporterFactoryInput Inputs required to wire reporting transports. .. py:attribute:: logger :type: LoggerProtocol .. py:attribute:: parameter_space :type: pyfebiopt.optimize.parameters.ParameterSpace .. py:attribute:: options :type: pyfebiopt.optimize.options.EngineOptions .. py:attribute:: workspace :type: pyfebiopt.optimize.storage.StorageWorkspace .. py:attribute:: case_descriptions :type: list[collections.abc.Mapping[str, Any]] .. py:attribute:: runner_command :type: tuple[str, Ellipsis] .. py:attribute:: runner_env :type: collections.abc.Mapping[str, str] | None .. py:attribute:: optimizer_adapter :type: str .. py:attribute:: reparam_enabled :type: bool .. py:function:: build_reporter(config: ReporterFactoryInput) -> Reporter Create and configure console/monitor reporters. :returns: Reporter fan-out used by the engine. .. py:class:: NullReporter No-op reporter used when monitoring/logging are disabled. .. py:method:: 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. .. py:method:: 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. .. py:method:: 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. .. py:method:: failed(_reason: str) -> None Ignore failure event. .. py:method:: close() -> None No resources to release. .. py:class:: 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. .. py:method:: 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. .. py:method:: 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. .. py:method:: 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. .. py:method:: failed(reason: str) -> None Relay failure event. .. py:method:: close() -> None Close all reporters. .. py:class:: 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. .. py:attribute:: logger .. py:attribute:: parameter_space .. py:attribute:: case_descriptions .. py:attribute:: workspace .. py:attribute:: reparam_enabled .. py:attribute:: log_optimizer_space :value: False .. py:method:: log_banner() -> None Print an ASCII banner on startup. .. py:method:: 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. .. py:method:: 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. .. py:method:: 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. .. py:method:: 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. .. py:method:: failed(reason: str) -> None Log a failure message. .. py:method:: close() -> None No resources to release for console logging. .. py:class:: 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. .. py:attribute:: parameter_space .. py:attribute:: workspace .. py:attribute:: case_descriptions .. py:attribute:: logger .. py:property:: enabled :type: bool True when the monitoring client is active. .. py:method:: 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. .. py:method:: 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. .. py:method:: 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. .. py:method:: failed(reason: str) -> None Notify the monitoring service of a failure. .. py:method:: close() -> None Detach the monitoring client.