Dock¶

Bases: StrEnum

The axis a dockview branch lays its children along.

Methods:

• flipped –

  The other orientation.

flipped() -> Orientation

42
43
44

def flipped(self) -> "Orientation":
    """The other orientation."""
    return Orientation.VERTICAL if self is Orientation.HORIZONTAL else Orientation.HORIZONTAL

Bases: StrEnum

Where add_panel puts a panel relative to its reference.

Bases: StrEnum

The keys of a dockview layout JSON.

Bases: StrEnum

The two node kinds of a dockview grid.

One tab: its title, the builder that makes its component once the shell exists, and its share of the axis.

Attributes:

• name (str) –

  The tab title, from which the panel id is derived when id is None.

• build (Builder) –

  Makes the panel's component from the shell, the panel id and the title, once the shell exists.

• weight (float) –

  [-]: The panel's share of its parent's axis, relative to its siblings' weights.

• id (str | None) –

  The panel id; None derives it from name.

• panel_id (str) –

  The id the dock knows the panel by: id when given, else name lowercased with dashes.

name: str

build: Builder

weight: float = 1

id: str | None = None

panel_id: str

Panels sharing one tab group, the first shown.

Attributes:

• panels (Sequence[Panel]) –

  The panels sharing the tab group, the first one shown.

• weight (float) –

  [-]: The group's share of its parent's axis, relative to its siblings' weights.

panels: Sequence[Panel]

weight: float = 1

Panels, tab groups or splits sharing one axis by weight; the axis alternates with nesting, horizontal at the root.

Attributes:

• children (Sequence[Node]) –

  The panels, tab groups and splits along the axis, in order.

• weight (float) –

  [-]: The split's share of its parent's axis, relative to its siblings' weights; ignored at the root.

children: Sequence[Node]

weight: float = 1

Where add_panel puts one panel: beside, below or within the group of a reference panel.

Methods:

• add_on –

  The position add-on for add_panel, empty for the first panel.

add_on() -> Layout

149
150
151
152
153
154

def add_on(self) -> Layout:
    """The position add-on for add_panel, empty for the first panel."""
    if self.reference is None:
        return {}

    return {Key.POSITION: {Key.REFERENCE_PANEL: self.reference, Key.DIRECTION: str(self.direction)}}

Bases: StrEnum

The documents the shell itself persists per workbench.

Bases: Protocol

Where a workbench keeps its UI state per user: layouts, dropdowns, brushes.

Methods:

• load –

  The document saved under key in scope, None when there is none.

• save –

  Saves data as the document under key in scope.

• flush –

  Writes every pending save now.

load(scope: Scope, key: str) -> Data | None

36
37
38

def load(self, scope: Scope, key: str) -> Data | None:
    """The document saved under `key` in `scope`, None when there is none."""
    ...

save(scope: Scope, key: str, data: Mapping[str, object]) -> None

40
41
42
43
44
45
46
47
48

def save(self, scope: Scope, key: str, data: Mapping[str, object]) -> None:
    """Saves `data` as the document under `key` in `scope`.

    Args:
        scope: The user and workbench the document belongs to.
        key: The document's name within the scope.
        data: The JSON-ready document.
    """
    ...

flush() -> None

50
51
52

def flush(self) -> None:
    """Writes every pending save now."""
    ...

One JSON file per (user, workbench, key), written at most once per debounce window; a load sees unflushed saves.

Methods:

• path –

  The JSON file of one document.

• load –

  The document: a pending save first, else the file's, None when neither exists.

• save –

  Queues the document and restarts the debounce timer that writes it.

• flush –

  Writes every pending document now.

68
69
70
71
72
73
74

def __init__(self, root: Path | None = None, *, debounce: float = DEBOUNCE_SECONDS) -> None:
    """Stores under `root`, the per-user config directory by default, writing at most once per `debounce` seconds."""
    self.root = root or default_root()
    self.debounce = debounce
    self.pending: dict[tuple[Scope, str], Data] = {}
    self.timer: threading.Timer | None = None
    self.lock = threading.Lock()

path(scope: Scope, key: str) -> Path

76
77
78
79
80

def path(self, scope: Scope, key: str) -> Path:
    """The JSON file of one document."""
    user, workbench = scope

    return self.root / user / workbench / f"{key}.json"

load(scope: Scope, key: str) -> Data | None

82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98

def load(self, scope: Scope, key: str) -> Data | None:
    """The document: a pending save first, else the file's, None when neither exists."""
    with self.lock:
        pending = self.pending.get((scope, key))

    if pending is not None:
        return dict(pending)

    path = self.path(scope, key)
    if not path.exists():
        return None

    loaded = cast("object", json.loads(path.read_text(encoding="utf-8")))
    if not isinstance(loaded, dict):
        return None

    return cast("Data", loaded)

save(scope: Scope, key: str, data: Mapping[str, object]) -> None

100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115

def save(self, scope: Scope, key: str, data: Mapping[str, object]) -> None:
    """Queues the document and restarts the debounce timer that writes it.

    Args:
        scope: The user and workbench the document belongs to.
        key: The document's name within the scope.
        data: The JSON-ready document.
    """
    with self.lock:
        self.pending[(scope, key)] = dict(data)
        if self.timer is not None:
            self.timer.cancel()

        self.timer = threading.Timer(self.debounce, self.flush)
        self.timer.daemon = True
        self.timer.start()

flush() -> None

117
118
119
120
121
122
123
124
125
126
127
128
129
130

def flush(self) -> None:
    """Writes every pending document now."""
    with self.lock:
        if self.timer is not None:
            self.timer.cancel()
            self.timer = None

        pending = self.pending
        self.pending = {}

    for (scope, key), data in pending.items():
        path = self.path(scope, key)
        path.parent.mkdir(parents=True, exist_ok=True)
        write_atomic(path, json.dumps(data, indent=1))

Runs poll in a worker thread every period seconds and hands what it returns to publish inside a state context.

Parameters:

• panel (PanelComponent) –

  The panel whose state the publish runs in.

• period (float) –

  Seconds between polls.

• poll (Callable[[], T]) –

  Reads what changed, off the server loop.

• publish (Callable[[T], None]) –

  Writes a poll's result into the panel, on the server loop.

Methods:

• start –

  Starts the loop unless it already runs; the first poll happens immediately.

• stop –

  Cancels the loop, so a hidden panel costs nothing.

• cycle –

  Poll, publish, sleep — until the task is cancelled; a failed poll is logged and the loop carries on.

Attributes:

• polls –

  How many polls have completed; a panel shows it as its refresh count and the tests wait on it.

• running (bool) –

  Whether the loop runs.

19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

def __init__(self, panel: "PanelComponent", period: float, poll: Callable[[], T], publish: Callable[[T], None]) -> None:
    """Sets the loop up stopped; `start` runs it.

    Args:
        panel: The panel whose state the publish runs in.
        period: Seconds between polls.
        poll: Reads what changed, off the server loop.
        publish: Writes a poll's result into the panel, on the server loop.
    """
    self.panel = panel
    self.period = period
    self.poll = poll
    self.publish = publish
    self.task: asyncio.Task[None] | None = None
    self.polls = 0
    """How many polls have completed; a panel shows it as its refresh count and the tests wait on it."""

polls = 0

running: bool

start() -> None

41
42
43
44
45
46
47
48

def start(self) -> None:
    """Starts the loop unless it already runs; the first poll happens immediately."""
    if self.running:
        return

    self.task = create_task(self.cycle())

    log.bind(panel=self.panel.id, period=self.period).info("Feed started.")

stop() -> None

50
51
52
53
54
55
56
57
58

def stop(self) -> None:
    """Cancels the loop, so a hidden panel costs nothing."""
    if self.task is None:
        return

    _ = self.task.cancel()
    self.task = None

    log.bind(panel=self.panel.id, polls=self.polls).info("Feed stopped.")

cycle() -> None

60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75

async def cycle(self) -> None:
    """Poll, publish, sleep — until the task is cancelled; a failed poll is logged and the loop carries on."""
    loop = asyncio.get_running_loop()
    while True:
        try:
            value: T = await loop.run_in_executor(None, self.poll)
        except Exception:
            log.opt(exception=True).warning("Feed poll failed.")
            await asyncio.sleep(self.period)
            continue

        self.polls += 1
        with self.panel.state:
            self.publish(value)

        await asyncio.sleep(self.period)

Drains an asyncio queue another party feeds on the server loop, handing each burst of items to publish inside a state context.

Parameters:

• panel (PanelComponent) –

  The panel whose state the publish runs in.

• subscribe (Callable[[AbstractEventLoop], Queue[T]]) –

  Hands out a queue fed on the given loop.

• unsubscribe (Callable[[Queue[T]], None]) –

  Stops feeding a queue subscribe handed out.

• publish (Callable[[list[T]], None]) –

  Writes a burst of items into the panel, on the server loop.

Methods:

• start –

  Subscribes and starts draining unless it already runs.

• stop –

  Cancels the drain and unsubscribes, so a hidden panel is no longer fed.

• cycle –

  Wait for an item, take every other one already queued, publish them together — until the task is cancelled.

Attributes:

• bursts –

  How many bursts have been published; the tests wait on it.

• running (bool) –

  Whether the drain runs.

 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103

def __init__(
    self,
    panel: "PanelComponent",
    subscribe: Callable[[asyncio.AbstractEventLoop], asyncio.Queue[T]],
    unsubscribe: Callable[[asyncio.Queue[T]], None],
    publish: Callable[[list[T]], None],
) -> None:
    """Sets the drain up stopped; `start` subscribes and runs it.

    Args:
        panel: The panel whose state the publish runs in.
        subscribe: Hands out a queue fed on the given loop.
        unsubscribe: Stops feeding a queue `subscribe` handed out.
        publish: Writes a burst of items into the panel, on the server loop.
    """
    self.panel = panel
    self.subscribe = subscribe
    self.unsubscribe = unsubscribe
    self.publish = publish
    self.task: asyncio.Task[None] | None = None
    self.queue: asyncio.Queue[T] | None = None
    self.bursts = 0
    """How many bursts have been published; the tests wait on it."""

bursts = 0

running: bool

start() -> None

110
111
112
113
114
115
116
117

def start(self) -> None:
    """Subscribes and starts draining unless it already runs."""
    if self.running:
        return

    self.task = create_task(self.cycle())

    log.bind(panel=self.panel.id).info("Queue feed started.")

stop() -> None

119
120
121
122
123
124
125
126
127
128
129
130

def stop(self) -> None:
    """Cancels the drain and unsubscribes, so a hidden panel is no longer fed."""
    if self.task is None:
        return

    _ = self.task.cancel()
    self.task = None
    if self.queue is not None:
        self.unsubscribe(self.queue)
        self.queue = None

    log.bind(panel=self.panel.id, bursts=self.bursts).info("Queue feed stopped.")

cycle() -> None

132
133
134
135
136
137
138
139
140
141
142
143

async def cycle(self) -> None:
    """Wait for an item, take every other one already queued, publish them together — until the task is cancelled."""
    queue = self.subscribe(asyncio.get_running_loop())
    self.queue = queue
    while True:
        batch: list[T] = [await queue.get()]
        while not queue.empty():
            batch.append(queue.get_nowait())

        self.bursts += 1
        with self.panel.state:
            self.publish(batch)