reel-driven-development API Documentation

adoc

Created on 2026-08-10.

asciidoc rendering of a reel

The reel file carries the Recording, the run configuration and the hops. The wiki templates render that model as wikitext; the templates here render the same model as asciidoc, so page and document say the same thing. The document is a build artefact of the folder - anyone holding the folder can rebuild it, with no wiki in the path.

see https://github.com/WolfgangFahl/reel-driven-development/issues/23

@author: wf

RecordingDoc

The asciidoc document of one reel.

One block per hop - the frame, when it was reached, the node and what happened there - so a reviewer reads their own walk and can write into the rendered pdf beside every step.

Source code in rdd/adoc.py

class RecordingDoc:
    """The asciidoc document of one reel.

    One block per hop - the frame, when it was reached, the node and what
    happened there - so a reviewer reads their own walk and can write
    into the rendered pdf beside every step.
    """

    def __init__(
        self,
        hop_set: HopSet,
        folder: str,
        persons: Optional[Dict[str, str]] = None,
        transcript: Optional[Transcript] = None,
        width: int = 640,
    ):
        """Initialize the document of the given reel.

        Args:
            hop_set: the reel with its Recording and its hops.
            folder: the recording folder the frames live in.
            persons: person name to the url they are identifiable by
                outside the wiki; a name that is not in it keeps its
                plain form, so a document renders before the mapping is
                complete and the gap is visible in it.
            transcript: the improved transcript, None where the folder
                carries none.
            width: width of an evidence frame in the document.
        """
        self.hop_set = hop_set
        self.recording = hop_set.recording
        self.folder = folder
        self.persons = persons if persons else {}
        self.transcript = transcript
        self.width = width

    @property
    def lang(self) -> str:
        """The language of the Recording, English where it carries none."""
        language = self.recording.language if self.recording else None
        doc_lang = language if language in LABELS else "en"
        return doc_lang

    def label(self, key: str) -> str:
        """Get the heading of the given section in the document language.

        Args:
            key: the section key.

        Returns:
            the heading.
        """
        heading = LABELS[self.lang][key]
        return heading

    @classmethod
    def of_folder(cls, folder: str, **kwargs) -> "RecordingDoc":
        """Get the document of the given recording folder.

        Args:
            folder: the recording folder.
            kwargs: passed on to the constructor, e.g. the frame width.

        Returns:
            the document of its reel file.

        Raises:
            ValueError: if the folder carries no reel file.
        """
        hop_set = HopSet.of_dir(folder)
        if hop_set is None:
            raise ValueError(f"{HopSet.path_of(folder)} not found")
        persons = cls.persons_of(os.path.join(folder, PERSONS_FILE))
        doc = cls(
            hop_set,
            folder,
            persons=persons,
            transcript=Transcript.of_dir(folder),
            **kwargs,
        )
        return doc

    @classmethod
    def persons_of(cls, yaml_path: str) -> Dict[str, str]:
        """Read the person mapping of the given file.

        Args:
            yaml_path: file of name to url entries.

        Returns:
            the mapping, empty where the file is not there - the mapping
            grows with the reels and is not a precondition of a document.
        """
        persons = {}
        if os.path.isfile(yaml_path):
            with open(yaml_path, encoding="utf-8") as yaml_file:
                persons = yaml.safe_load(yaml_file) or {}
        return persons

    @property
    def participants(self) -> List[str]:
        """The participants as the Recording names them."""
        names = []
        if self.recording and self.recording.participants:
            names = [
                name.strip()
                for name in self.recording.participants.split(",")
                if name.strip()
            ]
        return names

    def person_link(self, name: str) -> str:
        """Get the asciidoc form of the given person.

        Args:
            name: the person as the Recording names them.

        Returns:
            a link where the person has a url, the plain name otherwise.
        """
        person_url = self.persons.get(name)
        adoc_link = f"{person_url}[{name}]" if person_url else name
        return adoc_link

    @property
    def images_dir(self) -> str:
        """The directory the document takes its frames from.

        A frame is a full screen capture; embedding it at full
        resolution is what makes a document of two dozen hops too heavy
        to mail. The scaled copies live in a hidden directory of the
        recording folder, derived and rebuildable, so the evidence
        frames themselves stay untouched.
        """
        frames_dir = self.folder
        if self.width:
            frames_dir = os.path.join(self.folder, f".frames-{self.width}")
        return frames_dir

    def scale_frames(self) -> int:
        """Write the scaled copies of the evidence frames.

        Returns:
            the number of frames written.
        """
        scaled = 0
        if not self.width:
            return scaled
        os.makedirs(self.images_dir, exist_ok=True)
        for hop in self.hop_set.hops:
            source = self.frame_path(hop)
            if source is None:
                continue
            target = os.path.join(self.images_dir, hop.screenshot)
            if os.path.isfile(target):
                continue
            image = cv2.imread(source)
            if image is None:
                continue
            height, source_width = image.shape[:2]
            if source_width > self.width:
                height = int(round(height * self.width / source_width))
                image = cv2.resize(
                    image, (self.width, height), interpolation=cv2.INTER_AREA
                )
            cv2.imwrite(target, image)
            scaled += 1
        return scaled

    def frame_path(self, hop: HopContent) -> Optional[str]:
        """Get the file path of the evidence frame of the given hop.

        Args:
            hop: the hop record.

        Returns:
            the path of the frame, None where the hop has none or the
            frame is not in the folder - a missing frame is left out of
            the document rather than rendered as a broken image.
        """
        path = None
        if hop.screenshot:
            candidate = os.path.join(self.folder, hop.screenshot)
            if os.path.isfile(candidate):
                path = candidate
        return path

    def header(self) -> List[str]:
        """Get the document header lines."""
        rec = self.recording
        title = rec.name or rec.acronym or rec.videoFile or "reel"
        lines = [
            f"= {title}",
            ":doctype: article",
            ":toc: left",
            ":icons: font",
            f":lang: {self.lang}",
            f":toc-title: {self.label('toc')}",
            # the frames are named relative to the folder, so the document
            # travels with it and the same source renders inside the zip
            f":imagesdir: {os.path.abspath(self.images_dir)}",
            "",
        ]
        facts = [str(fact) for fact in (rec.date, rec.platform) if fact]
        if rec.durationMin is not None:
            facts.append(f"{rec.durationMin} min")
        facts.append(f"{self.hop_set.hopCount} {self.label('hops')}")
        lines += [" - ".join(facts), ""]
        if self.participants:
            links = [self.person_link(name) for name in self.participants]
            lines += [f"{self.label('participants')}: {', '.join(links)}", ""]
        if rec.summary:
            lines += [f"== {self.label('summary')}", "", rec.summary, ""]
        return lines

    def transcript_block(self) -> List[str]:
        """Get the asciidoc lines of the improved transcript.

        Returns:
            the lines, empty where the folder carries no transcript - a
            reel may be documented before its transcript is improved.
        """
        lines = []
        if self.transcript and self.transcript.segments:
            lines = [f"== {self.label('transcript')}", ""]
            for segment in self.transcript.segments:
                speaker = segment.speaker if segment.speaker else ""
                lines.append(f"`{segment.start}` {speaker}:: {segment.text}")
            lines.append("")
        return lines

    def hop_block(self, hop: HopContent) -> List[str]:
        """Get the asciidoc lines of one hop.

        Args:
            hop: the hop record.

        Returns:
            the lines of the hop block.
        """
        heading = hop.node if hop.node else hop.time
        lines = [f"=== {hop.pos}. {heading}", "", f"{hop.time}", ""]
        if self.frame_path(hop):
            lines += [f"image::{hop.screenshot}[{heading},width={self.width}]", ""]
        if hop.url:
            lines += [f"{hop.url}[{hop.url}]", ""]
        if hop.summary:
            lines += [hop.summary, ""]
        return lines

    def asciidoc(self) -> str:
        """Get the whole document as asciidoc.

        Returns:
            the asciidoc source.
        """
        lines = self.header()
        if self.hop_set.hops:
            lines += [f"== {self.label('walk')}", ""]
            for hop in self.hop_set.hops:
                lines += self.hop_block(hop)
        lines += self.transcript_block()
        doc = "\n".join(lines) + "\n"
        return doc

    def save(self, path: str) -> str:
        """Write the document to the given path.

        Args:
            path: the file to write.

        Returns:
            the path written.
        """
        self.scale_frames()
        with open(path, "w", encoding="utf-8") as adoc_file:
            adoc_file.write(self.asciidoc())
        return path

images_dir property

The directory the document takes its frames from.

A frame is a full screen capture; embedding it at full resolution is what makes a document of two dozen hops too heavy to mail. The scaled copies live in a hidden directory of the recording folder, derived and rebuildable, so the evidence frames themselves stay untouched.

lang property

The language of the Recording, English where it carries none.

participants property

The participants as the Recording names them.

__init__(hop_set, folder, persons=None, transcript=None, width=640)

Initialize the document of the given reel.

Parameters:

Name	Type	Description	Default
hop_set	HopSet	the reel with its Recording and its hops.	required
folder	str	the recording folder the frames live in.	required
persons	Optional[Dict[str, str]]	person name to the url they are identifiable by outside the wiki; a name that is not in it keeps its plain form, so a document renders before the mapping is complete and the gap is visible in it.	None
transcript	Optional[Transcript]	the improved transcript, None where the folder carries none.	None
width	int	width of an evidence frame in the document.	640

Source code in rdd/adoc.py

def __init__(
    self,
    hop_set: HopSet,
    folder: str,
    persons: Optional[Dict[str, str]] = None,
    transcript: Optional[Transcript] = None,
    width: int = 640,
):
    """Initialize the document of the given reel.

    Args:
        hop_set: the reel with its Recording and its hops.
        folder: the recording folder the frames live in.
        persons: person name to the url they are identifiable by
            outside the wiki; a name that is not in it keeps its
            plain form, so a document renders before the mapping is
            complete and the gap is visible in it.
        transcript: the improved transcript, None where the folder
            carries none.
        width: width of an evidence frame in the document.
    """
    self.hop_set = hop_set
    self.recording = hop_set.recording
    self.folder = folder
    self.persons = persons if persons else {}
    self.transcript = transcript
    self.width = width

asciidoc()

Get the whole document as asciidoc.

Returns:

Type	Description
str	the asciidoc source.

Source code in rdd/adoc.py

def asciidoc(self) -> str:
    """Get the whole document as asciidoc.

    Returns:
        the asciidoc source.
    """
    lines = self.header()
    if self.hop_set.hops:
        lines += [f"== {self.label('walk')}", ""]
        for hop in self.hop_set.hops:
            lines += self.hop_block(hop)
    lines += self.transcript_block()
    doc = "\n".join(lines) + "\n"
    return doc

frame_path(hop)

Get the file path of the evidence frame of the given hop.

Parameters:

Name	Type	Description	Default
hop	HopContent	the hop record.	required

Returns:

Type	Description
Optional[str]	the path of the frame, None where the hop has none or the
Optional[str]	frame is not in the folder - a missing frame is left out of
Optional[str]	the document rather than rendered as a broken image.

Source code in rdd/adoc.py

def frame_path(self, hop: HopContent) -> Optional[str]:
    """Get the file path of the evidence frame of the given hop.

    Args:
        hop: the hop record.

    Returns:
        the path of the frame, None where the hop has none or the
        frame is not in the folder - a missing frame is left out of
        the document rather than rendered as a broken image.
    """
    path = None
    if hop.screenshot:
        candidate = os.path.join(self.folder, hop.screenshot)
        if os.path.isfile(candidate):
            path = candidate
    return path

header()

Get the document header lines.

Source code in rdd/adoc.py

def header(self) -> List[str]:
    """Get the document header lines."""
    rec = self.recording
    title = rec.name or rec.acronym or rec.videoFile or "reel"
    lines = [
        f"= {title}",
        ":doctype: article",
        ":toc: left",
        ":icons: font",
        f":lang: {self.lang}",
        f":toc-title: {self.label('toc')}",
        # the frames are named relative to the folder, so the document
        # travels with it and the same source renders inside the zip
        f":imagesdir: {os.path.abspath(self.images_dir)}",
        "",
    ]
    facts = [str(fact) for fact in (rec.date, rec.platform) if fact]
    if rec.durationMin is not None:
        facts.append(f"{rec.durationMin} min")
    facts.append(f"{self.hop_set.hopCount} {self.label('hops')}")
    lines += [" - ".join(facts), ""]
    if self.participants:
        links = [self.person_link(name) for name in self.participants]
        lines += [f"{self.label('participants')}: {', '.join(links)}", ""]
    if rec.summary:
        lines += [f"== {self.label('summary')}", "", rec.summary, ""]
    return lines

hop_block(hop)

Get the asciidoc lines of one hop.

Parameters:

Name	Type	Description	Default
hop	HopContent	the hop record.	required

Returns:

Type	Description
List[str]	the lines of the hop block.

Source code in rdd/adoc.py

def hop_block(self, hop: HopContent) -> List[str]:
    """Get the asciidoc lines of one hop.

    Args:
        hop: the hop record.

    Returns:
        the lines of the hop block.
    """
    heading = hop.node if hop.node else hop.time
    lines = [f"=== {hop.pos}. {heading}", "", f"{hop.time}", ""]
    if self.frame_path(hop):
        lines += [f"image::{hop.screenshot}[{heading},width={self.width}]", ""]
    if hop.url:
        lines += [f"{hop.url}[{hop.url}]", ""]
    if hop.summary:
        lines += [hop.summary, ""]
    return lines

label(key)

Get the heading of the given section in the document language.

Parameters:

Name	Type	Description	Default
key	str	the section key.	required

Returns:

Type	Description
str	the heading.

Source code in rdd/adoc.py

def label(self, key: str) -> str:
    """Get the heading of the given section in the document language.

    Args:
        key: the section key.

    Returns:
        the heading.
    """
    heading = LABELS[self.lang][key]
    return heading

of_folder(folder, **kwargs) classmethod

Get the document of the given recording folder.

Parameters:

Name	Type	Description	Default
folder	str	the recording folder.	required
kwargs		passed on to the constructor, e.g. the frame width.	{}

Returns:

Type	Description
RecordingDoc	the document of its reel file.

Raises:

Type	Description
ValueError	if the folder carries no reel file.

Source code in rdd/adoc.py

@classmethod
def of_folder(cls, folder: str, **kwargs) -> "RecordingDoc":
    """Get the document of the given recording folder.

    Args:
        folder: the recording folder.
        kwargs: passed on to the constructor, e.g. the frame width.

    Returns:
        the document of its reel file.

    Raises:
        ValueError: if the folder carries no reel file.
    """
    hop_set = HopSet.of_dir(folder)
    if hop_set is None:
        raise ValueError(f"{HopSet.path_of(folder)} not found")
    persons = cls.persons_of(os.path.join(folder, PERSONS_FILE))
    doc = cls(
        hop_set,
        folder,
        persons=persons,
        transcript=Transcript.of_dir(folder),
        **kwargs,
    )
    return doc

person_link(name)

Get the asciidoc form of the given person.

Parameters:

Name	Type	Description	Default
name	str	the person as the Recording names them.	required

Returns:

Type	Description
str	a link where the person has a url, the plain name otherwise.

Source code in rdd/adoc.py

def person_link(self, name: str) -> str:
    """Get the asciidoc form of the given person.

    Args:
        name: the person as the Recording names them.

    Returns:
        a link where the person has a url, the plain name otherwise.
    """
    person_url = self.persons.get(name)
    adoc_link = f"{person_url}[{name}]" if person_url else name
    return adoc_link

persons_of(yaml_path) classmethod

Read the person mapping of the given file.

Parameters:

Name	Type	Description	Default
yaml_path	str	file of name to url entries.	required

Returns:

Type	Description
Dict[str, str]	the mapping, empty where the file is not there - the mapping
Dict[str, str]	grows with the reels and is not a precondition of a document.

Source code in rdd/adoc.py

@classmethod
def persons_of(cls, yaml_path: str) -> Dict[str, str]:
    """Read the person mapping of the given file.

    Args:
        yaml_path: file of name to url entries.

    Returns:
        the mapping, empty where the file is not there - the mapping
        grows with the reels and is not a precondition of a document.
    """
    persons = {}
    if os.path.isfile(yaml_path):
        with open(yaml_path, encoding="utf-8") as yaml_file:
            persons = yaml.safe_load(yaml_file) or {}
    return persons

save(path)

Write the document to the given path.

Parameters:

Name	Type	Description	Default
path	str	the file to write.	required

Returns:

Type	Description
str	the path written.

Source code in rdd/adoc.py

def save(self, path: str) -> str:
    """Write the document to the given path.

    Args:
        path: the file to write.

    Returns:
        the path written.
    """
    self.scale_frames()
    with open(path, "w", encoding="utf-8") as adoc_file:
        adoc_file.write(self.asciidoc())
    return path

scale_frames()

Write the scaled copies of the evidence frames.

Returns:

Type	Description
int	the number of frames written.

Source code in rdd/adoc.py

def scale_frames(self) -> int:
    """Write the scaled copies of the evidence frames.

    Returns:
        the number of frames written.
    """
    scaled = 0
    if not self.width:
        return scaled
    os.makedirs(self.images_dir, exist_ok=True)
    for hop in self.hop_set.hops:
        source = self.frame_path(hop)
        if source is None:
            continue
        target = os.path.join(self.images_dir, hop.screenshot)
        if os.path.isfile(target):
            continue
        image = cv2.imread(source)
        if image is None:
            continue
        height, source_width = image.shape[:2]
        if source_width > self.width:
            height = int(round(height * self.width / source_width))
            image = cv2.resize(
                image, (self.width, height), interpolation=cv2.INTER_AREA
            )
        cv2.imwrite(target, image)
        scaled += 1
    return scaled

transcript_block()

Get the asciidoc lines of the improved transcript.

Returns:

Type	Description
List[str]	the lines, empty where the folder carries no transcript - a
List[str]	reel may be documented before its transcript is improved.

Source code in rdd/adoc.py

def transcript_block(self) -> List[str]:
    """Get the asciidoc lines of the improved transcript.

    Returns:
        the lines, empty where the folder carries no transcript - a
        reel may be documented before its transcript is improved.
    """
    lines = []
    if self.transcript and self.transcript.segments:
        lines = [f"== {self.label('transcript')}", ""]
        for segment in self.transcript.segments:
            speaker = segment.speaker if segment.speaker else ""
            lines.append(f"`{segment.start}` {speaker}:: {segment.text}")
        lines.append("")
    return lines

adoc_cmd

Created on 2026-08-10.

command line interface of the asciidoc rendering of a reel

see https://github.com/WolfgangFahl/reel-driven-development/issues/23

@author: wf

ReelDocCmd

Bases: BaseCmd

Render the reel of a recording folder as an asciidoc document.

Source code in rdd/adoc_cmd.py

class ReelDocCmd(BaseCmd):
    """Render the reel of a recording folder as an asciidoc document."""

    def __init__(self):
        """Initialize with the reel-driven-development version info."""
        super().__init__(Version())

    def add_arguments(self, parser: argparse.ArgumentParser):
        """Add the document arguments to the given parser.

        Args:
            parser: the parser to add arguments to.
        """
        super().add_arguments(parser)
        parser.add_argument("folder", nargs="?", help="the recording folder")
        parser.add_argument(
            "-o", "--out", help="asciidoc file to write (default: in the folder)"
        )
        parser.add_argument(
            "--width",
            type=int,
            default=640,
            help="width of an evidence frame in the document (default: 640) "
            "- the lever on the size of the pdf",
        )

    def handle_args(self, args: argparse.Namespace) -> bool:
        """Handle the parsed arguments by rendering the document.

        Args:
            args: parsed argument namespace.

        Returns:
            True if the arguments were handled.

        Raises:
            ValueError: if the folder argument is missing.
        """
        handled = super().handle_args(args)
        if not handled:
            if args.folder is None:
                raise ValueError("the folder argument is required")
            self.render(args)
            handled = True
        return handled

    def render(self, args: argparse.Namespace):
        """Render the reel of the given folder.

        Args:
            args: parsed argument namespace.
        """
        doc = RecordingDoc.of_folder(args.folder, width=args.width)
        recording = doc.recording
        name = recording.acronym or os.path.basename(os.path.abspath(args.folder))
        out_path = args.out
        if out_path is None:
            out_path = os.path.join(args.folder, f"{name}.adoc")
        doc.save(out_path)
        if not args.quiet:
            missing = [hop.pos for hop in doc.hop_set.hops if not doc.frame_path(hop)]
            if missing:
                print(f"frames missing for hops: {missing}", file=sys.stderr)
            unnamed = [hop.pos for hop in doc.hop_set.hops if not hop.node]
            if unnamed:
                print(
                    f"{len(unnamed)} hops carry no node - the hop set is "
                    f"detected, not curated",
                    file=sys.stderr,
                )
        print(f"{name}: {doc.hop_set.hopCount} hops -> {os.path.abspath(out_path)}")

__init__()

Initialize with the reel-driven-development version info.

Source code in rdd/adoc_cmd.py

def __init__(self):
    """Initialize with the reel-driven-development version info."""
    super().__init__(Version())

add_arguments(parser)

Add the document arguments to the given parser.

Parameters:

Name	Type	Description	Default
parser	ArgumentParser	the parser to add arguments to.	required

Source code in rdd/adoc_cmd.py

def add_arguments(self, parser: argparse.ArgumentParser):
    """Add the document arguments to the given parser.

    Args:
        parser: the parser to add arguments to.
    """
    super().add_arguments(parser)
    parser.add_argument("folder", nargs="?", help="the recording folder")
    parser.add_argument(
        "-o", "--out", help="asciidoc file to write (default: in the folder)"
    )
    parser.add_argument(
        "--width",
        type=int,
        default=640,
        help="width of an evidence frame in the document (default: 640) "
        "- the lever on the size of the pdf",
    )

handle_args(args)

Handle the parsed arguments by rendering the document.

Parameters:

Name	Type	Description	Default
args	Namespace	parsed argument namespace.	required

Returns:

Type	Description
bool	True if the arguments were handled.

Raises:

Type	Description
ValueError	if the folder argument is missing.

Source code in rdd/adoc_cmd.py

def handle_args(self, args: argparse.Namespace) -> bool:
    """Handle the parsed arguments by rendering the document.

    Args:
        args: parsed argument namespace.

    Returns:
        True if the arguments were handled.

    Raises:
        ValueError: if the folder argument is missing.
    """
    handled = super().handle_args(args)
    if not handled:
        if args.folder is None:
            raise ValueError("the folder argument is required")
        self.render(args)
        handled = True
    return handled

render(args)

Render the reel of the given folder.

Parameters:

Name	Type	Description	Default
args	Namespace	parsed argument namespace.	required

Source code in rdd/adoc_cmd.py

def render(self, args: argparse.Namespace):
    """Render the reel of the given folder.

    Args:
        args: parsed argument namespace.
    """
    doc = RecordingDoc.of_folder(args.folder, width=args.width)
    recording = doc.recording
    name = recording.acronym or os.path.basename(os.path.abspath(args.folder))
    out_path = args.out
    if out_path is None:
        out_path = os.path.join(args.folder, f"{name}.adoc")
    doc.save(out_path)
    if not args.quiet:
        missing = [hop.pos for hop in doc.hop_set.hops if not doc.frame_path(hop)]
        if missing:
            print(f"frames missing for hops: {missing}", file=sys.stderr)
        unnamed = [hop.pos for hop in doc.hop_set.hops if not hop.node]
        if unnamed:
            print(
                f"{len(unnamed)} hops carry no node - the hop set is "
                f"detected, not curated",
                file=sys.stderr,
            )
    print(f"{name}: {doc.hop_set.hopCount} hops -> {os.path.abspath(out_path)}")

main(argv=None)

Command line entry point for the asciidoc rendering.

Parameters:

Name	Type	Description	Default
argv	Optional[List[str]]	command line arguments; defaults to sys.argv.	None

Returns:

Type	Description
int	exit code: 0 = OK, 1 = KeyboardInterrupt, 2 = Exception.

Source code in rdd/adoc_cmd.py

def main(argv: Optional[List[str]] = None) -> int:
    """Command line entry point for the asciidoc rendering.

    Args:
        argv: command line arguments; defaults to sys.argv.

    Returns:
        exit code: 0 = OK, 1 = KeyboardInterrupt, 2 = Exception.
    """
    cmd = ReelDocCmd()
    exit_code = cmd.run(argv)
    return exit_code

config

Created on 2026-08-09.

the values that decide a hop set

@author: wf

HopConfig

Every value that can change which hops are found - see issue #4.

A hop set is only evidence if the values that decided it are known, so the configuration is stored beside the hop set and a run is repeatable from it. The detector is named; the names are the offer of HopDetector.get_detectors.

Source code in rdd/config.py

@lod_storable
class HopConfig:
    """Every value that can change which hops are found - see issue #4.

    A hop set is only evidence if the values that decided it are known, so
    the configuration is stored beside the hop set and a run is repeatable
    from it. The detector is named; the names are the offer of
    HopDetector.get_detectors.
    """

    detector: str = "Adaptive"
    start_sec: Optional[float] = None
    end_sec: Optional[float] = None

frame

Created on 2026-08-08.

the frame module hides the technical datails of numpy

@author: wf

Frame

One picture of a recording together with its position in the recording.

The pixel representation is an implementation detail: callers ask a frame what it shows and where it sits in the recording, never how it is stored. The type of a picture is known to its suppliers - the video reader handing frames in, make building them - never to its users, who only ever pass the picture on as img.

Source code in rdd/frame.py

class Frame:
    """One picture of a recording together with its position in the recording.

    The pixel representation is an implementation detail: callers ask a
    frame what it shows and where it sits in the recording, never how it is
    stored. The type of a picture is known to its suppliers - the video
    reader handing frames in, make building them - never to its users,
    who only ever pass the picture on as img.
    """

    def __init__(
        self,
        img: np.ndarray,
        frame_num: int = 0,
        reel: Optional[Reel] = None,
    ):
        """Initialize the frame.

        Args:
            img: the picture as a HxW or HxWxC array.
            frame_num: the position of the frame in the recording.
            reel: the reel this picture belongs to; a frame without one is
                given a reel that has neither a stream nor a rate, so that
                a frame always knows its reel.
        """
        self._img = img
        self.frame_num = frame_num
        self.reel = reel if reel is not None else Reel()
        self._gray_cache: Optional[np.ndarray] = None

    @property
    def fps(self) -> Optional[float]:
        """Frames per second of the recording this frame belongs to."""
        fps = self.reel.fps
        return fps

    @property
    def img(self) -> np.ndarray:
        """The picture this frame shows.

        The handle a user passes on without asking what it is made of;
        only frame.py and the suppliers of a picture know its type.
        """
        img = self._img
        return img

    @property
    def width(self) -> int:
        """Width of the frame in pixels."""
        width = int(self._img.shape[1])
        return width

    @property
    def height(self) -> int:
        """Height of the frame in pixels."""
        height = int(self._img.shape[0])
        return height

    @property
    def time_sec(self) -> Optional[float]:
        """Position of the frame in the recording in seconds, if the recording
        knows its rate."""
        time_sec = self.reel.time_of(self.frame_num)
        return time_sec

    @property
    def timecode(self) -> str:
        """Position of the frame as mm:ss, or the frame number if no fps."""
        time_sec = self.time_sec
        if time_sec is None:
            timecode = f"frame {self.frame_num}"
        else:
            minutes, seconds = divmod(int(time_sec), 60)
            timecode = f"{minutes:02d}:{seconds:02d}"
        return timecode

    @property
    def _gray(self) -> np.ndarray:
        """The frame as a float32 grayscale array, computed once.

        Private on purpose: grayscale is how this module happens to
        compare pictures, not something a user of a frame should see.
        """
        if self._gray_cache is None:
            gray = self._img
            if gray.ndim == 3:
                gray = gray.mean(axis=2)
            self._gray_cache = gray.astype(np.float32)
        return self._gray_cache

    def crop(self, region: Optional[Region]) -> "Frame":
        """Restrict the frame to a region of interest.

        Args:
            region: the region; None returns the frame itself.

        Returns:
            a Frame showing only the region.
        """
        cropped = self
        if region is not None:
            y0, y1, x0, x1 = region.bounds(self.width, self.height)
            cropped = Frame(
                img=self._img[y0:y1, x0:x1],
                frame_num=self.frame_num,
                reel=self.reel,
            )
        return cropped

    def is_blank(self, tolerance: float = 1.0) -> bool:
        """Decide whether the frame shows a single uniform color.

        A blank frame is what a browser shows before a page has rendered;
        capturing it as evidence is a false hop - see issue #1.

        Args:
            tolerance: maximum spread in gray levels still counting as blank.

        Returns:
            True if the frame is uniform within the tolerance.
        """
        spread = float(self._gray.max() - self._gray.min())
        blank = spread <= tolerance
        return blank

    @classmethod
    def make(
        cls,
        frame_num: int = 0,
        fps: Optional[float] = 25.0,
        width: int = 1280,
        height: int = 720,
        value: int = 128,
        channels: Optional[int] = None,
    ) -> "Frame":
        """Create a frame of a single uniform color.

        Args:
            frame_num: the position of the frame in the recording.
            fps: frames per second of the recording; None leaves the recording unknown.
            width: frame width in pixels.
            height: frame height in pixels.
            value: the gray level or channel value to fill the frame with.
            channels: number of color channels; None creates a gray frame.

        Returns:
            the Frame.
        """
        shape = (height, width) if channels is None else (height, width, channels)
        img = np.full(shape, value, dtype=np.uint8)
        frame = cls(img=img, frame_num=frame_num, reel=Reel(fps=fps))
        return frame

    def with_rect(self, y0: int, y1: int, x0: int, x1: int, value: int) -> "Frame":
        """Copy the frame with a rectangle painted in a single value.

        Args:
            y0: first row of the rectangle.
            y1: row behind the last row of the rectangle.
            x0: first column of the rectangle.
            x1: column behind the last column of the rectangle.
            value: the gray level or channel value to paint with.

        Returns:
            a Frame showing the painted rectangle.
        """
        img = self._img.copy()
        img[y0:y1, x0:x1] = value
        painted = Frame(img=img, frame_num=self.frame_num, reel=self.reel)
        return painted

    def save(self, path: str) -> bool:
        """Write the frame to an image file.

        Args:
            path: the file path; the suffix selects the format.

        Returns:
            True if the file was written.
        """
        written = cv2.imwrite(path, self._img)
        return written

fps property

Frames per second of the recording this frame belongs to.

height property

Height of the frame in pixels.

img property

The picture this frame shows.

The handle a user passes on without asking what it is made of; only frame.py and the suppliers of a picture know its type.

time_sec property

Position of the frame in the recording in seconds, if the recording knows its rate.

timecode property

Position of the frame as mm:ss, or the frame number if no fps.

width property

Width of the frame in pixels.

__init__(img, frame_num=0, reel=None)

Initialize the frame.

Parameters:

Name	Type	Description	Default
img	ndarray	the picture as a HxW or HxWxC array.	required
frame_num	int	the position of the frame in the recording.	0
reel	Optional[Reel]	the reel this picture belongs to; a frame without one is given a reel that has neither a stream nor a rate, so that a frame always knows its reel.	None

Source code in rdd/frame.py

def __init__(
    self,
    img: np.ndarray,
    frame_num: int = 0,
    reel: Optional[Reel] = None,
):
    """Initialize the frame.

    Args:
        img: the picture as a HxW or HxWxC array.
        frame_num: the position of the frame in the recording.
        reel: the reel this picture belongs to; a frame without one is
            given a reel that has neither a stream nor a rate, so that
            a frame always knows its reel.
    """
    self._img = img
    self.frame_num = frame_num
    self.reel = reel if reel is not None else Reel()
    self._gray_cache: Optional[np.ndarray] = None

crop(region)

Restrict the frame to a region of interest.

Parameters:

Name	Type	Description	Default
region	Optional[Region]	the region; None returns the frame itself.	required

Returns:

Type	Description
Frame	a Frame showing only the region.

Source code in rdd/frame.py

def crop(self, region: Optional[Region]) -> "Frame":
    """Restrict the frame to a region of interest.

    Args:
        region: the region; None returns the frame itself.

    Returns:
        a Frame showing only the region.
    """
    cropped = self
    if region is not None:
        y0, y1, x0, x1 = region.bounds(self.width, self.height)
        cropped = Frame(
            img=self._img[y0:y1, x0:x1],
            frame_num=self.frame_num,
            reel=self.reel,
        )
    return cropped

is_blank(tolerance=1.0)

Decide whether the frame shows a single uniform color.

A blank frame is what a browser shows before a page has rendered; capturing it as evidence is a false hop - see issue #1.

Parameters:

Name	Type	Description	Default
tolerance	float	maximum spread in gray levels still counting as blank.	1.0

Returns:

Type	Description
bool	True if the frame is uniform within the tolerance.

Source code in rdd/frame.py

def is_blank(self, tolerance: float = 1.0) -> bool:
    """Decide whether the frame shows a single uniform color.

    A blank frame is what a browser shows before a page has rendered;
    capturing it as evidence is a false hop - see issue #1.

    Args:
        tolerance: maximum spread in gray levels still counting as blank.

    Returns:
        True if the frame is uniform within the tolerance.
    """
    spread = float(self._gray.max() - self._gray.min())
    blank = spread <= tolerance
    return blank

make(frame_num=0, fps=25.0, width=1280, height=720, value=128, channels=None) classmethod

Create a frame of a single uniform color.

Parameters:

Name	Type	Description	Default
frame_num	int	the position of the frame in the recording.	0
fps	Optional[float]	frames per second of the recording; None leaves the recording unknown.	25.0
width	int	frame width in pixels.	1280
height	int	frame height in pixels.	720
value	int	the gray level or channel value to fill the frame with.	128
channels	Optional[int]	number of color channels; None creates a gray frame.	None

Returns:

Type	Description
Frame	the Frame.

Source code in rdd/frame.py

@classmethod
def make(
    cls,
    frame_num: int = 0,
    fps: Optional[float] = 25.0,
    width: int = 1280,
    height: int = 720,
    value: int = 128,
    channels: Optional[int] = None,
) -> "Frame":
    """Create a frame of a single uniform color.

    Args:
        frame_num: the position of the frame in the recording.
        fps: frames per second of the recording; None leaves the recording unknown.
        width: frame width in pixels.
        height: frame height in pixels.
        value: the gray level or channel value to fill the frame with.
        channels: number of color channels; None creates a gray frame.

    Returns:
        the Frame.
    """
    shape = (height, width) if channels is None else (height, width, channels)
    img = np.full(shape, value, dtype=np.uint8)
    frame = cls(img=img, frame_num=frame_num, reel=Reel(fps=fps))
    return frame

save(path)

Write the frame to an image file.

Parameters:

Name	Type	Description	Default
path	str	the file path; the suffix selects the format.	required

Returns:

Type	Description
bool	True if the file was written.

Source code in rdd/frame.py

def save(self, path: str) -> bool:
    """Write the frame to an image file.

    Args:
        path: the file path; the suffix selects the format.

    Returns:
        True if the file was written.
    """
    written = cv2.imwrite(path, self._img)
    return written

with_rect(y0, y1, x0, x1, value)

Copy the frame with a rectangle painted in a single value.

Parameters:

Name	Type	Description	Default
y0	int	first row of the rectangle.	required
y1	int	row behind the last row of the rectangle.	required
x0	int	first column of the rectangle.	required
x1	int	column behind the last column of the rectangle.	required
value	int	the gray level or channel value to paint with.	required

Returns:

Type	Description
Frame	a Frame showing the painted rectangle.

Source code in rdd/frame.py

def with_rect(self, y0: int, y1: int, x0: int, x1: int, value: int) -> "Frame":
    """Copy the frame with a rectangle painted in a single value.

    Args:
        y0: first row of the rectangle.
        y1: row behind the last row of the rectangle.
        x0: first column of the rectangle.
        x1: column behind the last column of the rectangle.
        value: the gray level or channel value to paint with.

    Returns:
        a Frame showing the painted rectangle.
    """
    img = self._img.copy()
    img[y0:y1, x0:x1] = value
    painted = Frame(img=img, frame_num=self.frame_num, reel=self.reel)
    return painted

Reel

Bases: Recording

A Recording that can be played - the source of frames.

A reel knows how fast it runs and, where it has one, the stream it reads its pictures from. The stream is what is optional here, not the reel: a frame always belongs to a reel, while a reel built for a test has no video behind it and answers no pictures.

The video library stays inside this class - callers ask for a frame at a position and get a Frame, never a stream, a timecode object or a codec.

Source code in rdd/frame.py

class Reel(Recording):
    """A Recording that can be played - the source of frames.

    A reel knows how fast it runs and, where it has one, the stream it
    reads its pictures from. The stream is what is optional here, not the
    reel: a frame always belongs to a reel, while a reel built for a test
    has no video behind it and answers no pictures.

    The video library stays inside this class - callers ask for a frame at
    a position and get a Frame, never a stream, a timecode object or a codec.
    """

    def __init__(self, path: Optional[str] = None, fps: Optional[float] = None):
        """Open a reel, or create one that has no video.

        Args:
            path: path of the video file; None creates a reel without a
                stream, as used by tests and by synthetic frames.
            fps: frames per second; read from the video when a path is
                given, otherwise as given here and None where unknown.

        Raises:
            FileNotFoundError: if a path is given that holds no video file.
        """
        super().__init__()
        if path is not None and not os.path.isfile(path):
            raise FileNotFoundError(f"reel {path} not found in {os.getcwd()}")
        self.path = path
        self.videoFile = os.path.basename(path) if path else None
        self.stream = None
        self.fps = fps
        self.duration_sec = 0.0
        self.frame_count = 0
        self._keyframes: Optional[List[int]] = None
        if path is not None:
            self.stream = open_video(path)
            self.fps = float(self.stream.frame_rate)
            self.duration_sec = float(self.stream.duration.seconds)
            self.frame_count = int(round(self.duration_sec * self.fps))
            self.durationMin = self.duration_sec / 60.0

    def frame_num_of(self, time_sec: float) -> int:
        """Convert a position in seconds to a frame number.

        Args:
            time_sec: the position in seconds.

        Returns:
            the frame number.

        Raises:
            ValueError: if the recording does not know its rate.
        """
        if not self.fps:
            raise ValueError("the recording does not know its frame rate")
        frame_num = int(round(time_sec * self.fps))
        return frame_num

    def time_of(self, frame_num: int) -> Optional[float]:
        """Convert a frame number to a position in seconds.

        Args:
            frame_num: the frame number.

        Returns:
            the position in seconds, or None if the rate is unknown.
        """
        time_sec = None
        if self.fps:
            time_sec = frame_num / self.fps
        return time_sec

    @property
    def keyframes(self) -> List[int]:
        """The frame numbers of the key frames of this recording.

        Key frames are the positions that can be seeked to without
        decoding forward, so they are the cheap probes. The index is
        read once with ffprobe; without a stream or without ffprobe the
        list stays empty and the caller falls back to plain bisection.
        """
        if self._keyframes is None:
            self._keyframes = self._read_keyframes()
        return self._keyframes

    def _read_keyframes(self) -> List[int]:
        """Read the key frame positions with ffprobe.

        Returns:
            the key frame numbers, empty if there is no video or ffprobe
            is not available.
        """
        keyframes: List[int] = []
        if self.path is not None:
            cmd = [
                "ffprobe",
                "-v",
                "error",
                "-select_streams",
                "v:0",
                "-skip_frame",
                "nokey",
                "-show_entries",
                "frame=pts_time",
                "-of",
                "csv=p=0",
                self.path,
            ]
            try:
                output = subprocess.run(cmd, capture_output=True, text=True, check=True)
                for line in output.stdout.splitlines():
                    text = line.strip().rstrip(",")
                    if text:
                        keyframes.append(self.frame_num_of(float(text)))
            except (OSError, subprocess.CalledProcessError, ValueError):
                keyframes = []
        return keyframes

    def frame_at(self, frame_num: int) -> Optional["Frame"]:
        """Read the frame at the given position.

        Args:
            frame_num: the frame number to read.

        Returns:
            the Frame, or None without a stream or past the end of the recording.
        """
        frame = None
        if self.stream is not None:
            self.stream.seek(frame_num)
            img = self.stream.read()
            if img is not False and img is not None:
                frame = Frame(img=img, frame_num=frame_num, reel=self)
        return frame

keyframes property

The frame numbers of the key frames of this recording.

Key frames are the positions that can be seeked to without decoding forward, so they are the cheap probes. The index is read once with ffprobe; without a stream or without ffprobe the list stays empty and the caller falls back to plain bisection.

__init__(path=None, fps=None)

Open a reel, or create one that has no video.

Parameters:

Name	Type	Description	Default
path	Optional[str]	path of the video file; None creates a reel without a stream, as used by tests and by synthetic frames.	None
fps	Optional[float]	frames per second; read from the video when a path is given, otherwise as given here and None where unknown.	None

Raises:

Type	Description
FileNotFoundError	if a path is given that holds no video file.

Source code in rdd/frame.py

def __init__(self, path: Optional[str] = None, fps: Optional[float] = None):
    """Open a reel, or create one that has no video.

    Args:
        path: path of the video file; None creates a reel without a
            stream, as used by tests and by synthetic frames.
        fps: frames per second; read from the video when a path is
            given, otherwise as given here and None where unknown.

    Raises:
        FileNotFoundError: if a path is given that holds no video file.
    """
    super().__init__()
    if path is not None and not os.path.isfile(path):
        raise FileNotFoundError(f"reel {path} not found in {os.getcwd()}")
    self.path = path
    self.videoFile = os.path.basename(path) if path else None
    self.stream = None
    self.fps = fps
    self.duration_sec = 0.0
    self.frame_count = 0
    self._keyframes: Optional[List[int]] = None
    if path is not None:
        self.stream = open_video(path)
        self.fps = float(self.stream.frame_rate)
        self.duration_sec = float(self.stream.duration.seconds)
        self.frame_count = int(round(self.duration_sec * self.fps))
        self.durationMin = self.duration_sec / 60.0

frame_at(frame_num)

Read the frame at the given position.

Parameters:

Name	Type	Description	Default
frame_num	int	the frame number to read.	required

Returns:

Type	Description
Optional[Frame]	the Frame, or None without a stream or past the end of the recording.

Source code in rdd/frame.py

def frame_at(self, frame_num: int) -> Optional["Frame"]:
    """Read the frame at the given position.

    Args:
        frame_num: the frame number to read.

    Returns:
        the Frame, or None without a stream or past the end of the recording.
    """
    frame = None
    if self.stream is not None:
        self.stream.seek(frame_num)
        img = self.stream.read()
        if img is not False and img is not None:
            frame = Frame(img=img, frame_num=frame_num, reel=self)
    return frame

frame_num_of(time_sec)

Convert a position in seconds to a frame number.

Parameters:

Name	Type	Description	Default
time_sec	float	the position in seconds.	required

Returns:

Type	Description
int	the frame number.

Raises:

Type	Description
ValueError	if the recording does not know its rate.

Source code in rdd/frame.py

def frame_num_of(self, time_sec: float) -> int:
    """Convert a position in seconds to a frame number.

    Args:
        time_sec: the position in seconds.

    Returns:
        the frame number.

    Raises:
        ValueError: if the recording does not know its rate.
    """
    if not self.fps:
        raise ValueError("the recording does not know its frame rate")
    frame_num = int(round(time_sec * self.fps))
    return frame_num

time_of(frame_num)

Convert a frame number to a position in seconds.

Parameters:

Name	Type	Description	Default
frame_num	int	the frame number.	required

Returns:

Type	Description
Optional[float]	the position in seconds, or None if the rate is unknown.

Source code in rdd/frame.py

def time_of(self, frame_num: int) -> Optional[float]:
    """Convert a frame number to a position in seconds.

    Args:
        frame_num: the frame number.

    Returns:
        the position in seconds, or None if the rate is unknown.
    """
    time_sec = None
    if self.fps:
        time_sec = frame_num / self.fps
    return time_sec

Region dataclass

A rectangular part of a frame.

The region of interest restricts every judgement to the part of the screen that belongs to the walk, so that a permanently changing area outside it - live participant tiles, a clock, a scrolling log - can not defeat the detection - see issue #5.

Coordinates are either fractions of the frame (width and height <= 1) or pixels; the two forms describe the same area on a given frame.

Source code in rdd/frame.py

@dataclass
class Region:
    """A rectangular part of a frame.

    The region of interest restricts every judgement to the part of the
    screen that belongs to the walk, so that a permanently changing area
    outside it - live participant tiles, a clock, a scrolling log - can
    not defeat the detection - see issue #5.

    Coordinates are either fractions of the frame (width and height <= 1)
    or pixels; the two forms describe the same area on a given frame.
    """

    x: float = 0.0
    y: float = 0.0
    width: float = 1.0
    height: float = 1.0

    @classmethod
    def of_tuple(cls, values: Tuple[float, float, float, float]) -> "Region":
        """Create a region from an (x, y, width, height) tuple.

        Args:
            values: the four coordinates.

        Returns:
            the Region.
        """
        x, y, width, height = values
        region = cls(x=x, y=y, width=width, height=height)
        return region

    @classmethod
    def of_str(cls, text: str) -> "Region":
        """Create a region from a comma separated string.

        Args:
            text: "x,y,width,height" e.g. "0,0,0.875,1.0".

        Returns:
            the Region.

        Raises:
            ValueError: if the text does not hold four numbers.
        """
        parts = text.split(",")
        if len(parts) != 4:
            raise ValueError(f"region needs four values x,y,width,height - got {text}")
        values = tuple(float(part) for part in parts)
        region = cls.of_tuple(values)  # type: ignore[arg-type]
        return region

    @property
    def is_fractional(self) -> bool:
        """Decide whether the region is given as fractions of the frame.

        Returns:
            True if width and height are fractions.
        """
        fractional = self.width <= 1.0 and self.height <= 1.0
        return fractional

    def bounds(self, width: int, height: int) -> Tuple[int, int, int, int]:
        """Compute the pixel bounds of the region on a frame of the given size.

        Args:
            width: frame width in pixels.
            height: frame height in pixels.

        Returns:
            (y0, y1, x0, x1) crop bounds, clamped to the frame.
        """
        x, y, region_width, region_height = self.x, self.y, self.width, self.height
        if self.is_fractional:
            x, region_width = x * width, region_width * width
            y, region_height = y * height, region_height * height
        x0 = min(max(int(round(x)), 0), width)
        y0 = min(max(int(round(y)), 0), height)
        x1 = min(max(int(round(x + region_width)), x0), width)
        y1 = min(max(int(round(y + region_height)), y0), height)
        bounds = (y0, y1, x0, x1)
        return bounds

is_fractional property

Decide whether the region is given as fractions of the frame.

Returns:

Type	Description
bool	True if width and height are fractions.

bounds(width, height)

Compute the pixel bounds of the region on a frame of the given size.

Parameters:

Name	Type	Description	Default
width	int	frame width in pixels.	required
height	int	frame height in pixels.	required

Returns:

Type	Description
Tuple[int, int, int, int]	(y0, y1, x0, x1) crop bounds, clamped to the frame.

Source code in rdd/frame.py

def bounds(self, width: int, height: int) -> Tuple[int, int, int, int]:
    """Compute the pixel bounds of the region on a frame of the given size.

    Args:
        width: frame width in pixels.
        height: frame height in pixels.

    Returns:
        (y0, y1, x0, x1) crop bounds, clamped to the frame.
    """
    x, y, region_width, region_height = self.x, self.y, self.width, self.height
    if self.is_fractional:
        x, region_width = x * width, region_width * width
        y, region_height = y * height, region_height * height
    x0 = min(max(int(round(x)), 0), width)
    y0 = min(max(int(round(y)), 0), height)
    x1 = min(max(int(round(x + region_width)), x0), width)
    y1 = min(max(int(round(y + region_height)), y0), height)
    bounds = (y0, y1, x0, x1)
    return bounds

of_str(text) classmethod

Create a region from a comma separated string.

Parameters:

Name	Type	Description	Default
text	str	"x,y,width,height" e.g. "0,0,0.875,1.0".	required

Returns:

Type	Description
Region	the Region.

Raises:

Type	Description
ValueError	if the text does not hold four numbers.

Source code in rdd/frame.py

@classmethod
def of_str(cls, text: str) -> "Region":
    """Create a region from a comma separated string.

    Args:
        text: "x,y,width,height" e.g. "0,0,0.875,1.0".

    Returns:
        the Region.

    Raises:
        ValueError: if the text does not hold four numbers.
    """
    parts = text.split(",")
    if len(parts) != 4:
        raise ValueError(f"region needs four values x,y,width,height - got {text}")
    values = tuple(float(part) for part in parts)
    region = cls.of_tuple(values)  # type: ignore[arg-type]
    return region

of_tuple(values) classmethod

Create a region from an (x, y, width, height) tuple.

Parameters:

Name	Type	Description	Default
values	Tuple[float, float, float, float]	the four coordinates.	required

Returns:

Type	Description
Region	the Region.

Source code in rdd/frame.py

@classmethod
def of_tuple(cls, values: Tuple[float, float, float, float]) -> "Region":
    """Create a region from an (x, y, width, height) tuple.

    Args:
        values: the four coordinates.

    Returns:
        the Region.
    """
    x, y, width, height = values
    region = cls(x=x, y=y, width=width, height=height)
    return region

hop_frame_name(time_sec, with_ms=False)

The file name of the evidence frame of a hop at the given offset.

Issue #21: a running number cannot take an insert or a removal, so an evidence frame is named by the point in the reel it shows, which is the identity that survives curation. The fields are zero padded to a fixed width, so the names sort chronologically, and the name without the millisecond part is a prefix of the one with it, so that ordering holds across both forms. No prefix beyond "hop-" is added - the part directory says which reel the frame belongs to.

Parameters:

Name	Type	Description	Default
time_sec	float	the offset of the frame in the reel in seconds.	required
with_ms	bool	append the millisecond field; only needed where two hops fall in the same second.	False

Returns:

Type	Description
str	the file name, e.g. "hop-00h02m12s.jpg" or "hop-00h02m12s480ms.jpg".

Source code in rdd/frame.py

def hop_frame_name(time_sec: float, with_ms: bool = False) -> str:
    """The file name of the evidence frame of a hop at the given offset.

    Issue #21: a running number cannot take an insert or a removal, so an
    evidence frame is named by the point in the reel it shows, which is
    the identity that survives curation. The fields are zero padded to a
    fixed width, so the names sort chronologically, and the name without
    the millisecond part is a prefix of the one with it, so that ordering
    holds across both forms. No prefix beyond "hop-" is added - the part
    directory says which reel the frame belongs to.

    Args:
        time_sec: the offset of the frame in the reel in seconds.
        with_ms: append the millisecond field; only needed where two hops
            fall in the same second.

    Returns:
        the file name, e.g. "hop-00h02m12s.jpg" or "hop-00h02m12s480ms.jpg".
    """
    total_ms = int(round(time_sec * 1000))
    seconds, milliseconds = divmod(total_ms, 1000)
    hours, rest = divmod(seconds, 3600)
    minutes, seconds = divmod(rest, 60)
    name = f"hop-{hours:02d}h{minutes:02d}m{seconds:02d}s"
    if with_ms:
        name = f"{name}{milliseconds:03d}ms"
    name = f"{name}.jpg"
    return name

hop_frame_names(times_sec)

The evidence frame names of a whole hop set.

The millisecond field is emitted only where it is needed - where two or more hops fall in the same second - so the common case stays as short as it can be read.

Parameters:

Name	Type	Description	Default
times_sec	List[float]	the offsets of the hops in the reel in seconds.	required

Returns:

Type	Description
List[str]	one file name per offset, in the given order.

Source code in rdd/frame.py

def hop_frame_names(times_sec: List[float]) -> List[str]:
    """The evidence frame names of a whole hop set.

    The millisecond field is emitted only where it is needed - where two
    or more hops fall in the same second - so the common case stays as
    short as it can be read.

    Args:
        times_sec: the offsets of the hops in the reel in seconds.

    Returns:
        one file name per offset, in the given order.
    """
    seconds_names = [hop_frame_name(time_sec) for time_sec in times_sec]
    shared = set()
    seen = set()
    for name in seconds_names:
        if name in seen:
            shared.add(name)
        seen.add(name)
    names = [
        hop_frame_name(time_sec, with_ms=seconds_name in shared)
        for time_sec, seconds_name in zip(times_sec, seconds_names)
    ]
    return names

hopdetect_cmd

Created on 2026-08-08.

@author: wf

HopDetectCmd

Bases: BaseCmd

Reel Driven Development command line interface.

Source code in rdd/hopdetect_cmd.py

class HopDetectCmd(BaseCmd):
    """Reel Driven Development command line interface."""

    def __init__(self):
        """Initialize with the reel-driven-development version info."""
        super().__init__(Version())
        self.detector = None

    @staticmethod
    def time_of(text: Optional[str]) -> Optional[float]:
        """Convert a time argument to seconds.

        Args:
            text: mm:ss, hh:mm:ss or a number of seconds; None stays None.

        Returns:
            the time in seconds or None.
        """
        time_sec = None
        if text is not None:
            parts = text.split(":")
            time_sec = 0.0
            for part in parts:
                time_sec = time_sec * 60.0 + float(part)
        return time_sec

    def get_config(self, args: argparse.Namespace) -> HopConfig:
        """Get the configuration the given arguments select.

        Args:
            args: parsed argument namespace.

        Returns:
            the configuration of this run.
        """
        config = HopConfig(
            detector=args.detector,
            start_sec=self.time_of(args.start),
            end_sec=self.time_of(args.end),
        )
        return config

    def add_arguments(self, parser: argparse.ArgumentParser):
        """Add the hop detection arguments to the given parser.

        Args:
            parser: the parser to add arguments to.
        """
        super().add_arguments(parser)
        parser.add_argument("video", nargs="?", help="path of the reel to analyze")
        parser.add_argument("--start", help="segment start as mm:ss or seconds")
        parser.add_argument("--end", help="segment end as mm:ss or seconds")
        parser.add_argument(
            "-o", "--out", default="hops", help="output directory (default: hops)"
        )
        parser.add_argument(
            "--progress",
            action="store_true",
            help="show the progress bar - a run over a reel takes minutes "
            "and must not be silent",
        )
        parser.add_argument(
            "--detector",
            choices=HopDetector.get_detector_names(),
            default="Adaptive",
            help="scene detector to find the hops with (default: Adaptive) "
            "- see https://www.scenedetect.com/benchmarks/",
        )

    def handle_args(self, args: argparse.Namespace) -> bool:
        """Handle the parsed arguments by running the detection.

        Args:
            args: parsed argument namespace.

        Returns:
            True if the arguments were handled.

        Raises:
            ValueError: if the video argument is missing.
        """
        handled = super().handle_args(args)
        if not handled:
            if args.video is None:
                raise ValueError("the video argument is required")
            self.detect(args)
            handled = True
        return handled

    def detect(self, args: argparse.Namespace):
        """Run the hop detection on the given arguments.

        Args:
            args: parsed argument namespace.
        """
        reel = Reel(args.video)
        self.detector = HopDetector(reel)
        config = self.get_config(args)
        hops = self.detector.hops(
            config,
            out_dir=args.out,
            progress=args.progress,
            force=args.force,
        )
        if not args.quiet:
            for hop in hops.hops:
                print(f"{hop.pos:3d} {hop.time} {hop.screenshot}", file=sys.stderr)
        print(
            f"{reel.videoFile}: {hops.hopCount} hops "
            f"from {args.detector} over {reel.frame_count} frames -> {args.out}"
        )

__init__()

Initialize with the reel-driven-development version info.

Source code in rdd/hopdetect_cmd.py

def __init__(self):
    """Initialize with the reel-driven-development version info."""
    super().__init__(Version())
    self.detector = None

add_arguments(parser)

Add the hop detection arguments to the given parser.

Parameters:

Name	Type	Description	Default
parser	ArgumentParser	the parser to add arguments to.	required

Source code in rdd/hopdetect_cmd.py

def add_arguments(self, parser: argparse.ArgumentParser):
    """Add the hop detection arguments to the given parser.

    Args:
        parser: the parser to add arguments to.
    """
    super().add_arguments(parser)
    parser.add_argument("video", nargs="?", help="path of the reel to analyze")
    parser.add_argument("--start", help="segment start as mm:ss or seconds")
    parser.add_argument("--end", help="segment end as mm:ss or seconds")
    parser.add_argument(
        "-o", "--out", default="hops", help="output directory (default: hops)"
    )
    parser.add_argument(
        "--progress",
        action="store_true",
        help="show the progress bar - a run over a reel takes minutes "
        "and must not be silent",
    )
    parser.add_argument(
        "--detector",
        choices=HopDetector.get_detector_names(),
        default="Adaptive",
        help="scene detector to find the hops with (default: Adaptive) "
        "- see https://www.scenedetect.com/benchmarks/",
    )

detect(args)

Run the hop detection on the given arguments.

Parameters:

Name	Type	Description	Default
args	Namespace	parsed argument namespace.	required

Source code in rdd/hopdetect_cmd.py

def detect(self, args: argparse.Namespace):
    """Run the hop detection on the given arguments.

    Args:
        args: parsed argument namespace.
    """
    reel = Reel(args.video)
    self.detector = HopDetector(reel)
    config = self.get_config(args)
    hops = self.detector.hops(
        config,
        out_dir=args.out,
        progress=args.progress,
        force=args.force,
    )
    if not args.quiet:
        for hop in hops.hops:
            print(f"{hop.pos:3d} {hop.time} {hop.screenshot}", file=sys.stderr)
    print(
        f"{reel.videoFile}: {hops.hopCount} hops "
        f"from {args.detector} over {reel.frame_count} frames -> {args.out}"
    )

get_config(args)

Get the configuration the given arguments select.

Parameters:

Name	Type	Description	Default
args	Namespace	parsed argument namespace.	required

Returns:

Type	Description
HopConfig	the configuration of this run.

Source code in rdd/hopdetect_cmd.py

def get_config(self, args: argparse.Namespace) -> HopConfig:
    """Get the configuration the given arguments select.

    Args:
        args: parsed argument namespace.

    Returns:
        the configuration of this run.
    """
    config = HopConfig(
        detector=args.detector,
        start_sec=self.time_of(args.start),
        end_sec=self.time_of(args.end),
    )
    return config

handle_args(args)

Handle the parsed arguments by running the detection.

Parameters:

Name	Type	Description	Default
args	Namespace	parsed argument namespace.	required

Returns:

Type	Description
bool	True if the arguments were handled.

Raises:

Type	Description
ValueError	if the video argument is missing.

Source code in rdd/hopdetect_cmd.py

def handle_args(self, args: argparse.Namespace) -> bool:
    """Handle the parsed arguments by running the detection.

    Args:
        args: parsed argument namespace.

    Returns:
        True if the arguments were handled.

    Raises:
        ValueError: if the video argument is missing.
    """
    handled = super().handle_args(args)
    if not handled:
        if args.video is None:
            raise ValueError("the video argument is required")
        self.detect(args)
        handled = True
    return handled

time_of(text) staticmethod

Convert a time argument to seconds.

Parameters:

Name	Type	Description	Default
text	Optional[str]	mm:ss, hh:mm:ss or a number of seconds; None stays None.	required

Returns:

Type	Description
Optional[float]	the time in seconds or None.

Source code in rdd/hopdetect_cmd.py

@staticmethod
def time_of(text: Optional[str]) -> Optional[float]:
    """Convert a time argument to seconds.

    Args:
        text: mm:ss, hh:mm:ss or a number of seconds; None stays None.

    Returns:
        the time in seconds or None.
    """
    time_sec = None
    if text is not None:
        parts = text.split(":")
        time_sec = 0.0
        for part in parts:
            time_sec = time_sec * 60.0 + float(part)
    return time_sec

main(argv=None)

Command line entry point for hop detection.

Parameters:

Name	Type	Description	Default
argv	Optional[List[str]]	command line arguments; defaults to sys.argv.	None

Returns:

Type	Description
int	exit code: 0 = OK, 1 = KeyboardInterrupt, 2 = Exception.

Source code in rdd/hopdetect_cmd.py

def main(argv: Optional[List[str]] = None) -> int:
    """Command line entry point for hop detection.

    Args:
        argv: command line arguments; defaults to sys.argv.

    Returns:
        exit code: 0 = OK, 1 = KeyboardInterrupt, 2 = Exception.
    """
    cmd = HopDetectCmd()
    exit_code = cmd.run(argv)
    return exit_code

hopdetector

Created on 2026-08-08.

hop detector over a reel

@author: wf

HopDetector

Detect hops in the reel which e.g might be scene change.

Source code in rdd/hopdetector.py

class HopDetector:
    """Detect hops in the reel which e.g might be  scene change."""

    def __init__(self, reel: Reel):
        """Initialize the detector.

        Args:
            reel: the reel to analyze.
        """
        self.reel = reel

    @classmethod
    def get_detectors(cls) -> Generator[Tuple[str, SceneDetector], None, None]:
        """Get the detectors on offer.

        A detector at one threshold says nothing about how it answers to
        that threshold, so each is offered around its library default.
        Which value is right can only be decided against a labeled corpus
        - until we have one these are measurements, not claims.

        ThresholdDetector is not on offer: it detects fades to a
        near-black level, which our material of mostly zoom recordings
        does not have.

        The detectors and their measured quality are documented at
        https://www.scenedetect.com/benchmarks/

        Yields:
            the name of the detector and the detector itself
        """
        yield "Adaptive 1.5", AdaptiveDetector(adaptive_threshold=1.5)
        yield "Adaptive", AdaptiveDetector()
        yield "Adaptive 6", AdaptiveDetector(adaptive_threshold=6.0)
        yield "Content 13.5", ContentDetector(threshold=13.5)
        yield "Content", ContentDetector()
        yield "Content 54", ContentDetector(threshold=54.0)
        yield "Hash 0.2", HashDetector(threshold=0.2)
        yield "Hash", HashDetector()
        yield "Hash 0.79", HashDetector(threshold=0.79)
        yield "Histogram 0.025", HistogramDetector(threshold=0.025)
        yield "Histogram", HistogramDetector()
        yield "Histogram 0.1", HistogramDetector(threshold=0.1)

    @classmethod
    def get_detector_names(cls) -> List[str]:
        """Get the names the detectors are on offer under.

        Returns:
            the names, in the order they are offered.
        """
        names = [name for name, _detector in cls.get_detectors()]
        return names

    def get_detector(self, config: HopConfig) -> SceneDetector:
        """Get the detector the given configuration names.

        Args:
            config: the configuration naming the detector.

        Returns:
            the detector on offer under that name.

        Raises:
            ValueError: if the named detector is not on offer.
        """
        detector = None
        for name, candidate in self.get_detectors():
            if name == config.detector:
                detector = candidate
        if detector is None:
            raise ValueError(
                f"detector {config.detector} is not on offer - "
                f"choose one of {self.get_detector_names()}"
            )
        return detector

    def scenes(
        self,
        config: HopConfig,
        progress: bool = False,
    ) -> List[int]:
        """Candidate hop positions from the given scene detector.

        Which detector is used is the caller's choice - the library offers
        several and they are benchmarked against each other, so the
        detector is a parameter and never fixed here. Each detector
        carries its own thresholds in its own constructor.

        See https://www.scenedetect.com/benchmarks/

        Args:
            config: the values selecting and parameterizing the detector;
                end_sec None runs to one frame before the end, since the
                opencv backend fails on the last frame of some files with
                an undefined timestamp.
            progress: show the tqdm progress bar of the library - a run
                over a reel takes minutes and must not be silent.

        Returns:
            the frame numbers where the detector cuts, empty without a video.
        """
        candidates: List[int] = []
        if self.reel.path is not None:
            end_sec = config.end_sec
            if end_sec is None:
                end_sec = self.reel.duration_sec - 1.0 / self.reel.fps
            scenes = detect(
                self.reel.path,
                self.get_detector(config),
                start_time=config.start_sec,
                end_time=end_sec,
                show_progress=progress,
            )
            candidates = [int(start.frame_num) for start, _ in scenes]
        return candidates

    def clear(self, out_dir: str, force: bool) -> Optional[HopSet]:
        """Make sure a hop set is not silently mixed with an older one.

        Writing a hop set over an older one leaves the frames the new run
        does not cut at behind, and the directory then shows a hop set
        that never existed. An existing hop set is therefore kept unless
        it is replaced whole, and replacing it removes the frames the old
        hop set named.

        What a person wrote into the reel file - the name, the acronym,
        the participants - is not a hop set and survives the replacement:
        a detection replaces what it produced, never what it was given.

        Args:
            out_dir: the directory the hop set is written to.
            force: replace an existing hop set instead of keeping it.

        Returns:
            the reel file that was there, None where there was none.

        Raises:
            ValueError: if a hop set is there and force is not given.
        """
        old = HopSet.of_dir(out_dir)
        if old is not None and old.hops:
            if not force:
                raise ValueError(
                    f"{HopSet.path_of(out_dir)} already holds a hop set - "
                    f"use --force to replace it"
                )
            for hop in old.hops:
                if hop.screenshot:
                    frame_path = os.path.join(out_dir, hop.screenshot)
                    if os.path.isfile(frame_path):
                        os.remove(frame_path)
            old.hops = []
        return old

    def recording_of(self, given: Optional[Recording] = None) -> Recording:
        """The Recording record of the reel this run analyzed.

        Args:
            given: the Recording of a reel file written beforehand; its
                values win, because they are what a person knew and the
                reel cannot answer.

        Returns:
            the Recording, with the fields the reel itself can answer
            filled in where the given one leaves them open.
        """
        recording = given if given else Recording()
        if not recording.videoFile:
            recording.videoFile = self.reel.videoFile
        if recording.durationMin is None:
            recording.durationMin = round(self.reel.duration_sec / 60.0, 1)
        return recording

    def hops(
        self,
        config: HopConfig,
        out_dir: Optional[str] = None,
        progress: bool = False,
        force: bool = False,
    ) -> HopContents:
        """Turn the cuts of the given detector into hop records.

        The evidence frame of a hop is the frame the detector cuts at -
        the first frame of the new content. It is named by its offset in
        the reel and never by its position in the run - see issue #21 and
        hop_frame_names. node, url and summary of the walk stay empty:
        they come from the transcript and are never guessed from the
        picture.

        Args:
            config: the values selecting and parameterizing the detector.
            out_dir: directory the evidence frames and the reel.yaml that
                carries them with the values reproducing them are written
                to; a hopless reel.yaml already there is the input of the
                run and its recording values are kept. None writes
                nothing.
            progress: show the tqdm progress bar while detecting.
            force: overwrite a hop set that is already there.

        Returns:
            the hops of this run.

        Raises:
            ValueError: if the output directory already holds a hop set
                and force is not given.
        """
        # no back reference on the hop: the reel file it lives in is the reel
        hop_contents = HopContents()
        given = None
        if out_dir is not None:
            given = self.clear(out_dir, force)
            os.makedirs(out_dir, exist_ok=True)
        frame_nums = self.scenes(config, progress=progress)
        times_sec = [self.reel.time_of(frame_num) for frame_num in frame_nums]
        names = hop_frame_names(times_sec)
        for frame_num, name in zip(frame_nums, names):
            frame = self.reel.frame_at(frame_num)
            screenshot = None
            if out_dir is not None and frame is not None:
                screenshot = name
                frame.save(os.path.join(out_dir, screenshot))
            hop_contents.add(HopContent(time=frame.timecode, screenshot=screenshot))
        if out_dir is not None:
            hop_set = HopSet(
                recording=self.recording_of(given.recording if given else None),
                config=config,
                hops=hop_contents.hops,
            )
            hop_set.save(
                HopSet.path_of(out_dir),
                version=Version.version,
                date=datetime.now().strftime("%Y-%m-%d"),
            )
        return hop_contents

__init__(reel)

Initialize the detector.

Parameters:

Name	Type	Description	Default
reel	Reel	the reel to analyze.	required

Source code in rdd/hopdetector.py

def __init__(self, reel: Reel):
    """Initialize the detector.

    Args:
        reel: the reel to analyze.
    """
    self.reel = reel

clear(out_dir, force)

Make sure a hop set is not silently mixed with an older one.

Writing a hop set over an older one leaves the frames the new run does not cut at behind, and the directory then shows a hop set that never existed. An existing hop set is therefore kept unless it is replaced whole, and replacing it removes the frames the old hop set named.

What a person wrote into the reel file - the name, the acronym, the participants - is not a hop set and survives the replacement: a detection replaces what it produced, never what it was given.

Parameters:

Name	Type	Description	Default
out_dir	str	the directory the hop set is written to.	required
force	bool	replace an existing hop set instead of keeping it.	required

Returns:

Type	Description
Optional[HopSet]	the reel file that was there, None where there was none.

Raises:

Type	Description
ValueError	if a hop set is there and force is not given.

Source code in rdd/hopdetector.py

def clear(self, out_dir: str, force: bool) -> Optional[HopSet]:
    """Make sure a hop set is not silently mixed with an older one.

    Writing a hop set over an older one leaves the frames the new run
    does not cut at behind, and the directory then shows a hop set
    that never existed. An existing hop set is therefore kept unless
    it is replaced whole, and replacing it removes the frames the old
    hop set named.

    What a person wrote into the reel file - the name, the acronym,
    the participants - is not a hop set and survives the replacement:
    a detection replaces what it produced, never what it was given.

    Args:
        out_dir: the directory the hop set is written to.
        force: replace an existing hop set instead of keeping it.

    Returns:
        the reel file that was there, None where there was none.

    Raises:
        ValueError: if a hop set is there and force is not given.
    """
    old = HopSet.of_dir(out_dir)
    if old is not None and old.hops:
        if not force:
            raise ValueError(
                f"{HopSet.path_of(out_dir)} already holds a hop set - "
                f"use --force to replace it"
            )
        for hop in old.hops:
            if hop.screenshot:
                frame_path = os.path.join(out_dir, hop.screenshot)
                if os.path.isfile(frame_path):
                    os.remove(frame_path)
        old.hops = []
    return old

get_detector(config)

Get the detector the given configuration names.

Parameters:

Name	Type	Description	Default
config	HopConfig	the configuration naming the detector.	required

Returns:

Type	Description
SceneDetector	the detector on offer under that name.

Raises:

Type	Description
ValueError	if the named detector is not on offer.

Source code in rdd/hopdetector.py

def get_detector(self, config: HopConfig) -> SceneDetector:
    """Get the detector the given configuration names.

    Args:
        config: the configuration naming the detector.

    Returns:
        the detector on offer under that name.

    Raises:
        ValueError: if the named detector is not on offer.
    """
    detector = None
    for name, candidate in self.get_detectors():
        if name == config.detector:
            detector = candidate
    if detector is None:
        raise ValueError(
            f"detector {config.detector} is not on offer - "
            f"choose one of {self.get_detector_names()}"
        )
    return detector

get_detector_names() classmethod

Get the names the detectors are on offer under.

Returns:

Type	Description
List[str]	the names, in the order they are offered.

Source code in rdd/hopdetector.py

@classmethod
def get_detector_names(cls) -> List[str]:
    """Get the names the detectors are on offer under.

    Returns:
        the names, in the order they are offered.
    """
    names = [name for name, _detector in cls.get_detectors()]
    return names

get_detectors() classmethod

Get the detectors on offer.

A detector at one threshold says nothing about how it answers to that threshold, so each is offered around its library default. Which value is right can only be decided against a labeled corpus - until we have one these are measurements, not claims.

ThresholdDetector is not on offer: it detects fades to a near-black level, which our material of mostly zoom recordings does not have.

The detectors and their measured quality are documented at https://www.scenedetect.com/benchmarks/

Yields:

Type	Description
Tuple[str, SceneDetector]	the name of the detector and the detector itself

Source code in rdd/hopdetector.py

@classmethod
def get_detectors(cls) -> Generator[Tuple[str, SceneDetector], None, None]:
    """Get the detectors on offer.

    A detector at one threshold says nothing about how it answers to
    that threshold, so each is offered around its library default.
    Which value is right can only be decided against a labeled corpus
    - until we have one these are measurements, not claims.

    ThresholdDetector is not on offer: it detects fades to a
    near-black level, which our material of mostly zoom recordings
    does not have.

    The detectors and their measured quality are documented at
    https://www.scenedetect.com/benchmarks/

    Yields:
        the name of the detector and the detector itself
    """
    yield "Adaptive 1.5", AdaptiveDetector(adaptive_threshold=1.5)
    yield "Adaptive", AdaptiveDetector()
    yield "Adaptive 6", AdaptiveDetector(adaptive_threshold=6.0)
    yield "Content 13.5", ContentDetector(threshold=13.5)
    yield "Content", ContentDetector()
    yield "Content 54", ContentDetector(threshold=54.0)
    yield "Hash 0.2", HashDetector(threshold=0.2)
    yield "Hash", HashDetector()
    yield "Hash 0.79", HashDetector(threshold=0.79)
    yield "Histogram 0.025", HistogramDetector(threshold=0.025)
    yield "Histogram", HistogramDetector()
    yield "Histogram 0.1", HistogramDetector(threshold=0.1)

hops(config, out_dir=None, progress=False, force=False)

Turn the cuts of the given detector into hop records.

The evidence frame of a hop is the frame the detector cuts at - the first frame of the new content. It is named by its offset in the reel and never by its position in the run - see issue #21 and hop_frame_names. node, url and summary of the walk stay empty: they come from the transcript and are never guessed from the picture.

Parameters:

Name	Type	Description	Default
config	HopConfig	the values selecting and parameterizing the detector.	required
out_dir	Optional[str]	directory the evidence frames and the reel.yaml that carries them with the values reproducing them are written to; a hopless reel.yaml already there is the input of the run and its recording values are kept. None writes nothing.	None
progress	bool	show the tqdm progress bar while detecting.	False
force	bool	overwrite a hop set that is already there.	False

Returns:

Type	Description
HopContents	the hops of this run.

Raises:

Type	Description
ValueError	if the output directory already holds a hop set and force is not given.

Source code in rdd/hopdetector.py

def hops(
    self,
    config: HopConfig,
    out_dir: Optional[str] = None,
    progress: bool = False,
    force: bool = False,
) -> HopContents:
    """Turn the cuts of the given detector into hop records.

    The evidence frame of a hop is the frame the detector cuts at -
    the first frame of the new content. It is named by its offset in
    the reel and never by its position in the run - see issue #21 and
    hop_frame_names. node, url and summary of the walk stay empty:
    they come from the transcript and are never guessed from the
    picture.

    Args:
        config: the values selecting and parameterizing the detector.
        out_dir: directory the evidence frames and the reel.yaml that
            carries them with the values reproducing them are written
            to; a hopless reel.yaml already there is the input of the
            run and its recording values are kept. None writes
            nothing.
        progress: show the tqdm progress bar while detecting.
        force: overwrite a hop set that is already there.

    Returns:
        the hops of this run.

    Raises:
        ValueError: if the output directory already holds a hop set
            and force is not given.
    """
    # no back reference on the hop: the reel file it lives in is the reel
    hop_contents = HopContents()
    given = None
    if out_dir is not None:
        given = self.clear(out_dir, force)
        os.makedirs(out_dir, exist_ok=True)
    frame_nums = self.scenes(config, progress=progress)
    times_sec = [self.reel.time_of(frame_num) for frame_num in frame_nums]
    names = hop_frame_names(times_sec)
    for frame_num, name in zip(frame_nums, names):
        frame = self.reel.frame_at(frame_num)
        screenshot = None
        if out_dir is not None and frame is not None:
            screenshot = name
            frame.save(os.path.join(out_dir, screenshot))
        hop_contents.add(HopContent(time=frame.timecode, screenshot=screenshot))
    if out_dir is not None:
        hop_set = HopSet(
            recording=self.recording_of(given.recording if given else None),
            config=config,
            hops=hop_contents.hops,
        )
        hop_set.save(
            HopSet.path_of(out_dir),
            version=Version.version,
            date=datetime.now().strftime("%Y-%m-%d"),
        )
    return hop_contents

recording_of(given=None)

The Recording record of the reel this run analyzed.

Parameters:

Name	Type	Description	Default
given	Optional[Recording]	the Recording of a reel file written beforehand; its values win, because they are what a person knew and the reel cannot answer.	None

Returns:

Type	Description
Recording	the Recording, with the fields the reel itself can answer
Recording	filled in where the given one leaves them open.

Source code in rdd/hopdetector.py

def recording_of(self, given: Optional[Recording] = None) -> Recording:
    """The Recording record of the reel this run analyzed.

    Args:
        given: the Recording of a reel file written beforehand; its
            values win, because they are what a person knew and the
            reel cannot answer.

    Returns:
        the Recording, with the fields the reel itself can answer
        filled in where the given one leaves them open.
    """
    recording = given if given else Recording()
    if not recording.videoFile:
        recording.videoFile = self.reel.videoFile
    if recording.durationMin is None:
        recording.durationMin = round(self.reel.duration_sec / 60.0, 1)
    return recording

scenes(config, progress=False)

Candidate hop positions from the given scene detector.

Which detector is used is the caller's choice - the library offers several and they are benchmarked against each other, so the detector is a parameter and never fixed here. Each detector carries its own thresholds in its own constructor.

See https://www.scenedetect.com/benchmarks/

Parameters:

Name	Type	Description	Default
config	HopConfig	the values selecting and parameterizing the detector; end_sec None runs to one frame before the end, since the opencv backend fails on the last frame of some files with an undefined timestamp.	required
progress	bool	show the tqdm progress bar of the library - a run over a reel takes minutes and must not be silent.	False

Returns:

Type	Description
List[int]	the frame numbers where the detector cuts, empty without a video.

Source code in rdd/hopdetector.py

def scenes(
    self,
    config: HopConfig,
    progress: bool = False,
) -> List[int]:
    """Candidate hop positions from the given scene detector.

    Which detector is used is the caller's choice - the library offers
    several and they are benchmarked against each other, so the
    detector is a parameter and never fixed here. Each detector
    carries its own thresholds in its own constructor.

    See https://www.scenedetect.com/benchmarks/

    Args:
        config: the values selecting and parameterizing the detector;
            end_sec None runs to one frame before the end, since the
            opencv backend fails on the last frame of some files with
            an undefined timestamp.
        progress: show the tqdm progress bar of the library - a run
            over a reel takes minutes and must not be silent.

    Returns:
        the frame numbers where the detector cuts, empty without a video.
    """
    candidates: List[int] = []
    if self.reel.path is not None:
        end_sec = config.end_sec
        if end_sec is None:
            end_sec = self.reel.duration_sec - 1.0 / self.reel.fps
        scenes = detect(
            self.reel.path,
            self.get_detector(config),
            start_time=config.start_sec,
            end_time=end_sec,
            show_progress=progress,
        )
        candidates = [int(start.frame_num) for start, _ in scenes]
    return candidates

hopset

Created on 2026-08-09.

the reel file - the record of one reel and its hops

@author: wf

HopSet

The record of one reel: what it is, what produced its hops and the hops themselves.

The reel and the values that decided the hop set belong in the file that carries the hops: a hop set read a year later is only evidence if it says which reel it describes and what produced it. recording is the Recording of https://contexts.bitplan.com/index.php/Concept:Recording so the file reads straight back into the model.

The file is also the input of a run: what only a person knows - the name, the acronym, the date, the participants, the language - is written into a hopless file beforehand and kept when the hops are added, so a detection never invents the identity of a reel and never loses it (issue #24).

hopCount is not stored: in this file it would be the length of the hop list a second time and could only ever disagree with it. The equality it controls is the one on the wiki page, where the hops are subobjects.

Source code in rdd/hopset.py

@lod_storable
class HopSet:
    """The record of one reel: what it is, what produced its hops and the
    hops themselves.

    The reel and the values that decided the hop set belong in the file
    that carries the hops: a hop set read a year later is only evidence
    if it says which reel it describes and what produced it. recording is
    the Recording of https://contexts.bitplan.com/index.php/Concept:Recording
    so the file reads straight back into the model.

    The file is also the input of a run: what only a person knows - the
    name, the acronym, the date, the participants, the language - is
    written into a hopless file beforehand and kept when the hops are
    added, so a detection never invents the identity of a reel and never
    loses it (issue #24).

    hopCount is not stored: in this file it would be the length of the
    hop list a second time and could only ever disagree with it. The
    equality it controls is the one on the wiki page, where the hops are
    subobjects.
    """

    FILE_NAME = "reel.yaml"

    recording: Optional[Recording] = None
    config: Optional[HopConfig] = None
    hops: List[HopContent] = None

    def __post_init__(self):
        """Start with an empty hop list where none was given."""
        if self.hops is None:
            self.hops = []

    @property
    def hopCount(self) -> int:
        """The number of hops of this reel."""
        hop_count = len(self.hops)
        return hop_count

    @classmethod
    def path_of(cls, out_dir: str) -> str:
        """Get the path of the reel file in the given directory.

        Args:
            out_dir: the recording directory.

        Returns:
            the path of the reel file.
        """
        reel_path = os.path.join(out_dir, cls.FILE_NAME)
        return reel_path

    @classmethod
    def of_dir(cls, out_dir: str) -> Optional["HopSet"]:
        """Get the reel file of the given directory.

        Args:
            out_dir: the recording directory.

        Returns:
            the hop set, None where the directory carries no reel file.
        """
        reel_path = cls.path_of(out_dir)
        hop_set = None
        if os.path.isfile(reel_path):
            hop_set = cls.load_from_yaml_file(reel_path)
        return hop_set

    def save(self, path: str, version: str, date: str) -> None:
        """Save this reel with the header that says where it comes from.

        Args:
            path: the file to write.
            version: the version of hopdetect that produced this hop set.
            date: the ISO date of the run.

        Returns:
            None
        """
        header = HEADER.format(version=version, date=date)
        yaml_text = self.to_yaml()
        with open(path, "w", encoding="utf-8") as reel_file:
            reel_file.write(header)
            reel_file.write("\n")
            reel_file.write(yaml_text)

hopCount property

The number of hops of this reel.

__post_init__()

Start with an empty hop list where none was given.

Source code in rdd/hopset.py

def __post_init__(self):
    """Start with an empty hop list where none was given."""
    if self.hops is None:
        self.hops = []

of_dir(out_dir) classmethod

Get the reel file of the given directory.

Parameters:

Name	Type	Description	Default
out_dir	str	the recording directory.	required

Returns:

Type	Description
Optional[HopSet]	the hop set, None where the directory carries no reel file.

Source code in rdd/hopset.py

@classmethod
def of_dir(cls, out_dir: str) -> Optional["HopSet"]:
    """Get the reel file of the given directory.

    Args:
        out_dir: the recording directory.

    Returns:
        the hop set, None where the directory carries no reel file.
    """
    reel_path = cls.path_of(out_dir)
    hop_set = None
    if os.path.isfile(reel_path):
        hop_set = cls.load_from_yaml_file(reel_path)
    return hop_set

path_of(out_dir) classmethod

Get the path of the reel file in the given directory.

Parameters:

Name	Type	Description	Default
out_dir	str	the recording directory.	required

Returns:

Type	Description
str	the path of the reel file.

Source code in rdd/hopset.py

@classmethod
def path_of(cls, out_dir: str) -> str:
    """Get the path of the reel file in the given directory.

    Args:
        out_dir: the recording directory.

    Returns:
        the path of the reel file.
    """
    reel_path = os.path.join(out_dir, cls.FILE_NAME)
    return reel_path

save(path, version, date)

Save this reel with the header that says where it comes from.

Parameters:

Name	Type	Description	Default
path	str	the file to write.	required
version	str	the version of hopdetect that produced this hop set.	required
date	str	the ISO date of the run.	required

Returns:

Type	Description
None	None

Source code in rdd/hopset.py

def save(self, path: str, version: str, date: str) -> None:
    """Save this reel with the header that says where it comes from.

    Args:
        path: the file to write.
        version: the version of hopdetect that produced this hop set.
        date: the ISO date of the run.

    Returns:
        None
    """
    header = HEADER.format(version=version, date=date)
    yaml_text = self.to_yaml()
    with open(path, "w", encoding="utf-8") as reel_file:
        reel_file.write(header)
        reel_file.write("\n")
        reel_file.write(yaml_text)

i18n

Created on 2026-08-14.

i18n of the reel site - de and en for a start, the default being the browser setting and a selector with flag per the i18n issue. The texts are a resource of the package: rdd/resources/i18n.yaml.

@author: wf

I18n

The i18n texts of a reel site.

Loaded from the i18n.yaml resource the package ships - the texts of the site pages and of the packaged review page, the languages and their flags.

Source code in rdd/i18n.py

@lod_storable
class I18n:
    """The i18n texts of a reel site.

    Loaded from the i18n.yaml resource the package ships - the texts
    of the site pages and of the packaged review page, the languages
    and their flags.
    """

    languages: List[str] = field(default_factory=list)
    flags: Dict[str, str] = field(default_factory=dict)
    texts: Dict[str, Dict[str, str]] = field(default_factory=dict)
    review: Dict[str, Dict[str, str]] = field(default_factory=dict)

    _instance: ClassVar[Optional["I18n"]] = None

    @classmethod
    def resource_path(cls) -> Path:
        """Path of the i18n texts shipped with the package."""
        path = Path(__file__).parent / "resources" / "i18n.yaml"
        return path

    @classmethod
    def of_resource(cls) -> "I18n":
        """Load the i18n texts shipped with the package."""
        i18n = cls.load_from_yaml_file(str(cls.resource_path()))
        return i18n

    @classmethod
    def get_instance(cls) -> "I18n":
        """Get the shared instance, loaded once from the resource."""
        if cls._instance is None:
            cls._instance = cls.of_resource()
        return cls._instance

get_instance() classmethod

Get the shared instance, loaded once from the resource.

Source code in rdd/i18n.py

@classmethod
def get_instance(cls) -> "I18n":
    """Get the shared instance, loaded once from the resource."""
    if cls._instance is None:
        cls._instance = cls.of_resource()
    return cls._instance

of_resource() classmethod

Load the i18n texts shipped with the package.

Source code in rdd/i18n.py

@classmethod
def of_resource(cls) -> "I18n":
    """Load the i18n texts shipped with the package."""
    i18n = cls.load_from_yaml_file(str(cls.resource_path()))
    return i18n

resource_path() classmethod

Path of the i18n texts shipped with the package.

Source code in rdd/i18n.py

@classmethod
def resource_path(cls) -> Path:
    """Path of the i18n texts shipped with the package."""
    path = Path(__file__).parent / "resources" / "i18n.yaml"
    return path

pick_language(query_lang=None, cookie_lang=None, accept_language=None)

Pick the language of a request.

The explicit choice wins, then the remembered one, then the browser setting - per the i18n issue the default is the browser setting.

Parameters:

Name	Type	Description	Default
query_lang	Optional[str]	the ?lang= parameter, if any.	None
cookie_lang	Optional[str]	the remembered choice, if any.	None
accept_language	Optional[str]	the Accept-Language header, if any.	None

Returns:

Type	Description
str	the language code; en where nothing decides.

Source code in rdd/i18n.py

def pick_language(
    query_lang: Optional[str] = None,
    cookie_lang: Optional[str] = None,
    accept_language: Optional[str] = None,
) -> str:
    """Pick the language of a request.

    The explicit choice wins, then the remembered one, then the browser
    setting - per the i18n issue the default is the browser setting.

    Args:
        query_lang: the ?lang= parameter, if any.
        cookie_lang: the remembered choice, if any.
        accept_language: the Accept-Language header, if any.

    Returns:
        the language code; en where nothing decides.
    """
    lang = "en"
    if query_lang in LANGUAGES:
        lang = query_lang
    elif cookie_lang in LANGUAGES:
        lang = cookie_lang
    elif accept_language:
        for part in accept_language.split(","):
            code = part.split(";")[0].strip().lower()[:2]
            if code in LANGUAGES:
                lang = code
                break
    return lang

review_texts(lang)

The review page texts of the given language.

Parameters:

Name	Type	Description	Default
lang	str	the language code.	required

Returns:

Type	Description
Dict[str, str]	the texts; english where the language is not carried.

Source code in rdd/i18n.py

def review_texts(lang: str) -> Dict[str, str]:
    """The review page texts of the given language.

    Args:
        lang: the language code.

    Returns:
        the texts; english where the language is not carried.
    """
    i18n = I18n.get_instance()
    lang_texts = i18n.review.get(lang, i18n.review["en"])
    return lang_texts

texts(lang)

The site page texts of the given language.

Parameters:

Name	Type	Description	Default
lang	str	the language code.	required

Returns:

Type	Description
Dict[str, str]	the texts; english where the language is not carried.

Source code in rdd/i18n.py

def texts(lang: str) -> Dict[str, str]:
    """The site page texts of the given language.

    Args:
        lang: the language code.

    Returns:
        the texts; english where the language is not carried.
    """
    i18n = I18n.get_instance()
    lang_texts = i18n.texts.get(lang, i18n.texts["en"])
    return lang_texts

icons

Created on 2026-08-12.

the material icons a reel site draws in its menu

The BITPlan applications wear the material icon set that quasar loads for nicegui. A reel site may not load a font from a foreign host - the Review UI stack decision requires that the browser loads nothing but the reel site - so the handful of icons the menu needs travel as inline svg path data taken from https://github.com/google/material-design-icons (Apache-2.0), same names, same shapes.

@author: wf

svg(name, size='1.2em')

Render the named material icon as an inline svg.

Parameters:

Name	Type	Description	Default
name	str	material icon name e.g. home.	required
size	str	css length for width and height.	'1.2em'

Returns:

Type	Description
str	the svg markup, drawn in the current text color.

Raises:

Type	Description
ValueError	if the icon is not one of the shipped ones.

Source code in rdd/icons.py

def svg(name: str, size: str = "1.2em") -> str:
    """Render the named material icon as an inline svg.

    Args:
        name: material icon name e.g. home.
        size: css length for width and height.

    Returns:
        the svg markup, drawn in the current text color.

    Raises:
        ValueError: if the icon is not one of the shipped ones.
    """
    path = ICONS.get(name)
    if path is None:
        available = ", ".join(sorted(ICONS.keys()))
        raise ValueError(f"unknown icon {name} - available: {available}")
    markup = (
        f'<svg viewBox="0 0 24 24" width="{size}" height="{size}" '
        f'fill="currentColor" aria-hidden="true"><path d="{path}"/></svg>'
    )
    return markup

mint

Created on 2026-08-13.

owner bootstrap and token minting of a reel site

per the Owner bootstrap and minting ADR on https://media.bitplan.com/index.php/Talk:Rdd.bitplan.com minting is a CLI matter - the webservice never mints.

@author: wf

Mint

Mint review tokens - the CLI side of access rights.

A site without reviews.yaml is in installation mode: init_site seeds the owner and mints the wildcard owner token. mint_review adds a reviewer. Tokens are 128-bit random - possession is the right per the Reel Review decision.

Source code in rdd/mint.py

class Mint:
    """Mint review tokens - the CLI side of access rights.

    A site without reviews.yaml is in installation mode: init_site
    seeds the owner and mints the wildcard owner token. mint_review
    adds a reviewer. Tokens are 128-bit random - possession is the
    right per the Reel Review decision.
    """

    OWNER_LINK_FILE = "owner_link.txt"

    def __init__(self, config: RddSiteConfig, rdd_path: str = "~/.rdd"):
        """Initialize with the site configuration.

        Args:
            config: the site configuration - names the url the links carry.
            rdd_path: the directory beside the site configuration where
                persons.yaml, reviews.yaml and the owner link live.
        """
        self.config = config
        self.rdd_dir = os.path.expanduser(rdd_path)
        self.reviews_path = os.path.join(self.rdd_dir, "reviews.yaml")
        self.persons_path = os.path.join(self.rdd_dir, "persons.yaml")
        self.owner_link_path = os.path.join(self.rdd_dir, self.OWNER_LINK_FILE)

    @property
    def base_url(self) -> str:
        """The base url of the site - the configured public url, localhost
        where the configuration names none."""
        base_url = self.config.url or f"http://127.0.0.1:{self.config.port}"
        return base_url

    def token(self) -> str:
        """Mint a 128-bit token.

        Returns:
            32 hex characters of cryptographic randomness.
        """
        token = secrets.token_hex(16)
        return token

    def review_url(self, review: Review) -> str:
        """The link of the given review.

        Args:
            review: the review.

        Returns:
            the url whose possession is the right.
        """
        url = f"{self.base_url}{self.config.reels_url_prefix}{review.token}"
        return url

    def init_site(self, username: str, name: str, email: str, url: str) -> str:
        """Initialize the site: seed the owner and mint the owner token.

        Per the Owner bootstrap decision the owner is written into
        persons.yaml, a wildcard Review into reviews.yaml, and the
        owner link into a file of mode 600 beside the site
        configuration - the caller shows it once on the interactive
        terminal and nowhere else.

        Args:
            username: the username of the owner.
            name: the full name of the owner.
            email: the email of the owner.
            url: the url of the owner.

        Returns:
            the owner link.

        Raises:
            ValueError: where reviews.yaml already exists - a site is
                initialized exactly once.
        """
        if os.path.isfile(self.reviews_path):
            raise ValueError(
                f"{self.reviews_path} exists - the site is already initialized"
            )
        os.makedirs(self.rdd_dir, exist_ok=True)
        persons = Persons.of_path(self.persons_path)
        persons.persons.append(
            Person(username=username, name=name, email=email, url=url)
        )
        persons.save_to_yaml_file(self.persons_path)
        review = Review(
            token=self.token(),
            person=username,
            meeting="owner bootstrap",
            reels=[Review.WILDCARD],
        )
        reviews = Reviews(reviews=[review])
        self.save_reviews(reviews)
        owner_url = self.review_url(review)
        descriptor = os.open(
            self.owner_link_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600
        )
        with os.fdopen(descriptor, "w") as owner_link_file:
            owner_link_file.write(f"{owner_url}\n")
        return owner_url

    def save_reviews(self, reviews: Reviews):
        """Save the given reviews with owner-only permissions.

        The tokens in reviews.yaml are the rights themselves, so the
        file is as protected as the owner link.

        Args:
            reviews: the reviews to save.
        """
        reviews.save_to_yaml_file(self.reviews_path)
        os.chmod(self.reviews_path, 0o600)

    def mint_review(
        self,
        person: str,
        meeting: str = "",
        reels: Optional[List[str]] = None,
    ) -> str:
        """Mint a review token for the given person.

        Args:
            person: the person the link is for.
            meeting: the meeting the review belongs to.
            reels: the acronyms the review grants.

        Returns:
            the review link.

        Raises:
            ValueError: where the site is not initialized - init_site
                comes first.
        """
        if not os.path.isfile(self.reviews_path):
            raise ValueError(
                f"no {self.reviews_path} - initialize the site with rdd site --init"
            )
        reviews = Reviews.of_path(self.reviews_path)
        review = Review(
            token=self.token(),
            person=person,
            meeting=meeting,
            reels=list(reels or []),
        )
        reviews.reviews.append(review)
        self.save_reviews(reviews)
        url = self.review_url(review)
        return url

base_url property

The base url of the site - the configured public url, localhost where the configuration names none.

__init__(config, rdd_path='~/.rdd')

Initialize with the site configuration.

Parameters:

Name	Type	Description	Default
config	RddSiteConfig	the site configuration - names the url the links carry.	required
rdd_path	str	the directory beside the site configuration where persons.yaml, reviews.yaml and the owner link live.	'~/.rdd'

Source code in rdd/mint.py

def __init__(self, config: RddSiteConfig, rdd_path: str = "~/.rdd"):
    """Initialize with the site configuration.

    Args:
        config: the site configuration - names the url the links carry.
        rdd_path: the directory beside the site configuration where
            persons.yaml, reviews.yaml and the owner link live.
    """
    self.config = config
    self.rdd_dir = os.path.expanduser(rdd_path)
    self.reviews_path = os.path.join(self.rdd_dir, "reviews.yaml")
    self.persons_path = os.path.join(self.rdd_dir, "persons.yaml")
    self.owner_link_path = os.path.join(self.rdd_dir, self.OWNER_LINK_FILE)

init_site(username, name, email, url)

Initialize the site: seed the owner and mint the owner token.

Per the Owner bootstrap decision the owner is written into persons.yaml, a wildcard Review into reviews.yaml, and the owner link into a file of mode 600 beside the site configuration - the caller shows it once on the interactive terminal and nowhere else.

Parameters:

Name	Type	Description	Default
username	str	the username of the owner.	required
name	str	the full name of the owner.	required
email	str	the email of the owner.	required
url	str	the url of the owner.	required

Returns:

Type	Description
str	the owner link.

Raises:

Type	Description
ValueError	where reviews.yaml already exists - a site is initialized exactly once.

Source code in rdd/mint.py

def init_site(self, username: str, name: str, email: str, url: str) -> str:
    """Initialize the site: seed the owner and mint the owner token.

    Per the Owner bootstrap decision the owner is written into
    persons.yaml, a wildcard Review into reviews.yaml, and the
    owner link into a file of mode 600 beside the site
    configuration - the caller shows it once on the interactive
    terminal and nowhere else.

    Args:
        username: the username of the owner.
        name: the full name of the owner.
        email: the email of the owner.
        url: the url of the owner.

    Returns:
        the owner link.

    Raises:
        ValueError: where reviews.yaml already exists - a site is
            initialized exactly once.
    """
    if os.path.isfile(self.reviews_path):
        raise ValueError(
            f"{self.reviews_path} exists - the site is already initialized"
        )
    os.makedirs(self.rdd_dir, exist_ok=True)
    persons = Persons.of_path(self.persons_path)
    persons.persons.append(
        Person(username=username, name=name, email=email, url=url)
    )
    persons.save_to_yaml_file(self.persons_path)
    review = Review(
        token=self.token(),
        person=username,
        meeting="owner bootstrap",
        reels=[Review.WILDCARD],
    )
    reviews = Reviews(reviews=[review])
    self.save_reviews(reviews)
    owner_url = self.review_url(review)
    descriptor = os.open(
        self.owner_link_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600
    )
    with os.fdopen(descriptor, "w") as owner_link_file:
        owner_link_file.write(f"{owner_url}\n")
    return owner_url

mint_review(person, meeting='', reels=None)

Mint a review token for the given person.

Parameters:

Name	Type	Description	Default
person	str	the person the link is for.	required
meeting	str	the meeting the review belongs to.	''
reels	Optional[List[str]]	the acronyms the review grants.	None

Returns:

Type	Description
str	the review link.

Raises:

Type	Description
ValueError	where the site is not initialized - init_site comes first.

Source code in rdd/mint.py

def mint_review(
    self,
    person: str,
    meeting: str = "",
    reels: Optional[List[str]] = None,
) -> str:
    """Mint a review token for the given person.

    Args:
        person: the person the link is for.
        meeting: the meeting the review belongs to.
        reels: the acronyms the review grants.

    Returns:
        the review link.

    Raises:
        ValueError: where the site is not initialized - init_site
            comes first.
    """
    if not os.path.isfile(self.reviews_path):
        raise ValueError(
            f"no {self.reviews_path} - initialize the site with rdd site --init"
        )
    reviews = Reviews.of_path(self.reviews_path)
    review = Review(
        token=self.token(),
        person=person,
        meeting=meeting,
        reels=list(reels or []),
    )
    reviews.reviews.append(review)
    self.save_reviews(reviews)
    url = self.review_url(review)
    return url

review_url(review)

The link of the given review.

Parameters:

Name	Type	Description	Default
review	Review	the review.	required

Returns:

Type	Description
str	the url whose possession is the right.

Source code in rdd/mint.py

def review_url(self, review: Review) -> str:
    """The link of the given review.

    Args:
        review: the review.

    Returns:
        the url whose possession is the right.
    """
    url = f"{self.base_url}{self.config.reels_url_prefix}{review.token}"
    return url

save_reviews(reviews)

Save the given reviews with owner-only permissions.

The tokens in reviews.yaml are the rights themselves, so the file is as protected as the owner link.

Parameters:

Name	Type	Description	Default
reviews	Reviews	the reviews to save.	required

Source code in rdd/mint.py

def save_reviews(self, reviews: Reviews):
    """Save the given reviews with owner-only permissions.

    The tokens in reviews.yaml are the rights themselves, so the
    file is as protected as the owner link.

    Args:
        reviews: the reviews to save.
    """
    reviews.save_to_yaml_file(self.reviews_path)
    os.chmod(self.reviews_path, 0o600)

token()

Mint a 128-bit token.

Returns:

Type	Description
str	32 hex characters of cryptographic randomness.

Source code in rdd/mint.py

def token(self) -> str:
    """Mint a 128-bit token.

    Returns:
        32 hex characters of cryptographic randomness.
    """
    token = secrets.token_hex(16)
    return token

palette

Created on 2026-08-12.

the color palette of a reel site

@author: wf

Palette dataclass

One Material palette schema.

The eight values are the ones the ColorSchema of ngwidgets carries, so a reel site looks like the other BITPlan applications without depending on the library.

Source code in rdd/palette.py

@dataclass
class Palette:
    """One Material palette schema.

    The eight values are the ones the ColorSchema of ngwidgets carries,
    so a reel site looks like the other BITPlan applications without
    depending on the library.
    """

    primary: str = "#5898d4"
    secondary: str = "#26a69a"
    accent: str = "#9c27b0"
    dark: str = "#1d1d1d"
    positive: str = "#21ba45"
    negative: str = "#c10015"
    info: str = "#31ccec"
    warning: str = "#f2c037"

    def as_css(self) -> str:
        """Render the palette as CSS custom properties.

        Returns:
            the eight values as --name: value declarations.
        """
        css = "\n".join(
            f"  --{name}: {value};" for name, value in self.__dict__.items()
        )
        return css

as_css()

Render the palette as CSS custom properties.

Returns:

Type	Description
str	the eight values as --name: value declarations.

Source code in rdd/palette.py

def as_css(self) -> str:
    """Render the palette as CSS custom properties.

    Returns:
        the eight values as --name: value declarations.
    """
    css = "\n".join(
        f"  --{name}: {value};" for name, value in self.__dict__.items()
    )
    return css

Palettes

The palette schemas a reel site may name.

Source code in rdd/palette.py

@lod_storable
class Palettes:
    """The palette schemas a reel site may name."""

    palettes: Dict[str, Palette] = field(default_factory=dict)

    @classmethod
    def resource_path(cls) -> Path:
        """Path of the palettes shipped with the package."""
        path = Path(__file__).parent / "resources" / "palettes.yaml"
        return path

    @classmethod
    def of_resource(cls) -> "Palettes":
        """Load the palettes shipped with the package."""
        palettes = cls.load_from_yaml_file(str(cls.resource_path()))
        return palettes

    def by_name(self, name: str) -> Palette:
        """Get the palette of the given name.

        Args:
            name: name of a Material palette schema e.g. blue_grey.

        Returns:
            the palette.

        Raises:
            ValueError: if no palette of that name is shipped.
        """
        palette = self.palettes.get(name)
        if palette is None:
            available = ", ".join(sorted(self.palettes.keys()))
            raise ValueError(f"unknown palette {name} - available: {available}")
        return palette

by_name(name)

Get the palette of the given name.

Parameters:

Name	Type	Description	Default
name	str	name of a Material palette schema e.g. blue_grey.	required

Returns:

Type	Description
Palette	the palette.

Raises:

Type	Description
ValueError	if no palette of that name is shipped.

Source code in rdd/palette.py

def by_name(self, name: str) -> Palette:
    """Get the palette of the given name.

    Args:
        name: name of a Material palette schema e.g. blue_grey.

    Returns:
        the palette.

    Raises:
        ValueError: if no palette of that name is shipped.
    """
    palette = self.palettes.get(name)
    if palette is None:
        available = ", ".join(sorted(self.palettes.keys()))
        raise ValueError(f"unknown palette {name} - available: {available}")
    return palette

of_resource() classmethod

Load the palettes shipped with the package.

Source code in rdd/palette.py

@classmethod
def of_resource(cls) -> "Palettes":
    """Load the palettes shipped with the package."""
    palettes = cls.load_from_yaml_file(str(cls.resource_path()))
    return palettes

resource_path() classmethod

Path of the palettes shipped with the package.

Source code in rdd/palette.py

@classmethod
def resource_path(cls) -> Path:
    """Path of the palettes shipped with the package."""
    path = Path(__file__).parent / "resources" / "palettes.yaml"
    return path

rdd_cmd

Created on 2026-08-14.

rdd - the dispatcher command of Reel Driven Development

rdd is the name of what we do, so rdd is the one command name a user has to know; each subcommand forwards to the tool of the pipeline and the tool names stay available as entry points of their own.

@author: wf

RddCmd

Bases: BaseCmd

The rdd dispatcher - answers what rdd can do.

Dispatching itself happens in main before argument parsing, so a subcommand owns its own arguments; this class only serves the case of no subcommand - the standard options and the list of subcommands.

Source code in rdd/rdd_cmd.py

class RddCmd(BaseCmd):
    """The rdd dispatcher - answers what rdd can do.

    Dispatching itself happens in main before argument parsing, so a
    subcommand owns its own arguments; this class only serves the case
    of no subcommand - the standard options and the list of
    subcommands.
    """

    def __init__(self):
        """Initialize with the reel-driven-development version info."""
        super().__init__(Version())

    def add_arguments(self, parser: argparse.ArgumentParser):
        """Add the subcommand overview to the given parser.

        Args:
            parser: the parser to add arguments to.
        """
        super().add_arguments(parser)
        parser.description = (
            "Reel Driven Development - subcommands: "
            "detect (find the hops of a reel), "
            "doc (generate the reel document), "
            "review (serve one reel for review), "
            "site (serve, initialize or mint for the reel site); "
            "rdd <subcommand> --help shows the arguments of a subcommand"
        )

    def handle_args(self, args: argparse.Namespace) -> bool:
        """Handle the parsed arguments - without a subcommand the help is
        the answer.

        Args:
            args: parsed argument namespace.

        Returns:
            True if the arguments were handled.
        """
        handled = super().handle_args(args)
        if not handled:
            self.parser.print_help()
            handled = True
        return handled

__init__()

Initialize with the reel-driven-development version info.

Source code in rdd/rdd_cmd.py

def __init__(self):
    """Initialize with the reel-driven-development version info."""
    super().__init__(Version())

add_arguments(parser)

Add the subcommand overview to the given parser.

Parameters:

Name	Type	Description	Default
parser	ArgumentParser	the parser to add arguments to.	required

Source code in rdd/rdd_cmd.py

def add_arguments(self, parser: argparse.ArgumentParser):
    """Add the subcommand overview to the given parser.

    Args:
        parser: the parser to add arguments to.
    """
    super().add_arguments(parser)
    parser.description = (
        "Reel Driven Development - subcommands: "
        "detect (find the hops of a reel), "
        "doc (generate the reel document), "
        "review (serve one reel for review), "
        "site (serve, initialize or mint for the reel site); "
        "rdd <subcommand> --help shows the arguments of a subcommand"
    )

handle_args(args)

Handle the parsed arguments - without a subcommand the help is the answer.

Parameters:

Name	Type	Description	Default
args	Namespace	parsed argument namespace.	required

Returns:

Type	Description
bool	True if the arguments were handled.

Source code in rdd/rdd_cmd.py

def handle_args(self, args: argparse.Namespace) -> bool:
    """Handle the parsed arguments - without a subcommand the help is
    the answer.

    Args:
        args: parsed argument namespace.

    Returns:
        True if the arguments were handled.
    """
    handled = super().handle_args(args)
    if not handled:
        self.parser.print_help()
        handled = True
    return handled

main(argv=None)

Command line entry point of the rdd dispatcher.

Parameters:

Name	Type	Description	Default
argv	Optional[List[str]]	command line arguments; defaults to sys.argv.	None

Returns:

Type	Description
int	the exit code of the subcommand, or 0 = OK, 1 =
int	KeyboardInterrupt, 2 = Exception of the dispatcher itself.

Source code in rdd/rdd_cmd.py

def main(argv: Optional[List[str]] = None) -> int:
    """Command line entry point of the rdd dispatcher.

    Args:
        argv: command line arguments; defaults to sys.argv.

    Returns:
        the exit code of the subcommand, or 0 = OK, 1 =
        KeyboardInterrupt, 2 = Exception of the dispatcher itself.
    """
    args = sys.argv[1:] if argv is None else argv
    module_name = SUBCOMMANDS.get(args[0]) if args else None
    if module_name is not None:
        module = importlib.import_module(module_name)
        exit_code = module.main(args[1:])
    else:
        cmd = RddCmd()
        exit_code = cmd.run(args)
    return exit_code

rdd_site

Created on 2026-08-12.

the site of an organization's reel driven development videos - home page, menu and about

see https://media.bitplan.com/index.php/Talk:Rdd.bitplan.com ADRs: Review UI stack, Home page and menu

@author: wf

MenuEntry dataclass

One entry of the menu.

Source code in rdd/rdd_site.py

@dataclass
class MenuEntry:
    """One entry of the menu."""

    name: str
    target: str
    icon: str
    new_tab: bool = False

    def as_html(self) -> str:
        """Render the entry as an icon labelled link button.

        Returns:
            the anchor markup carrying the material icon and the name.
        """
        target = html.escape(self.target)
        new_tab = " target=_blank" if self.new_tab else ""
        markup = (
            f'      <a href="{target}"{new_tab}>{svg(self.icon)}'
            f"<span>{html.escape(self.name)}</span></a>"
        )
        return markup

as_html()

Render the entry as an icon labelled link button.

Returns:

Type	Description
str	the anchor markup carrying the material icon and the name.

Source code in rdd/rdd_site.py

def as_html(self) -> str:
    """Render the entry as an icon labelled link button.

    Returns:
        the anchor markup carrying the material icon and the name.
    """
    target = html.escape(self.target)
    new_tab = " target=_blank" if self.new_tab else ""
    markup = (
        f'      <a href="{target}"{new_tab}>{svg(self.icon)}'
        f"<span>{html.escape(self.name)}</span></a>"
    )
    return markup

Person

One person of a site - the seed minimum of the Owner bootstrap decision.

The username is the key; full name, email and url are what a site needs to address its people.

Source code in rdd/rdd_site.py

@lod_storable
class Person:
    """One person of a site - the seed minimum of the Owner bootstrap
    decision.

    The username is the key; full name, email and url are what a site
    needs to address its people.
    """

    username: str = ""
    name: str = ""
    email: str = ""
    url: str = ""

Persons

The persons of a site.

Source code in rdd/rdd_site.py

@lod_storable
class Persons:
    """The persons of a site."""

    DEFAULT_PATH = "~/.rdd/persons.yaml"

    persons: List[Person] = field(default_factory=list)

    @classmethod
    def of_path(cls, path: Optional[str] = None) -> "Persons":
        """Load the persons from the given yaml file.

        Args:
            path: the persons file; the default path when None.

        Returns:
            the persons; none where no file exists.
        """
        persons_path = os.path.expanduser(path or cls.DEFAULT_PATH)
        if os.path.isfile(persons_path):
            persons = cls.load_from_yaml_file(persons_path)
        else:
            persons = cls()
        return persons

of_path(path=None) classmethod

Load the persons from the given yaml file.

Parameters:

Name	Type	Description	Default
path	Optional[str]	the persons file; the default path when None.	None

Returns:

Type	Description
Persons	the persons; none where no file exists.

Source code in rdd/rdd_site.py

@classmethod
def of_path(cls, path: Optional[str] = None) -> "Persons":
    """Load the persons from the given yaml file.

    Args:
        path: the persons file; the default path when None.

    Returns:
        the persons; none where no file exists.
    """
    persons_path = os.path.expanduser(path or cls.DEFAULT_PATH)
    if os.path.isfile(persons_path):
        persons = cls.load_from_yaml_file(persons_path)
    else:
        persons = cls()
    return persons

RddSiteConfig

Everything an organization configures to run its own reel site.

One yaml is the whole configuration - the site names itself, its repository, its documentation, the palette it wears and its recordings directory, so a stranger's site names their project and not ours.

The reels are not configured: the site names the recordings directory and the reels are what that directory holds, so publishing a reel is putting its folder there. main_demo names the demo the home page offers - a site has at least one reel in demo status.

Source code in rdd/rdd_site.py

@lod_storable
class RddSiteConfig:
    """Everything an organization configures to run its own reel site.

    One yaml is the whole configuration - the site names itself, its
    repository, its documentation, the palette it wears and its recordings
    directory, so a stranger's site names their project and not ours.

    The reels are not configured: the site names the recordings directory
    and the reels are what that directory holds, so publishing a reel is
    putting its folder there. main_demo names the demo the home page
    offers - a site has at least one reel in demo status.
    """

    DEFAULT_PATH = "~/.rdd/rdd_site.yaml"

    name: str = "reels"
    title: str = "Reels"
    intro: str = ""
    url: str = ""
    palette: str = "indigo"
    copy_right: str = ""
    cm_url: str = Version.cm_url
    doc_url: str = Version.doc_url
    port: int = 9925
    recordings_path: str = "~/.rdd/recordings"
    main_demo: str = ""
    max_reels_space_gb: int = 100
    reels_url_prefix: str = "/reels/"

    @classmethod
    def of_file(cls, path: str) -> "RddSiteConfig":
        """Load the site configuration from the given yaml file."""
        config = cls.load_from_yaml_file(path)
        return config

    @classmethod
    def of_path(cls, path: Optional[str] = None) -> "RddSiteConfig":
        """Load the site configuration from the given yaml file.

        Args:
            path: the configuration file; the default path when None.

        Returns:
            the configuration; the default configuration where no file exists.
        """
        config_path = os.path.expanduser(path or cls.DEFAULT_PATH)
        if os.path.isfile(config_path):
            config = cls.of_file(config_path)
        else:
            config = cls()
        return config

    @property
    def recordings_dir(self) -> str:
        """The recordings directory with the user's home resolved."""
        recordings_dir = os.path.expanduser(self.recordings_path)
        return recordings_dir

recordings_dir property

The recordings directory with the user's home resolved.

of_file(path) classmethod

Load the site configuration from the given yaml file.

Source code in rdd/rdd_site.py

@classmethod
def of_file(cls, path: str) -> "RddSiteConfig":
    """Load the site configuration from the given yaml file."""
    config = cls.load_from_yaml_file(path)
    return config

of_path(path=None) classmethod

Load the site configuration from the given yaml file.

Parameters:

Name	Type	Description	Default
path	Optional[str]	the configuration file; the default path when None.	None

Returns:

Type	Description
RddSiteConfig	the configuration; the default configuration where no file exists.

Source code in rdd/rdd_site.py

@classmethod
def of_path(cls, path: Optional[str] = None) -> "RddSiteConfig":
    """Load the site configuration from the given yaml file.

    Args:
        path: the configuration file; the default path when None.

    Returns:
        the configuration; the default configuration where no file exists.
    """
    config_path = os.path.expanduser(path or cls.DEFAULT_PATH)
    if os.path.isfile(config_path):
        config = cls.of_file(config_path)
    else:
        config = cls()
    return config

ReelSite

The pages of a reel site.

The layout is the one the BITPlan applications share - a menu with home, github, help and about, a footer with copyright and version, per the Home page and menu decision. It is rendered here rather than imported so that a reel site needs python and nothing else.

Source code in rdd/rdd_site.py

class ReelSite:
    """The pages of a reel site.

    The layout is the one the BITPlan applications share - a menu with home,
    github, help and about, a footer with copyright and version, per the Home
    page and menu decision. It is rendered here rather than imported so that a
    reel site needs python and nothing else.
    """

    def __init__(
        self,
        config: RddSiteConfig,
        version: Optional[Version] = None,
        reels: Optional[Reels] = None,
        reviews: Optional[Reviews] = None,
    ):
        """Initialize with the site configuration.

        Args:
            config: the configuration of this site.
            version: version info of the software; defaults to the package version.
            reels: the reels of this site; scanned from the configuration when None.
            reviews: the review rights of this site; loaded from the default
                path when None.
        """
        self.config = config
        self.version = version or Version()
        self.palette: Palette = Palettes.of_resource().by_name(config.palette)
        self.reels_found = reels if reels is not None else self.scan()
        self.reviews = reviews if reviews is not None else Reviews.of_path()

    def scan(self) -> Reels:
        """Scan the recordings directory of this site into its directory of
        reels.

        Returns:
            the reels the recordings directory holds.
        """
        reels = Reels.of_dir(self.config.recordings_dir)
        return reels

    def check_main_demo(self) -> Reel:
        """Get the mandatory main demo of this site.

        Returns:
            the reel the configuration names as main_demo.

        Raises:
            ValueError: where main_demo is unset, unknown or not in demo
                status - a site has at least one demo.
        """
        directory = self.reels_found.by_acronym()
        main_demo = directory.get(self.config.main_demo)
        if main_demo is None:
            raise ValueError(
                f"main_demo '{self.config.main_demo}' is not a reel of "
                f"{self.config.recordings_dir}"
            )
        if not main_demo.is_demo:
            raise ValueError(
                f"main_demo '{self.config.main_demo}' has status "
                f"'{main_demo.status}' - demo is required"
            )
        return main_demo

    def reel_url(self, reel: Reel, review: Optional[Review] = None) -> str:
        """The url this site serves the given reel under.

        Per the Delivery decision the acronym is the only address a url
        needs; a review link carries its token before the acronym.

        Args:
            reel: the reel.
            review: the review whose token the url carries; None for the
                public url.

        Returns:
            the url of the reel.
        """
        prefix = self.config.reels_url_prefix
        if review:
            prefix = f"{prefix}{review.token}/"
        url = f"{prefix}{reel.acronym}/"
        return url

    def allowed(self, reel: Reel, review: Optional[Review] = None) -> bool:
        """Whether the holder of the given right may inspect the reel.

        Args:
            reel: the reel.
            review: the review right; None for anonymous.

        Returns:
            True for a public or demo reel, or a reel the review grants -
            every reel where the review grants the wildcard.
        """
        granted = review.reels if review else []
        allowed = (
            reel.is_public or Review.WILDCARD in granted or reel.acronym in granted
        )
        return allowed

    def resolve_reel(self, parts: List[str]):
        """Resolve the addressed reel - shortcut or lengthy form.

        Per the Hop url decision the acronym is the shortcut address
        and year/month/acronym the lengthy form disambiguating
        non-unique acronyms.

        Args:
            parts: the path parts after /reels/ with the token stripped.

        Returns:
            the reel or None, and the remaining path parts.
        """
        reel = None
        file_parts: List[str] = []
        if (
            len(parts) >= 3
            and re.match(r"\d{4}$", parts[0])
            and re.match(r"\d{2}$", parts[1])
        ):
            reel = self.reels_found.by_pid().get("/".join(parts[:3]))
            file_parts = [part for part in parts[3:] if part]
        elif parts:
            reel = self.reels_found.by_acronym().get(parts[0])
            file_parts = [part for part in parts[1:] if part]
        return reel, file_parts

    def reel_files(self, reel: Reel) -> List[str]:
        """The files of the given reel folder.

        Args:
            reel: the reel.

        Returns:
            the sorted file names, hidden files and the review page
            excluded - the review page is code of the package, never
            data of a reel.
        """
        names = [
            name
            for name in sorted(os.listdir(reel.path))
            if not name.startswith(".")
            and name != "reelreview.html"
            and os.path.isfile(os.path.join(reel.path, name))
        ]
        return names

    def reel_page(self, reel: Reel, lang: str = "en") -> str:
        """The page of one reel - what it is and the files it carries.

        The review and file links are relative so a review link keeps
        its token.

        Args:
            reel: the reel.
            lang: the language of the page.

        Returns:
            the reel page.
        """
        recording = reel.recording
        summary = recording.summary if recording and recording.summary else ""
        rows = "\n".join(
            f'<tr><td><a href="{html.escape(urllib.parse.quote(name))}">'
            f"{html.escape(name)}</a></td>"
            f"<td>{os.path.getsize(os.path.join(reel.path, name))}</td></tr>"
            for name in self.reel_files(reel)
        )
        t = texts(lang)
        content = (
            f"<h2>{html.escape(reel.title)}</h2>\n"
            f'<p><a href="review">{t["review"]}</a></p>\n'
            f'<div class="card">\n<b>{t["summary"]}</b><br>\n'
            f"{html.escape(summary)}\n</div>\n"
            f'<div class="card">\n<table>\n'
            f'<tr><th>{t["file"]}</th><th>{t["bytes"]}</th></tr>\n'
            f"{rows}\n</table>\n</div>"
        )
        page = self.page(reel.acronym, content, lang)
        return page

    def review_page(self, lang: str = "en") -> str:
        """The review page of this site.

        Per the Review UI stack decision the packaged page is one
        self-contained file; serving it, the site derives the CSS
        variables from its named palette, puts its own menu into
        the marked header slot and speaks the visitor's language via
        the marked i18n slot, so the review wears the same palette,
        menu and language as every other page of the site - one
        source, no second copy to keep in step.

        Args:
            lang: the language of the page.

        Returns:
            the review page in the site's palette, menu and language.
        """
        page = page_path().read_text()
        for name, value in self.palette.__dict__.items():
            page = re.sub(rf"--{name}: #[0-9A-Fa-f]+;", f"--{name}: {value};", page)
        links = "\n".join(entry.as_html() for entry in self.menu(lang))
        page = re.sub(
            r"<!-- menu -->.*?<!-- /menu -->",
            f"<!-- menu -->\n{links}\n  <!-- /menu -->",
            page,
            flags=re.DOTALL,
        )
        translations = json.dumps(review_texts(lang), ensure_ascii=False)
        page = re.sub(
            r"/\* <i18n> \*/.*?/\* </i18n> \*/",
            f'/* <i18n> */\nconst LANG = "{lang}";\n'
            f"const T = {translations};\n/* </i18n> */",
            page,
            flags=re.DOTALL,
        )
        page = page.replace('<html lang="en">', f'<html lang="{lang}">', 1)
        return page

    def reel_zip(self, reel: Reel) -> str:
        """Zip the folder of the given reel into a temporary file.

        Per the Reel verdict decision the verdict page offers the reel
        folder as one download; videos do not compress, so the entries
        are stored.

        Args:
            reel: the reel.

        Returns:
            the path of the temporary zip file - the caller removes it
            after delivery.
        """
        handle, zip_path = tempfile.mkstemp(suffix=".zip", prefix=f"{reel.acronym}-")
        os.close(handle)
        with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_STORED) as zip_file:
            for root, _dirs, files in os.walk(reel.path):
                for file_name in sorted(files):
                    file_path = os.path.join(root, file_name)
                    arcname = os.path.join(
                        reel.acronym, os.path.relpath(file_path, reel.path)
                    )
                    zip_file.write(file_path, arcname)
        return zip_path

    def menu(self, lang: str = "en") -> List[MenuEntry]:
        """The menu entries - settings and chat are dropped, a visitor has
        neither.

        Args:
            lang: the language of the entry names.
        """
        t = texts(lang)
        entries = [
            MenuEntry(t["home"], "/", "home"),
            MenuEntry(t["reels"], "/reels", "movie"),
            MenuEntry(t["github"], self.config.cm_url, "bug_report", new_tab=True),
            MenuEntry(t["help"], self.config.doc_url, "help", new_tab=True),
            MenuEntry(t["about"], "/about", "info"),
        ]
        return entries

    def style(self) -> str:
        """The stylesheet, derived from the named palette."""
        css = f""":root {{
{self.palette.as_css()}
  --bg: #fafafa; --fg: #222; --card: #fff; --border: #ccc;
}}
@media (prefers-color-scheme: dark) {{
  :root {{ --bg: var(--dark); --fg: #ddd; --card: #2a2a2a; --border: #555; }}
}}
body {{ margin: 0; font-family: system-ui, sans-serif; background: var(--bg); color: var(--fg); }}
header {{ display: flex; gap: .8em; align-items: center; padding: .6em .8em; background: var(--primary); color: #fff; flex-wrap: wrap; }}
header h1 {{ font-size: 1.1em; margin: 0 .4em 0 0; }}
header a {{ display: inline-flex; align-items: center; gap: .5em; color: #fff; text-decoration: none; background: var(--primary); border-radius: 4px; padding: .5em 1em; font-size: .85em; font-weight: 500; text-transform: uppercase; letter-spacing: .05em; box-shadow: 0 1px 5px rgba(0,0,0,.2), 0 2px 2px rgba(0,0,0,.14), 0 3px 1px -2px rgba(0,0,0,.12); }}
header a:hover {{ background: var(--accent); }}
header a.flag {{ box-shadow: none; text-transform: none; background: none; font-size: 1.1em; padding: .2em; margin-left: auto; }}
.hamburger {{ display: inline-flex; align-items: center; background: var(--primary); border: none; color: #fff; cursor: pointer; border-radius: 4px; padding: .45em .6em; box-shadow: 0 1px 5px rgba(0,0,0,.2), 0 2px 2px rgba(0,0,0,.14), 0 3px 1px -2px rgba(0,0,0,.12); }}
.hamburger.collapsed {{ position: fixed; top: .4em; left: .4em; z-index: 10; }}
.hamburger[hidden] {{ display: none; }}
body.collapsed header, body.collapsed footer {{ display: none; }}
main {{ padding: 1em; max-width: 55em; }}
main h2 {{ font-size: 1.2em; }}
main a {{ color: var(--primary); }}
.card {{ background: var(--card); border: 1px solid var(--border); border-radius: 6px; padding: .8em; margin-bottom: .8em; }}
footer {{ padding: .6em .8em; background: var(--primary); color: #fff; font-size: .85em; }}
footer a {{ color: #fff; }}
table {{ border-collapse: collapse; }}
td, th {{ text-align: left; padding: .2em .8em .2em 0; }}
"""
        return css

    def flag_selector(self, lang: str) -> str:
        """The language selector with flag - the other languages as links.

        Args:
            lang: the language of the current page.

        Returns:
            the selector markup.
        """
        flags = " ".join(
            f'<a class="flag" href="?lang={other}" title="{other}">{FLAGS[other]}</a>'
            for other in LANGUAGES
            if other != lang
        )
        return flags

    def page(self, title: str, content: str, lang: str = "en") -> str:
        """Render a page with menu and footer.

        Args:
            title: title of the page.
            content: the html of the page body.
            lang: the language of the page.

        Returns:
            the complete html page.
        """
        links = "\n".join(entry.as_html() for entry in self.menu(lang))
        site_title = html.escape(self.config.title)
        menu_icon = svg("menu")
        page = f"""<!DOCTYPE html>
<html lang="{lang}">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>{site_title} - {html.escape(title)}</title>
<style>
{self.style()}</style>
</head>
<body>
<button class="hamburger collapsed" id="unhide" onclick="toggleMenu()" title="show menu" hidden>{menu_icon}</button>
<header>
  <button class="hamburger" onclick="toggleMenu()" title="hide menu">{menu_icon}</button>
  <h1>{site_title}</h1>
{links}
  {self.flag_selector(lang)}
</header>
<main>
{content}
</main>
<footer>{html.escape(self.config.copy_right)} - {self.version.name} {self.version.version}</footer>
<script>
function toggleMenu() {{
  const collapsed = document.body.classList.toggle('collapsed');
  document.getElementById('unhide').hidden = !collapsed;
}}
</script>
</body>
</html>
"""
        return page

    def demo_card(self, lang: str = "en") -> str:
        """The demo section of the home page - the main_demo of this site.

        Args:
            lang: the language of the card.

        Returns:
            the demo card markup; empty where the main_demo does not
            resolve, so a misconfigured site still serves its home page.
        """
        t = texts(lang)
        card = ""
        directory = self.reels_found.by_acronym()
        main_demo = directory.get(self.config.main_demo)
        if main_demo and main_demo.is_demo:
            url = self.reel_url(main_demo)
            card = (
                f'<h2>{t["demo"]}</h2>\n<div class="card">\n'
                f'{t["demo_text"]}: <a href="{html.escape(url)}">'
                f"{html.escape(main_demo.title)}</a> "
                f'({main_demo.hop_count} {t["hops"]}) - {t["demo_hint"]}\n</div>'
            )
        return card

    def home(self, lang: str = "en") -> str:
        """The home page - what this site is and the two ways in.

        Args:
            lang: the language of the page.
        """
        t = texts(lang)
        intro = html.escape(self.config.intro) if self.config.intro else t["intro"]
        content = f"""<div class="card">
{intro}
</div>
<h2>{t["reviewing"]}</h2>
<div class="card">
{t["reviewing_text"]}
</div>
<h2>{t["browsing"]}</h2>
<div class="card">
{t["browsing_text"]}
</div>
{self.demo_card(lang)}
<h2>Reel Driven Development</h2>
<div class="card">
{t["rdd_text"]}
<a href="{html.escape(self.config.cm_url)}" target=_blank>reel-driven-development</a>,
{t["rdd_text2"]}
</div>
"""
        page = self.page(t["home"], content, lang)
        return page

    def reels(self, review: Optional[Review] = None, lang: str = "en") -> str:
        """The reels directory as the holder of the given right sees it.

        Anyone sees the public and demo reels; a Review right adds its
        private reels under the same directory.

        Args:
            review: the review right; None for the anonymous directory.
            lang: the language of the page.

        Returns:
            the reels directory page.
        """
        t = texts(lang)
        granted = review.reels if review else None
        visible_reels = self.reels_found.visible(granted)
        heading = "Reels"
        if review:
            heading = t["review_by"].format(person=review.person)
        if visible_reels:
            rows = "\n".join(
                f'<tr><td><a href="{html.escape(self.reel_url(reel, review))}">'
                f"{html.escape(reel.acronym)}</a></td>"
                f"<td>{html.escape(reel.title)}</td>"
                f"<td>{reel.hop_count}</td>"
                f"<td>{html.escape(reel.status)}</td></tr>"
                for reel in visible_reels
            )
            content = (
                f'<h2>{html.escape(heading)}</h2>\n<div class="card">\n<table>\n'
                f'<tr><th>{t["reel"]}</th><th>{t["title"]}</th>'
                f'<th>{t["hops"]}</th><th>{t["status"]}</th></tr>\n'
                f"{rows}\n</table>\n</div>"
            )
        else:
            content = (
                f'<h2>{html.escape(heading)}</h2>\n<div class="card">\n'
                f'{t["no_reels"]}\n</div>'
            )
        page = self.page(t["reels"], content, lang)
        return page

    def installation(self, reviews_file: str) -> str:
        """The installation mode page - the state a visitor sees while the
        site is not initialized.

        Per the Owner bootstrap decision a site without reviews.yaml
        refuses to serve reels and names the init command instead; the
        state is shown, never hidden behind a dead backend.

        Args:
            reviews_file: the reviews file whose absence is the state.

        Returns:
            the installation mode page.
        """
        content = f"""<h2>Installation mode</h2>
<div class="card">
This reel site is not initialized: <code>{html.escape(reviews_file)}</code>
does not exist, so no review right exists yet - not even the owner's.
No reel is served in this state.
</div>
<div class="card">
The owner initializes the site on its host - access administration needs
ssh and nothing else:
<pre>{html.escape(init_command())}</pre>
The command asks for username, full name, email and url, seeds the owner
and mints the wildcard owner token. The token is shown once on the
terminal and written beside the site configuration with mode 600; it is
never mailed, never logged and never minted via this webservice.
</div>
"""
        page = self.page("installation mode", content)
        return page

    def not_found(self, path: str, lang: str = "en") -> str:
        """The framed 404 page - it shows like any other page, with an
        example of a valid address.

        Args:
            path: the path that has no page.
            lang: the language of the page.

        Returns:
            the framed 404 page.
        """
        t = texts(lang)
        example = "/reels"
        directory = self.reels_found.by_acronym()
        main_demo = directory.get(self.config.main_demo)
        if main_demo and main_demo.is_demo:
            example = self.reel_url(main_demo)
        content = (
            f'<h2>404 - {t["not_found"]}</h2>\n'
            f'<div class="card">\n{t["no_page"]} '
            f"<code>{html.escape(path)}</code>.\n</div>\n"
            f'<div class="card">\n{t["example"]}: '
            f'<a href="{html.escape(example)}">{html.escape(example)}</a>; '
            f'{t["reels_listed"]}'
            "\n</div>"
        )
        page = self.page(t["not_found"], content, lang)
        return page

    def about(self, lang: str = "en") -> str:
        """The about page - version, license and repository.

        Args:
            lang: the language of the page.
        """
        t = texts(lang)
        version = self.version
        content = f"""<h2>{t["about_heading"]}</h2>
<div class="card">
<table>
<tr><th>{t["site"]}</th><td>{html.escape(self.config.title)}</td></tr>
<tr><th>{t["software"]}</th><td>{html.escape(version.name)}</td></tr>
<tr><th>{t["version"]}</th><td>{html.escape(version.version)}</td></tr>
<tr><th>{t["updated"]}</th><td>{html.escape(version.updated)}</td></tr>
<tr><th>{t["license"]}</th><td>Apache-2.0</td></tr>
<tr><th>{t["source"]}</th><td><a href="{html.escape(self.config.cm_url)}" target=_blank>{html.escape(self.config.cm_url)}</a></td></tr>
<tr><th>{t["documentation"]}</th><td><a href="{html.escape(self.config.doc_url)}" target=_blank>{html.escape(self.config.doc_url)}</a></td></tr>
</table>
</div>
"""
        page = self.page(t["about"], content, lang)
        return page

__init__(config, version=None, reels=None, reviews=None)

Initialize with the site configuration.

Parameters:

Name	Type	Description	Default
config	RddSiteConfig	the configuration of this site.	required
version	Optional[Version]	version info of the software; defaults to the package version.	None
reels	Optional[Reels]	the reels of this site; scanned from the configuration when None.	None
reviews	Optional[Reviews]	the review rights of this site; loaded from the default path when None.	None

Source code in rdd/rdd_site.py

def __init__(
    self,
    config: RddSiteConfig,
    version: Optional[Version] = None,
    reels: Optional[Reels] = None,
    reviews: Optional[Reviews] = None,
):
    """Initialize with the site configuration.

    Args:
        config: the configuration of this site.
        version: version info of the software; defaults to the package version.
        reels: the reels of this site; scanned from the configuration when None.
        reviews: the review rights of this site; loaded from the default
            path when None.
    """
    self.config = config
    self.version = version or Version()
    self.palette: Palette = Palettes.of_resource().by_name(config.palette)
    self.reels_found = reels if reels is not None else self.scan()
    self.reviews = reviews if reviews is not None else Reviews.of_path()

about(lang='en')

The about page - version, license and repository.

Parameters:

Name	Type	Description	Default
lang	str	the language of the page.	'en'

Source code in rdd/rdd_site.py

    def about(self, lang: str = "en") -> str:
        """The about page - version, license and repository.

        Args:
            lang: the language of the page.
        """
        t = texts(lang)
        version = self.version
        content = f"""<h2>{t["about_heading"]}</h2>
<div class="card">
<table>
<tr><th>{t["site"]}</th><td>{html.escape(self.config.title)}</td></tr>
<tr><th>{t["software"]}</th><td>{html.escape(version.name)}</td></tr>
<tr><th>{t["version"]}</th><td>{html.escape(version.version)}</td></tr>
<tr><th>{t["updated"]}</th><td>{html.escape(version.updated)}</td></tr>
<tr><th>{t["license"]}</th><td>Apache-2.0</td></tr>
<tr><th>{t["source"]}</th><td><a href="{html.escape(self.config.cm_url)}" target=_blank>{html.escape(self.config.cm_url)}</a></td></tr>
<tr><th>{t["documentation"]}</th><td><a href="{html.escape(self.config.doc_url)}" target=_blank>{html.escape(self.config.doc_url)}</a></td></tr>
</table>
</div>
"""
        page = self.page(t["about"], content, lang)
        return page

allowed(reel, review=None)

Whether the holder of the given right may inspect the reel.

Parameters:

Name	Type	Description	Default
reel	Reel	the reel.	required
review	Optional[Review]	the review right; None for anonymous.	None

Returns:

Type	Description
bool	True for a public or demo reel, or a reel the review grants -
bool	every reel where the review grants the wildcard.

Source code in rdd/rdd_site.py

def allowed(self, reel: Reel, review: Optional[Review] = None) -> bool:
    """Whether the holder of the given right may inspect the reel.

    Args:
        reel: the reel.
        review: the review right; None for anonymous.

    Returns:
        True for a public or demo reel, or a reel the review grants -
        every reel where the review grants the wildcard.
    """
    granted = review.reels if review else []
    allowed = (
        reel.is_public or Review.WILDCARD in granted or reel.acronym in granted
    )
    return allowed

check_main_demo()

Get the mandatory main demo of this site.

Returns:

Type	Description
Reel	the reel the configuration names as main_demo.

Raises:

Type	Description
ValueError	where main_demo is unset, unknown or not in demo status - a site has at least one demo.

Source code in rdd/rdd_site.py

def check_main_demo(self) -> Reel:
    """Get the mandatory main demo of this site.

    Returns:
        the reel the configuration names as main_demo.

    Raises:
        ValueError: where main_demo is unset, unknown or not in demo
            status - a site has at least one demo.
    """
    directory = self.reels_found.by_acronym()
    main_demo = directory.get(self.config.main_demo)
    if main_demo is None:
        raise ValueError(
            f"main_demo '{self.config.main_demo}' is not a reel of "
            f"{self.config.recordings_dir}"
        )
    if not main_demo.is_demo:
        raise ValueError(
            f"main_demo '{self.config.main_demo}' has status "
            f"'{main_demo.status}' - demo is required"
        )
    return main_demo

demo_card(lang='en')

The demo section of the home page - the main_demo of this site.

Parameters:

Name	Type	Description	Default
lang	str	the language of the card.	'en'

Returns:

Type	Description
str	the demo card markup; empty where the main_demo does not
str	resolve, so a misconfigured site still serves its home page.

Source code in rdd/rdd_site.py

def demo_card(self, lang: str = "en") -> str:
    """The demo section of the home page - the main_demo of this site.

    Args:
        lang: the language of the card.

    Returns:
        the demo card markup; empty where the main_demo does not
        resolve, so a misconfigured site still serves its home page.
    """
    t = texts(lang)
    card = ""
    directory = self.reels_found.by_acronym()
    main_demo = directory.get(self.config.main_demo)
    if main_demo and main_demo.is_demo:
        url = self.reel_url(main_demo)
        card = (
            f'<h2>{t["demo"]}</h2>\n<div class="card">\n'
            f'{t["demo_text"]}: <a href="{html.escape(url)}">'
            f"{html.escape(main_demo.title)}</a> "
            f'({main_demo.hop_count} {t["hops"]}) - {t["demo_hint"]}\n</div>'
        )
    return card

flag_selector(lang)

The language selector with flag - the other languages as links.

Parameters:

Name	Type	Description	Default
lang	str	the language of the current page.	required

Returns:

Type	Description
str	the selector markup.

Source code in rdd/rdd_site.py

def flag_selector(self, lang: str) -> str:
    """The language selector with flag - the other languages as links.

    Args:
        lang: the language of the current page.

    Returns:
        the selector markup.
    """
    flags = " ".join(
        f'<a class="flag" href="?lang={other}" title="{other}">{FLAGS[other]}</a>'
        for other in LANGUAGES
        if other != lang
    )
    return flags

home(lang='en')

The home page - what this site is and the two ways in.

Parameters:

Name	Type	Description	Default
lang	str	the language of the page.	'en'

Source code in rdd/rdd_site.py

    def home(self, lang: str = "en") -> str:
        """The home page - what this site is and the two ways in.

        Args:
            lang: the language of the page.
        """
        t = texts(lang)
        intro = html.escape(self.config.intro) if self.config.intro else t["intro"]
        content = f"""<div class="card">
{intro}
</div>
<h2>{t["reviewing"]}</h2>
<div class="card">
{t["reviewing_text"]}
</div>
<h2>{t["browsing"]}</h2>
<div class="card">
{t["browsing_text"]}
</div>
{self.demo_card(lang)}
<h2>Reel Driven Development</h2>
<div class="card">
{t["rdd_text"]}
<a href="{html.escape(self.config.cm_url)}" target=_blank>reel-driven-development</a>,
{t["rdd_text2"]}
</div>
"""
        page = self.page(t["home"], content, lang)
        return page

installation(reviews_file)

The installation mode page - the state a visitor sees while the site is not initialized.

Per the Owner bootstrap decision a site without reviews.yaml refuses to serve reels and names the init command instead; the state is shown, never hidden behind a dead backend.

Parameters:

Name	Type	Description	Default
reviews_file	str	the reviews file whose absence is the state.	required

Returns:

Type	Description
str	the installation mode page.

Source code in rdd/rdd_site.py

    def installation(self, reviews_file: str) -> str:
        """The installation mode page - the state a visitor sees while the
        site is not initialized.

        Per the Owner bootstrap decision a site without reviews.yaml
        refuses to serve reels and names the init command instead; the
        state is shown, never hidden behind a dead backend.

        Args:
            reviews_file: the reviews file whose absence is the state.

        Returns:
            the installation mode page.
        """
        content = f"""<h2>Installation mode</h2>
<div class="card">
This reel site is not initialized: <code>{html.escape(reviews_file)}</code>
does not exist, so no review right exists yet - not even the owner's.
No reel is served in this state.
</div>
<div class="card">
The owner initializes the site on its host - access administration needs
ssh and nothing else:
<pre>{html.escape(init_command())}</pre>
The command asks for username, full name, email and url, seeds the owner
and mints the wildcard owner token. The token is shown once on the
terminal and written beside the site configuration with mode 600; it is
never mailed, never logged and never minted via this webservice.
</div>
"""
        page = self.page("installation mode", content)
        return page

menu(lang='en')

The menu entries - settings and chat are dropped, a visitor has neither.

Parameters:

Name	Type	Description	Default
lang	str	the language of the entry names.	'en'

Source code in rdd/rdd_site.py

def menu(self, lang: str = "en") -> List[MenuEntry]:
    """The menu entries - settings and chat are dropped, a visitor has
    neither.

    Args:
        lang: the language of the entry names.
    """
    t = texts(lang)
    entries = [
        MenuEntry(t["home"], "/", "home"),
        MenuEntry(t["reels"], "/reels", "movie"),
        MenuEntry(t["github"], self.config.cm_url, "bug_report", new_tab=True),
        MenuEntry(t["help"], self.config.doc_url, "help", new_tab=True),
        MenuEntry(t["about"], "/about", "info"),
    ]
    return entries

not_found(path, lang='en')

The framed 404 page - it shows like any other page, with an example of a valid address.

Parameters:

Name	Type	Description	Default
path	str	the path that has no page.	required
lang	str	the language of the page.	'en'

Returns:

Type	Description
str	the framed 404 page.

Source code in rdd/rdd_site.py

def not_found(self, path: str, lang: str = "en") -> str:
    """The framed 404 page - it shows like any other page, with an
    example of a valid address.

    Args:
        path: the path that has no page.
        lang: the language of the page.

    Returns:
        the framed 404 page.
    """
    t = texts(lang)
    example = "/reels"
    directory = self.reels_found.by_acronym()
    main_demo = directory.get(self.config.main_demo)
    if main_demo and main_demo.is_demo:
        example = self.reel_url(main_demo)
    content = (
        f'<h2>404 - {t["not_found"]}</h2>\n'
        f'<div class="card">\n{t["no_page"]} '
        f"<code>{html.escape(path)}</code>.\n</div>\n"
        f'<div class="card">\n{t["example"]}: '
        f'<a href="{html.escape(example)}">{html.escape(example)}</a>; '
        f'{t["reels_listed"]}'
        "\n</div>"
    )
    page = self.page(t["not_found"], content, lang)
    return page

page(title, content, lang='en')

Render a page with menu and footer.

Parameters:

Name	Type	Description	Default
title	str	title of the page.	required
content	str	the html of the page body.	required
lang	str	the language of the page.	'en'

Returns:

Type	Description
str	the complete html page.

Source code in rdd/rdd_site.py

    def page(self, title: str, content: str, lang: str = "en") -> str:
        """Render a page with menu and footer.

        Args:
            title: title of the page.
            content: the html of the page body.
            lang: the language of the page.

        Returns:
            the complete html page.
        """
        links = "\n".join(entry.as_html() for entry in self.menu(lang))
        site_title = html.escape(self.config.title)
        menu_icon = svg("menu")
        page = f"""<!DOCTYPE html>
<html lang="{lang}">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>{site_title} - {html.escape(title)}</title>
<style>
{self.style()}</style>
</head>
<body>
<button class="hamburger collapsed" id="unhide" onclick="toggleMenu()" title="show menu" hidden>{menu_icon}</button>
<header>
  <button class="hamburger" onclick="toggleMenu()" title="hide menu">{menu_icon}</button>
  <h1>{site_title}</h1>
{links}
  {self.flag_selector(lang)}
</header>
<main>
{content}
</main>
<footer>{html.escape(self.config.copy_right)} - {self.version.name} {self.version.version}</footer>
<script>
function toggleMenu() {{
  const collapsed = document.body.classList.toggle('collapsed');
  document.getElementById('unhide').hidden = !collapsed;
}}
</script>
</body>
</html>
"""
        return page

reel_files(reel)

The files of the given reel folder.

Parameters:

Name	Type	Description	Default
reel	Reel	the reel.	required

Returns:

Type	Description
List[str]	the sorted file names, hidden files and the review page
List[str]	excluded - the review page is code of the package, never
List[str]	data of a reel.

Source code in rdd/rdd_site.py

def reel_files(self, reel: Reel) -> List[str]:
    """The files of the given reel folder.

    Args:
        reel: the reel.

    Returns:
        the sorted file names, hidden files and the review page
        excluded - the review page is code of the package, never
        data of a reel.
    """
    names = [
        name
        for name in sorted(os.listdir(reel.path))
        if not name.startswith(".")
        and name != "reelreview.html"
        and os.path.isfile(os.path.join(reel.path, name))
    ]
    return names

reel_page(reel, lang='en')

The page of one reel - what it is and the files it carries.

The review and file links are relative so a review link keeps its token.

Parameters:

Name	Type	Description	Default
reel	Reel	the reel.	required
lang	str	the language of the page.	'en'

Returns:

Type	Description
str	the reel page.

Source code in rdd/rdd_site.py

def reel_page(self, reel: Reel, lang: str = "en") -> str:
    """The page of one reel - what it is and the files it carries.

    The review and file links are relative so a review link keeps
    its token.

    Args:
        reel: the reel.
        lang: the language of the page.

    Returns:
        the reel page.
    """
    recording = reel.recording
    summary = recording.summary if recording and recording.summary else ""
    rows = "\n".join(
        f'<tr><td><a href="{html.escape(urllib.parse.quote(name))}">'
        f"{html.escape(name)}</a></td>"
        f"<td>{os.path.getsize(os.path.join(reel.path, name))}</td></tr>"
        for name in self.reel_files(reel)
    )
    t = texts(lang)
    content = (
        f"<h2>{html.escape(reel.title)}</h2>\n"
        f'<p><a href="review">{t["review"]}</a></p>\n'
        f'<div class="card">\n<b>{t["summary"]}</b><br>\n'
        f"{html.escape(summary)}\n</div>\n"
        f'<div class="card">\n<table>\n'
        f'<tr><th>{t["file"]}</th><th>{t["bytes"]}</th></tr>\n'
        f"{rows}\n</table>\n</div>"
    )
    page = self.page(reel.acronym, content, lang)
    return page

reel_url(reel, review=None)

The url this site serves the given reel under.

Per the Delivery decision the acronym is the only address a url needs; a review link carries its token before the acronym.

Parameters:

Name	Type	Description	Default
reel	Reel	the reel.	required
review	Optional[Review]	the review whose token the url carries; None for the public url.	None

Returns:

Type	Description
str	the url of the reel.

Source code in rdd/rdd_site.py

def reel_url(self, reel: Reel, review: Optional[Review] = None) -> str:
    """The url this site serves the given reel under.

    Per the Delivery decision the acronym is the only address a url
    needs; a review link carries its token before the acronym.

    Args:
        reel: the reel.
        review: the review whose token the url carries; None for the
            public url.

    Returns:
        the url of the reel.
    """
    prefix = self.config.reels_url_prefix
    if review:
        prefix = f"{prefix}{review.token}/"
    url = f"{prefix}{reel.acronym}/"
    return url

reel_zip(reel)

Zip the folder of the given reel into a temporary file.

Per the Reel verdict decision the verdict page offers the reel folder as one download; videos do not compress, so the entries are stored.

Parameters:

Name	Type	Description	Default
reel	Reel	the reel.	required

Returns:

Type	Description
str	the path of the temporary zip file - the caller removes it
str	after delivery.

Source code in rdd/rdd_site.py

def reel_zip(self, reel: Reel) -> str:
    """Zip the folder of the given reel into a temporary file.

    Per the Reel verdict decision the verdict page offers the reel
    folder as one download; videos do not compress, so the entries
    are stored.

    Args:
        reel: the reel.

    Returns:
        the path of the temporary zip file - the caller removes it
        after delivery.
    """
    handle, zip_path = tempfile.mkstemp(suffix=".zip", prefix=f"{reel.acronym}-")
    os.close(handle)
    with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_STORED) as zip_file:
        for root, _dirs, files in os.walk(reel.path):
            for file_name in sorted(files):
                file_path = os.path.join(root, file_name)
                arcname = os.path.join(
                    reel.acronym, os.path.relpath(file_path, reel.path)
                )
                zip_file.write(file_path, arcname)
    return zip_path

reels(review=None, lang='en')

The reels directory as the holder of the given right sees it.

Anyone sees the public and demo reels; a Review right adds its private reels under the same directory.

Parameters:

Name	Type	Description	Default
review	Optional[Review]	the review right; None for the anonymous directory.	None
lang	str	the language of the page.	'en'

Returns:

Type	Description
str	the reels directory page.

Source code in rdd/rdd_site.py

def reels(self, review: Optional[Review] = None, lang: str = "en") -> str:
    """The reels directory as the holder of the given right sees it.

    Anyone sees the public and demo reels; a Review right adds its
    private reels under the same directory.

    Args:
        review: the review right; None for the anonymous directory.
        lang: the language of the page.

    Returns:
        the reels directory page.
    """
    t = texts(lang)
    granted = review.reels if review else None
    visible_reels = self.reels_found.visible(granted)
    heading = "Reels"
    if review:
        heading = t["review_by"].format(person=review.person)
    if visible_reels:
        rows = "\n".join(
            f'<tr><td><a href="{html.escape(self.reel_url(reel, review))}">'
            f"{html.escape(reel.acronym)}</a></td>"
            f"<td>{html.escape(reel.title)}</td>"
            f"<td>{reel.hop_count}</td>"
            f"<td>{html.escape(reel.status)}</td></tr>"
            for reel in visible_reels
        )
        content = (
            f'<h2>{html.escape(heading)}</h2>\n<div class="card">\n<table>\n'
            f'<tr><th>{t["reel"]}</th><th>{t["title"]}</th>'
            f'<th>{t["hops"]}</th><th>{t["status"]}</th></tr>\n'
            f"{rows}\n</table>\n</div>"
        )
    else:
        content = (
            f'<h2>{html.escape(heading)}</h2>\n<div class="card">\n'
            f'{t["no_reels"]}\n</div>'
        )
    page = self.page(t["reels"], content, lang)
    return page

resolve_reel(parts)

Resolve the addressed reel - shortcut or lengthy form.

Per the Hop url decision the acronym is the shortcut address and year/month/acronym the lengthy form disambiguating non-unique acronyms.

Parameters:

Name	Type	Description	Default
parts	List[str]	the path parts after /reels/ with the token stripped.	required

Returns:

Type	Description
	the reel or None, and the remaining path parts.

Source code in rdd/rdd_site.py

def resolve_reel(self, parts: List[str]):
    """Resolve the addressed reel - shortcut or lengthy form.

    Per the Hop url decision the acronym is the shortcut address
    and year/month/acronym the lengthy form disambiguating
    non-unique acronyms.

    Args:
        parts: the path parts after /reels/ with the token stripped.

    Returns:
        the reel or None, and the remaining path parts.
    """
    reel = None
    file_parts: List[str] = []
    if (
        len(parts) >= 3
        and re.match(r"\d{4}$", parts[0])
        and re.match(r"\d{2}$", parts[1])
    ):
        reel = self.reels_found.by_pid().get("/".join(parts[:3]))
        file_parts = [part for part in parts[3:] if part]
    elif parts:
        reel = self.reels_found.by_acronym().get(parts[0])
        file_parts = [part for part in parts[1:] if part]
    return reel, file_parts

review_page(lang='en')

The review page of this site.

Per the Review UI stack decision the packaged page is one self-contained file; serving it, the site derives the CSS variables from its named palette, puts its own menu into the marked header slot and speaks the visitor's language via the marked i18n slot, so the review wears the same palette, menu and language as every other page of the site - one source, no second copy to keep in step.

Parameters:

Name	Type	Description	Default
lang	str	the language of the page.	'en'

Returns:

Type	Description
str	the review page in the site's palette, menu and language.

Source code in rdd/rdd_site.py

def review_page(self, lang: str = "en") -> str:
    """The review page of this site.

    Per the Review UI stack decision the packaged page is one
    self-contained file; serving it, the site derives the CSS
    variables from its named palette, puts its own menu into
    the marked header slot and speaks the visitor's language via
    the marked i18n slot, so the review wears the same palette,
    menu and language as every other page of the site - one
    source, no second copy to keep in step.

    Args:
        lang: the language of the page.

    Returns:
        the review page in the site's palette, menu and language.
    """
    page = page_path().read_text()
    for name, value in self.palette.__dict__.items():
        page = re.sub(rf"--{name}: #[0-9A-Fa-f]+;", f"--{name}: {value};", page)
    links = "\n".join(entry.as_html() for entry in self.menu(lang))
    page = re.sub(
        r"<!-- menu -->.*?<!-- /menu -->",
        f"<!-- menu -->\n{links}\n  <!-- /menu -->",
        page,
        flags=re.DOTALL,
    )
    translations = json.dumps(review_texts(lang), ensure_ascii=False)
    page = re.sub(
        r"/\* <i18n> \*/.*?/\* </i18n> \*/",
        f'/* <i18n> */\nconst LANG = "{lang}";\n'
        f"const T = {translations};\n/* </i18n> */",
        page,
        flags=re.DOTALL,
    )
    page = page.replace('<html lang="en">', f'<html lang="{lang}">', 1)
    return page

scan()

Scan the recordings directory of this site into its directory of reels.

Returns:

Type	Description
Reels	the reels the recordings directory holds.

Source code in rdd/rdd_site.py

def scan(self) -> Reels:
    """Scan the recordings directory of this site into its directory of
    reels.

    Returns:
        the reels the recordings directory holds.
    """
    reels = Reels.of_dir(self.config.recordings_dir)
    return reels

style()

The stylesheet, derived from the named palette.

Source code in rdd/rdd_site.py

    def style(self) -> str:
        """The stylesheet, derived from the named palette."""
        css = f""":root {{
{self.palette.as_css()}
  --bg: #fafafa; --fg: #222; --card: #fff; --border: #ccc;
}}
@media (prefers-color-scheme: dark) {{
  :root {{ --bg: var(--dark); --fg: #ddd; --card: #2a2a2a; --border: #555; }}
}}
body {{ margin: 0; font-family: system-ui, sans-serif; background: var(--bg); color: var(--fg); }}
header {{ display: flex; gap: .8em; align-items: center; padding: .6em .8em; background: var(--primary); color: #fff; flex-wrap: wrap; }}
header h1 {{ font-size: 1.1em; margin: 0 .4em 0 0; }}
header a {{ display: inline-flex; align-items: center; gap: .5em; color: #fff; text-decoration: none; background: var(--primary); border-radius: 4px; padding: .5em 1em; font-size: .85em; font-weight: 500; text-transform: uppercase; letter-spacing: .05em; box-shadow: 0 1px 5px rgba(0,0,0,.2), 0 2px 2px rgba(0,0,0,.14), 0 3px 1px -2px rgba(0,0,0,.12); }}
header a:hover {{ background: var(--accent); }}
header a.flag {{ box-shadow: none; text-transform: none; background: none; font-size: 1.1em; padding: .2em; margin-left: auto; }}
.hamburger {{ display: inline-flex; align-items: center; background: var(--primary); border: none; color: #fff; cursor: pointer; border-radius: 4px; padding: .45em .6em; box-shadow: 0 1px 5px rgba(0,0,0,.2), 0 2px 2px rgba(0,0,0,.14), 0 3px 1px -2px rgba(0,0,0,.12); }}
.hamburger.collapsed {{ position: fixed; top: .4em; left: .4em; z-index: 10; }}
.hamburger[hidden] {{ display: none; }}
body.collapsed header, body.collapsed footer {{ display: none; }}
main {{ padding: 1em; max-width: 55em; }}
main h2 {{ font-size: 1.2em; }}
main a {{ color: var(--primary); }}
.card {{ background: var(--card); border: 1px solid var(--border); border-radius: 6px; padding: .8em; margin-bottom: .8em; }}
footer {{ padding: .6em .8em; background: var(--primary); color: #fff; font-size: .85em; }}
footer a {{ color: #fff; }}
table {{ border-collapse: collapse; }}
td, th {{ text-align: left; padding: .2em .8em .2em 0; }}
"""
        return css

Review

One review right - a reviewer and the reels their link grants.

Per the Reel Review decision the review rights are not modeled in SMW (yet) but kept as entities the rdd site must keep track of. Per the Owner bootstrap decision the owner's Review grants the wildcard '*' - every reel, including future ones.

Source code in rdd/rdd_site.py

@lod_storable
class Review:
    """One review right - a reviewer and the reels their link grants.

    Per the Reel Review decision the review rights are not modeled in
    SMW (yet) but kept as entities the rdd site must keep track of.
    Per the Owner bootstrap decision the owner's Review grants the
    wildcard '*' - every reel, including future ones.
    """

    WILDCARD = "*"

    token: str = ""
    person: str = ""
    meeting: str = ""
    reels: List[str] = field(default_factory=list)

    @property
    def is_wildcard(self) -> bool:
        """Whether this review grants every reel."""
        is_wildcard = self.WILDCARD in self.reels
        return is_wildcard

is_wildcard property

Whether this review grants every reel.

Reviews

The review rights of a site.

Source code in rdd/rdd_site.py

@lod_storable
class Reviews:
    """The review rights of a site."""

    DEFAULT_PATH = "~/.rdd/reviews.yaml"

    reviews: List[Review] = field(default_factory=list)

    @classmethod
    def of_path(cls, path: Optional[str] = None) -> "Reviews":
        """Load the reviews from the given yaml file.

        Args:
            path: the reviews file; the default path when None.

        Returns:
            the reviews; no reviews where no file exists.
        """
        reviews_path = os.path.expanduser(path or cls.DEFAULT_PATH)
        if os.path.isfile(reviews_path):
            reviews = cls.load_from_yaml_file(reviews_path)
        else:
            reviews = cls()
        return reviews

    def by_token(self) -> Dict[str, Review]:
        """The lookup from token to review.

        Returns:
            the lookup from token to review.
        """
        lookup = {review.token: review for review in self.reviews}
        return lookup

by_token()

The lookup from token to review.

Returns:

Type	Description
Dict[str, Review]	the lookup from token to review.

Source code in rdd/rdd_site.py

def by_token(self) -> Dict[str, Review]:
    """The lookup from token to review.

    Returns:
        the lookup from token to review.
    """
    lookup = {review.token: review for review in self.reviews}
    return lookup

of_path(path=None) classmethod

Load the reviews from the given yaml file.

Parameters:

Name	Type	Description	Default
path	Optional[str]	the reviews file; the default path when None.	None

Returns:

Type	Description
Reviews	the reviews; no reviews where no file exists.

Source code in rdd/rdd_site.py

@classmethod
def of_path(cls, path: Optional[str] = None) -> "Reviews":
    """Load the reviews from the given yaml file.

    Args:
        path: the reviews file; the default path when None.

    Returns:
        the reviews; no reviews where no file exists.
    """
    reviews_path = os.path.expanduser(path or cls.DEFAULT_PATH)
    if os.path.isfile(reviews_path):
        reviews = cls.load_from_yaml_file(reviews_path)
    else:
        reviews = cls()
    return reviews

init_command()

The init command as this installation runs it.

A user cannot be expected to know a command name, and the venv of a service is not on anybody's PATH - so the command is named with the absolute path of the running installation where the rdd dispatcher lies beside the interpreter, and by its bare name otherwise.

Returns:

Type	Description
str	the command that initializes this site.

Source code in rdd/rdd_site.py

def init_command() -> str:
    """The init command as this installation runs it.

    A user cannot be expected to know a command name, and the venv of a
    service is not on anybody's PATH - so the command is named with the
    absolute path of the running installation where the rdd dispatcher
    lies beside the interpreter, and by its bare name otherwise.

    Returns:
        the command that initializes this site.
    """
    rdd_command = os.path.join(os.path.dirname(sys.executable), "rdd")
    if not os.path.isfile(rdd_command):
        rdd_command = "rdd"
    command = f"{rdd_command} site --init"
    return command

serve(config, host='127.0.0.1', reviews_path=None)

Serve the site of the given configuration with uvicorn.

Per the Owner bootstrap decision a site without reviews.yaml is in installation mode: it stays up, refuses to serve reels and names the init command on every request, so the state is never hidden behind a dead backend.

Parameters:

Name	Type	Description	Default
config	RddSiteConfig	the site configuration.	required
host	str	the interface to listen on; localhost by default - the web server in front is what the internet talks to.	'127.0.0.1'
reviews_path	Optional[str]	the reviews file; the default path when None.	None

Source code in rdd/rdd_site.py

def serve(
    config: RddSiteConfig,
    host: str = "127.0.0.1",
    reviews_path: Optional[str] = None,
) -> None:
    """Serve the site of the given configuration with uvicorn.

    Per the Owner bootstrap decision a site without reviews.yaml is in
    installation mode: it stays up, refuses to serve reels and names the
    init command on every request, so the state is never hidden behind a
    dead backend.

    Args:
        config: the site configuration.
        host: the interface to listen on; localhost by default - the web
            server in front is what the internet talks to.
        reviews_path: the reviews file; the default path when None.
    """
    import uvicorn

    from rdd.webapp import create_app, create_installation_app

    reviews_file = os.path.expanduser(reviews_path or Reviews.DEFAULT_PATH)
    if not os.path.isfile(reviews_file):
        install_site = ReelSite(config, reels=Reels(), reviews=Reviews())
        page = install_site.installation(reviews_file)
        print(
            f"rdd_site: installation mode - no {reviews_file}; run {init_command()}",
            flush=True,
        )
        print(
            f"rdd_site: {config.title} on http://{host}:{config.port}/ "
            "(installation mode)",
            flush=True,
        )
        uvicorn.run(create_installation_app(page), host=host, port=config.port)
        return
    site = ReelSite(config, reviews=Reviews.of_path(reviews_file))
    main_demo = site.check_main_demo()
    print(f"rdd_site: {site.reels_found.as_summary()}", flush=True)
    print(f"rdd_site: main_demo {main_demo.acronym}", flush=True)
    print(f"rdd_site: {config.title} on http://{host}:{config.port}/", flush=True)
    uvicorn.run(create_app(site), host=host, port=config.port)

recording

Created on 2026-08-08.

hop records of a graph walk

Schema: the Meeting context of https://contexts.bitplan.com * https://contexts.bitplan.com/index.php/Concept:HopContent * https://contexts.bitplan.com/index.php/Concept:Recording

@author: wf

HopContent

One node visit in the graph walk of a Recording.

The node is the page, screen or application reached by a context switch; the record says when it was reached and what happened there.

The field names are the property names of https://contexts.bitplan.com/index.php/Concept:HopContent and are never renamed: a record is stored as a subobject on its Recording page and read back by the same names.

Source code in rdd/recording.py

@lod_storable
class HopContent:
    """One node visit in the graph walk of a Recording.

    The node is the page, screen or application reached by a context
    switch; the record says when it was reached and what happened there.

    The field names are the property names of
    https://contexts.bitplan.com/index.php/Concept:HopContent
    and are never renamed: a record is stored as a
    subobject on its Recording page and read back by the same names.
    """

    pos: int = 0
    time: str = ""
    node: Optional[str] = None
    url: Optional[str] = None
    summary: Optional[str] = None
    screenshot: Optional[str] = None
    recording: Optional[str] = None

HopContents

The hops of one Recording.

https://contexts.bitplan.com/index.php/Concept:Recording is linked to its hops 1:n via the recordingHops TopicLink, and the hopCount of the Recording must equal the number of hop records - that equality is the mechanical completeness control of the walk.

Source code in rdd/recording.py

@lod_storable
class HopContents:
    """The hops of one Recording.

    https://contexts.bitplan.com/index.php/Concept:Recording is linked to
    its hops 1:n via the recordingHops TopicLink, and the hopCount of the
    Recording must equal the number of hop records - that equality is the
    mechanical completeness control of the walk.
    """

    recording: Optional[str] = None
    hops: List[HopContent] = None

    def __post_init__(self):
        """Start with an empty hop list where none was given."""
        if self.hops is None:
            self.hops = []

    @property
    def hopCount(self) -> int:
        """The number of hop records, to be compared with the hopCount of the
        Recording."""
        hop_count = len(self.hops)
        return hop_count

    def add(self, hop: HopContent) -> HopContent:
        """Add a hop, giving it the next position and this recording.

        Args:
            hop: the hop record to add.

        Returns:
            the added hop.
        """
        hop.pos = len(self.hops) + 1
        hop.recording = self.recording
        self.hops.append(hop)
        return hop

hopCount property

The number of hop records, to be compared with the hopCount of the Recording.

__post_init__()

Start with an empty hop list where none was given.

Source code in rdd/recording.py

def __post_init__(self):
    """Start with an empty hop list where none was given."""
    if self.hops is None:
        self.hops = []

add(hop)

Add a hop, giving it the next position and this recording.

Parameters:

Name	Type	Description	Default
hop	HopContent	the hop record to add.	required

Returns:

Type	Description
HopContent	the added hop.

Source code in rdd/recording.py

def add(self, hop: HopContent) -> HopContent:
    """Add a hop, giving it the next position and this recording.

    Args:
        hop: the hop record to add.

    Returns:
        the added hop.
    """
    hop.pos = len(self.hops) + 1
    hop.recording = self.recording
    self.hops.append(hop)
    return hop

Recording

One recorded video of a session, with its processing state.

The field names are the property names of https://contexts.bitplan.com/index.php/Concept:Recording and are never renamed: a Recording is a page in the Meeting context and is read back by the same names.

A Recording is the record of a video, not the video itself; reading pictures from the file is the business of the specialization in rdd.frame.

The acronym is the short ASCII identifier every artefact of this Recording carries as its name prefix, so an artefact can be traced back to its Recording by name alone.

Source code in rdd/recording.py

@lod_storable
class Recording:
    """One recorded video of a session, with its processing state.

    The field names are the property names of
    https://contexts.bitplan.com/index.php/Concept:Recording
    and are never renamed: a Recording is a page in the Meeting context
    and is read back by the same names.

    A Recording is the record of a video, not the video itself; reading
    pictures from the file is the business of the specialization in
    rdd.frame.

    The acronym is the short ASCII identifier every artefact of this
    Recording carries as its name prefix, so an artefact can be traced
    back to its Recording by name alone.
    """

    name: Optional[str] = None
    acronym: Optional[str] = None
    date: Optional[str] = None
    durationMin: Optional[float] = None
    videoFile: Optional[str] = None
    platform: Optional[str] = None
    meeting: Optional[str] = None
    participants: Optional[str] = None
    language: Optional[str] = None
    computer: Optional[str] = None
    user: Optional[str] = None
    state: Optional[str] = None
    transcript: Optional[str] = None
    driveLink: Optional[str] = None
    hopCount: Optional[int] = None
    # what the recording is, in the language the Recording denotes -
    # a document without it says what was clicked but not what it was about
    summary: Optional[str] = None

reelreview

Reel Driven Development - reelreview server https://github.com/WolfgangFahl/reel-driven-development

Serves a recording folder for the reelreview.html human-in-the-loop verdict pass and accepts the curated reel.yaml back:

GET  /            -> reelreview.html
GET  /<file>      -> static file from the recording folder
GET  /api/files   -> JSON list of the folder's file names
GET  /api/info    -> JSON folder name and acronym of the reviewed reel
GET  /api/reel    -> JSON of the reel.yaml hop set parsed by the model
POST /api/save     -> body replaces reel.yaml (RCS checkpoint first)
POST /api/feedback -> body replaces reel-feedback.yaml (RCS checkpoint first)
POST /api/upload   -> forward reel-feedback.yaml body to the feedback_url
                      configured in reel.yaml's config block

The YAML work is done by the HopSet model - the page never parses YAML.

Usage

reelreview [folder][--port PORT]

ReelReviewHandler

Bases: SimpleHTTPRequestHandler

Serve one recording folder for the review pass.

The whole folder is readable and reel.yaml and reel-feedback.yaml are writable through the api, with no authentication of any kind - this is a single user tool for the person curating a reel on their own machine. It is bound to localhost for that reason; a --host that opens it to a network hands that write access to everyone who can reach the port.

Source code in rdd/reelreview.py

class ReelReviewHandler(http.server.SimpleHTTPRequestHandler):
    """Serve one recording folder for the review pass.

    The whole folder is readable and reel.yaml and reel-feedback.yaml are
    writable through the api, with no authentication of any kind - this is a
    single user tool for the person curating a reel on their own machine. It
    is bound to localhost for that reason; a --host that opens it to a network
    hands that write access to everyone who can reach the port.
    """

    def do_GET(self):
        """Answer a file or api request.

        /, /index.html and /reelreview.html serve the packaged review
        page - the page is never data of the folder, so a stale copy in
        the folder is shadowed; an extensionless path is a hop url per
        the Hop url decision and serves the page too, positioned by the
        page itself. /api/info answers the folder name and acronym of
        the reviewed reel, /api/files the file names of the folder,
        /api/reel the hop set of reel.yaml parsed by the model;
        anything else is a static file of the folder.
        """
        hop_url = re.match(r"/[^/.]+$", self.path) and not self.path.startswith("/api/")
        if self.path in ("/", "/index.html", "/reelreview.html") or hop_url:
            body = page_path().read_bytes()
            self.send_response(200)
            self.send_header("Content-Type", "text/html; charset=utf-8")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)
            return
        if self.path == "/api/info":
            folder = Path(self.directory)
            info = {"folder": folder.name}
            reel = folder / "reel.yaml"
            if reel.exists():
                match = re.search(
                    r"^\s+acronym:\s*(\S+)", reel.read_text(), re.MULTILINE
                )
                if match:
                    info["acronym"] = match.group(1).strip("\"'")
            body = json.dumps(info).encode()
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)
            return
        if self.path == "/api/reel":
            hop_set = HopSet.of_dir(self.directory)
            body = json.dumps(hop_set.to_dict() if hop_set else {}).encode()
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)
            return
        if self.path == "/api/files":
            names = sorted(
                p.name for p in Path(self.directory).iterdir() if p.is_file()
            )
            body = json.dumps(names).encode()
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)
            return
        super().do_GET()

    def do_POST(self):
        """Take the curated yaml back.

        /api/save replaces reel.yaml and /api/feedback reel-
        feedback.yaml, both after an RCS checkpoint of the file being
        replaced; /api/upload forwards the body to the feedback_url of
        reel.yaml. Anything else is a 404.
        """
        targets = {"/api/save": "reel.yaml", "/api/feedback": "reel-feedback.yaml"}
        length = int(self.headers.get("Content-Length", 0))
        content = self.rfile.read(length)
        if self.path in targets:
            target = Path(self.directory) / targets[self.path]
            self.checkpoint(target)
            target.write_bytes(content)
            self.send_response(200)
            self.send_header("Content-Length", "0")
            self.end_headers()
        elif self.path == "/api/upload":
            self.upload(content)
        else:
            self.send_error(404)

    def upload(self, content: bytes):
        """Forward the feedback to the feedback_url configured in reel.yaml."""
        import re
        import urllib.request

        reel = (Path(self.directory) / "reel.yaml").read_text()
        match = re.search(r"^\s+feedback_url:\s*(\S+)", reel, re.MULTILINE)
        if not match:
            self.send_error(409, "no feedback_url in reel.yaml config")
            return
        request = urllib.request.Request(
            match.group(1), data=content, headers={"Content-Type": "text/yaml"}
        )
        try:
            with urllib.request.urlopen(request, timeout=30) as response:
                self.send_response(response.status)
                self.send_header("Content-Length", "0")
                self.end_headers()
        except Exception as ex:
            self.send_error(502, str(ex))

    def checkpoint(self, target: Path):
        """Version the current reel.yaml with RCS before overwriting."""
        if target.exists() and shutil.which("ci"):
            subprocess.run(
                [
                    "ci",
                    "-l",
                    f"-t-{target.name}",
                    "-m",
                    "reelreview checkpoint",
                    str(target),
                ],
                cwd=self.directory,
                capture_output=True,
            )

checkpoint(target)

Version the current reel.yaml with RCS before overwriting.

Source code in rdd/reelreview.py

def checkpoint(self, target: Path):
    """Version the current reel.yaml with RCS before overwriting."""
    if target.exists() and shutil.which("ci"):
        subprocess.run(
            [
                "ci",
                "-l",
                f"-t-{target.name}",
                "-m",
                "reelreview checkpoint",
                str(target),
            ],
            cwd=self.directory,
            capture_output=True,
        )

do_GET()

Answer a file or api request.

/, /index.html and /reelreview.html serve the packaged review page - the page is never data of the folder, so a stale copy in the folder is shadowed; an extensionless path is a hop url per the Hop url decision and serves the page too, positioned by the page itself. /api/info answers the folder name and acronym of the reviewed reel, /api/files the file names of the folder, /api/reel the hop set of reel.yaml parsed by the model; anything else is a static file of the folder.

Source code in rdd/reelreview.py

def do_GET(self):
    """Answer a file or api request.

    /, /index.html and /reelreview.html serve the packaged review
    page - the page is never data of the folder, so a stale copy in
    the folder is shadowed; an extensionless path is a hop url per
    the Hop url decision and serves the page too, positioned by the
    page itself. /api/info answers the folder name and acronym of
    the reviewed reel, /api/files the file names of the folder,
    /api/reel the hop set of reel.yaml parsed by the model;
    anything else is a static file of the folder.
    """
    hop_url = re.match(r"/[^/.]+$", self.path) and not self.path.startswith("/api/")
    if self.path in ("/", "/index.html", "/reelreview.html") or hop_url:
        body = page_path().read_bytes()
        self.send_response(200)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)
        return
    if self.path == "/api/info":
        folder = Path(self.directory)
        info = {"folder": folder.name}
        reel = folder / "reel.yaml"
        if reel.exists():
            match = re.search(
                r"^\s+acronym:\s*(\S+)", reel.read_text(), re.MULTILINE
            )
            if match:
                info["acronym"] = match.group(1).strip("\"'")
        body = json.dumps(info).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)
        return
    if self.path == "/api/reel":
        hop_set = HopSet.of_dir(self.directory)
        body = json.dumps(hop_set.to_dict() if hop_set else {}).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)
        return
    if self.path == "/api/files":
        names = sorted(
            p.name for p in Path(self.directory).iterdir() if p.is_file()
        )
        body = json.dumps(names).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)
        return
    super().do_GET()

do_POST()

Take the curated yaml back.

/api/save replaces reel.yaml and /api/feedback reel- feedback.yaml, both after an RCS checkpoint of the file being replaced; /api/upload forwards the body to the feedback_url of reel.yaml. Anything else is a 404.

Source code in rdd/reelreview.py

def do_POST(self):
    """Take the curated yaml back.

    /api/save replaces reel.yaml and /api/feedback reel-
    feedback.yaml, both after an RCS checkpoint of the file being
    replaced; /api/upload forwards the body to the feedback_url of
    reel.yaml. Anything else is a 404.
    """
    targets = {"/api/save": "reel.yaml", "/api/feedback": "reel-feedback.yaml"}
    length = int(self.headers.get("Content-Length", 0))
    content = self.rfile.read(length)
    if self.path in targets:
        target = Path(self.directory) / targets[self.path]
        self.checkpoint(target)
        target.write_bytes(content)
        self.send_response(200)
        self.send_header("Content-Length", "0")
        self.end_headers()
    elif self.path == "/api/upload":
        self.upload(content)
    else:
        self.send_error(404)

upload(content)

Forward the feedback to the feedback_url configured in reel.yaml.

Source code in rdd/reelreview.py

def upload(self, content: bytes):
    """Forward the feedback to the feedback_url configured in reel.yaml."""
    import re
    import urllib.request

    reel = (Path(self.directory) / "reel.yaml").read_text()
    match = re.search(r"^\s+feedback_url:\s*(\S+)", reel, re.MULTILINE)
    if not match:
        self.send_error(409, "no feedback_url in reel.yaml config")
        return
    request = urllib.request.Request(
        match.group(1), data=content, headers={"Content-Type": "text/yaml"}
    )
    try:
        with urllib.request.urlopen(request, timeout=30) as response:
            self.send_response(response.status)
            self.send_header("Content-Length", "0")
            self.end_headers()
    except Exception as ex:
        self.send_error(502, str(ex))

page_path()

Path of the review page shipped with the package.

Source code in rdd/reelreview.py

def page_path() -> Path:
    """Path of the review page shipped with the package."""
    path = Path(__file__).parent / "resources" / "reelreview.html"
    return path

serve(folder, port=DEFAULT_PORT, host=DEFAULT_HOST)

Serve the given recording folder for the review pass.

Parameters:

Name	Type	Description	Default
folder	Path	the recording folder holding reel.yaml.	required
port	int	port to serve on.	DEFAULT_PORT
host	str	interface to listen on; localhost by default - the api writes reel.yaml without authentication, so binding a reachable interface is an explicit choice of the person starting the server.	DEFAULT_HOST

Raises:

Type	Description
ValueError	if the folder holds no reel.yaml or the page is missing.

Source code in rdd/reelreview.py

def serve(folder: Path, port: int = DEFAULT_PORT, host: str = DEFAULT_HOST) -> None:
    """Serve the given recording folder for the review pass.

    Args:
        folder: the recording folder holding reel.yaml.
        port: port to serve on.
        host: interface to listen on; localhost by default - the api writes
            reel.yaml without authentication, so binding a reachable interface
            is an explicit choice of the person starting the server.

    Raises:
        ValueError: if the folder holds no reel.yaml or the page is missing.
    """
    if not (folder / "reel.yaml").exists():
        raise ValueError(f"no reel.yaml in {folder}")
    if not page_path().exists():
        raise ValueError(f"no packaged reelreview.html at {page_path()}")
    handler = lambda *a, **kw: ReelReviewHandler(*a, directory=str(folder), **kw)
    with http.server.ThreadingHTTPServer((host, port), handler) as httpd:
        print(f"reelreview: serving {folder} on http://{host}:{port}/")
        httpd.serve_forever()

reelreview_cmd

Created on 2026-08-12.

command line interface of the reel review pass

see https://github.com/WolfgangFahl/reel-driven-development/issues/27

@author: wf

ReelReviewCmd

Bases: BaseCmd

Serve a recording folder for the human in the loop review pass.

Source code in rdd/reelreview_cmd.py

class ReelReviewCmd(BaseCmd):
    """Serve a recording folder for the human in the loop review pass."""

    def __init__(self):
        """Initialize with the reel-driven-development version info."""
        super().__init__(Version())

    def add_arguments(self, parser: argparse.ArgumentParser):
        """Add the review arguments to the given parser.

        Args:
            parser: the parser to add arguments to.
        """
        super().add_arguments(parser)
        parser.add_argument(
            "folder", nargs="?", default=".", help="recording folder (default: .)"
        )
        parser.add_argument(
            "--host",
            default=DEFAULT_HOST,
            help="interface to listen on; the api needs no authentication, so"
            " opening this beyond localhost shares write access to the reel"
            " [default: %(default)s]",
        )
        parser.add_argument(
            "-p",
            "--port",
            type=int,
            default=DEFAULT_PORT,
            help="port to serve on [default: %(default)s]",
        )

    def handle_args(self, args: argparse.Namespace) -> bool:
        """Handle the parsed arguments by serving the folder.

        Args:
            args: parsed argument namespace.

        Returns:
            True if the arguments were handled.
        """
        handled = super().handle_args(args)
        if not handled:
            folder = Path(args.folder).expanduser().resolve()
            serve(folder, port=args.port, host=args.host)
            handled = True
        return handled

__init__()

Initialize with the reel-driven-development version info.

Source code in rdd/reelreview_cmd.py

def __init__(self):
    """Initialize with the reel-driven-development version info."""
    super().__init__(Version())

add_arguments(parser)

Add the review arguments to the given parser.

Parameters:

Name	Type	Description	Default
parser	ArgumentParser	the parser to add arguments to.	required

Source code in rdd/reelreview_cmd.py

def add_arguments(self, parser: argparse.ArgumentParser):
    """Add the review arguments to the given parser.

    Args:
        parser: the parser to add arguments to.
    """
    super().add_arguments(parser)
    parser.add_argument(
        "folder", nargs="?", default=".", help="recording folder (default: .)"
    )
    parser.add_argument(
        "--host",
        default=DEFAULT_HOST,
        help="interface to listen on; the api needs no authentication, so"
        " opening this beyond localhost shares write access to the reel"
        " [default: %(default)s]",
    )
    parser.add_argument(
        "-p",
        "--port",
        type=int,
        default=DEFAULT_PORT,
        help="port to serve on [default: %(default)s]",
    )

handle_args(args)

Handle the parsed arguments by serving the folder.

Parameters:

Name	Type	Description	Default
args	Namespace	parsed argument namespace.	required

Returns:

Type	Description
bool	True if the arguments were handled.

Source code in rdd/reelreview_cmd.py

def handle_args(self, args: argparse.Namespace) -> bool:
    """Handle the parsed arguments by serving the folder.

    Args:
        args: parsed argument namespace.

    Returns:
        True if the arguments were handled.
    """
    handled = super().handle_args(args)
    if not handled:
        folder = Path(args.folder).expanduser().resolve()
        serve(folder, port=args.port, host=args.host)
        handled = True
    return handled

main(argv=None)

Command line entry point of the review pass.

Parameters:

Name	Type	Description	Default
argv	Optional[List[str]]	command line arguments; defaults to sys.argv.	None

Returns:

Type	Description
int	exit code: 0 = OK, 1 = KeyboardInterrupt, 2 = Exception.

Source code in rdd/reelreview_cmd.py

def main(argv: Optional[List[str]] = None) -> int:
    """Command line entry point of the review pass.

    Args:
        argv: command line arguments; defaults to sys.argv.

    Returns:
        exit code: 0 = OK, 1 = KeyboardInterrupt, 2 = Exception.
    """
    cmd = ReelReviewCmd()
    exit_code = cmd.run(argv)
    return exit_code

reels

Created on 2026-08-13.

the directory of reels of an installation - found by scanning a recordings directory

see the Directory of reels and Reel Review ADRs on https://media.bitplan.com/index.php/Talk:Rdd.bitplan.com

@author: wf

Reel

One reel of an installation - the published form of a Recording.

Per the Unit of reel publication decision the reel is the folder with its reel.yaml; the folder name is the identifier a url can carry, the hop set is the record the reviews and the document pass read back.

Source code in rdd/reels.py

@lod_storable
class Reel:
    """One reel of an installation - the published form of a Recording.

    Per the Unit of reel publication decision the reel is the folder
    with its reel.yaml; the folder name is the identifier a url can
    carry, the hop set is the record the reviews and the document pass
    read back.
    """

    path: str = ""
    hop_set: Optional[HopSet] = None

    @property
    def folder(self) -> str:
        """The name of the reel folder."""
        folder = os.path.basename(self.path)
        return folder

    @property
    def recording(self):
        """The Recording of this reel, None where the reel names none."""
        recording = self.hop_set.recording if self.hop_set else None
        return recording

    @property
    def acronym(self) -> str:
        """The acronym of the reel, the folder name where the reel names
        none."""
        acronym = self.folder
        recording = self.recording
        if recording and recording.acronym:
            acronym = recording.acronym
        return acronym

    @property
    def title(self) -> str:
        """The name of the recording, the acronym where the reel names none."""
        title = self.acronym
        recording = self.recording
        if recording and recording.name:
            title = recording.name
        return title

    @property
    def hop_count(self) -> int:
        """The number of hops of this reel."""
        hop_count = self.hop_set.hopCount if self.hop_set else 0
        return hop_count

    @property
    def status(self) -> str:
        """The status of this reel - the state of its Recording.

        The status continues from the processing states into publication:
        public and demo per the Reel Review decision.
        """
        status = ""
        recording = self.recording
        if recording and recording.state:
            status = recording.state
        return status

    @property
    def is_public(self) -> bool:
        """Whether anyone may inspect this reel."""
        is_public = self.status in PUBLIC_STATUSES
        return is_public

    @property
    def is_demo(self) -> bool:
        """Whether this reel is offered in true inspection mode."""
        is_demo = self.status == "demo"
        return is_demo

    @property
    def year_month(self) -> Optional[str]:
        """The year/month of the recording date - the lengthy address
        part of the Hop url decision; None where the reel names no date."""
        year_month = None
        recording = self.recording
        if recording and recording.date and len(recording.date) >= 7:
            year_month = f"{recording.date[:4]}/{recording.date[5:7]}"
        return year_month

    def hop_slugs(self) -> List[str]:
        """The persistent identifier slugs of the hops of this reel.

        Per #21 the slug is the evidence frame name minus its extension -
        the base name, so frame, hop and url carry one identity even
        when the frame lies in a subfolder.

        Returns:
            the slugs of the hops carrying a screenshot.
        """
        hops = self.hop_set.hops if self.hop_set else []
        slugs = [
            os.path.splitext(os.path.basename(hop.screenshot))[0]
            for hop in hops
            if hop.screenshot
        ]
        return slugs

acronym property

The acronym of the reel, the folder name where the reel names none.

folder property

The name of the reel folder.

hop_count property

The number of hops of this reel.

is_demo property

Whether this reel is offered in true inspection mode.

is_public property

Whether anyone may inspect this reel.

recording property

The Recording of this reel, None where the reel names none.

status property

The status of this reel - the state of its Recording.

The status continues from the processing states into publication: public and demo per the Reel Review decision.

title property

The name of the recording, the acronym where the reel names none.

year_month property

The year/month of the recording date - the lengthy address part of the Hop url decision; None where the reel names no date.

hop_slugs()

The persistent identifier slugs of the hops of this reel.

Per #21 the slug is the evidence frame name minus its extension - the base name, so frame, hop and url carry one identity even when the frame lies in a subfolder.

Returns:

Type	Description
List[str]	the slugs of the hops carrying a screenshot.

Source code in rdd/reels.py

def hop_slugs(self) -> List[str]:
    """The persistent identifier slugs of the hops of this reel.

    Per #21 the slug is the evidence frame name minus its extension -
    the base name, so frame, hop and url carry one identity even
    when the frame lies in a subfolder.

    Returns:
        the slugs of the hops carrying a screenshot.
    """
    hops = self.hop_set.hops if self.hop_set else []
    slugs = [
        os.path.splitext(os.path.basename(hop.screenshot))[0]
        for hop in hops
        if hop.screenshot
    ]
    return slugs

Reels

The directory of reels below a recordings directory.

The directory is built at startup by walking for reel.yaml files and reading each of them, so what the site offers is what the disk has. The walk and the read are timed - a scan that grows with the number of reels has to be measurable before a cache is worth its complexity.

Source code in rdd/reels.py

@lod_storable
class Reels:
    """The directory of reels below a recordings directory.

    The directory is built at startup by walking for reel.yaml files and
    reading each of them, so what the site offers is what the disk has.
    The walk and the read are timed - a scan that grows with the number
    of reels has to be measurable before a cache is worth its complexity.
    """

    recordings_path: str = ""
    reels: List[Reel] = field(default_factory=list)
    found: int = 0
    walk_time: float = 0.0
    read_time: float = 0.0
    errors: Dict[str, str] = field(default_factory=dict)

    @property
    def count(self) -> int:
        """The number of reels held in memory."""
        count = len(self.reels)
        return count

    @property
    def total_time(self) -> float:
        """The time the whole scan took."""
        total_time = self.walk_time + self.read_time
        return total_time

    @classmethod
    def paths_of(cls, recordings_dir: str) -> List[str]:
        """Get the sorted paths of the reel folders below the given directory.

        A reel folder is a directory carrying a reel.yaml; the walk does
        not descend into a reel folder, so files beside the reel - frames,
        video, document - cost nothing but their directory entry.

        Args:
            recordings_dir: the directory to walk.

        Returns:
            the sorted paths of the reel folders.
        """
        paths = []
        for dir_path, dir_names, file_names in os.walk(recordings_dir):
            if HopSet.FILE_NAME in file_names:
                paths.append(dir_path)
                dir_names.clear()
        paths = sorted(paths)
        return paths

    @classmethod
    def of_dir(cls, recordings_dir: str) -> "Reels":
        """Scan the given directory into the directory of reels.

        Args:
            recordings_dir: the directory holding the reel folders.

        Returns:
            the reels found, with the timings of the scan.
        """
        reels = cls(recordings_path=recordings_dir)
        walk_start = time.time()
        paths = cls.paths_of(recordings_dir)
        reels.walk_time = time.time() - walk_start
        reels.found = len(paths)
        read_start = time.time()
        for path in paths:
            try:
                hop_set = HopSet.of_dir(path)
                reels.reels.append(Reel(path=path, hop_set=hop_set))
            except Exception as ex:
                reels.errors[path] = str(ex)
        reels.read_time = time.time() - read_start
        return reels

    def by_acronym(self) -> Dict[str, Reel]:
        """The directory of reels - the lookup from acronym to reel.

        Returns:
            the lookup from acronym to reel.
        """
        lookup = {reel.acronym: reel for reel in self.reels}
        return lookup

    def by_pid(self) -> Dict[str, Reel]:
        """The lookup by the lengthy address - year/month/acronym.

        Per the Hop url decision the acronym is a shortcut that holds
        while acronyms are unique; the lengthy form disambiguates by
        the year and month of the Recording date.

        Returns:
            the lookup from year/month/acronym to reel.
        """
        lookup = {
            f"{reel.year_month}/{reel.acronym}": reel
            for reel in self.reels
            if reel.year_month
        }
        return lookup

    def visible(self, granted: Optional[List[str]] = None) -> List[Reel]:
        """The reels the holder of the given right may see.

        Anyone sees the public and demo reels; a Review right adds its
        private reels - per the Reel Review decision the access right
        changes the visibility of reels in the reels directory. The
        wildcard '*' grants every reel per the Owner bootstrap decision.

        Args:
            granted: the acronyms a Review grants; None for anonymous.

        Returns:
            the visible reels, in the order of the scan.
        """
        granted_set = set(granted) if granted else set()
        if "*" in granted_set:
            visible_reels = list(self.reels)
        else:
            visible_reels = [
                reel
                for reel in self.reels
                if reel.is_public or reel.acronym in granted_set
            ]
        return visible_reels

    def as_summary(self) -> str:
        """A one line summary of the scan for the service log.

        Returns:
            the counts and timings of this scan.
        """
        summary = (
            f"{self.count} of {self.found} reels from {self.recordings_path} "
            f"in {self.total_time:.3f}s "
            f"(walk {self.walk_time:.3f}s read {self.read_time:.3f}s)"
        )
        if self.errors:
            summary += f" - {len(self.errors)} unreadable"
        return summary

count property

The number of reels held in memory.

total_time property

The time the whole scan took.

as_summary()

A one line summary of the scan for the service log.

Returns:

Type	Description
str	the counts and timings of this scan.

Source code in rdd/reels.py

def as_summary(self) -> str:
    """A one line summary of the scan for the service log.

    Returns:
        the counts and timings of this scan.
    """
    summary = (
        f"{self.count} of {self.found} reels from {self.recordings_path} "
        f"in {self.total_time:.3f}s "
        f"(walk {self.walk_time:.3f}s read {self.read_time:.3f}s)"
    )
    if self.errors:
        summary += f" - {len(self.errors)} unreadable"
    return summary

by_acronym()

The directory of reels - the lookup from acronym to reel.

Returns:

Type	Description
Dict[str, Reel]	the lookup from acronym to reel.

Source code in rdd/reels.py

def by_acronym(self) -> Dict[str, Reel]:
    """The directory of reels - the lookup from acronym to reel.

    Returns:
        the lookup from acronym to reel.
    """
    lookup = {reel.acronym: reel for reel in self.reels}
    return lookup

by_pid()

The lookup by the lengthy address - year/month/acronym.

Per the Hop url decision the acronym is a shortcut that holds while acronyms are unique; the lengthy form disambiguates by the year and month of the Recording date.

Returns:

Type	Description
Dict[str, Reel]	the lookup from year/month/acronym to reel.

Source code in rdd/reels.py

def by_pid(self) -> Dict[str, Reel]:
    """The lookup by the lengthy address - year/month/acronym.

    Per the Hop url decision the acronym is a shortcut that holds
    while acronyms are unique; the lengthy form disambiguates by
    the year and month of the Recording date.

    Returns:
        the lookup from year/month/acronym to reel.
    """
    lookup = {
        f"{reel.year_month}/{reel.acronym}": reel
        for reel in self.reels
        if reel.year_month
    }
    return lookup

of_dir(recordings_dir) classmethod

Scan the given directory into the directory of reels.

Parameters:

Name	Type	Description	Default
recordings_dir	str	the directory holding the reel folders.	required

Returns:

Type	Description
Reels	the reels found, with the timings of the scan.

Source code in rdd/reels.py

@classmethod
def of_dir(cls, recordings_dir: str) -> "Reels":
    """Scan the given directory into the directory of reels.

    Args:
        recordings_dir: the directory holding the reel folders.

    Returns:
        the reels found, with the timings of the scan.
    """
    reels = cls(recordings_path=recordings_dir)
    walk_start = time.time()
    paths = cls.paths_of(recordings_dir)
    reels.walk_time = time.time() - walk_start
    reels.found = len(paths)
    read_start = time.time()
    for path in paths:
        try:
            hop_set = HopSet.of_dir(path)
            reels.reels.append(Reel(path=path, hop_set=hop_set))
        except Exception as ex:
            reels.errors[path] = str(ex)
    reels.read_time = time.time() - read_start
    return reels

paths_of(recordings_dir) classmethod

Get the sorted paths of the reel folders below the given directory.

A reel folder is a directory carrying a reel.yaml; the walk does not descend into a reel folder, so files beside the reel - frames, video, document - cost nothing but their directory entry.

Parameters:

Name	Type	Description	Default
recordings_dir	str	the directory to walk.	required

Returns:

Type	Description
List[str]	the sorted paths of the reel folders.

Source code in rdd/reels.py

@classmethod
def paths_of(cls, recordings_dir: str) -> List[str]:
    """Get the sorted paths of the reel folders below the given directory.

    A reel folder is a directory carrying a reel.yaml; the walk does
    not descend into a reel folder, so files beside the reel - frames,
    video, document - cost nothing but their directory entry.

    Args:
        recordings_dir: the directory to walk.

    Returns:
        the sorted paths of the reel folders.
    """
    paths = []
    for dir_path, dir_names, file_names in os.walk(recordings_dir):
        if HopSet.FILE_NAME in file_names:
            paths.append(dir_path)
            dir_names.clear()
    paths = sorted(paths)
    return paths

visible(granted=None)

The reels the holder of the given right may see.

Anyone sees the public and demo reels; a Review right adds its private reels - per the Reel Review decision the access right changes the visibility of reels in the reels directory. The wildcard '*' grants every reel per the Owner bootstrap decision.

Parameters:

Name	Type	Description	Default
granted	Optional[List[str]]	the acronyms a Review grants; None for anonymous.	None

Returns:

Type	Description
List[Reel]	the visible reels, in the order of the scan.

Source code in rdd/reels.py

def visible(self, granted: Optional[List[str]] = None) -> List[Reel]:
    """The reels the holder of the given right may see.

    Anyone sees the public and demo reels; a Review right adds its
    private reels - per the Reel Review decision the access right
    changes the visibility of reels in the reels directory. The
    wildcard '*' grants every reel per the Owner bootstrap decision.

    Args:
        granted: the acronyms a Review grants; None for anonymous.

    Returns:
        the visible reels, in the order of the scan.
    """
    granted_set = set(granted) if granted else set()
    if "*" in granted_set:
        visible_reels = list(self.reels)
    else:
        visible_reels = [
            reel
            for reel in self.reels
            if reel.is_public or reel.acronym in granted_set
        ]
    return visible_reels

reelsite_cmd

Created on 2026-08-12.

command line interface of the reel site

@author: wf

ReelSiteCmd

Bases: BaseCmd

Serve the reel site of an organization.

Source code in rdd/reelsite_cmd.py

class ReelSiteCmd(BaseCmd):
    """Serve the reel site of an organization."""

    def __init__(self):
        """Initialize with the reel-driven-development version info."""
        super().__init__(Version())

    def add_arguments(self, parser: argparse.ArgumentParser):
        """Add the site arguments to the given parser.

        Args:
            parser: the parser to add arguments to.
        """
        super().add_arguments(parser)
        parser.add_argument(
            "-c",
            "--config",
            default=RddSiteConfig.DEFAULT_PATH,
            help="site configuration yaml [default: %(default)s]",
        )
        parser.add_argument(
            "--host", default="127.0.0.1", help="interface to listen on"
        )
        parser.add_argument("-p", "--port", type=int, help="port to serve on")
        parser.add_argument("-s", "--serve", action="store_true", help="serve the site")
        parser.add_argument(
            "--init",
            action="store_true",
            help="initialize the site: seed the owner and mint the owner token",
        )
        parser.add_argument(
            "--mint",
            metavar="PERSON",
            help="mint a review token for the given person",
        )
        parser.add_argument(
            "--meeting", default="", help="the meeting a minted review belongs to"
        )
        parser.add_argument(
            "--reels",
            nargs="+",
            default=[],
            metavar="ACRONYM",
            help="the reels a minted review grants",
        )

    def handle_args(self, args: argparse.Namespace) -> bool:
        """Handle the parsed arguments - init, mint or serve.

        Args:
            args: parsed argument namespace.

        Returns:
            True if the arguments were handled.
        """
        handled = super().handle_args(args)
        if handled:
            return handled
        config = RddSiteConfig.of_path(args.config)
        if args.port:
            config.port = args.port
        if args.init:
            self.init_site(config)
            handled = True
        elif args.mint:
            mint = Mint(config)
            url = mint.mint_review(args.mint, args.meeting, args.reels)
            print(url)
            handled = True
        elif args.serve:
            config_path = Path(args.config).expanduser()
            if not config_path.exists():
                raise ValueError(f"no site configuration at {config_path}")
            reviews_file = os.path.expanduser(Reviews.DEFAULT_PATH)
            if not os.path.isfile(reviews_file) and sys.stdin.isatty():
                # per the Owner bootstrap decision the first interactive
                # start is installation mode - the owner is asked here;
                # a non-interactive start serves the installation state
                self.init_site(config)
            serve(config, host=args.host)
            handled = True
        return handled

    def init_site(self, config: RddSiteConfig):
        """Run installation mode - seed the owner interactively.

        Per the Owner bootstrap decision the owner link is shown once
        on the interactive terminal and written to a mode 600 file; it
        goes nowhere else.

        Args:
            config: the site configuration.
        """
        mint = Mint(config)
        username = input("owner username: ").strip()
        name = input("owner full name: ").strip()
        email = input("owner email: ").strip()
        url = input("owner url: ").strip()
        owner_url = mint.init_site(username, name, email, url)
        print(f"owner link (also in {mint.owner_link_path}, mode 600):")
        print(owner_url)

__init__()

Initialize with the reel-driven-development version info.

Source code in rdd/reelsite_cmd.py

def __init__(self):
    """Initialize with the reel-driven-development version info."""
    super().__init__(Version())

add_arguments(parser)

Add the site arguments to the given parser.

Parameters:

Name	Type	Description	Default
parser	ArgumentParser	the parser to add arguments to.	required

Source code in rdd/reelsite_cmd.py

def add_arguments(self, parser: argparse.ArgumentParser):
    """Add the site arguments to the given parser.

    Args:
        parser: the parser to add arguments to.
    """
    super().add_arguments(parser)
    parser.add_argument(
        "-c",
        "--config",
        default=RddSiteConfig.DEFAULT_PATH,
        help="site configuration yaml [default: %(default)s]",
    )
    parser.add_argument(
        "--host", default="127.0.0.1", help="interface to listen on"
    )
    parser.add_argument("-p", "--port", type=int, help="port to serve on")
    parser.add_argument("-s", "--serve", action="store_true", help="serve the site")
    parser.add_argument(
        "--init",
        action="store_true",
        help="initialize the site: seed the owner and mint the owner token",
    )
    parser.add_argument(
        "--mint",
        metavar="PERSON",
        help="mint a review token for the given person",
    )
    parser.add_argument(
        "--meeting", default="", help="the meeting a minted review belongs to"
    )
    parser.add_argument(
        "--reels",
        nargs="+",
        default=[],
        metavar="ACRONYM",
        help="the reels a minted review grants",
    )

handle_args(args)

Handle the parsed arguments - init, mint or serve.

Parameters:

Name	Type	Description	Default
args	Namespace	parsed argument namespace.	required

Returns:

Type	Description
bool	True if the arguments were handled.

Source code in rdd/reelsite_cmd.py

def handle_args(self, args: argparse.Namespace) -> bool:
    """Handle the parsed arguments - init, mint or serve.

    Args:
        args: parsed argument namespace.

    Returns:
        True if the arguments were handled.
    """
    handled = super().handle_args(args)
    if handled:
        return handled
    config = RddSiteConfig.of_path(args.config)
    if args.port:
        config.port = args.port
    if args.init:
        self.init_site(config)
        handled = True
    elif args.mint:
        mint = Mint(config)
        url = mint.mint_review(args.mint, args.meeting, args.reels)
        print(url)
        handled = True
    elif args.serve:
        config_path = Path(args.config).expanduser()
        if not config_path.exists():
            raise ValueError(f"no site configuration at {config_path}")
        reviews_file = os.path.expanduser(Reviews.DEFAULT_PATH)
        if not os.path.isfile(reviews_file) and sys.stdin.isatty():
            # per the Owner bootstrap decision the first interactive
            # start is installation mode - the owner is asked here;
            # a non-interactive start serves the installation state
            self.init_site(config)
        serve(config, host=args.host)
        handled = True
    return handled

init_site(config)

Run installation mode - seed the owner interactively.

Per the Owner bootstrap decision the owner link is shown once on the interactive terminal and written to a mode 600 file; it goes nowhere else.

Parameters:

Name	Type	Description	Default
config	RddSiteConfig	the site configuration.	required

Source code in rdd/reelsite_cmd.py

def init_site(self, config: RddSiteConfig):
    """Run installation mode - seed the owner interactively.

    Per the Owner bootstrap decision the owner link is shown once
    on the interactive terminal and written to a mode 600 file; it
    goes nowhere else.

    Args:
        config: the site configuration.
    """
    mint = Mint(config)
    username = input("owner username: ").strip()
    name = input("owner full name: ").strip()
    email = input("owner email: ").strip()
    url = input("owner url: ").strip()
    owner_url = mint.init_site(username, name, email, url)
    print(f"owner link (also in {mint.owner_link_path}, mode 600):")
    print(owner_url)

main(argv=None)

Command line entry point of the reel site.

Parameters:

Name	Type	Description	Default
argv	Optional[List[str]]	command line arguments; defaults to sys.argv.	None

Returns:

Type	Description
int	exit code: 0 = OK, 1 = KeyboardInterrupt, 2 = Exception.

Source code in rdd/reelsite_cmd.py

def main(argv: Optional[List[str]] = None) -> int:
    """Command line entry point of the reel site.

    Args:
        argv: command line arguments; defaults to sys.argv.

    Returns:
        exit code: 0 = OK, 1 = KeyboardInterrupt, 2 = Exception.
    """
    cmd = ReelSiteCmd()
    exit_code = cmd.run(argv)
    return exit_code

transcript

Created on 2026-08-10.

the transcript of a reel

Schema: the Meeting context of https://contexts.bitplan.com * https://contexts.bitplan.com/index.php/Concept:TranscriptSegment

The segments live in a file of their own beside the reel: the reel file is what a person curates by hand, and a transcript of a few hundred segments would drown the few dozen hops in it.

see https://github.com/WolfgangFahl/reel-driven-development/issues/25

@author: wf

Transcript

The improved transcript of one reel.

The raw result of the transcription stays as it came out of the tool; this is the corrected reading of it, and the two stay diffable.

Source code in rdd/transcript.py

@lod_storable
class Transcript:
    """The improved transcript of one reel.

    The raw result of the transcription stays as it came out of the
    tool; this is the corrected reading of it, and the two stay
    diffable.
    """

    FILE_NAME = "segments.yaml"

    segments: List[TranscriptSegment] = None

    def __post_init__(self):
        """Start with an empty segment list where none was given."""
        if self.segments is None:
            self.segments = []

    @property
    def segmentCount(self) -> int:
        """The number of segments of this transcript."""
        segment_count = len(self.segments)
        return segment_count

    @classmethod
    def path_of(cls, folder: str) -> str:
        """Get the path of the transcript file in the given folder.

        Args:
            folder: the recording folder.

        Returns:
            the path of the transcript file.
        """
        transcript_path = os.path.join(folder, cls.FILE_NAME)
        return transcript_path

    @classmethod
    def of_dir(cls, folder: str) -> Optional["Transcript"]:
        """Get the transcript of the given folder.

        Args:
            folder: the recording folder.

        Returns:
            the transcript, None where the folder carries none - a reel
            may be documented before its transcript is improved.
        """
        transcript_path = cls.path_of(folder)
        transcript = None
        if os.path.isfile(transcript_path):
            transcript = cls.load_from_yaml_file(transcript_path)
        return transcript

segmentCount property

The number of segments of this transcript.

__post_init__()

Start with an empty segment list where none was given.

Source code in rdd/transcript.py

def __post_init__(self):
    """Start with an empty segment list where none was given."""
    if self.segments is None:
        self.segments = []

of_dir(folder) classmethod

Get the transcript of the given folder.

Parameters:

Name	Type	Description	Default
folder	str	the recording folder.	required

Returns:

Type	Description
Optional[Transcript]	the transcript, None where the folder carries none - a reel
Optional[Transcript]	may be documented before its transcript is improved.

Source code in rdd/transcript.py

@classmethod
def of_dir(cls, folder: str) -> Optional["Transcript"]:
    """Get the transcript of the given folder.

    Args:
        folder: the recording folder.

    Returns:
        the transcript, None where the folder carries none - a reel
        may be documented before its transcript is improved.
    """
    transcript_path = cls.path_of(folder)
    transcript = None
    if os.path.isfile(transcript_path):
        transcript = cls.load_from_yaml_file(transcript_path)
    return transcript

path_of(folder) classmethod

Get the path of the transcript file in the given folder.

Parameters:

Name	Type	Description	Default
folder	str	the recording folder.	required

Returns:

Type	Description
str	the path of the transcript file.

Source code in rdd/transcript.py

@classmethod
def path_of(cls, folder: str) -> str:
    """Get the path of the transcript file in the given folder.

    Args:
        folder: the recording folder.

    Returns:
        the path of the transcript file.
    """
    transcript_path = os.path.join(folder, cls.FILE_NAME)
    return transcript_path

TranscriptSegment

One timed segment of the improved transcript of a Recording.

The field names are the property names of https://contexts.bitplan.com/index.php/Concept:TranscriptSegment and are never renamed. end stays empty where the source provides only starts; speaker stays empty where diarization and content disagree - an unattributed segment is the honest form, a guessed one is not.

Source code in rdd/transcript.py

@lod_storable
class TranscriptSegment:
    """One timed segment of the improved transcript of a Recording.

    The field names are the property names of
    https://contexts.bitplan.com/index.php/Concept:TranscriptSegment
    and are never renamed. end stays empty where the source provides only
    starts; speaker stays empty where diarization and content disagree -
    an unattributed segment is the honest form, a guessed one is not.
    """

    pos: int = 0
    start: str = ""
    end: Optional[str] = None
    speaker: Optional[str] = None
    text: str = ""

version

Created on 2026-08-02.

@author: wf

Version dataclass

Version information for reel-driven-development.

Source code in rdd/version.py

@dataclass
class Version:
    """Version information for reel-driven-development."""

    name: str = "reel-driven-development"
    version: str = rdd.__version__
    date: str = "2026-07-30"
    updated: str = "2026-08-14"
    description: str = (
        "Reel Driven Development - turn recorded user walks "
        "into domain stories and outcome objects"
    )
    authors: str = "Wolfgang Fahl"
    doc_url: str = "https://wiki.bitplan.com/index.php/Reel_Driven_Development"
    chat_url: str = (
        "https://github.com/WolfgangFahl/reel-driven-development/discussions"
    )
    cm_url: str = "https://github.com/WolfgangFahl/reel-driven-development"

webapp

Created on 2026-08-14.

the fastapi application of a reel site - the pages and the api as decided, plus /docs and /openapi.json per the OpenAPI docs issue

Per the Delivery decision the site itself answers below /reels/; the web server in front only proxies. Per the framed 404 issue every miss answers the framed page. The docs are self-contained like every other page: the swagger assets are served by the site, never by a CDN.

@author: wf

InstallationApp

The installation mode application.

Per the Owner bootstrap decision a site without reviews.yaml refuses to serve reels and names the init command on every request - the state is shown, never hidden behind a dead backend.

Source code in rdd/webapp.py

class InstallationApp:
    """The installation mode application.

    Per the Owner bootstrap decision a site without reviews.yaml refuses
    to serve reels and names the init command on every request - the
    state is shown, never hidden behind a dead backend.
    """

    def __init__(self, page: str):
        """Initialize with the installation page.

        Args:
            page: the installation mode page.
        """
        self.page = page
        self.app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
        self.add_routes()

    def add_routes(self) -> None:
        """Every request answers the installation state."""
        app = self.app
        page = self.page

        @app.api_route("/{rest:path}", methods=["GET", "POST"], include_in_schema=False)
        def installation(rest: str) -> HTMLResponse:
            """The installation mode state as service unavailable."""
            return page_response(page, status=503)

__init__(page)

Initialize with the installation page.

Parameters:

Name	Type	Description	Default
page	str	the installation mode page.	required

Source code in rdd/webapp.py

def __init__(self, page: str):
    """Initialize with the installation page.

    Args:
        page: the installation mode page.
    """
    self.page = page
    self.app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
    self.add_routes()

add_routes()

Every request answers the installation state.

Source code in rdd/webapp.py

def add_routes(self) -> None:
    """Every request answers the installation state."""
    app = self.app
    page = self.page

    @app.api_route("/{rest:path}", methods=["GET", "POST"], include_in_schema=False)
    def installation(rest: str) -> HTMLResponse:
        """The installation mode state as service unavailable."""
        return page_response(page, status=503)

RateLimit

Per-client limit on missed lookups.

Per the Reel Review decision unknown tokens are rate-limited: every miss is tarpitted, and a client whose misses exceed the limit within the window answers 429 until the window has passed.

Source code in rdd/webapp.py

class RateLimit:
    """Per-client limit on missed lookups.

    Per the Reel Review decision unknown tokens are rate-limited: every
    miss is tarpitted, and a client whose misses exceed the limit within
    the window answers 429 until the window has passed.
    """

    def __init__(self, max_misses: int = 10, window_seconds: float = 60.0):
        """Initialize the limit.

        Args:
            max_misses: the misses a client may accumulate per window.
            window_seconds: the sliding window in seconds.
        """
        self.max_misses = max_misses
        self.window_seconds = window_seconds
        self.misses: Dict[str, List[float]] = {}

    def miss(self, client: str, now: Optional[float] = None) -> bool:
        """Record a miss for the given client.

        Args:
            client: the client address the miss counts against.
            now: the time of the miss; the monotonic clock by default.

        Returns:
            True where the client is over the limit.
        """
        if now is None:
            now = time.monotonic()
        cutoff = now - self.window_seconds
        timestamps = [t for t in self.misses.get(client, []) if t > cutoff]
        timestamps.append(now)
        self.misses[client] = timestamps
        over_limit = len(timestamps) > self.max_misses
        return over_limit

__init__(max_misses=10, window_seconds=60.0)

Initialize the limit.

Parameters:

Name	Type	Description	Default
max_misses	int	the misses a client may accumulate per window.	10
window_seconds	float	the sliding window in seconds.	60.0

Source code in rdd/webapp.py

def __init__(self, max_misses: int = 10, window_seconds: float = 60.0):
    """Initialize the limit.

    Args:
        max_misses: the misses a client may accumulate per window.
        window_seconds: the sliding window in seconds.
    """
    self.max_misses = max_misses
    self.window_seconds = window_seconds
    self.misses: Dict[str, List[float]] = {}

miss(client, now=None)

Record a miss for the given client.

Parameters:

Name	Type	Description	Default
client	str	the client address the miss counts against.	required
now	Optional[float]	the time of the miss; the monotonic clock by default.	None

Returns:

Type	Description
bool	True where the client is over the limit.

Source code in rdd/webapp.py

def miss(self, client: str, now: Optional[float] = None) -> bool:
    """Record a miss for the given client.

    Args:
        client: the client address the miss counts against.
        now: the time of the miss; the monotonic clock by default.

    Returns:
        True where the client is over the limit.
    """
    if now is None:
        now = time.monotonic()
    cutoff = now - self.window_seconds
    timestamps = [t for t in self.misses.get(client, []) if t > cutoff]
    timestamps.append(now)
    self.misses[client] = timestamps
    over_limit = len(timestamps) > self.max_misses
    return over_limit

ReelApp

The fastapi application of a reel site.

One instance wires the routes of one ReelSite; the app is what uvicorn serves and what /docs documents.

Source code in rdd/webapp.py

class ReelApp:
    """The fastapi application of a reel site.

    One instance wires the routes of one ReelSite; the app is what
    uvicorn serves and what /docs documents.
    """

    def __init__(self, site: ReelSite):
        """Initialize with the site to serve.

        Args:
            site: the reel site.
        """
        self.site = site
        self.tarpit_seconds = TARPIT_SECONDS
        self.rate_limit = RateLimit()
        self.app = FastAPI(
            title=site.config.title,
            version=site.version.version,
            description=(
                "The api of a reel site per the Delivery and Reel Review "
                "decisions: a reel and its files are served by acronym and "
                "right. Unknown tokens, unknown acronyms and denied reels "
                "answer alike - a tarpitted framed 404 - so neither tokens "
                "nor private acronyms can be probed."
            ),
            docs_url=None,
            redoc_url=None,
        )
        self.app.mount(
            "/static/swagger",
            StaticFiles(directory=swagger_ui_path),
            name="swagger",
        )
        self.add_routes()

    def not_found(self, request: Request, tarpit: bool = False) -> HTMLResponse:
        """The framed 404 response.

        Args:
            request: the request that has no page.
            tarpit: delay the answer so tokens and private acronyms
                cannot be probed; a client over the rate limit
                answers 429 instead.

        Returns:
            the framed 404 page as a response; 429 over the limit.
        """
        path = request.url.path
        lang = lang_of(request)
        page = self.site.not_found(path, lang)
        status = 404
        if tarpit:
            client = request.client.host if request.client else "?"
            if self.rate_limit.miss(client):
                status = 429
            else:
                time.sleep(self.tarpit_seconds)
        response = page_response(page, status=status)
        if status == 429:
            response.headers["Retry-After"] = str(int(self.rate_limit.window_seconds))
        return response

    def address_parts(
        self, address: str
    ) -> Tuple[Optional[Review], Optional[Reel], List[str]]:
        """Resolve an address below /reels/ into right, reel and rest.

        Args:
            address: the path after /reels/, optionally token first.

        Returns:
            the review right or None, the reel or None, and the
            remaining path parts.
        """
        parts = [urllib.parse.unquote(part) for part in address.split("/")]
        review = self.site.reviews.by_token().get(parts[0])
        if review is not None:
            parts = parts[1:]
        reel, file_parts = self.site.resolve_reel(parts)
        return review, reel, file_parts

    def checked_reel(
        self, address: str
    ) -> Tuple[Optional[Reel], Optional[Review], List[str]]:
        """The reel of the given address where the right allows it.

        Args:
            address: the path after /reels/, optionally token first.

        Returns:
            reel, review and remaining parts; reel is None where the
            address resolves to nothing the right allows.
        """
        review, reel, file_parts = self.address_parts(address)
        if reel is not None and not self.site.allowed(reel, review):
            reel = None
        return reel, review, file_parts

    def add_routes(self) -> None:
        """Wire the routes of the site."""
        app = self.app
        site = self.site

        @app.get("/", response_class=HTMLResponse, summary="the home page")
        @app.get("/index.html", response_class=HTMLResponse, include_in_schema=False)
        def home(request: Request) -> HTMLResponse:
            """The home page - what this site is and the ways in."""
            lang = lang_of(request)
            return remember_lang(request, page_response(site.home(lang)))

        @app.get("/reels", response_class=HTMLResponse, summary="the reels directory")
        def reels(request: Request) -> HTMLResponse:
            """The public reels directory."""
            lang = lang_of(request)
            return remember_lang(request, page_response(site.reels(lang=lang)))

        @app.get("/about", response_class=HTMLResponse, summary="the about page")
        def about(request: Request) -> HTMLResponse:
            """The about page - version, license and repository."""
            lang = lang_of(request)
            return remember_lang(request, page_response(site.about(lang)))

        @app.get("/docs", include_in_schema=False)
        def docs() -> HTMLResponse:
            """The api documentation - swagger assets served by the site."""
            return get_swagger_ui_html(
                openapi_url="/openapi.json",
                title=f"{site.config.title} - api",
                swagger_js_url="/static/swagger/swagger-ui-bundle.js",
                swagger_css_url="/static/swagger/swagger-ui.css",
                swagger_favicon_url="/static/swagger/favicon-32x32.png",
            )

        @app.get(
            "/reels/{address:path}/api/files",
            summary="the files of a reel",
        )
        def api_files(address: str, request: Request):
            """The sorted file names of the reel - the review page's read api."""
            reel, _review, file_parts = self.checked_reel(address)
            if reel is None or file_parts:
                return self.not_found(request, tarpit=True)
            return JSONResponse(site.reel_files(reel))

        @app.get(
            "/reels/{address:path}/api/info",
            summary="folder and acronym of a reel",
        )
        def api_info(address: str, request: Request):
            """Folder and acronym of the reel."""
            reel, _review, file_parts = self.checked_reel(address)
            if reel is None or file_parts:
                return self.not_found(request, tarpit=True)
            return JSONResponse({"folder": reel.folder, "acronym": reel.acronym})

        @app.get(
            "/reels/{address:path}/api/reel",
            summary="the hop set of a reel",
        )
        def api_reel(address: str, request: Request):
            """The hop set parsed by the model - the page never parses YAML."""
            reel, _review, file_parts = self.checked_reel(address)
            if reel is None or file_parts:
                return self.not_found(request, tarpit=True)
            return JSONResponse(reel.hop_set.to_dict() if reel.hop_set else {})

        @app.get(
            "/reels/{address:path}/api/zip",
            summary="the reel folder as one zip",
        )
        def api_zip(address: str, request: Request):
            """The reel folder zipped - the verdict page's download per the
            Reel verdict decision."""
            reel, _review, file_parts = self.checked_reel(address)
            if reel is None or file_parts:
                return self.not_found(request, tarpit=True)
            zip_path = site.reel_zip(reel)
            return FileResponse(
                zip_path,
                media_type="application/zip",
                filename=f"{reel.acronym}.zip",
                background=BackgroundTask(os.remove, zip_path),
            )

        @app.post(
            "/reels/{address:path}/api/{action}",
            summary="the write api - true inspection mode",
        )
        async def api_write(address: str, action: str, request: Request):
            """Per the Reel Review decision a save on this site answers success
            and stores nothing; the request must name an allowed reel, so the
            write api reveals no more than the read api."""
            reel, _review, file_parts = self.checked_reel(address)
            if (
                reel is None
                or file_parts
                or action not in ("save", "feedback", "upload")
            ):
                return self.not_found(request, tarpit=True)
            await request.body()
            return JSONResponse({})

        @app.get(
            "/reels/{rest:path}",
            response_class=HTMLResponse,
            summary="a reel, its review or one of its files",
        )
        def reel_route(rest: str, request: Request):
            """Delivery per the Delivery and Hop url decisions.

            The url is /reels/[token/][yyyy/mm/]acronym/[file|review|hop-slug].
            A bare token answers the reels directory of its review.
            """
            path = request.url.path
            lang = lang_of(request)
            parts = [urllib.parse.unquote(part) for part in rest.split("/")]
            review = site.reviews.by_token().get(parts[0])
            if review is not None:
                parts = parts[1:]
                if not parts or parts == [""]:
                    return remember_lang(
                        request, page_response(site.reels(review, lang=lang))
                    )
            elif not rest or rest == "":
                return remember_lang(request, page_response(site.reels(lang=lang)))
            reel, file_parts = site.resolve_reel(parts)
            if reel is None or not site.allowed(reel, review):
                return self.not_found(request, tarpit=True)
            if not file_parts:
                if not path.endswith("/"):
                    # the reel page needs its trailing slash so its relative
                    # review and file links resolve below the reel
                    return RedirectResponse(path + "/", status_code=301)
                return remember_lang(request, page_response(site.reel_page(reel, lang)))
            if file_parts in (["review"], ["reelreview.html"], ["verdict"]):
                return remember_lang(request, page_response(site.review_page(lang)))
            if len(file_parts) == 1 and file_parts[0] in reel.hop_slugs():
                return remember_lang(request, page_response(site.review_page(lang)))
            file_path = os.path.realpath(os.path.join(reel.path, *file_parts))
            reel_dir = os.path.realpath(reel.path)
            if not file_path.startswith(reel_dir + os.sep) or not os.path.isfile(
                file_path
            ):
                return self.not_found(request)
            return FileResponse(file_path)

        @app.exception_handler(404)
        async def framed_404(request: Request, _exception) -> HTMLResponse:
            """Any miss answers the framed 404 page."""
            page = site.not_found(request.url.path, lang_of(request))
            return page_response(page, status=404)

__init__(site)

Initialize with the site to serve.

Parameters:

Name	Type	Description	Default
site	ReelSite	the reel site.	required

Source code in rdd/webapp.py

def __init__(self, site: ReelSite):
    """Initialize with the site to serve.

    Args:
        site: the reel site.
    """
    self.site = site
    self.tarpit_seconds = TARPIT_SECONDS
    self.rate_limit = RateLimit()
    self.app = FastAPI(
        title=site.config.title,
        version=site.version.version,
        description=(
            "The api of a reel site per the Delivery and Reel Review "
            "decisions: a reel and its files are served by acronym and "
            "right. Unknown tokens, unknown acronyms and denied reels "
            "answer alike - a tarpitted framed 404 - so neither tokens "
            "nor private acronyms can be probed."
        ),
        docs_url=None,
        redoc_url=None,
    )
    self.app.mount(
        "/static/swagger",
        StaticFiles(directory=swagger_ui_path),
        name="swagger",
    )
    self.add_routes()

add_routes()

Wire the routes of the site.

Source code in rdd/webapp.py

def add_routes(self) -> None:
    """Wire the routes of the site."""
    app = self.app
    site = self.site

    @app.get("/", response_class=HTMLResponse, summary="the home page")
    @app.get("/index.html", response_class=HTMLResponse, include_in_schema=False)
    def home(request: Request) -> HTMLResponse:
        """The home page - what this site is and the ways in."""
        lang = lang_of(request)
        return remember_lang(request, page_response(site.home(lang)))

    @app.get("/reels", response_class=HTMLResponse, summary="the reels directory")
    def reels(request: Request) -> HTMLResponse:
        """The public reels directory."""
        lang = lang_of(request)
        return remember_lang(request, page_response(site.reels(lang=lang)))

    @app.get("/about", response_class=HTMLResponse, summary="the about page")
    def about(request: Request) -> HTMLResponse:
        """The about page - version, license and repository."""
        lang = lang_of(request)
        return remember_lang(request, page_response(site.about(lang)))

    @app.get("/docs", include_in_schema=False)
    def docs() -> HTMLResponse:
        """The api documentation - swagger assets served by the site."""
        return get_swagger_ui_html(
            openapi_url="/openapi.json",
            title=f"{site.config.title} - api",
            swagger_js_url="/static/swagger/swagger-ui-bundle.js",
            swagger_css_url="/static/swagger/swagger-ui.css",
            swagger_favicon_url="/static/swagger/favicon-32x32.png",
        )

    @app.get(
        "/reels/{address:path}/api/files",
        summary="the files of a reel",
    )
    def api_files(address: str, request: Request):
        """The sorted file names of the reel - the review page's read api."""
        reel, _review, file_parts = self.checked_reel(address)
        if reel is None or file_parts:
            return self.not_found(request, tarpit=True)
        return JSONResponse(site.reel_files(reel))

    @app.get(
        "/reels/{address:path}/api/info",
        summary="folder and acronym of a reel",
    )
    def api_info(address: str, request: Request):
        """Folder and acronym of the reel."""
        reel, _review, file_parts = self.checked_reel(address)
        if reel is None or file_parts:
            return self.not_found(request, tarpit=True)
        return JSONResponse({"folder": reel.folder, "acronym": reel.acronym})

    @app.get(
        "/reels/{address:path}/api/reel",
        summary="the hop set of a reel",
    )
    def api_reel(address: str, request: Request):
        """The hop set parsed by the model - the page never parses YAML."""
        reel, _review, file_parts = self.checked_reel(address)
        if reel is None or file_parts:
            return self.not_found(request, tarpit=True)
        return JSONResponse(reel.hop_set.to_dict() if reel.hop_set else {})

    @app.get(
        "/reels/{address:path}/api/zip",
        summary="the reel folder as one zip",
    )
    def api_zip(address: str, request: Request):
        """The reel folder zipped - the verdict page's download per the
        Reel verdict decision."""
        reel, _review, file_parts = self.checked_reel(address)
        if reel is None or file_parts:
            return self.not_found(request, tarpit=True)
        zip_path = site.reel_zip(reel)
        return FileResponse(
            zip_path,
            media_type="application/zip",
            filename=f"{reel.acronym}.zip",
            background=BackgroundTask(os.remove, zip_path),
        )

    @app.post(
        "/reels/{address:path}/api/{action}",
        summary="the write api - true inspection mode",
    )
    async def api_write(address: str, action: str, request: Request):
        """Per the Reel Review decision a save on this site answers success
        and stores nothing; the request must name an allowed reel, so the
        write api reveals no more than the read api."""
        reel, _review, file_parts = self.checked_reel(address)
        if (
            reel is None
            or file_parts
            or action not in ("save", "feedback", "upload")
        ):
            return self.not_found(request, tarpit=True)
        await request.body()
        return JSONResponse({})

    @app.get(
        "/reels/{rest:path}",
        response_class=HTMLResponse,
        summary="a reel, its review or one of its files",
    )
    def reel_route(rest: str, request: Request):
        """Delivery per the Delivery and Hop url decisions.

        The url is /reels/[token/][yyyy/mm/]acronym/[file|review|hop-slug].
        A bare token answers the reels directory of its review.
        """
        path = request.url.path
        lang = lang_of(request)
        parts = [urllib.parse.unquote(part) for part in rest.split("/")]
        review = site.reviews.by_token().get(parts[0])
        if review is not None:
            parts = parts[1:]
            if not parts or parts == [""]:
                return remember_lang(
                    request, page_response(site.reels(review, lang=lang))
                )
        elif not rest or rest == "":
            return remember_lang(request, page_response(site.reels(lang=lang)))
        reel, file_parts = site.resolve_reel(parts)
        if reel is None or not site.allowed(reel, review):
            return self.not_found(request, tarpit=True)
        if not file_parts:
            if not path.endswith("/"):
                # the reel page needs its trailing slash so its relative
                # review and file links resolve below the reel
                return RedirectResponse(path + "/", status_code=301)
            return remember_lang(request, page_response(site.reel_page(reel, lang)))
        if file_parts in (["review"], ["reelreview.html"], ["verdict"]):
            return remember_lang(request, page_response(site.review_page(lang)))
        if len(file_parts) == 1 and file_parts[0] in reel.hop_slugs():
            return remember_lang(request, page_response(site.review_page(lang)))
        file_path = os.path.realpath(os.path.join(reel.path, *file_parts))
        reel_dir = os.path.realpath(reel.path)
        if not file_path.startswith(reel_dir + os.sep) or not os.path.isfile(
            file_path
        ):
            return self.not_found(request)
        return FileResponse(file_path)

    @app.exception_handler(404)
    async def framed_404(request: Request, _exception) -> HTMLResponse:
        """Any miss answers the framed 404 page."""
        page = site.not_found(request.url.path, lang_of(request))
        return page_response(page, status=404)

address_parts(address)

Resolve an address below /reels/ into right, reel and rest.

Parameters:

Name	Type	Description	Default
address	str	the path after /reels/, optionally token first.	required

Returns:

Type	Description
Optional[Review]	the review right or None, the reel or None, and the
Optional[Reel]	remaining path parts.

Source code in rdd/webapp.py

def address_parts(
    self, address: str
) -> Tuple[Optional[Review], Optional[Reel], List[str]]:
    """Resolve an address below /reels/ into right, reel and rest.

    Args:
        address: the path after /reels/, optionally token first.

    Returns:
        the review right or None, the reel or None, and the
        remaining path parts.
    """
    parts = [urllib.parse.unquote(part) for part in address.split("/")]
    review = self.site.reviews.by_token().get(parts[0])
    if review is not None:
        parts = parts[1:]
    reel, file_parts = self.site.resolve_reel(parts)
    return review, reel, file_parts

checked_reel(address)

The reel of the given address where the right allows it.

Parameters:

Name	Type	Description	Default
address	str	the path after /reels/, optionally token first.	required

Returns:

Type	Description
Optional[Reel]	reel, review and remaining parts; reel is None where the
Optional[Review]	address resolves to nothing the right allows.

Source code in rdd/webapp.py

def checked_reel(
    self, address: str
) -> Tuple[Optional[Reel], Optional[Review], List[str]]:
    """The reel of the given address where the right allows it.

    Args:
        address: the path after /reels/, optionally token first.

    Returns:
        reel, review and remaining parts; reel is None where the
        address resolves to nothing the right allows.
    """
    review, reel, file_parts = self.address_parts(address)
    if reel is not None and not self.site.allowed(reel, review):
        reel = None
    return reel, review, file_parts

not_found(request, tarpit=False)

The framed 404 response.

Parameters:

Name	Type	Description	Default
request	Request	the request that has no page.	required
tarpit	bool	delay the answer so tokens and private acronyms cannot be probed; a client over the rate limit answers 429 instead.	False

Returns:

Type	Description
HTMLResponse	the framed 404 page as a response; 429 over the limit.

Source code in rdd/webapp.py

def not_found(self, request: Request, tarpit: bool = False) -> HTMLResponse:
    """The framed 404 response.

    Args:
        request: the request that has no page.
        tarpit: delay the answer so tokens and private acronyms
            cannot be probed; a client over the rate limit
            answers 429 instead.

    Returns:
        the framed 404 page as a response; 429 over the limit.
    """
    path = request.url.path
    lang = lang_of(request)
    page = self.site.not_found(path, lang)
    status = 404
    if tarpit:
        client = request.client.host if request.client else "?"
        if self.rate_limit.miss(client):
            status = 429
        else:
            time.sleep(self.tarpit_seconds)
    response = page_response(page, status=status)
    if status == 429:
        response.headers["Retry-After"] = str(int(self.rate_limit.window_seconds))
    return response

create_app(site)

Create the fastapi application of the given site.

Parameters:

Name	Type	Description	Default
site	ReelSite	the reel site.	required

Returns:

Type	Description
FastAPI	the application.

Source code in rdd/webapp.py

def create_app(site: ReelSite) -> FastAPI:
    """Create the fastapi application of the given site.

    Args:
        site: the reel site.

    Returns:
        the application.
    """
    app = ReelApp(site).app
    return app

create_installation_app(page)

Create the installation mode application.

Parameters:

Name	Type	Description	Default
page	str	the installation mode page.	required

Returns:

Type	Description
FastAPI	the application.

Source code in rdd/webapp.py

def create_installation_app(page: str) -> FastAPI:
    """Create the installation mode application.

    Args:
        page: the installation mode page.

    Returns:
        the application.
    """
    app = InstallationApp(page).app
    return app

lang_of(request)

The language of the given request.

Per the i18n issue the default is the browser setting; an explicit ?lang= wins and is remembered by the cookie the response sets.

Parameters:

Name	Type	Description	Default
request	Request	the request.	required

Returns:

Type	Description
str	the language code.

Source code in rdd/webapp.py

def lang_of(request: Request) -> str:
    """The language of the given request.

    Per the i18n issue the default is the browser setting; an explicit
    ?lang= wins and is remembered by the cookie the response sets.

    Args:
        request: the request.

    Returns:
        the language code.
    """
    lang = pick_language(
        query_lang=request.query_params.get("lang"),
        cookie_lang=request.cookies.get("lang"),
        accept_language=request.headers.get("accept-language"),
    )
    return lang

page_response(page, status=200)

The given page as an html response.

Parameters:

Name	Type	Description	Default
page	str	the html page.	required
status	int	the http status; 200 by default.	200

Returns:

Type	Description
HTMLResponse	the response; no-cache so a browser never shows a stale page.

Source code in rdd/webapp.py

def page_response(page: str, status: int = 200) -> HTMLResponse:
    """The given page as an html response.

    Args:
        page: the html page.
        status: the http status; 200 by default.

    Returns:
        the response; no-cache so a browser never shows a stale page.
    """
    response = HTMLResponse(page, status_code=status)
    response.headers["Cache-Control"] = "no-cache"
    return response

remember_lang(request, response)

Remember an explicit language choice in the cookie.

Parameters:

Name	Type	Description	Default
request	Request	the request whose ?lang= is the choice, if any.	required
response	HTMLResponse	the response to carry the cookie.	required

Returns:

Type	Description
HTMLResponse	the response.

Source code in rdd/webapp.py

def remember_lang(request: Request, response: HTMLResponse) -> HTMLResponse:
    """Remember an explicit language choice in the cookie.

    Args:
        request: the request whose ?lang= is the choice, if any.
        response: the response to carry the cookie.

    Returns:
        the response.
    """
    query_lang = request.query_params.get("lang")
    if query_lang in LANGUAGES:
        response.set_cookie("lang", query_lang)
    return response