1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
|
# deck-it - the module contract and registry
# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
"""The module contract and the registry that dispatches to modules.
A module is a data source plus an allow-listed set of actions. The client sends
a module name and an action id; it never sends a command. The registry is the
single place that maps ids to code, so adding a module never touches the server.
"""
from dataclasses import dataclass, field
from typing import Callable
class ModuleError(Exception):
"""A bad module/action id, or a module that cannot serve the request."""
@dataclass
class Module:
name: str
title: str
state: Callable[[], dict]
actions: dict[str, Callable[[dict], dict | None]] = field(default_factory=dict)
start: Callable[[], None] | None = None
stop: Callable[[], None] | None = None
class Registry:
def __init__(self, modules):
self._modules = {m.name: m for m in modules}
def names(self) -> list[dict]:
return [{"name": m.name, "title": m.title} for m in self._modules.values()]
def get(self, name: str) -> Module:
if name not in self._modules:
raise ModuleError(f"unknown module: {name}")
return self._modules[name]
_get = get
def start_all(self) -> None:
for m in self._modules.values():
if m.start:
try:
m.start()
except Exception:
pass
def stop_all(self) -> None:
for m in self._modules.values():
if m.stop:
try:
m.stop()
except Exception:
pass
def state(self, name: str) -> dict:
return self._get(name).state()
def run_action(self, name: str, action: str, body: dict) -> dict | None:
module = self._get(name)
if action not in module.actions:
raise ModuleError(f"unknown action: {name}/{action}")
return module.actions[action](body)
|
