"""``pysnap report`` subcommands."""
from __future__ import annotations
import argparse
import os
from pathlib import Path
import re
import shutil
from typing import Sequence, TextIO
from pysnap.core.service import PySnapService
from pysnap.errors import PySnapError, ReportFormatError
from pysnap.report.highlight import highlight_transcript
from pysnap.report.models import (
CellStyle,
CommandRecord,
Recording,
StyledRun,
Transcript,
)
from pysnap.report.player import DEFAULT_MAX_DELAY, PlaybackController, ReplayPlayer
from pysnap.report.recording import load_report
from pysnap.report.render import render_transcript
from pysnap.report.checkfile import load_check
from pysnap.report.compare import LEVELS, CompareResult, compare_reports
from pysnap.report.extract import extract_file
from pysnap.report.matcher import CheckResult, run_check
from pysnap.report.stats import ReportStats, compute_stats
_ANSI_COLORS = {
"black": 0,
"red": 1,
"green": 2,
"brown": 3,
"yellow": 3,
"blue": 4,
"magenta": 5,
"cyan": 6,
"white": 7,
}
_HEX_COLOR = re.compile(r"^[0-9a-fA-F]{6}$")
_RESET = "\x1b[0m"
[docs]
def build_report_parser(stdout: TextIO, stderr: TextIO) -> argparse.ArgumentParser:
"""Build the ``pysnap report`` parser with its subcommands.
:param stdout: Stream used for help output.
:param stderr: Stream used for parser errors.
:returns: Configured parser.
"""
# ``pysnap.cli.app`` imports this module to dispatch ``report``, so the
# shared parser class is imported lazily to avoid an import cycle.
from pysnap.cli.app import CliArgumentParser, _examples
parser = CliArgumentParser(
prog="pysnap report",
description=(
"Read session reports (report.NN.HOST archives) recorded by the "
"report utility inside the lab VMs. Reports are read safely: nothing "
"is extracted to disk and nothing from a report reaches the terminal "
"as a raw control sequence."
),
epilog=(
'Run "pysnap report SUBCOMMAND --help" for the arguments of a subcommand.\n\n'
+ _examples(
"pysnap report text report.01.first",
"pysnap report check report.01.first lab01-first.check.toml",
"pysnap report compare reports/",
)
),
stdout=stdout,
stderr=stderr,
)
subcommands = parser.add_subparsers(dest="subcommand", metavar="SUBCOMMAND")
subcommands.required = True
text = subcommands.add_parser(
"text",
help="Print the session text with highlighted commands.",
description=(
"Print the text of a recorded session without timing, as the student "
"saw it. The report prompt is highlighted and entered commands are "
"shown in bold; colors produced in the VM are kept. Text longer than "
"the terminal opens in a pager: arrows scroll in all directions, "
"PgUp/PgDn and Home/End jump, q quits."
),
epilog=_examples(
"pysnap report text report.01.first",
"pysnap report text report.01.first --commands",
"pysnap report text report.01.first --color never --no-pager > first.txt",
),
stdout=stdout,
stderr=stderr,
)
text.add_argument("report", metavar="REPORT", help="Report file, for example report.01.first.")
text.add_argument(
"--commands",
action="store_true",
help="Print only the list of entered commands.",
)
text.add_argument(
"--color",
choices=("auto", "always", "never"),
default="auto",
metavar="WHEN",
help="When to color: auto (default, only on a terminal), always or never.",
)
text.add_argument(
"--no-pager",
action="store_true",
help="Print directly even when the text is longer than the terminal.",
)
check = subcommands.add_parser(
"check",
help="Show report information and check it against a check file.",
description=(
"Show where and when a report was recorded, its commands, pauses, "
"typing, pasted input, addresses and integrity checks. With a "
"check file, also check the expected commands and output blocks "
"and grade the report."
),
epilog=_examples(
"pysnap report check report.01.first",
"pysnap report check report.01.first lab01-first.check.toml",
),
stdout=stdout,
stderr=stderr,
)
check.add_argument("report", metavar="REPORT", help="Report file, for example report.01.first.")
check.add_argument(
"check_file",
nargs="?",
metavar="CHECK",
help="Check file in TOML format, for example lab02-first.check.toml.",
)
compare = subcommands.add_parser(
"compare",
help="Compare reports pairwise and show signs of a shared origin.",
description=(
"Compare every pair of the given reports and list the signals "
"that they share their origin: identical files or output, the "
"same MAC addresses, start times, keyboard delays, random output "
"values, CPU.txt or erroneous commands. Directories are searched "
"recursively for report.NN.HOST files. Whether two reports come "
"from different students is up to the reader. Pairs with signals are "
"listed first with their signals, other pairs as OK."
),
epilog=_examples(
"pysnap report compare reports/",
"pysnap report compare ivanov/report.01.first petrov/report.01.first",
),
stdout=stdout,
stderr=stderr,
)
compare.add_argument(
"paths", nargs="+", metavar="PATH", help="Report files and directories with reports."
)
extract = subcommands.add_parser(
"extract",
help="Copy a report file from a running VM through its console or ssh.",
description=(
"Copy a file from a running VM to the host. A VM reached through ssh "
"is read through a separate ssh session with the PySnap key (see "
"pysnap ssh-key); the student's terminal is not touched. Other VMs "
"use the UART1 serial console: it must be at a shell prompt, without "
"a running report recording, and an attached pysnap connect session "
"runs the transfer on its own connection. A name without / is looked "
"up only in the home directory of the user logged in on the console, "
"or of the ssh user of the VM (pysnap set VM ssh-user=NAME, root by "
"default); give the full path for a report of another user."
),
epilog=_examples(
"pysnap report extract first report.01.first",
"pysnap report extract first /tmp/report.03.pc1 --output reports/pc1 --force",
),
stdout=stdout,
stderr=stderr,
)
extract.add_argument("vm", metavar="VM", help="Name of the running virtual machine.")
extract.add_argument(
"name", metavar="NAME", help="Report file in the VM, for example report.01.first."
)
extract.add_argument(
"--output",
metavar="PATH",
help="Host file to write; defaults to the same name in the current directory.",
)
extract.add_argument(
"--force",
action="store_true",
help="Replace an existing host file.",
)
show = subcommands.add_parser(
"show",
help="Replay a report with its timing in a safe terminal view.",
description=(
"Replay a recorded session with its timing. Keys: Space pause, "
"+/- speed, n/p next or previous command, Home/End start or end, "
"q quit; while paused Alt+Up/Down scroll the history."
),
epilog=_examples(
"pysnap report show report.01.first",
"pysnap report show report.01.first --speed 4 --max-delay 0.5",
),
stdout=stdout,
stderr=stderr,
)
show.add_argument("report", metavar="REPORT", help="Report file, for example report.01.first.")
show.add_argument(
"--speed",
type=float,
default=1.0,
metavar="X",
help="Initial speed: 0.25, 0.5, 1, 2, 4, 8 or 16 (default 1).",
)
show.add_argument(
"--max-delay",
type=float,
default=DEFAULT_MAX_DELAY,
metavar="SECONDS",
help=(
"Shorten pauses longer than this, like scriptreplay -m "
f"(default {DEFAULT_MAX_DELAY:g}; 0 keeps real pauses)."
),
)
return parser
[docs]
def run_report_command(
arguments: Sequence[str],
service: PySnapService,
stdout: TextIO,
stderr: TextIO,
) -> int:
"""Run ``pysnap report``.
:param arguments: Arguments after ``report``.
:param service: Application service.
:param stdout: Output stream.
:param stderr: Error stream.
:returns: Process exit code.
"""
namespace = build_report_parser(stdout, stderr).parse_args(list(arguments))
if namespace.subcommand == "text":
return _run_text(namespace, stdout, stderr)
if namespace.subcommand == "show":
return _run_show(namespace, stdout, stderr)
if namespace.subcommand == "check":
return _run_check(namespace, stdout)
if namespace.subcommand == "extract":
return _run_extract(namespace, service, stdout, stderr)
if namespace.subcommand == "compare":
return _run_compare(namespace, stdout, stderr)
return 1
def _run_compare(namespace: argparse.Namespace, stdout: TextIO, stderr: TextIO) -> int:
"""Run ``pysnap report compare``.
:param namespace: Parsed arguments.
:param stdout: Output stream.
:param stderr: Error stream.
:returns: Process exit code.
"""
result = compare_reports(namespace.paths)
for warning in result.warnings:
print(f"Warning: {warning}", file=stderr)
if not result.reports:
raise ReportFormatError("No readable reports to compare.")
print(format_compare_result(result), file=stdout)
return 0
def _plural(count: int, noun: str) -> str:
"""Return a count with a regular English plural."""
return f"{count} {noun}" if count == 1 else f"{count} {noun}s"
def _run_extract(
namespace: argparse.Namespace,
service: PySnapService,
stdout: TextIO,
stderr: TextIO,
) -> int:
"""Run ``pysnap report extract``.
:param namespace: Parsed arguments.
:param service: Application service.
:param stdout: Output stream.
:param stderr: Error stream.
:returns: Process exit code.
"""
result = extract_file(
service,
namespace.vm,
namespace.name,
destination=Path(namespace.output) if namespace.output else None,
force=namespace.force,
)
print(
f"Extracted {result.remote_path} from {result.vm_name} to {result.destination} "
f"({result.size} bytes, sha256 {result.sha256}).",
file=stdout,
)
try:
load_report(result.destination)
except PySnapError as error:
print(f"Warning: the file is not a readable report: {error}", file=stderr)
return 0
def _run_check(namespace: argparse.Namespace, stdout: TextIO) -> int:
"""Run ``pysnap report check``.
:param namespace: Parsed arguments.
:param stdout: Output stream.
:returns: Process exit code.
"""
spec = load_check(namespace.check_file) if namespace.check_file else None
recording = load_report(namespace.report)
transcript = render_transcript(recording)
print(format_report_stats(compute_stats(recording, transcript)), file=stdout)
if spec is not None:
result = run_check(spec, transcript, report_name=recording.name)
print("", file=stdout)
print(format_check_result(result, transcript), file=stdout)
return 0
def _count(items, attribute: str) -> int:
"""Count items whose boolean attribute is set."""
return sum(1 for item in items if getattr(item, attribute))
def _format_value(value) -> str:
"""Format an optional value."""
return "unknown" if value is None else str(value)
def _format_datetime(value) -> str:
"""Format an optional timestamp with its UTC offset."""
return "unknown" if value is None else value.isoformat(sep=" ", timespec="seconds")
def _format_seconds(value: float | None) -> str:
"""Format an optional duration in seconds."""
return "unknown" if value is None else f"{value:.2f} s"
def _format_rate(value: float | None) -> str:
"""Format an optional typing speed."""
return "unknown" if value is None else f"{value:.2f} keys/s"
def _format_share(value: float | None) -> str:
"""Format an optional share as a percentage."""
return "unknown" if value is None else f"{value * 100:.1f}% of keys"
def _run_text(namespace: argparse.Namespace, stdout: TextIO, stderr: TextIO) -> int:
"""Run ``pysnap report text``.
:param namespace: Parsed arguments.
:param stdout: Output stream.
:param stderr: Error stream.
:returns: Process exit code.
"""
recording = load_report(namespace.report)
transcript = render_transcript(recording)
_print_diagnostics(recording, transcript, stderr)
if namespace.commands:
print(format_command_list(transcript.commands), file=stdout)
return 0
color = _use_color(namespace.color, stdout)
if not namespace.no_pager and _needs_pager(len(transcript.lines), stdout):
from pysnap.report.viewer import run_pager
highlighted = highlight_transcript(transcript)
if not color:
highlighted = [(StyledRun("".join(run.text for run in runs)),) for runs in highlighted]
run_pager(recording.source_name, highlighted)
return 0
for line in format_transcript(transcript, color=color):
print(line, file=stdout)
return 0
def _run_show(namespace: argparse.Namespace, stdout: TextIO, stderr: TextIO) -> int:
"""Run ``pysnap report show``.
:param namespace: Parsed arguments.
:param stdout: Output stream.
:param stderr: Error stream.
:returns: Process exit code.
:raises PySnapError: When the output is not an interactive terminal.
"""
if not _is_terminal(stdout):
raise PySnapError(
"pysnap report show needs an interactive terminal; "
"use pysnap report text for redirected output."
)
recording = load_report(namespace.report)
transcript = render_transcript(recording)
_print_diagnostics(recording, transcript, stderr)
player = ReplayPlayer(recording, transcript, max_delay=namespace.max_delay)
from pysnap.report.viewer import run_player
run_player(PlaybackController(player, speed=namespace.speed), recording.source_name)
return 0
def _render_runs(runs: Sequence[StyledRun]) -> str:
"""Render styled runs with ANSI SGR sequences built from known values."""
pieces: list[str] = []
for run in runs:
codes = _sgr_codes(run.style)
if codes:
pieces.append(f"\x1b[{';'.join(codes)}m{run.text}{_RESET}")
else:
pieces.append(run.text)
return "".join(pieces)
def _sgr_codes(style: CellStyle) -> list[str]:
"""Translate a cell style into SGR parameters."""
codes: list[str] = []
if style.bold:
codes.append("1")
if style.italics:
codes.append("3")
if style.underscore:
codes.append("4")
if style.reverse:
codes.append("7")
if style.strikethrough:
codes.append("9")
codes.extend(_color_codes(style.fg, foreground=True))
codes.extend(_color_codes(style.bg, foreground=False))
return codes
def _color_codes(color: str, *, foreground: bool) -> list[str]:
"""Translate one ``pyte`` color into SGR parameters."""
base, bright_base, extended = (30, 90, "38") if foreground else (40, 100, "48")
name = color.lower()
if name in _ANSI_COLORS:
return [str(base + _ANSI_COLORS[name])]
if name.startswith("bright") and name[6:] in _ANSI_COLORS:
return [str(bright_base + _ANSI_COLORS[name[6:]])]
if _HEX_COLOR.match(name):
red, green, blue = (int(name[i:i + 2], 16) for i in (0, 2, 4))
return [extended, "2", str(red), str(green), str(blue)]
return []
def _use_color(mode: str, stdout: TextIO) -> bool:
"""Decide whether to emit ANSI styles."""
if mode == "always":
return True
if mode == "never" or os.environ.get("NO_COLOR"):
return False
return _is_terminal(stdout)
def _is_terminal(stream: TextIO) -> bool:
"""Return whether a stream is an interactive terminal."""
isatty = getattr(stream, "isatty", None)
return bool(isatty and isatty())
def _needs_pager(line_count: int, stdout: TextIO) -> bool:
"""Return whether ``text`` output should open the pager.
The pager opens only for a terminal and only when the text does not fit
on the screen.
"""
if not _is_terminal(stdout):
return False
return line_count > shutil.get_terminal_size(fallback=(80, 24)).lines
def _print_diagnostics(recording: Recording, transcript: Transcript, stderr: TextIO) -> None:
"""Print report diagnostics as warnings."""
for diagnostic in (*recording.diagnostics, *transcript.diagnostics):
print(f"Warning: {diagnostic.message}", file=stderr)