How to write a frontend¶

To show a session with something other than Qt, you define the placements a view can ask for, a Frontend that lists them, and a session class that puts the views in place and runs the loop that delivers their calls. How a frontend shows a session on screen explains what a frontend is responsible for.

Prerequisites¶

You need the toolkit your frontend shows views with. The example below prints one line of text for each view to the terminal, so it needs nothing beyond redsun. The blocks below are parts of one script, and the whole script is at the end.

Define the placements¶

A placement is a frozen dataclass subclassing Placement. Define one for each place your toolkit can put a view, plus a base class that the views of that place must be a subclass of:

@dataclass(frozen=True)
class Line(Placement):
    order: int = 0


class ConsoleView:
    def text(self) -> str:
        raise NotImplementedError

Define the frontend¶

Subclass Frontend. requires maps each placement to the class a view asking for it must be, and the session refuses a view that asks for another placement, or is of another class, before building it. thread_of names the thread the slots of a view run on when the slot doesn't say:

class Console(Frontend):
    requires: ClassVar[Mapping[type[Placement], type]] = {Line: ConsoleView}

    @classmethod
    def thread_of(cls, consumer: object) -> SlotThread:
        return "main" if isinstance(consumer, ConsoleView) else None

Most toolkits allow their objects to be used from one thread only, so views run on "main". If thread_of returns None, the slot runs on the thread that sent the signal.

To refuse a view class for another reason, such as a constructor your toolkit can't call, override check_view and raise TypeError.

Define the session¶

Subclass Session, set frontend, and fill present, which runs once every view is built and linked:

class ConsoleSession(Session):
    frontend = Console

    def present(self) -> None:
        views = [v for v in self.views.values() if isinstance(v, ConsoleView)]
        self.lines = sorted(views, key=self.order_of)

    @staticmethod
    def order_of(view: ConsoleView) -> int:
        placement = getattr(view, "placement", None)
        return placement.order if isinstance(placement, Line) else 0

    def run(self, interval: float = 0.5) -> None:
        self.build()
        try:
            while True:
                psygnal.emit_queued()
                print(" | ".join(view.text() for view in self.lines), flush=True)
                time.sleep(interval)
        except KeyboardInterrupt:
            pass
        finally:
            self.shutdown()

run builds the session, then loops until the user stops it. Each turn it calls psygnal.emit_queued(), which delivers the calls held for the main thread, and then shows the views. If your toolkit has an event loop of its own, call psygnal.emit_queued() from a timer of that loop instead.

To pass more arguments to every view's constructor, as QtSession passes parent, override the property view_arguments.
To make toolkit objects before any component exists, such as the application object of the toolkit, override start_runtime and call super().start_runtime() first, which sets the backend that coroutine slots run on.

Write a view for it¶

A view subclasses the base class of its placement and names the placement:

class PositionLine(ConsoleView):
    placement: Placement = Line(order=0)

    def __init__(self, name: str) -> None:
        self.name = name
        self.position = 0.0

    @slot
    def show_reading(self, reading: dict[str, Reading[float]]) -> None:
        for entry in reading.values():
            self.position = entry["value"]

    def text(self) -> str:
        return f"{self.name}: {self.position:.2f}"

class MyApp(ConsoleSession):
    stage: AsDevice[MyStage]
    position: AsView[PositionLine]

    def wire(self) -> Iterator[Link]:
        yield self.stage.position, self.position.show_reading


if __name__ == "__main__":
    MyApp().run()

Let a session file name it¶

Register the session class in the redsun.frontends entry point group of your package:

[project.entry-points."redsun.frontends"]
console = "mylab.console:ConsoleSession"

A session file then names it with frontend: console.

The example in full¶

The whole script

"""The frontend of the guide "How to write a frontend"."""

from __future__ import annotations

import time
from collections.abc import Iterator, Mapping  # noqa: TC003
from dataclasses import dataclass
from typing import ClassVar

import psygnal
from bluesky.protocols import Reading  # noqa: TC002
from ophyd_async.core import StandardReadable, soft_signal_rw

from redsun import AsDevice, AsView, Frontend, Link, Placement, Session, slot
from redsun.ports import SlotThread  # noqa: TC001


@dataclass(frozen=True)
class Line(Placement):
    order: int = 0


class ConsoleView:
    def text(self) -> str:
        raise NotImplementedError


class Console(Frontend):
    requires: ClassVar[Mapping[type[Placement], type]] = {Line: ConsoleView}

    @classmethod
    def thread_of(cls, consumer: object) -> SlotThread:
        return "main" if isinstance(consumer, ConsoleView) else None


class ConsoleSession(Session):
    frontend = Console

    def present(self) -> None:
        views = [v for v in self.views.values() if isinstance(v, ConsoleView)]
        self.lines = sorted(views, key=self.order_of)

    @staticmethod
    def order_of(view: ConsoleView) -> int:
        placement = getattr(view, "placement", None)
        return placement.order if isinstance(placement, Line) else 0

    def run(self, interval: float = 0.5) -> None:
        self.build()
        try:
            while True:
                psygnal.emit_queued()
                print(" | ".join(view.text() for view in self.lines), flush=True)
                time.sleep(interval)
        except KeyboardInterrupt:
            pass
        finally:
            self.shutdown()


class MyStage(StandardReadable):
    def __init__(self, name: str = "") -> None:
        with self.add_children_as_readables():
            self.position = soft_signal_rw(float)
        super().__init__(name=name)


class PositionLine(ConsoleView):
    placement: Placement = Line(order=0)

    def __init__(self, name: str) -> None:
        self.name = name
        self.position = 0.0

    @slot
    def show_reading(self, reading: dict[str, Reading[float]]) -> None:
        for entry in reading.values():
            self.position = entry["value"]

    def text(self) -> str:
        return f"{self.name}: {self.position:.2f}"


class MyApp(ConsoleSession):
    stage: AsDevice[MyStage]
    position: AsView[PositionLine]

    def wire(self) -> Iterator[Link]:
        yield self.stage.position, self.position.show_reading


if __name__ == "__main__":
    MyApp().run()