Terminus 2
asciinema_handler.py 103 lines
import jsonfrom pathlib import Pathfrom typing import TextIOclass AsciinemaHandler:    """    Handles operations related to asciinema recordings, including marker management.    """    def __init__(self, markers: list[tuple[float, str]], recording_path: Path):        """        Initialize the AsciinemaHandler.        Args:            markers: List of tuples containing (timestamp, label) for each marker            recording_path: Path to the asciinema recording file        """        self._markers = sorted(markers, key=lambda x: x[0]) if markers else []        self._recording_path = recording_path    def merge_markers(self) -> None:        """        Merge asciinema markers into a recording.        Inserts marker events into an asciinema recording file at specified timestamps.        Markers are added as special events with type 'm' in the recording format.        The original recording is preserved until the merge is successful.        In the future Asciinema might support adding markers via RCP:        https://discourse.asciinema.org/t/add-markers-with-title-from-cli/861        """        if not self._markers or not self._recording_path.exists():            return        # Create a temporary file in the same directory as the recording        temp_path = self._recording_path.with_suffix(".tmp")        self._write_merged_recording(temp_path)        temp_path.replace(self._recording_path)    def _write_merged_recording(self, output_path: Path) -> None:        """        Write a new recording file with markers merged in at the correct timestamps.        """        marker_index = 0        with (            open(self._recording_path) as input_file,            open(output_path, "w") as output_file,        ):            # Preserve header            output_file.write(input_file.readline())            for line in input_file:                marker_index = self._process_recording_line(                    line, output_file, marker_index                )            # Add any remaining markers at the end            self._write_remaining_markers(output_file, self._markers[marker_index:])    def _process_recording_line(        self,        line: str,        output_file: TextIO,        marker_index: int,    ) -> int:        """Process a single line from the recording, inserting markers as needed."""        if not line.startswith("["):            output_file.write(line)            return marker_index        try:            data = json.loads(line)            timestamp = float(data[0])            # Insert any markers that should appear before this timestamp            while (                marker_index < len(self._markers)                and self._markers[marker_index][0] <= timestamp            ):                self._write_marker(output_file, self._markers[marker_index])                marker_index += 1        except (json.JSONDecodeError, ValueError, IndexError):            # If we can't parse the line, preserve it as-is            pass        output_file.write(line)        return marker_index    def _write_marker(self, output_file: TextIO, marker: tuple[float, str]) -> None:        """Write a single marker event to the output file."""        marker_time, marker_label = marker        marker_data = [marker_time, "m", marker_label]        output_file.write(json.dumps(marker_data) + "\n")    def _write_remaining_markers(        self, output_file: TextIO, markers: list[tuple[float, str]]    ) -> None:        """Write any remaining markers that come after all recorded events."""        for marker in markers:            self._write_marker(output_file, marker)