Mirrors/asciinema
1
0
Fork 0
mirror of https://github.com/asciinema/asciinema.git synced 2025-12-15 19:28:00 +01:00
Code Issues Packages Projects Releases Wiki Activity

Add --stream option to play command

Browse Source
This commit is contained in:
Marcin Kulik
2022-05-14 14:12:16 +02:00
parent e81ce0ee49
commit 675b37d535
8 changed files with 67 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
README.md
View File

@@ -236,7 +236,7 @@ Available options:
Stdin recording allows for capturing of all characters typed in by the user in
the currently recorded shell. This may be used by a player (e.g.
[asciinema-player](https://github.com/asciinema/asciinema-player)) to display
pressed keys. Because it's basically a key-logging (scoped to a single shell
pressed keys. Because it's basically key-logging (scoped to a single shell
instance), it's disabled by default, and has to be explicitly enabled via
`--stdin` option.
@@ -293,6 +293,11 @@ Available options:
- `-i, --idle-time-limit=<sec>` - Limit replayed terminal inactivity to max `<sec>` seconds
- `-s, --speed=<factor>` - Playback speed (can be fractional)
- `--stream=<stream>` - Recorded stream to play (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 <filename>`.
> 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

5
asciinema/__main__.py
View File

@@ -178,6 +178,11 @@ For help on a specific command run:
type=positive_float,
default=cfg.play_speed,
)
parser_play.add_argument(
"--stream",
help="recorded stream to play (o, i)",
default="o",
)
parser_play.add_argument(
"filename", help='local path, http/ipfs URL or "-" (read from stdin)'
)

12
asciinema/asciicast/events.py
View File

@@ -1,8 +1,8 @@
from typing import Any, Generator, List, Optional
from typing import Any, Generator, Iterable, List, Optional
def to_relative_time(
events: Generator[List[Any], None, None]
events: Iterable[Any],
) -> Generator[List[Any], None, None]:
prev_time = 0
@@ -14,7 +14,7 @@ def to_relative_time(
def to_absolute_time(
events: Generator[List[Any], None, None]
events: Iterable[Any],
) -> Generator[List[Any], None, None]:
time = 0
@@ -25,8 +25,8 @@ def to_absolute_time(
def cap_relative_time(
events: Generator[List[Any], None, None], time_limit: Optional[float]
) -> Generator[List[Any], None, None]:
events: Iterable[Any], time_limit: Optional[float]
) -> Iterable[Any]:
if time_limit:
return (
[min(delay, time_limit), type_, data]
@@ -36,6 +36,6 @@ def cap_relative_time(
def adjust_speed(
events: Generator[List[Any], None, None], speed: Any
events: Iterable[Any], speed: Any
) -> Generator[List[Any], None, None]:
return ([delay / speed, type_, data] for delay, type_, data in events)

23
asciinema/asciicast/v1.py
View File

@@ -1,7 +1,16 @@
import json
from codecs import StreamReader
from json.decoder import JSONDecodeError
from typing import Any, Dict, Generator, List, Optional, TextIO, Union
from typing import (
Any,
Dict,
Generator,
Iterable,
List,
Optional,
TextIO,
Union,
)
from .events import to_absolute_time
@@ -26,16 +35,16 @@ class Asciicast:
}
return header
def events(self, type_: Optional[str]) -> Iterable[List[Any]]:
if type_ in [None, "o"]:
return to_absolute_time(self.__stdout_events())
else:
return []
def __stdout_events(self) -> Generator[List[Any], None, None]:
for time, data in self.__attrs["stdout"]:
yield [time, "o", data]
def events(self) -> Any:
return self.stdout_events()
def stdout_events(self) -> Generator[List[Any], None, None]:
return to_absolute_time(self.__stdout_events())
class open_from_file:
FORMAT_ERROR: str = "only asciicast v1 format can be opened"

19
asciinema/asciicast/v2.py
View File

@@ -31,14 +31,15 @@ class Asciicast:
self.v2_header = header
self.idle_time_limit = header.get("idle_time_limit")
def events(self) -> Generator[Any, None, None]:
for line in self.__file:
yield json.loads(line)
def stdout_events(self) -> Generator[List[Any], None, None]:
for time, type_, data in self.events():
if type_ == "o":
yield [time, type_, data]
def events(self, type_: Optional[str]) -> Generator[List[Any], None, None]:
if type_ is None:
for line in self.__file:
yield json.loads(line)
else:
for line in self.__file:
event = json.loads(line)
if event[1] == type_:
yield event
def build_from_header_and_file(
@@ -76,7 +77,7 @@ def get_duration(path_: str) -> Any:
first_line = f.readline()
with open_from_file(first_line, f) as a:
last_frame = None
for last_frame in a.stdout_events():
for last_frame in a.events("o"):
pass
return last_frame[0]

2
asciinema/commands/cat.py
View File

@@ -17,7 +17,7 @@ class CatCommand(Command):
with open("/dev/tty", "rt", encoding="utf-8") as stdin:
with raw(stdin.fileno()):
with asciicast.open_from_url(self.filename) as a:
for _, _type, text in a.stdout_events():
for _, _type, text in a.events("o"):
sys.stdout.write(text)
sys.stdout.flush()

7
asciinema/commands/play.py
View File

@@ -18,6 +18,7 @@ class PlayCommand(Command):
self.filename = args.filename
self.idle_time_limit = args.idle_time_limit
self.speed = args.speed
self.stream = args.stream
self.player = player if player is not None else Player()
self.key_bindings = {
"pause": config.play_pause_key,
@@ -28,7 +29,11 @@ class PlayCommand(Command):
try:
with asciicast.open_from_url(self.filename) as a:
self.player.play(
a, self.idle_time_limit, self.speed, self.key_bindings
a,
idle_time_limit=self.idle_time_limit,
speed=self.speed,
key_bindings=self.key_bindings,
stream=self.stream,
)
except asciicast.LoadError as e:

25
asciinema/player.py
View File

@@ -15,6 +15,7 @@ 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",
) -> None:
if key_bindings is None:
key_bindings = {}
@@ -22,10 +23,17 @@ class Player: # pylint: disable=too-few-public-methods
with open("/dev/tty", "rt", encoding="utf-8") as stdin:
with raw(stdin.fileno()):
self._play(
asciicast, idle_time_limit, speed, stdin, key_bindings
asciicast,
idle_time_limit,
speed,
stdin,
key_bindings,
stream,
)
except Exception: # pylint: disable=broad-except
self._play(asciicast, idle_time_limit, speed, None, key_bindings)
self._play(
asciicast, idle_time_limit, speed, None, key_bindings, stream
)
@staticmethod
def _play( # pylint: disable=too-many-locals
@@ -34,23 +42,24 @@ class Player: # pylint: disable=too-few-public-methods
speed: float,
stdin: Optional[TextIO],
key_bindings: Dict[str, Any],
stream: str,
) -> None:
idle_time_limit = idle_time_limit or asciicast.idle_time_limit
pause_key = key_bindings.get("pause")
step_key = key_bindings.get("step")
stdout = asciicast.stdout_events()
stdout = ev.to_relative_time(stdout)
stdout = ev.cap_relative_time(stdout, idle_time_limit)
stdout = ev.to_absolute_time(stdout)
stdout = ev.adjust_speed(stdout, speed)
events = asciicast.events(stream)
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)
base_time = time.time()
ctrl_c = False
paused = False
pause_time: Optional[float] = None
for t, _type, text in stdout:
for t, _type, text in events:
delay = t - (time.time() - base_time)
while stdin and not ctrl_c and delay > 0: