solpbc.org
linux observer
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""Stream identity for observer segments.
5
6Extracted from solstone's think/streams.py — only the pure naming functions
7needed by standalone observers.
8
9Naming convention (separator is '.'):
10 Local Linux: {hostname} e.g. "archon"
11 Observer: {observer_name} e.g. "desktop"
12"""
13
14from __future__ import annotations
15
16import re
17
18_STREAM_NAME_RE = re.compile(r"^[a-z0-9][a-z0-9._-]*$")
19
20
21def _strip_hostname(name: str) -> str:
22 """Strip domain suffix from a hostname, keeping only the first label.
23
24 Dots in stream names are reserved for qualifiers (e.g., '.tmux').
25 Hostnames like 'ja1r.local' or '192.168.1.1' must be reduced to a
26 dot-free base name.
27
28 Examples: 'ja1r.local' -> 'ja1r', '192.168.1.1' -> '192-168-1-1',
29 'archon' -> 'archon', 'my.host.example.com' -> 'my'
30 """
31 name = name.strip()
32 if not name:
33 return name
34 parts = name.split(".")
35 if all(p.isdigit() for p in parts if p):
36 return "-".join(p for p in parts if p)
37 return parts[0]
38
39
40def stream_name(
41 *,
42 host: str | None = None,
43 observer: str | None = None,
44 qualifier: str | None = None,
45) -> str:
46 """Derive canonical stream name from source characteristics.
47
48 Parameters
49 ----------
50 host : str, optional
51 Local hostname (e.g., "archon").
52 observer : str, optional
53 Observer name (e.g., "desktop").
54 qualifier : str, optional
55 Sub-stream qualifier. Appended with dot separator.
56
57 Returns
58 -------
59 str
60 Canonical stream name.
61
62 Raises
63 ------
64 ValueError
65 If no source is provided, or the resulting name is invalid.
66 """
67 if host:
68 base = _strip_hostname(host)
69 elif observer:
70 base = _strip_hostname(observer)
71 else:
72 raise ValueError("stream_name requires host or observer")
73
74 name = base.lower().strip()
75 name = re.sub(r"[\s/\\]+", "-", name)
76
77 if qualifier:
78 qualifier = qualifier.lower().strip()
79 qualifier = re.sub(r"[\s/\\]+", "-", qualifier)
80 name = f"{name}.{qualifier}"
81
82 if not name or ".." in name:
83 raise ValueError(f"Invalid stream name: {name!r}")
84 if not _STREAM_NAME_RE.match(name):
85 raise ValueError(f"Invalid stream name: {name!r}")
86
87 return name