solpbc.org
linux observer
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""CLI entry point for solstone-linux.
5
6Subcommands:
7 run Start capture loop + sync service (default)
8 setup Interactive configuration
9 install-service Write systemd user unit, enable, start
10 status Show capture and sync state
11"""
12
13from __future__ import annotations
14
15import argparse
16import asyncio
17import json
18import logging
19import shutil
20import socket
21import subprocess
22import sys
23from pathlib import Path
24
25from .config import load_config, save_config
26from .streams import stream_name
27
28
29def _setup_logging(verbose: bool = False) -> None:
30 level = logging.DEBUG if verbose else logging.INFO
31 logging.basicConfig(
32 level=level,
33 format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
34 datefmt="%H:%M:%S",
35 )
36
37
38def cmd_run(args: argparse.Namespace) -> int:
39 """Start the capture loop + sync service."""
40 from .observer import async_run
41 from .recovery import recover_incomplete_segments
42
43 config = load_config()
44 config.ensure_dirs()
45
46 if not config.stream:
47 try:
48 config.stream = stream_name(host=socket.gethostname())
49 except ValueError as e:
50 print(f"Error: {e}", file=sys.stderr)
51 return 1
52
53 if args.interval:
54 config.segment_interval = args.interval
55
56 # Crash recovery before starting
57 recovered = recover_incomplete_segments(config.captures_dir)
58 if recovered:
59 print(f"Recovered {recovered} incomplete segment(s)")
60
61 try:
62 return asyncio.run(async_run(config))
63 except KeyboardInterrupt:
64 return 0
65
66
67def cmd_setup(args: argparse.Namespace) -> int:
68 """Interactive setup — configure server URL and register."""
69 from .upload import UploadClient
70
71 config = load_config()
72
73 # Prompt for server URL
74 default_url = config.server_url or ""
75 url = input(f"Solstone server URL [{default_url}]: ").strip()
76 if url:
77 config.server_url = url
78 elif not config.server_url:
79 print("Error: server URL is required", file=sys.stderr)
80 return 1
81
82 # Derive stream name
83 if not config.stream:
84 try:
85 config.stream = stream_name(host=socket.gethostname())
86 except ValueError as e:
87 print(f"Error deriving stream name: {e}", file=sys.stderr)
88 return 1
89 print(f"Stream: {config.stream}")
90
91 # Save config before registration (so URL is persisted)
92 config.ensure_dirs()
93 save_config(config)
94
95 # Auto-register — try sol CLI first (no server needed), fall back to HTTP
96 if not config.key:
97 sol = shutil.which("sol")
98 if sol:
99 print("Registering via sol CLI...")
100 try:
101 result = subprocess.run(
102 [sol, "observer", "--json", "create", config.stream],
103 capture_output=True,
104 text=True,
105 timeout=10,
106 )
107 if result.returncode == 0:
108 data = json.loads(result.stdout)
109 config.key = data["key"]
110 save_config(config)
111 print(f"Registered (key: {config.key[:8]}...)")
112 else:
113 print("CLI registration failed, trying HTTP...")
114 except (subprocess.TimeoutExpired, json.JSONDecodeError, KeyError, OSError):
115 print("CLI registration failed, trying HTTP...")
116
117 if not config.key:
118 print("Registering with server...")
119 client = UploadClient(config)
120 if client.ensure_registered(config):
121 config = load_config()
122 print(f"Registered (key: {config.key[:8]}...)")
123 else:
124 print(
125 "Warning: registration failed. Run setup again when server is available."
126 )
127 else:
128 print(f"Already registered (key: {config.key[:8]}...)")
129
130 print(f"\nConfig saved to {config.config_path}")
131 print(f"Captures will go to {config.captures_dir}")
132 print(
133 "\nRun 'solstone-linux run' to start, or 'solstone-linux install-service' for systemd."
134 )
135 return 0
136
137
138def cmd_install_service(args: argparse.Namespace) -> int:
139 """Write systemd user unit file, enable, and start the service."""
140 binary = shutil.which("solstone-linux")
141 if not binary:
142 print("Error: solstone-linux not found on PATH", file=sys.stderr)
143 print(
144 "Install with: pipx install --system-site-packages solstone-linux",
145 file=sys.stderr,
146 )
147 return 1
148
149 unit_dir = Path.home() / ".config" / "systemd" / "user"
150 unit_dir.mkdir(parents=True, exist_ok=True)
151 unit_path = unit_dir / "solstone-linux.service"
152
153 unit_content = f"""\
154[Unit]
155Description=Solstone Linux Desktop Observer
156After=graphical-session.target
157BindsTo=graphical-session.target
158
159[Service]
160Type=simple
161ExecStart={binary} run
162PassEnvironment=DISPLAY WAYLAND_DISPLAY DBUS_SESSION_BUS_ADDRESS XDG_RUNTIME_DIR XDG_CURRENT_DESKTOP
163Restart=on-failure
164RestartSec=10
165StartLimitIntervalSec=300
166StartLimitBurst=5
167
168[Install]
169WantedBy=graphical-session.target
170"""
171
172 unit_path.write_text(unit_content)
173 print(f"Wrote {unit_path}")
174
175 # Reload, enable, start
176 try:
177 subprocess.run(["systemctl", "--user", "daemon-reload"], check=True)
178 subprocess.run(
179 ["systemctl", "--user", "enable", "--now", "solstone-linux.service"],
180 check=True,
181 )
182 print("Service enabled and started.")
183 subprocess.run(
184 ["systemctl", "--user", "status", "solstone-linux.service"],
185 check=False,
186 )
187 except FileNotFoundError:
188 print("Warning: systemctl not found. Enable the service manually.")
189 except subprocess.CalledProcessError as e:
190 print(f"Warning: systemctl command failed: {e}")
191
192 return 0
193
194
195def cmd_status(args: argparse.Namespace) -> int:
196 """Show capture and sync state."""
197 config = load_config()
198
199 print(f"Config: {config.config_path}")
200 print(f"Server: {config.server_url or '(not configured)'}")
201 print(f"Key: {config.key[:8] + '...' if config.key else '(not registered)'}")
202 print(f"Stream: {config.stream or '(not set)'}")
203 print()
204
205 # Cache size
206 captures_dir = config.captures_dir
207 if captures_dir.exists():
208 total_size = 0
209 segment_count = 0
210 day_count = 0
211 incomplete_count = 0
212
213 for day_dir in sorted(captures_dir.iterdir()):
214 if not day_dir.is_dir():
215 continue
216 day_count += 1
217 for stream_dir in day_dir.iterdir():
218 if not stream_dir.is_dir():
219 continue
220 for seg_dir in stream_dir.iterdir():
221 if not seg_dir.is_dir():
222 continue
223 if seg_dir.name.endswith(".incomplete"):
224 incomplete_count += 1
225 continue
226 if seg_dir.name.endswith(".failed"):
227 continue
228 segment_count += 1
229 for f in seg_dir.iterdir():
230 if f.is_file():
231 total_size += f.stat().st_size
232
233 size_mb = total_size / (1024 * 1024)
234 print(f"Cache: {captures_dir}")
235 print(
236 f" {segment_count} segments across {day_count} day(s), {size_mb:.1f} MB"
237 )
238 if incomplete_count:
239 print(f" {incomplete_count} incomplete segment(s)")
240 else:
241 print(f"Cache: {captures_dir} (not created yet)")
242
243 # Synced days
244 synced_path = config.state_dir / "synced_days.json"
245 if synced_path.exists():
246 try:
247 with open(synced_path) as f:
248 synced = json.load(f)
249 print(f"Synced: {len(synced)} day(s) fully synced")
250 except (json.JSONDecodeError, OSError):
251 pass
252
253 # Systemd status
254 try:
255 result = subprocess.run(
256 ["systemctl", "--user", "is-active", "solstone-linux.service"],
257 capture_output=True,
258 text=True,
259 )
260 state = result.stdout.strip()
261 print(f"\nService: {state}")
262 except FileNotFoundError:
263 pass
264
265 return 0
266
267
268def main() -> None:
269 """CLI entry point."""
270 parser = argparse.ArgumentParser(
271 prog="solstone-linux",
272 description="Standalone Linux desktop observer for solstone",
273 )
274 parser.add_argument(
275 "-v", "--verbose", action="store_true", help="Enable debug logging"
276 )
277 subparsers = parser.add_subparsers(dest="command")
278
279 # run
280 run_parser = subparsers.add_parser("run", help="Start capture + sync")
281 run_parser.add_argument(
282 "--interval",
283 type=int,
284 default=None,
285 help="Segment duration in seconds (default: 300)",
286 )
287
288 # setup
289 subparsers.add_parser("setup", help="Interactive configuration")
290
291 # install-service
292 subparsers.add_parser("install-service", help="Install systemd user service")
293
294 # status
295 subparsers.add_parser("status", help="Show capture and sync state")
296
297 args = parser.parse_args()
298 _setup_logging(args.verbose)
299
300 # Default to run if no subcommand
301 command = args.command or "run"
302
303 commands = {
304 "run": cmd_run,
305 "setup": cmd_setup,
306 "install-service": cmd_install_service,
307 "status": cmd_status,
308 }
309
310 handler = commands.get(command)
311 if handler:
312 sys.exit(handler(args))
313 else:
314 parser.print_help()
315 sys.exit(1)