Tracer API

This module contains the centralized logging utility for the simulation engine.

class ml.tracer.LocalLogRecord(level, component_id, event, details, tick, event_id, thread_name, process_name, local_event_id)

Bases: tuple

component_id: Alias for field number 1

details: Alias for field number 3

event: Alias for field number 2

event_id: Alias for field number 5

level: Alias for field number 0

local_event_id: Alias for field number 8

process_name: Alias for field number 7

thread_name: Alias for field number 6

tick: Alias for field number 4

class ml.tracer.LogRecord(timestamp, thread_name, component_id, event, details, level)

Bases: tuple

component_id: Alias for field number 2

details: Alias for field number 4

event: Alias for field number 3

level: Alias for field number 5

thread_name: Alias for field number 1

timestamp: Alias for field number 0

class ml.tracer.TimedEvent(timestamp, component_id, event_type, thread_name, details, tick)

Bases: object

Represents a timed event with a start, end, and children.

complete(end_time, end_tick): Marks the event as complete and calculates its duration.

class ml.tracer.Tracer

Bases: object

A thread-safe, centralized logging utility for the simulation engine.

It buffers log records and periodically flushes them to the console and/or a file in a formatted, column-aligned manner.

classmethod configure_for_worker(level: LogLevel, log_queue: Queue)

Configures the Tracer instance within a worker process to send logs to the shared multiprocessing.Queue. This should be called once at the start of each worker process.

Parameters:

level – The minimum log level to record.
log_queue – The multiprocessing queue to send logs to.

classmethod get_current_event_id() → int | None

Retrieves the current event ID from the thread-local context.

Returns:: The current parent event ID, or None if not set.

classmethod get_external_queues() → Dict[str, Queue | None]

Returns the multiprocessing queues used by the tracer. This is used by the engine to pass queues to worker processes.

Returns:: A dictionary containing the log and error queues.

classmethod log(level: LogLevel, component_id: str, event: str, details: Dict[str, Any] | None = None, tick: int | None = None, event_id: int | None = None)

Adds a log record to the buffer.

This method is thread-safe.

Parameters:

level – The severity level of the log.
component_id – The identifier of the component logging the event.
event – The name of the event being logged.
details – An optional dictionary of structured data. If a string is passed, it will be wrapped in a ‘message’ key.
tick – The optional simulation tick number.
event_id – An optional ID to link this event to a previous START event. If provided, a new ID will not be generated.

Returns:

A dictionary containing the log entry details (including timestamp and event_id) if successful, or None if the log level is too low.

classmethod set_parent_event_id(parent_event_id: int | None)

Sets the parent event ID in a thread-local context. This is used to link events across different execution contexts, especially between processes.

Parameters:: parent_event_id – The ID of the parent event.

classmethod start(level: LogLevel, flush_interval_seconds: float, output_file: str | None, error_file: str | None, log_to_console: bool = True, log_queue: Queue | None = None, error_queue: Queue | None = None)

Initializes and starts the tracer thread.

Parameters:

level – The minimum log level to record.
flush_interval_seconds – How often to flush logs.
output_file – An optional file path to write logs to.
error_file – An optional file path to write detailed exception tracebacks to.
log_to_console – If True, logs will also be printed to the console.
log_queue – An optional multiprocessing.Queue. If provided, the tracer will read from this queue (main process only).
error_queue – An optional multiprocessing.Queue for receiving errors from worker processes.

classmethod stop(): Stops the tracer thread and ensures all logs are flushed.

ml.tracer.analyze_trace_log(input_file: str, output_format: str = 'text', output_file: str | None = None, title: str = 'Schedule')

Main analysis function. Parses the log and prints a performance report.

Parameters:

input_file – Path to the simulation trace log file.
output_format – The desired output format. Can be ‘text’, ‘json’, or ‘json:perfetto’.
output_file – Optional path to write the output to. If None, prints to stdout.
title – The title for the Perfetto trace (used if output_format is ‘json:perfetto’).

ml.tracer.merge_trace_logs(input_file_1: str, input_file_2: str, output_file: str)

Merges two Perfetto JSON trace files into a single file with two distinct processes.

Process 1 (from file 1) is assigned PID 1. Process 2 (from file 2) is assigned PID 2.

Parameters:

input_file_1 – Path to the first input trace file.
input_file_2 – Path to the second input trace file.
output_file – Path to the output merged trace file.