babylon.utils

Utility modules for Babylon/Babylon.

These are the tools of production - the means by which the engine transforms inputs into outputs.

exception babylon.utils.BabylonError(message, error_code=None, details=None)[source]

Bases: Exception

Base exception for all Babylon/Babylon errors.

All exceptions in the system inherit from this class, allowing for unified error handling at the boundary.

Parameters:

message (str)
error_code (str | None)
details (dict[str, object] | None)

Return type:

None

message: Human-readable error description

error_code: Machine-readable error identifier

details: Additional context for debugging

__init__(message, error_code=None, details=None)[source]

Parameters:

message (str)
error_code (str | None)
details (dict[str, object] | None)

Return type:

None

log(logger, level=logging.ERROR, exc_info=True)[source]

Log this exception with full structured context.

Convenience method for logging exceptions with their structured data included in the log record’s extra fields.

Parameters:

logger (Logger) – The logger instance to use.
level (int) – Log level (default: ERROR).
exc_info (bool) – Whether to include exception info (default: True).

Return type:

None

Example

try:: risky_operation()
except BabylonError as e:: e.log(logger) # Logs with ERROR level and full context

to_dict()[source]

Serialize exception for logging/API responses.

Return type:: dict[str, object]

exception babylon.utils.ConfigurationError(message, error_code=None, details=None)[source]

Bases: ValidationError

Raised when configuration is invalid or missing.

The material conditions for operation have not been established.

Parameters:

message (str)
error_code (str | None)
details (dict[str, object] | None)

Return type:

None

__init__(message, error_code=None, details=None)[source]

Parameters:

message (str)
error_code (str | None)
details (dict[str, object] | None)

Return type:

None

exception babylon.utils.DatabaseError(message, error_code=None, details=None)[source]

Bases: InfrastructureError

Raised when database operations fail.

The Ledger cannot record the material reality.

Parameters:

message (str)
error_code (str | None)
details (dict[str, object] | None)

Return type:

None

__init__(message, error_code=None, details=None)[source]

Parameters:

message (str)
error_code (str | None)
details (dict[str, object] | None)

Return type:

None

exception babylon.utils.EmbeddingError(message, error_code=None, details=None)[source]

Bases: BabylonError

DEPRECATED: Use RagError instead.

Raised when embedding operations fail. The Archive cannot encode semantic meaning.

Parameters:

message (str)
error_code (str | None)
details (dict[str, object] | None)

Return type:

None

__init__(message, error_code=None, details=None)[source]

Parameters:

message (str)
error_code (str | None)
details (dict[str, object] | None)

Return type:

None

exception babylon.utils.ValidationError(message, error_code=None, details=None)[source]

Bases: BabylonError

Raised when data validation fails.

The input does not conform to the schema of material reality.

Parameters:

message (str)
error_code (str | None)
details (dict[str, object] | None)

Return type:

None

__init__(message, error_code=None, details=None)[source]

Parameters:

message (str)
error_code (str | None)
details (dict[str, object] | None)

Return type:

None

babylon.utils.retry_on_exception(max_retries=3, delay=1.0, backoff=2.0, exceptions=Exception, logger=None)[source]

Decorator that retries a function on specified exceptions.

Uses exponential backoff to avoid hammering failing resources. This is a materialist approach: we respect the physical constraints of the systems we depend on.

Parameters:

max_retries (int) – Maximum number of retry attempts
delay (float) – Initial delay between retries (seconds)
backoff (float) – Multiplier applied to delay after each retry
exceptions (type[Exception] | tuple[type[Exception], ...]) – Exception types that trigger a retry
logger (Logger | None) – Logger instance for retry messages

Return type:

Callable[[TypeVar(F, bound= Callable[..., Any])], TypeVar(F, bound= Callable[..., Any])]

Returns:

Decorated function with retry logic

Example

@retry_on_exception(max_retries=3, delay=1.0, exceptions=(ConnectionError,)) def fetch_data():

return requests.get(”https://api.example.com”)

babylon.utils.quantize(value)[source]

Snap float to precision grid (ROUND_HALF_UP).

Parameters:: value (float) – Float to quantize (None treated as 0.0).
Return type:: float
Returns:: Value on 10^-precision grid.

Examples

>>> quantize(0.123456789)
0.123457
>>> quantize(-0.123456789)
-0.123457

babylon.utils.get_precision()[source]

Current decimal places for quantization (default 6).

Return type:: int

babylon.utils.set_precision(decimal_places)[source]

Set quantization precision (1-10 decimal places).

Parameters:: decimal_places (int) – 1=coarse (0.1), 6=default (0.000001), 10=ultra.
Raises:: ValueError – If not in range [1, 10].
Return type:: None

class babylon.utils.SessionRecorder(metrics_collector, narrative_director=None, base_dir=None)[source]

Bases: object

Black box recorder for simulation forensics.

Implements SimulationObserver protocol to persist tick-by-tick simulation state to JSONL files for debugging and replay.

Each session creates a unique timestamped directory containing: - metrics.jsonl: TickMetrics snapshots (one per line) - events.jsonl: SimulationEvents (one per line) - narrative.jsonl: Narrative entries with tick numbers - summary.json: Session metadata on completion

Parameters:

metrics_collector (TickStateRecorder) – TickStateRecorder observer to extract TickMetrics.
narrative_director (NarrativeDirector | None) – Optional NarrativeDirector for narrative log.
base_dir (Path | None) – Base directory for sessions (default: logs/sessions).

Example

>>> from babylon.utils.recorder import SessionRecorder
>>> recorder = SessionRecorder(metrics_collector=collector)
>>> # Recorder is registered as observer with Simulation
>>> # Files written automatically during simulation
>>> zip_path = recorder.export_package()  # Create debug archive

__init__(metrics_collector, narrative_director=None, base_dir=None)[source]

Initialize SessionRecorder with observer references.

Parameters:

metrics_collector (TickStateRecorder)
narrative_director (NarrativeDirector | None)
base_dir (Path | None)

Return type:

None

export_package()[source]

Create ZIP archive of session files.

Flushes all buffers and creates a compressed archive containing all session JSONL files for sharing and debugging.

Return type:: Path
Returns:: Path to the created ZIP file.
Raises:: RuntimeError – If no session has been started.

property name: str: Observer identifier for logging and debugging.

on_simulation_end(final_state)[source]

Close file handles and write summary.

Parameters:: final_state (WorldState) – Final WorldState when simulation ends.
Return type:: None

on_simulation_start(_initial_state, _config)[source]

Create session directory and open file handles.

Parameters:

_initial_state (WorldState) – WorldState at tick 0 (unused, required by protocol).
_config (SimulationConfig) – SimulationConfig for this run (unused, required by protocol).

Return type:

None

on_tick(_previous_state, new_state)[source]

Record tick data to JSONL files.

Parameters:

_previous_state (WorldState) – WorldState before tick (unused, required by protocol).
new_state (WorldState) – WorldState after tick (contains events).

Return type:

None

Modules

`exceptions`	Exception hierarchy for Babylon/Babylon.
`log`	Custom logging utilities for Babylon.
`math`	Mathematical utilities for simulation precision.
`recorder`	Black box recorder for simulation forensics.
`retry`	Retry decorator for transient failure handling.