From 4550bdb7394e6660a2cfb0db910add6835031efd Mon Sep 17 00:00:00 2001 From: Marcin Kulik Date: Sat, 14 May 2022 18:10:15 +0200 Subject: [PATCH] Add `--out-fmt` option to `play` command --- README.md | 10 +++++- asciinema/__main__.py | 6 ++++ asciinema/asciicast/v1.py | 2 +- asciinema/asciicast/v2.py | 4 ++- asciinema/commands/play.py | 2 ++ asciinema/player.py | 68 ++++++++++++++++++++++++++++++++++---- 6 files changed, 82 insertions(+), 10 deletions(-) diff --git a/README.md b/README.md index 10c8593..aec9d6b 100644 --- a/README.md +++ b/README.md @@ -293,12 +293,20 @@ Available options: - `-i, --idle-time-limit=` - Limit replayed terminal inactivity to max `` seconds - `-s, --speed=` - Playback speed (can be fractional) -- `--stream=` - Recorded stream to play (see below) +- `--stream=` - Select stream to play (see below) +- `--out-fmt=` - Select output format (see below) By default the output stream (`o`) is played. This is what you want in most cases. If you recorded the input stream (`i`) with `asciinema rec --stdin` then you can replay it with `asciinema play --stream=i `. +By default the selected stream is written to stdout in original, raw data form. +This is also what you want in majority of cases. However you can change the +output format to asciicast (newline delimited JSON) with `asciinema play +--out-fmt=asciicast `. This allows delegating actual rendering to +another place (e.g. outside of your terminal) by piping output of `asciinema +play` to a tool of your choice. + > For the best playback experience it is recommended to run `asciinema play` in > a terminal of dimensions not smaller than the one used for recording, as > there's no "transcoding" of control sequences for new terminal size. diff --git a/asciinema/__main__.py b/asciinema/__main__.py index 51b2a78..61b90a1 100644 --- a/asciinema/__main__.py +++ b/asciinema/__main__.py @@ -178,6 +178,12 @@ For help on a specific command run: type=positive_float, default=cfg.play_speed, ) + parser_play.add_argument( + "--out-fmt", + help="output format", + choices=["raw", "asciicast"], + default="raw", + ) parser_play.add_argument( "--stream", help="recorded stream to play (o, i)", diff --git a/asciinema/asciicast/v1.py b/asciinema/asciicast/v1.py index 07db4f8..d188c12 100644 --- a/asciinema/asciicast/v1.py +++ b/asciinema/asciicast/v1.py @@ -35,7 +35,7 @@ class Asciicast: } return header - def events(self, type_: Optional[str]) -> Iterable[List[Any]]: + def events(self, type_: Optional[str] = None) -> Iterable[List[Any]]: if type_ in [None, "o"]: return to_absolute_time(self.__stdout_events()) else: diff --git a/asciinema/asciicast/v2.py b/asciinema/asciicast/v2.py index c97ab01..4113c91 100644 --- a/asciinema/asciicast/v2.py +++ b/asciinema/asciicast/v2.py @@ -31,7 +31,9 @@ class Asciicast: self.v2_header = header self.idle_time_limit = header.get("idle_time_limit") - def events(self, type_: Optional[str]) -> Generator[List[Any], None, None]: + def events( + self, type_: Optional[str] = None + ) -> Generator[List[Any], None, None]: if type_ is None: for line in self.__file: yield json.loads(line) diff --git a/asciinema/commands/play.py b/asciinema/commands/play.py index 4c96c73..68466e8 100644 --- a/asciinema/commands/play.py +++ b/asciinema/commands/play.py @@ -18,6 +18,7 @@ class PlayCommand(Command): self.filename = args.filename self.idle_time_limit = args.idle_time_limit self.speed = args.speed + self.out_fmt = args.out_fmt self.stream = args.stream self.player = player if player is not None else Player() self.key_bindings = { @@ -33,6 +34,7 @@ class PlayCommand(Command): idle_time_limit=self.idle_time_limit, speed=self.speed, key_bindings=self.key_bindings, + out_fmt=self.out_fmt, stream=self.stream, ) diff --git a/asciinema/player.py b/asciinema/player.py index 608f055..a72c810 100644 --- a/asciinema/player.py +++ b/asciinema/player.py @@ -1,3 +1,4 @@ +import json import sys import time from typing import Any, Dict, Optional, TextIO, Union @@ -7,6 +8,44 @@ from .asciicast.v1 import Asciicast as v1 from .asciicast.v2 import Asciicast as v2 from .tty_ import raw, read_blocking +Header = Dict[str, Any] + + +class RawOutput: + def __init__(self, stream: Optional[str]) -> None: + self.stream = stream or "o" + + def start(self, _header: Header) -> None: + pass + + def write(self, _time: float, event_type: str, data: str) -> None: + if event_type == self.stream: + sys.stdout.write(data) + sys.stdout.flush() + + +class AsciicastOutput: + def __init__(self, stream: Optional[str]) -> None: + self.stream = stream + + def start(self, header: Header) -> None: + self.__write_line(header) + + def write(self, time: float, event_type: str, data: str) -> None: + if self.stream in [None, event_type]: + self.__write_line([time, event_type, data]) + + def __write_line(self, obj: Any) -> None: + line = json.dumps( + obj, ensure_ascii=False, indent=None, separators=(", ", ": ") + ) + + sys.stdout.write(f"{line}\r\n") + sys.stdout.flush() + + +Output = Union[RawOutput, AsciicastOutput] + class Player: # pylint: disable=too-few-public-methods def play( @@ -15,10 +54,16 @@ class Player: # pylint: disable=too-few-public-methods idle_time_limit: Optional[int] = None, speed: float = 1.0, key_bindings: Optional[Dict[str, Any]] = None, - stream: str = "o", + out_fmt: str = "raw", + stream: Optional[str] = None, ) -> None: if key_bindings is None: key_bindings = {} + + output: Output = ( + RawOutput(stream) if out_fmt == "raw" else AsciicastOutput(stream) + ) + try: with open("/dev/tty", "rt", encoding="utf-8") as stdin: with raw(stdin.fileno()): @@ -29,10 +74,17 @@ class Player: # pylint: disable=too-few-public-methods stdin, key_bindings, stream, + output, ) except Exception: # pylint: disable=broad-except self._play( - asciicast, idle_time_limit, speed, None, key_bindings, stream + asciicast, + idle_time_limit, + speed, + None, + key_bindings, + stream, + output, ) @staticmethod @@ -42,24 +94,27 @@ class Player: # pylint: disable=too-few-public-methods speed: float, stdin: Optional[TextIO], key_bindings: Dict[str, Any], - stream: str, + stream: Optional[str], + output: Output, ) -> None: idle_time_limit = idle_time_limit or asciicast.idle_time_limit pause_key = key_bindings.get("pause") step_key = key_bindings.get("step") - events = asciicast.events(stream) + events = asciicast.events() events = ev.to_relative_time(events) events = ev.cap_relative_time(events, idle_time_limit) events = ev.to_absolute_time(events) events = ev.adjust_speed(events, speed) + output.start(asciicast.v2_header) + base_time = time.time() ctrl_c = False paused = False pause_time: Optional[float] = None - for t, _type, text in events: + for t, event_type, text in events: delay = t - (time.time() - base_time) while stdin and not ctrl_c and delay > 0: @@ -101,5 +156,4 @@ class Player: # pylint: disable=too-few-public-methods if ctrl_c: break - sys.stdout.write(text) - sys.stdout.flush() + output.write(t, event_type, text)