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)