modlink_core

View Source

 1from .bus import FrameStream, FrameStreamOverflowError, StreamBus
 2from .event_stream import (
 3    BackendEventBroker,
 4    EventStream,
 5    EventStreamOverflowError,
 6    StreamClosedError,
 7)
 8from .events import (
 9    DriverConnectionLostEvent,
10    DriverExecutorFailedEvent,
11    RecordingFailedEvent,
12    SettingChangedEvent,
13)
14from .logging_setup import configure_host_logging
15from .models import (
16    DriverSnapshot,
17    ExportJobSnapshot,
18    RecordingSnapshot,
19    RecordingStartSummary,
20    RecordingStopSummary,
21    ReplayMarker,
22    ReplayRecordingSummary,
23    ReplaySegment,
24    ReplaySnapshot,
25)
26from .recording import RecordingBackend
27from .replay import ReplayBackend
28from .runtime import ModLinkEngine
29from .settings import (
30    SettingsStore,
31)
32
33__all__ = [
34    "RecordingBackend",
35    "RecordingSnapshot",
36    "RecordingStartSummary",
37    "RecordingStopSummary",
38    "ReplayBackend",
39    "ReplayMarker",
40    "ReplayRecordingSummary",
41    "ReplaySegment",
42    "ReplaySnapshot",
43    "ExportJobSnapshot",
44    "BackendEventBroker",
45    "DriverConnectionLostEvent",
46    "DriverExecutorFailedEvent",
47    "EventStream",
48    "EventStreamOverflowError",
49    "FrameStream",
50    "FrameStreamOverflowError",
51    "configure_host_logging",
52    "DriverSnapshot",
53    "ModLinkEngine",
54    "RecordingFailedEvent",
55    "SettingChangedEvent",
56    "SettingsStore",
57    "StreamClosedError",
58    "StreamBus",
59]

@dataclass(frozen=True, slots=True)

class RecordingSnapshot: View Source

17@dataclass(frozen=True, slots=True)
18class RecordingSnapshot:
19    state: str
20    is_started: bool
21    is_recording: bool
22    root_dir: str

RecordingSnapshot(state: str, is_started: bool, is_recording: bool, root_dir: str)

state: str

is_started: bool

is_recording: bool

root_dir: str

@dataclass(frozen=True, slots=True)

class RecordingStartSummary: View Source

25@dataclass(frozen=True, slots=True)
26class RecordingStartSummary:
27    recording_id: str
28    recording_path: str
29    started_at_ns: int

RecordingStartSummary(recording_id: str, recording_path: str, started_at_ns: int)

recording_id: str

recording_path: str

started_at_ns: int

@dataclass(frozen=True, slots=True)

class RecordingStopSummary: View Source

32@dataclass(frozen=True, slots=True)
33class RecordingStopSummary:
34    recording_id: str
35    recording_path: str
36    started_at_ns: int
37    stopped_at_ns: int
38    status: str
39    frame_counts_by_stream: dict[str, int]

RecordingStopSummary( recording_id: str, recording_path: str, started_at_ns: int, stopped_at_ns: int, status: str, frame_counts_by_stream: dict[str, int])

recording_id: str

recording_path: str

started_at_ns: int

stopped_at_ns: int

status: str

frame_counts_by_stream: dict[str, int]

@dataclass(frozen=True, slots=True)

class ReplayMarker: View Source

67@dataclass(frozen=True, slots=True)
68class ReplayMarker:
69    timestamp_ns: int
70    label: str | None

ReplayMarker(timestamp_ns: int, label: str | None)

timestamp_ns: int

label: str | None

@dataclass(frozen=True, slots=True)

class ReplayRecordingSummary: View Source

53@dataclass(frozen=True, slots=True)
54class ReplayRecordingSummary:
55    recording_id: str
56    recording_label: str | None
57    recording_path: str
58    stream_ids: tuple[str, ...]
59    session_name: str | None = None
60    experiment_name: str | None = None
61    started_at_ns: int | None = None
62    duration_ns: int | None = None
63    status: str | None = None
64    total_frames: int | None = None

recording_id: str

recording_label: str | None

recording_path: str

stream_ids: tuple[str, ...]

session_name: str | None

experiment_name: str | None

started_at_ns: int | None

duration_ns: int | None

status: str | None

total_frames: int | None

@dataclass(frozen=True, slots=True)

class ReplaySegment: View Source

73@dataclass(frozen=True, slots=True)
74class ReplaySegment:
75    start_ns: int
76    end_ns: int
77    label: str | None

ReplaySegment(start_ns: int, end_ns: int, label: str | None)

start_ns: int

end_ns: int

label: str | None

@dataclass(frozen=True, slots=True)

class ReplaySnapshot: View Source

42@dataclass(frozen=True, slots=True)
43class ReplaySnapshot:
44    state: str
45    is_started: bool
46    recording_id: str | None
47    recording_path: str | None
48    position_ns: int
49    duration_ns: int
50    speed_multiplier: float

ReplaySnapshot( state: str, is_started: bool, recording_id: str | None, recording_path: str | None, position_ns: int, duration_ns: int, speed_multiplier: float)

state: str

is_started: bool

recording_id: str | None

recording_path: str | None

position_ns: int

duration_ns: int

speed_multiplier: float

@dataclass(frozen=True, slots=True)

class ExportJobSnapshot: View Source

80@dataclass(frozen=True, slots=True)
81class ExportJobSnapshot:
82    job_id: str
83    recording_id: str
84    state: str
85    progress: float
86    output_path: str | None
87    error: str | None
88    request_summary: str = ""

ExportJobSnapshot( job_id: str, recording_id: str, state: str, progress: float, output_path: str | None, error: str | None, request_summary: str = '')

job_id: str

recording_id: str

state: str

progress: float

output_path: str | None

error: str | None

request_summary: str

class BackendEventBroker: View Source

119class BackendEventBroker:
120    def __init__(self) -> None:
121        self._streams: list[EventStream] = []
122        self._lock = RLock()
123
124    def open_stream(self, *, maxsize: int = 1024) -> EventStream:
125        stream = EventStream(self, maxsize=maxsize)
126        with self._lock:
127            self._streams.append(stream)
128        return stream
129
130    def publish(self, event: BackendEvent) -> None:
131        with self._lock:
132            streams = tuple(self._streams)
133        for stream in streams:
134            stream._publish(event)
135
136    def _close_stream(self, stream: EventStream) -> None:
137        with self._lock:
138            try:
139                self._streams.remove(stream)
140            except ValueError:
141                pass
142        stream._push_closed()

def open_stream(self, *, maxsize: int = 1024) -> EventStream: View Source

124    def open_stream(self, *, maxsize: int = 1024) -> EventStream:
125        stream = EventStream(self, maxsize=maxsize)
126        with self._lock:
127            self._streams.append(stream)
128        return stream

def publish(self, event: BackendEvent) -> None: View Source

130    def publish(self, event: BackendEvent) -> None:
131        with self._lock:
132            streams = tuple(self._streams)
133        for stream in streams:
134            stream._publish(event)

@dataclass(frozen=True, slots=True)

class DriverConnectionLostEvent: View Source

 8@dataclass(frozen=True, slots=True)
 9class DriverConnectionLostEvent:
10    driver_id: str
11    detail: object | None = None
12    kind: Literal["driver_connection_lost"] = "driver_connection_lost"

DriverConnectionLostEvent( driver_id: str, detail: object | None = None, kind: Literal['driver_connection_lost'] = 'driver_connection_lost')

driver_id: str

detail: object | None

kind: Literal['driver_connection_lost']

@dataclass(frozen=True, slots=True)

class DriverExecutorFailedEvent: View Source

15@dataclass(frozen=True, slots=True)
16class DriverExecutorFailedEvent:
17    driver_id: str
18    detail: object
19    kind: Literal["driver_executor_failed"] = "driver_executor_failed"

DriverExecutorFailedEvent( driver_id: str, detail: object, kind: Literal['driver_executor_failed'] = 'driver_executor_failed')

driver_id: str

detail: object

kind: Literal['driver_executor_failed']

class EventStream: View Source

 22class EventStream:
 23    def __init__(
 24        self,
 25        broker: BackendEventBroker,
 26        *,
 27        maxsize: int = 1024,
 28    ) -> None:
 29        normalized_maxsize = max(1, int(maxsize))
 30        self._broker = broker
 31        self._queue: queue.Queue[BackendEvent | object] = queue.Queue(maxsize=normalized_maxsize)
 32        self._lock = RLock()
 33        self._closed = False
 34
 35    @property
 36    def closed(self) -> bool:
 37        return self._closed
 38
 39    def read(
 40        self,
 41        *,
 42        block: bool = True,
 43        timeout: float | None = None,
 44    ) -> BackendEvent:
 45        item = self._queue.get(block=block, timeout=timeout)
 46        if item is _STREAM_CLOSED:
 47            raise StreamClosedError("event stream is closed")
 48        if item is _STREAM_OVERFLOW:
 49            raise EventStreamOverflowError("event stream overflowed")
 50        return item  # type: ignore[return-value]
 51
 52    def read_many(self, *, max_items: int | None = None) -> list[BackendEvent]:
 53        if max_items is not None and max_items <= 0:
 54            return []
 55
 56        items: list[BackendEvent] = []
 57        while max_items is None or len(items) < max_items:
 58            try:
 59                item = self._queue.get_nowait()
 60            except queue.Empty:
 61                break
 62            if item is _STREAM_CLOSED:
 63                if items:
 64                    return items
 65                raise StreamClosedError("event stream is closed")
 66            if item is _STREAM_OVERFLOW:
 67                if items:
 68                    return items
 69                raise EventStreamOverflowError("event stream overflowed")
 70            items.append(item)  # type: ignore[arg-type]
 71        return items
 72
 73    def close(self) -> None:
 74        with self._lock:
 75            if self._closed:
 76                return
 77            self._closed = True
 78        self._broker._close_stream(self)
 79
 80    def _publish(self, event: BackendEvent) -> None:
 81        with self._lock:
 82            if self._closed:
 83                return
 84            try:
 85                self._queue.put_nowait(event)
 86                return
 87            except queue.Full:
 88                pass
 89
 90            try:
 91                self._queue.get_nowait()
 92            except queue.Empty:
 93                pass
 94
 95            try:
 96                self._queue.put_nowait(_STREAM_OVERFLOW)
 97            except queue.Full:
 98                pass
 99
100    def _push_closed(self) -> None:
101        with self._lock:
102            try:
103                self._queue.put_nowait(_STREAM_CLOSED)
104                return
105            except queue.Full:
106                pass
107
108            try:
109                self._queue.get_nowait()
110            except queue.Empty:
111                pass
112
113            try:
114                self._queue.put_nowait(_STREAM_CLOSED)
115            except queue.Full:
116                pass

EventStream( broker: BackendEventBroker, *, maxsize: int = 1024) View Source

23    def __init__(
24        self,
25        broker: BackendEventBroker,
26        *,
27        maxsize: int = 1024,
28    ) -> None:
29        normalized_maxsize = max(1, int(maxsize))
30        self._broker = broker
31        self._queue: queue.Queue[BackendEvent | object] = queue.Queue(maxsize=normalized_maxsize)
32        self._lock = RLock()
33        self._closed = False

closed: bool View Source

35    @property
36    def closed(self) -> bool:
37        return self._closed

def read( self, *, block: bool = True, timeout: float | None = None) -> BackendEvent: View Source

39    def read(
40        self,
41        *,
42        block: bool = True,
43        timeout: float | None = None,
44    ) -> BackendEvent:
45        item = self._queue.get(block=block, timeout=timeout)
46        if item is _STREAM_CLOSED:
47            raise StreamClosedError("event stream is closed")
48        if item is _STREAM_OVERFLOW:
49            raise EventStreamOverflowError("event stream overflowed")
50        return item  # type: ignore[return-value]

def read_many(self, *, max_items: int | None = None) -> list[BackendEvent]: View Source

52    def read_many(self, *, max_items: int | None = None) -> list[BackendEvent]:
53        if max_items is not None and max_items <= 0:
54            return []
55
56        items: list[BackendEvent] = []
57        while max_items is None or len(items) < max_items:
58            try:
59                item = self._queue.get_nowait()
60            except queue.Empty:
61                break
62            if item is _STREAM_CLOSED:
63                if items:
64                    return items
65                raise StreamClosedError("event stream is closed")
66            if item is _STREAM_OVERFLOW:
67                if items:
68                    return items
69                raise EventStreamOverflowError("event stream overflowed")
70            items.append(item)  # type: ignore[arg-type]
71        return items

def close(self) -> None: View Source

73    def close(self) -> None:
74        with self._lock:
75            if self._closed:
76                return
77            self._closed = True
78        self._broker._close_stream(self)

class EventStreamOverflowError(builtins.Exception): View Source

14class EventStreamOverflowError(Exception):
15    """Raised when the low-frequency event stream overflows."""

Raised when the low-frequency event stream overflows.

class FrameStream: View Source

 26class FrameStream:
 27    def __init__(
 28        self,
 29        bus: StreamBus,
 30        *,
 31        maxsize: int = 256,
 32        drop_policy: FrameDropPolicy = "drop_oldest",
 33        consumer_name: str,
 34    ) -> None:
 35        if drop_policy not in {"drop_oldest", "error"}:
 36            raise ValueError(f"unsupported frame drop policy: {drop_policy}")
 37        normalized_maxsize = max(1, int(maxsize))
 38        self._bus = bus
 39        self._queue: queue.Queue[FrameEnvelope | object] = queue.Queue(maxsize=normalized_maxsize)
 40        self._drop_policy = drop_policy
 41        self._consumer_name = consumer_name
 42        self._dropped_count = 0
 43        self._closed = False
 44        self._overflowed = False
 45        self._lock = RLock()
 46
 47    @property
 48    def consumer_name(self) -> str:
 49        return self._consumer_name
 50
 51    @property
 52    def dropped_count(self) -> int:
 53        return self._dropped_count
 54
 55    @property
 56    def overflowed(self) -> bool:
 57        return self._overflowed
 58
 59    @property
 60    def closed(self) -> bool:
 61        return self._closed
 62
 63    def read(
 64        self,
 65        *,
 66        block: bool = True,
 67        timeout: float | None = None,
 68    ) -> FrameEnvelope:
 69        item = self._queue.get(block=block, timeout=timeout)
 70        if item is _FRAME_STREAM_CLOSED:
 71            raise StreamClosedError("frame stream is closed")
 72        if item is _FRAME_STREAM_OVERFLOW:
 73            raise FrameStreamOverflowError(self._consumer_name)
 74        return item  # type: ignore[return-value]
 75
 76    def read_many(self, *, max_items: int | None = None) -> list[FrameEnvelope]:
 77        if max_items is not None and max_items <= 0:
 78            return []
 79
 80        items: list[FrameEnvelope] = []
 81        while max_items is None or len(items) < max_items:
 82            try:
 83                item = self._queue.get_nowait()
 84            except queue.Empty:
 85                break
 86            if item is _FRAME_STREAM_CLOSED:
 87                if items:
 88                    return items
 89                raise StreamClosedError("frame stream is closed")
 90            if item is _FRAME_STREAM_OVERFLOW:
 91                if items:
 92                    return items
 93                raise FrameStreamOverflowError(self._consumer_name)
 94            items.append(item)  # type: ignore[arg-type]
 95        return items
 96
 97    def close(self) -> None:
 98        with self._lock:
 99            if self._closed:
100                return
101            self._closed = True
102        self._bus._close_frame_stream(self)
103
104    def _publish(self, frame: FrameEnvelope) -> None:
105        with self._lock:
106            if self._closed:
107                return
108            if self._overflowed and self._drop_policy == "error":
109                self._dropped_count += 1
110                return
111            try:
112                self._queue.put_nowait(frame)
113                return
114            except queue.Full:
115                self._dropped_count += 1
116
117                if self._drop_policy == "drop_oldest":
118                    self._drop_oldest_and_enqueue(frame)
119                    return
120
121                self._overflowed = True
122                self._clear_queue_locked()
123                self._push_control_locked(_FRAME_STREAM_OVERFLOW)
124
125    def _drop_oldest_and_enqueue(self, frame: FrameEnvelope) -> None:
126        try:
127            self._queue.get_nowait()
128        except queue.Empty:
129            pass
130        try:
131            self._queue.put_nowait(frame)
132        except queue.Full:
133            pass
134
135    def _clear_queue_locked(self) -> None:
136        while True:
137            try:
138                self._queue.get_nowait()
139            except queue.Empty:
140                break
141
142    def _push_closed(self) -> None:
143        with self._lock:
144            self._push_control_locked(_FRAME_STREAM_CLOSED)
145
146    def _push_control_locked(self, item: object) -> None:
147        try:
148            self._queue.put_nowait(item)
149            return
150        except queue.Full:
151            pass
152
153        try:
154            self._queue.get_nowait()
155        except queue.Empty:
156            pass
157
158        try:
159            self._queue.put_nowait(item)
160        except queue.Full:
161            pass

FrameStream( bus: StreamBus, *, maxsize: int = 256, drop_policy: Literal['drop_oldest', 'error'] = 'drop_oldest', consumer_name: str) View Source

27    def __init__(
28        self,
29        bus: StreamBus,
30        *,
31        maxsize: int = 256,
32        drop_policy: FrameDropPolicy = "drop_oldest",
33        consumer_name: str,
34    ) -> None:
35        if drop_policy not in {"drop_oldest", "error"}:
36            raise ValueError(f"unsupported frame drop policy: {drop_policy}")
37        normalized_maxsize = max(1, int(maxsize))
38        self._bus = bus
39        self._queue: queue.Queue[FrameEnvelope | object] = queue.Queue(maxsize=normalized_maxsize)
40        self._drop_policy = drop_policy
41        self._consumer_name = consumer_name
42        self._dropped_count = 0
43        self._closed = False
44        self._overflowed = False
45        self._lock = RLock()

consumer_name: str View Source

47    @property
48    def consumer_name(self) -> str:
49        return self._consumer_name

dropped_count: int View Source

51    @property
52    def dropped_count(self) -> int:
53        return self._dropped_count

overflowed: bool View Source

55    @property
56    def overflowed(self) -> bool:
57        return self._overflowed

closed: bool View Source

59    @property
60    def closed(self) -> bool:
61        return self._closed

def read( self, *, block: bool = True, timeout: float | None = None) -> modlink_sdk.FrameEnvelope: View Source

63    def read(
64        self,
65        *,
66        block: bool = True,
67        timeout: float | None = None,
68    ) -> FrameEnvelope:
69        item = self._queue.get(block=block, timeout=timeout)
70        if item is _FRAME_STREAM_CLOSED:
71            raise StreamClosedError("frame stream is closed")
72        if item is _FRAME_STREAM_OVERFLOW:
73            raise FrameStreamOverflowError(self._consumer_name)
74        return item  # type: ignore[return-value]

def read_many( self, *, max_items: int | None = None) -> list[modlink_sdk.FrameEnvelope]: View Source

76    def read_many(self, *, max_items: int | None = None) -> list[FrameEnvelope]:
77        if max_items is not None and max_items <= 0:
78            return []
79
80        items: list[FrameEnvelope] = []
81        while max_items is None or len(items) < max_items:
82            try:
83                item = self._queue.get_nowait()
84            except queue.Empty:
85                break
86            if item is _FRAME_STREAM_CLOSED:
87                if items:
88                    return items
89                raise StreamClosedError("frame stream is closed")
90            if item is _FRAME_STREAM_OVERFLOW:
91                if items:
92                    return items
93                raise FrameStreamOverflowError(self._consumer_name)
94            items.append(item)  # type: ignore[arg-type]
95        return items

def close(self) -> None: View Source

 97    def close(self) -> None:
 98        with self._lock:
 99            if self._closed:
100                return
101            self._closed = True
102        self._bus._close_frame_stream(self)

class FrameStreamOverflowError(builtins.Exception): View Source

20class FrameStreamOverflowError(Exception):
21    def __init__(self, consumer_name: str) -> None:
22        self.consumer_name = consumer_name
23        super().__init__(f"frame stream overflowed: consumer_name={consumer_name}")

Common base class for all non-exit exceptions.

FrameStreamOverflowError(consumer_name: str) View Source

21    def __init__(self, consumer_name: str) -> None:
22        self.consumer_name = consumer_name
23        super().__init__(f"frame stream overflowed: consumer_name={consumer_name}")

consumer_name

def configure_host_logging( *, log_path: str | pathlib._local.Path | None = None, app_name: str = 'ModLink Studio', app_author: str = 'ModLink', log_filename: str = 'modlink.log', console: bool = True, debug: bool = False, max_bytes: int = 5242880, backup_count: int = 5) -> pathlib._local.Path: View Source

26def configure_host_logging(
27    *,
28    log_path: str | Path | None = None,
29    app_name: str = "ModLink Studio",
30    app_author: str = "ModLink",
31    log_filename: str = "modlink.log",
32    console: bool = True,
33    debug: bool = False,
34    max_bytes: int = DEFAULT_MAX_LOG_BYTES,
35    backup_count: int = DEFAULT_BACKUP_COUNT,
36) -> Path:
37    resolved_path = (
38        Path(log_path)
39        if log_path is not None
40        else Path(user_log_path(app_name, app_author)) / log_filename
41    )
42    resolved_path.parent.mkdir(parents=True, exist_ok=True)
43
44    root_logger = logging.getLogger()
45    _remove_managed_handlers(root_logger)
46    root_logger.setLevel(logging.WARNING)
47    log_level = logging.DEBUG if debug else logging.INFO
48
49    formatter = logging.Formatter(DEFAULT_LOG_FORMAT)
50    file_handler = RotatingFileHandler(
51        resolved_path,
52        maxBytes=max(1, int(max_bytes)),
53        backupCount=max(1, int(backup_count)),
54        encoding="utf-8",
55    )
56    file_handler.setLevel(log_level)
57    file_handler.setFormatter(formatter)
58    _mark_managed(file_handler)
59    root_logger.addHandler(file_handler)
60
61    if console:
62        console_handler = logging.StreamHandler(sys.stderr)
63        console_handler.setLevel(log_level)
64        console_handler.setFormatter(formatter)
65        _mark_managed(console_handler)
66        root_logger.addHandler(console_handler)
67
68    for logger_name in _MANAGED_LOGGER_NAMES:
69        logger = logging.getLogger(logger_name)
70        logger.setLevel(log_level)
71        logger.propagate = True
72
73    logging.captureWarnings(True)
74    logging.getLogger(__name__).info(
75        "Configured logging at %s (level=%s)",
76        resolved_path,
77        logging.getLevelName(log_level),
78    )
79    return resolved_path

@dataclass(frozen=True, slots=True)

class DriverSnapshot: View Source

 7@dataclass(frozen=True, slots=True)
 8class DriverSnapshot:
 9    driver_id: str
10    display_name: str
11    supported_providers: tuple[str, ...]
12    is_running: bool
13    is_connected: bool
14    is_streaming: bool

DriverSnapshot( driver_id: str, display_name: str, supported_providers: tuple[str, ...], is_running: bool, is_connected: bool, is_streaming: bool)

driver_id: str

display_name: str

supported_providers: tuple[str, ...]

is_running: bool

is_connected: bool

is_streaming: bool

@dataclass(frozen=True, slots=True)

class RecordingFailedEvent: View Source

22@dataclass(frozen=True, slots=True)
23class RecordingFailedEvent:
24    recording_id: str
25    recording_path: str
26    frame_counts_by_stream: dict[str, int]
27    reason: str
28    ts_ns: int
29    kind: Literal["recording_failed"] = "recording_failed"

RecordingFailedEvent( recording_id: str, recording_path: str, frame_counts_by_stream: dict[str, int], reason: str, ts_ns: int, kind: Literal['recording_failed'] = 'recording_failed')

recording_id: str

recording_path: str

frame_counts_by_stream: dict[str, int]

reason: str

ts_ns: int

kind: Literal['recording_failed']

@dataclass(frozen=True, slots=True)

class SettingChangedEvent: View Source

32@dataclass(frozen=True, slots=True)
33class SettingChangedEvent:
34    key: str
35    value: Any
36    ts: float
37    kind: Literal["setting_changed"] = "setting_changed"

SettingChangedEvent( key: str, value: Any, ts: float, kind: Literal['setting_changed'] = 'setting_changed')

key: str

value: Any

ts: float

kind: Literal['setting_changed']

class SettingsStore: View Source

230class SettingsStore:
231    ROOT_RESERVED = frozenset({"add", "load", "path", "save", "snapshot"})
232
233    def __init__(
234        self,
235        *,
236        path: str | Path | None = None,
237        version: int = 1,
238        on_change: Any = None,
239    ) -> None:
240        object.__setattr__(self, "_path", None if path is None else Path(path))
241        object.__setattr__(self, "_version", version)
242        object.__setattr__(self, "_on_change", on_change)
243        object.__setattr__(self, "_lock", RLock())
244        object.__setattr__(self, "_root", SettingsNode(name="", parent=None, store=self))
245
246    @property
247    def path(self) -> Path | None:
248        return self._path
249
250    def add(self, **specs: SettingsSpec) -> SettingsStore:
251        with self._lock:
252            overlap = self.ROOT_RESERVED.intersection(specs)
253            if overlap:
254                reserved_name = next(iter(sorted(overlap)))
255                raise ValueError(f"reserved key: {reserved_name!r}")
256            self._root.add(**specs)
257        return self
258
259    def __getattr__(self, name: str) -> SettingsNode | SettingItem:
260        if name.startswith("_"):
261            raise AttributeError(name)
262        with self._lock:
263            return self._root.get_child(name)
264
265    def __setattr__(self, name: str, value: object) -> None:
266        if name.startswith("_"):
267            object.__setattr__(self, name, value)
268            return
269        with self._lock:
270            setattr(self._root, name, value)
271
272    def snapshot(self) -> dict[str, object]:
273        with self._lock:
274            return self._root.dump_dict()
275
276    def save(self, path: str | Path | None = None) -> None:
277        with self._lock:
278            resolved_path = self._resolve_path(path)
279            resolved_path.parent.mkdir(parents=True, exist_ok=True)
280            payload = {
281                "_version": self._version,
282                "values": self._root.dump_dict(),
283            }
284            _atomic_write_text(
285                resolved_path,
286                json.dumps(payload, indent=2, ensure_ascii=False),
287            )
288
289    def load(
290        self,
291        path: str | Path | None = None,
292        *,
293        ignore_unknown: bool = False,
294    ) -> None:
295        with self._lock:
296            resolved_path = self._resolve_path(path)
297            payload = json.loads(resolved_path.read_text(encoding="utf-8"))
298            if not isinstance(payload, dict):
299                raise TypeError("settings file must contain a JSON object")
300
301            file_version = payload.get("_version")
302            if file_version is not None and file_version != self._version:
303                raise ValueError(
304                    f"settings version mismatch: file={file_version}, expected={self._version}"
305                )
306
307            values = payload.get("values", {})
308            previous_snapshot = self.snapshot()
309            previous = self._root.collect_leaf_values()
310            self._root.reset(notify=False)
311            try:
312                self._root.apply_dict(values, notify=False, ignore_unknown=ignore_unknown)
313            except Exception:
314                self._root.reset(notify=False)
315                self._root.apply_dict(previous_snapshot, notify=False)
316                raise
317            current = self._root.collect_leaf_values()
318            for key, value in current.items():
319                if previous.get(key) != value:
320                    self._emit_change(key, value)
321
322    def _resolve_path(self, path: str | Path | None) -> Path:
323        resolved_path = self._path if path is None else Path(path)
324        if resolved_path is None:
325            raise ValueError("path is required")
326        return resolved_path
327
328    def _emit_change(self, key: str, value: object) -> None:
329        on_change = self._on_change
330        if on_change is None:
331            return
332        on_change(key, value)

SettingsStore( *, path: str | pathlib._local.Path | None = None, version: int = 1, on_change: Any = None) View Source

233    def __init__(
234        self,
235        *,
236        path: str | Path | None = None,
237        version: int = 1,
238        on_change: Any = None,
239    ) -> None:
240        object.__setattr__(self, "_path", None if path is None else Path(path))
241        object.__setattr__(self, "_version", version)
242        object.__setattr__(self, "_on_change", on_change)
243        object.__setattr__(self, "_lock", RLock())
244        object.__setattr__(self, "_root", SettingsNode(name="", parent=None, store=self))

ROOT_RESERVED = frozenset({'load', 'add', 'save', 'snapshot', 'path'})

path: pathlib._local.Path | None View Source

246    @property
247    def path(self) -> Path | None:
248        return self._path

def add( self, **specs: modlink_core.settings.spec.SettingsSpec) -> SettingsStore: View Source

250    def add(self, **specs: SettingsSpec) -> SettingsStore:
251        with self._lock:
252            overlap = self.ROOT_RESERVED.intersection(specs)
253            if overlap:
254                reserved_name = next(iter(sorted(overlap)))
255                raise ValueError(f"reserved key: {reserved_name!r}")
256            self._root.add(**specs)
257        return self

def snapshot(self) -> dict[str, object]: View Source

272    def snapshot(self) -> dict[str, object]:
273        with self._lock:
274            return self._root.dump_dict()

def save(self, path: str | pathlib._local.Path | None = None) -> None: View Source

276    def save(self, path: str | Path | None = None) -> None:
277        with self._lock:
278            resolved_path = self._resolve_path(path)
279            resolved_path.parent.mkdir(parents=True, exist_ok=True)
280            payload = {
281                "_version": self._version,
282                "values": self._root.dump_dict(),
283            }
284            _atomic_write_text(
285                resolved_path,
286                json.dumps(payload, indent=2, ensure_ascii=False),
287            )

def load( self, path: str | pathlib._local.Path | None = None, *, ignore_unknown: bool = False) -> None: View Source

289    def load(
290        self,
291        path: str | Path | None = None,
292        *,
293        ignore_unknown: bool = False,
294    ) -> None:
295        with self._lock:
296            resolved_path = self._resolve_path(path)
297            payload = json.loads(resolved_path.read_text(encoding="utf-8"))
298            if not isinstance(payload, dict):
299                raise TypeError("settings file must contain a JSON object")
300
301            file_version = payload.get("_version")
302            if file_version is not None and file_version != self._version:
303                raise ValueError(
304                    f"settings version mismatch: file={file_version}, expected={self._version}"
305                )
306
307            values = payload.get("values", {})
308            previous_snapshot = self.snapshot()
309            previous = self._root.collect_leaf_values()
310            self._root.reset(notify=False)
311            try:
312                self._root.apply_dict(values, notify=False, ignore_unknown=ignore_unknown)
313            except Exception:
314                self._root.reset(notify=False)
315                self._root.apply_dict(previous_snapshot, notify=False)
316                raise
317            current = self._root.collect_leaf_values()
318            for key, value in current.items():
319                if previous.get(key) != value:
320                    self._emit_change(key, value)

class StreamClosedError(builtins.Exception): View Source

10class StreamClosedError(Exception):
11    """Raised when a blocking stream read is interrupted by stream closure."""

Raised when a blocking stream read is interrupted by stream closure.

class StreamBus: View Source

164class StreamBus:
165    """Stores stream descriptors and fans out accepted frames to frame streams."""
166
167    _STREAM_COUNTER = count(1)
168
169    def __init__(
170        self,
171        *,
172        event_broker: BackendEventBroker,
173        parent: object | None = None,
174    ) -> None:
175        self._event_broker = event_broker
176        self._parent = parent
177        self._descriptors: dict[str, StreamDescriptor] = {}
178        self._frame_streams: list[FrameStream] = []
179        self._lock = RLock()
180
181    def add_descriptor(self, descriptor: StreamDescriptor) -> None:
182        existing_descriptor = self._descriptors.get(descriptor.stream_id)
183        if existing_descriptor is not None:
184            if existing_descriptor == descriptor:
185                return
186
187            raise ValueError(f"conflicting descriptor for stream_id '{descriptor.stream_id}'")
188
189        self._descriptors[descriptor.stream_id] = descriptor
190
191    def add_descriptors(self, descriptors: Iterable[StreamDescriptor]) -> None:
192        for descriptor in descriptors:
193            self.add_descriptor(descriptor)
194
195    def remove_descriptor(self, stream_id: str) -> None:
196        self._descriptors.pop(str(stream_id), None)
197
198    def ingest_frame(self, frame: object) -> bool:
199        if not isinstance(frame, FrameEnvelope):
200            return False
201
202        if frame.stream_id not in self._descriptors:
203            return False
204
205        with self._lock:
206            streams = tuple(self._frame_streams)
207        for stream in streams:
208            stream._publish(frame)
209        return True
210
211    def open_frame_stream(
212        self,
213        *,
214        maxsize: int = 256,
215        drop_policy: FrameDropPolicy = "drop_oldest",
216        consumer_name: str | None = None,
217    ) -> FrameStream:
218        stream = FrameStream(
219            self,
220            maxsize=maxsize,
221            drop_policy=drop_policy,
222            consumer_name=consumer_name or self._next_consumer_name(),
223        )
224        with self._lock:
225            self._frame_streams.append(stream)
226        return stream
227
228    def descriptor(self, stream_id: str) -> StreamDescriptor | None:
229        return self._descriptors.get(stream_id)
230
231    def descriptors(self) -> dict[str, StreamDescriptor]:
232        return dict(self._descriptors)
233
234    def _close_frame_stream(self, stream: FrameStream) -> None:
235        with self._lock:
236            try:
237                self._frame_streams.remove(stream)
238            except ValueError:
239                pass
240        stream._push_closed()
241
242    @classmethod
243    def _next_consumer_name(cls) -> str:
244        return f"frame_stream_{next(cls._STREAM_COUNTER)}"

Stores stream descriptors and fans out accepted frames to frame streams.

StreamBus( *, event_broker: BackendEventBroker, parent: object | None = None) View Source

169    def __init__(
170        self,
171        *,
172        event_broker: BackendEventBroker,
173        parent: object | None = None,
174    ) -> None:
175        self._event_broker = event_broker
176        self._parent = parent
177        self._descriptors: dict[str, StreamDescriptor] = {}
178        self._frame_streams: list[FrameStream] = []
179        self._lock = RLock()

def add_descriptor(self, descriptor: modlink_sdk.StreamDescriptor) -> None: View Source

181    def add_descriptor(self, descriptor: StreamDescriptor) -> None:
182        existing_descriptor = self._descriptors.get(descriptor.stream_id)
183        if existing_descriptor is not None:
184            if existing_descriptor == descriptor:
185                return
186
187            raise ValueError(f"conflicting descriptor for stream_id '{descriptor.stream_id}'")
188
189        self._descriptors[descriptor.stream_id] = descriptor

def add_descriptors(self, descriptors: Iterable[modlink_sdk.StreamDescriptor]) -> None: View Source

191    def add_descriptors(self, descriptors: Iterable[StreamDescriptor]) -> None:
192        for descriptor in descriptors:
193            self.add_descriptor(descriptor)

def remove_descriptor(self, stream_id: str) -> None: View Source

195    def remove_descriptor(self, stream_id: str) -> None:
196        self._descriptors.pop(str(stream_id), None)

def ingest_frame(self, frame: object) -> bool: View Source

198    def ingest_frame(self, frame: object) -> bool:
199        if not isinstance(frame, FrameEnvelope):
200            return False
201
202        if frame.stream_id not in self._descriptors:
203            return False
204
205        with self._lock:
206            streams = tuple(self._frame_streams)
207        for stream in streams:
208            stream._publish(frame)
209        return True

def open_frame_stream( self, *, maxsize: int = 256, drop_policy: Literal['drop_oldest', 'error'] = 'drop_oldest', consumer_name: str | None = None) -> FrameStream: View Source

211    def open_frame_stream(
212        self,
213        *,
214        maxsize: int = 256,
215        drop_policy: FrameDropPolicy = "drop_oldest",
216        consumer_name: str | None = None,
217    ) -> FrameStream:
218        stream = FrameStream(
219            self,
220            maxsize=maxsize,
221            drop_policy=drop_policy,
222            consumer_name=consumer_name or self._next_consumer_name(),
223        )
224        with self._lock:
225            self._frame_streams.append(stream)
226        return stream

def descriptor(self, stream_id: str) -> modlink_sdk.StreamDescriptor | None: View Source

228    def descriptor(self, stream_id: str) -> StreamDescriptor | None:
229        return self._descriptors.get(stream_id)

def descriptors(self) -> dict[str, modlink_sdk.StreamDescriptor]: View Source

231    def descriptors(self) -> dict[str, StreamDescriptor]:
232        return dict(self._descriptors)