MCAP Handler

McapHandler

Bases: Handler

A handler class which writes log messages to MCAP files using ProtoBuf serialization.

Source code in mcap_logger/mcap_handler.py

class McapHandler(logging.Handler):
    """
    A handler class which writes log messages
    to MCAP files using ProtoBuf serialization.
    """

    def __init__(self, file: Path) -> None:
        """
        Open the specified file and use it as the stream for logging.

        If the file already exists, it will be overwritten.

        If the parent directory does not exist, it will be created.

        Args:
            file: The MCAP file to store the logs.
        """
        super().__init__()
        self._file: Path | TextIOWrapper = file
        self.writer: None | Writer = None
        self._open()

    def emit(self, record: logging.LogRecord) -> None:
        """
        Emit a record.

        The record will be emitted under the `/log` topic.

        Args:
            record: The record to emit into the log file.
        """
        time_sec, time_ns = _split_time(record.created)

        log_message = Log(
            timestamp=Timestamp(nanos=time_ns, seconds=time_sec),
            level=record.levelname,
            message=record.getMessage(),
            name=record.module,
            file=record.filename,
            line=record.lineno,
        )

        self.writer.write_message(
            topic="/log",
            message=log_message,
            log_time=int(record.created * 1_000_000_000),
            publish_time=time.time_ns(),
        )

    def close(self) -> None:
        """
        Close the file.
        """
        self.writer.finish()
        self._file.close()
        super().close()

    def _open(self) -> None:
        """
        Open the log file and initialize the ProtoBuf writer.

        If the parent directory does not exist, it will be created.
        """
        if self.writer is None:
            self._file.parent.mkdir(parents=True, exist_ok=True)
            self._file = self._file.open("wb")
            self.writer = Writer(self._file)

__init__(file)

Open the specified file and use it as the stream for logging.

If the file already exists, it will be overwritten.

If the parent directory does not exist, it will be created.

Parameters:

Name	Type	Description	Default
file	Path	The MCAP file to store the logs.	required

Source code in mcap_logger/mcap_handler.py

def __init__(self, file: Path) -> None:
    """
    Open the specified file and use it as the stream for logging.

    If the file already exists, it will be overwritten.

    If the parent directory does not exist, it will be created.

    Args:
        file: The MCAP file to store the logs.
    """
    super().__init__()
    self._file: Path | TextIOWrapper = file
    self.writer: None | Writer = None
    self._open()

close()

Close the file.

Source code in mcap_logger/mcap_handler.py

def close(self) -> None:
    """
    Close the file.
    """
    self.writer.finish()
    self._file.close()
    super().close()

emit(record)

Emit a record.

The record will be emitted under the /log topic.

Parameters:

Name	Type	Description	Default
record	LogRecord	The record to emit into the log file.	required

Source code in mcap_logger/mcap_handler.py

def emit(self, record: logging.LogRecord) -> None:
    """
    Emit a record.

    The record will be emitted under the `/log` topic.

    Args:
        record: The record to emit into the log file.
    """
    time_sec, time_ns = _split_time(record.created)

    log_message = Log(
        timestamp=Timestamp(nanos=time_ns, seconds=time_sec),
        level=record.levelname,
        message=record.getMessage(),
        name=record.module,
        file=record.filename,
        line=record.lineno,
    )

    self.writer.write_message(
        topic="/log",
        message=log_message,
        log_time=int(record.created * 1_000_000_000),
        publish_time=time.time_ns(),
    )