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 importlib.resources
18import json
19import logging
20import os
21import shutil
22import socket
23import subprocess
24import sys
25from pathlib import Path
26
27from . import doctor, streams
28from .config import load_config, save_config
29from .streams import stream_name
30
31
32def _setup_logging(verbose: bool = False) -> None:
33 level = logging.DEBUG if verbose else logging.INFO
34 logging.basicConfig(
35 level=level,
36 format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
37 datefmt="%H:%M:%S",
38 )
39
40
41def cmd_run(args: argparse.Namespace) -> int:
42 """Start the capture loop + sync service."""
43 from .observer import async_run
44 from .recovery import recover_incomplete_segments
45
46 config = load_config()
47 config.ensure_dirs()
48
49 if not config.stream:
50 try:
51 config.stream = stream_name(host=socket.gethostname())
52 except ValueError as e:
53 print(f"Error: {e}", file=sys.stderr)
54 return 1
55
56 if args.interval:
57 config.segment_interval = args.interval
58
59 # Crash recovery before starting
60 recovered = recover_incomplete_segments(config.captures_dir)
61 if recovered:
62 print(f"Recovered {recovered} incomplete segment(s)")
63
64 try:
65 return asyncio.run(async_run(config))
66 except KeyboardInterrupt:
67 return 0
68
69
70def cmd_setup(args: argparse.Namespace) -> int:
71 """Interactive setup — configure server URL and register."""
72 cli_token = args.token if getattr(args, "token", None) else None
73 env_token = os.environ.get("SOLSTONE_TOKEN")
74 token = cli_token or env_token
75 non_interactive = getattr(args, "non_interactive", False)
76
77 if (
78 cli_token is None
79 and env_token is None
80 and getattr(args, "server_url", None) is None
81 and getattr(args, "stream_name", None) is None
82 and not non_interactive
83 ):
84 return _cmd_setup_interactive()
85
86 if cli_token:
87 print(
88 "warning: --token on the command line may be visible in shell history and /proc on shared machines",
89 file=sys.stderr,
90 )
91
92 from .upload import UploadClient
93
94 config = load_config()
95
96 server_url = getattr(args, "server_url", None) or config.server_url
97 if not server_url:
98 if non_interactive:
99 print(
100 "error: --server-url required with --non-interactive", file=sys.stderr
101 )
102 return 2
103 default_url = config.server_url or ""
104 url = input(f"Solstone server URL [{default_url}]: ").strip()
105 if url:
106 server_url = url
107 elif not config.server_url:
108 print("Error: server URL is required", file=sys.stderr)
109 return 1
110 config.server_url = server_url
111
112 stream_override = getattr(args, "stream_name", None)
113 if stream_override:
114 config.stream = stream_override
115 elif not config.stream:
116 try:
117 config.stream = streams.stream_name(host=socket.gethostname())
118 except ValueError as e:
119 print(f"Error deriving stream name: {e}", file=sys.stderr)
120 return 1
121
122 config.ensure_dirs()
123
124 if token:
125 config.key = token
126 save_config(config)
127 print(f"Server: {config.server_url}")
128 print(f"Stream: {config.stream}")
129 print("Using provided token; skipping registration.")
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 print(f"Stream: {config.stream}")
138 save_config(config)
139
140 if not config.key:
141 sol = shutil.which("sol")
142 if sol:
143 print("Registering via sol CLI...")
144 try:
145 result = subprocess.run(
146 [sol, "observer", "--json", "create", config.stream],
147 capture_output=True,
148 text=True,
149 timeout=10,
150 )
151 if result.returncode == 0:
152 data = json.loads(result.stdout)
153 config.key = data["key"]
154 save_config(config)
155 print(f"Registered (key: {config.key[:8]}...)")
156 else:
157 print("CLI registration failed, trying HTTP...")
158 except (subprocess.TimeoutExpired, json.JSONDecodeError, KeyError, OSError):
159 print("CLI registration failed, trying HTTP...")
160
161 if not config.key:
162 print("Registering with server...")
163 client = UploadClient(config)
164 if client.ensure_registered(config):
165 config = load_config()
166 print(f"Registered (key: {config.key[:8]}...)")
167 else:
168 print(
169 "Warning: registration failed. Run setup again when server is available."
170 )
171 if non_interactive:
172 return 1
173 else:
174 print(f"Already registered (key: {config.key[:8]}...)")
175
176 print(f"\nConfig saved to {config.config_path}")
177 print(f"Captures will go to {config.captures_dir}")
178 print(
179 "\nRun 'solstone-linux run' to start, or 'solstone-linux install-service' for systemd."
180 )
181 return 0
182
183
184def _cmd_setup_interactive() -> int:
185 # Keep the legacy no-flags setup path separate so its prompt/output stays byte-identical.
186 from .upload import UploadClient
187
188 config = load_config()
189
190 # Prompt for server URL
191 default_url = config.server_url or ""
192 url = input(f"Solstone server URL [{default_url}]: ").strip()
193 if url:
194 config.server_url = url
195 elif not config.server_url:
196 print("Error: server URL is required", file=sys.stderr)
197 return 1
198
199 # Derive stream name
200 if not config.stream:
201 try:
202 config.stream = stream_name(host=socket.gethostname())
203 except ValueError as e:
204 print(f"Error deriving stream name: {e}", file=sys.stderr)
205 return 1
206 print(f"Stream: {config.stream}")
207
208 # Save config before registration (so URL is persisted)
209 config.ensure_dirs()
210 save_config(config)
211
212 # Auto-register — try sol CLI first (no server needed), fall back to HTTP
213 if not config.key:
214 sol = shutil.which("sol")
215 if sol:
216 print("Registering via sol CLI...")
217 try:
218 result = subprocess.run(
219 [sol, "observer", "--json", "create", config.stream],
220 capture_output=True,
221 text=True,
222 timeout=10,
223 )
224 if result.returncode == 0:
225 data = json.loads(result.stdout)
226 config.key = data["key"]
227 save_config(config)
228 print(f"Registered (key: {config.key[:8]}...)")
229 else:
230 print("CLI registration failed, trying HTTP...")
231 except (subprocess.TimeoutExpired, json.JSONDecodeError, KeyError, OSError):
232 print("CLI registration failed, trying HTTP...")
233
234 if not config.key:
235 print("Registering with server...")
236 client = UploadClient(config)
237 if client.ensure_registered(config):
238 config = load_config()
239 print(f"Registered (key: {config.key[:8]}...)")
240 else:
241 print(
242 "Warning: registration failed. Run setup again when server is available."
243 )
244 else:
245 print(f"Already registered (key: {config.key[:8]}...)")
246
247 print(f"\nConfig saved to {config.config_path}")
248 print(f"Captures will go to {config.captures_dir}")
249 print(
250 "\nRun 'solstone-linux run' to start, or 'solstone-linux install-service' for systemd."
251 )
252 return 0
253
254
255def cmd_doctor(args: argparse.Namespace) -> int:
256 return doctor.run_doctor()
257
258
259def cmd_install_service(args: argparse.Namespace) -> int:
260 """Write systemd user unit file, enable, and start the service."""
261 binary = shutil.which("solstone-linux")
262 if not binary:
263 print("Error: solstone-linux not found on PATH", file=sys.stderr)
264 print(
265 "Install with: pipx install --system-site-packages solstone-linux",
266 file=sys.stderr,
267 )
268 return 1
269
270 venv_bin = str(Path(binary).resolve().parent)
271 raw_path = os.environ.get("PATH") or "/usr/local/bin:/usr/bin:/bin"
272 path_entries = [venv_bin] + raw_path.split(":")
273 service_path = ":".join(dict.fromkeys(path_entries))
274
275 unit_dir = Path.home() / ".config" / "systemd" / "user"
276 unit_path = unit_dir / "solstone-linux.service"
277 template = (
278 importlib.resources.files("solstone_linux")
279 .joinpath("solstone-linux.service.in")
280 .read_text()
281 )
282 unit = template.replace("{BINARY}", binary).replace("{PATH}", service_path)
283 unit_dir.mkdir(parents=True, exist_ok=True)
284 unit_path.write_text(unit)
285 print(f"Wrote {unit_path}")
286
287 # Reload, enable, restart, and show status
288 try:
289 subprocess.run(["systemctl", "--user", "daemon-reload"], check=True)
290 subprocess.run(
291 ["systemctl", "--user", "enable", "--now", "solstone-linux.service"],
292 check=True,
293 )
294 subprocess.run(
295 ["systemctl", "--user", "restart", "solstone-linux.service"],
296 check=True,
297 )
298 subprocess.run(
299 [
300 "systemctl",
301 "--user",
302 "--no-pager",
303 "status",
304 "solstone-linux.service",
305 ],
306 check=False,
307 )
308 except FileNotFoundError:
309 print("Warning: systemctl not found. Enable the service manually.")
310 except subprocess.CalledProcessError as e:
311 print(f"Warning: systemctl command failed: {e}")
312
313 icon_source = Path(__file__).resolve().parent / "icons" / "hicolor"
314 if icon_source.is_dir():
315 icon_dest = Path.home() / ".local" / "share" / "icons" / "hicolor"
316 status_dir = icon_dest / "scalable" / "status"
317 status_dir.mkdir(parents=True, exist_ok=True)
318
319 for svg in sorted((icon_source / "scalable" / "status").iterdir()):
320 if svg.suffix == ".svg":
321 shutil.copy2(svg, status_dir / svg.name)
322 print(f"Installed {status_dir / svg.name}")
323
324 # Copy index.theme only if one doesn't already exist
325 index_dest = icon_dest / "index.theme"
326 if not index_dest.exists():
327 shutil.copy2(icon_source / "index.theme", index_dest)
328 print(f"Wrote {index_dest}")
329
330 # Update icon cache (non-fatal)
331 try:
332 subprocess.run(["gtk-update-icon-cache", str(icon_dest)], check=False)
333 except FileNotFoundError:
334 pass
335
336 return 0
337
338
339def cmd_status(args: argparse.Namespace) -> int:
340 """Show capture and sync state."""
341 config = load_config()
342
343 print(f"Config: {config.config_path}")
344 print(f"Server: {config.server_url or '(not configured)'}")
345 print(f"Key: {config.key[:8] + '...' if config.key else '(not registered)'}")
346 print(f"Stream: {config.stream or '(not set)'}")
347 print()
348
349 # Cache size
350 captures_dir = config.captures_dir
351 if captures_dir.exists():
352 total_size = 0
353 segment_count = 0
354 day_count = 0
355 incomplete_count = 0
356
357 for day_dir in sorted(captures_dir.iterdir()):
358 if not day_dir.is_dir():
359 continue
360 day_count += 1
361 for stream_dir in day_dir.iterdir():
362 if not stream_dir.is_dir():
363 continue
364 for seg_dir in stream_dir.iterdir():
365 if not seg_dir.is_dir():
366 continue
367 if seg_dir.name.endswith(".incomplete"):
368 incomplete_count += 1
369 continue
370 if seg_dir.name.endswith(".failed"):
371 continue
372 segment_count += 1
373 for f in seg_dir.iterdir():
374 if f.is_file():
375 total_size += f.stat().st_size
376
377 size_mb = total_size / (1024 * 1024)
378 print(f"Cache: {captures_dir}")
379 print(
380 f" {segment_count} segments across {day_count} day(s), {size_mb:.1f} MB"
381 )
382 if incomplete_count:
383 print(f" {incomplete_count} incomplete segment(s)")
384 else:
385 print(f"Cache: {captures_dir} (not created yet)")
386
387 # Retention policy
388 retention = config.cache_retention_days
389 if retention < 0:
390 print("Retain: forever")
391 elif retention == 0:
392 print("Retain: delete after sync")
393 else:
394 print(f"Retain: {retention} day(s)")
395
396 # Synced days
397 synced_path = config.state_dir / "synced_days.json"
398 if synced_path.exists():
399 try:
400 with open(synced_path) as f:
401 synced = json.load(f)
402 print(f"Synced: {len(synced)} day(s) fully synced")
403 except (json.JSONDecodeError, OSError):
404 pass
405
406 # Systemd status
407 try:
408 result = subprocess.run(
409 ["systemctl", "--user", "is-active", "solstone-linux.service"],
410 capture_output=True,
411 text=True,
412 )
413 state = result.stdout.strip()
414 print(f"\nService: {state}")
415 except FileNotFoundError:
416 pass
417
418 return 0
419
420
421def main() -> None:
422 """CLI entry point."""
423 parser = argparse.ArgumentParser(
424 prog="solstone-linux",
425 description="Standalone Linux desktop observer for solstone",
426 )
427 parser.add_argument(
428 "-v", "--verbose", action="store_true", help="Enable debug logging"
429 )
430 subparsers = parser.add_subparsers(dest="command")
431
432 # run
433 run_parser = subparsers.add_parser("run", help="Start capture + sync")
434 run_parser.add_argument(
435 "--interval",
436 type=int,
437 default=None,
438 help="Segment duration in seconds (default: 300)",
439 )
440
441 # setup
442 setup_parser = subparsers.add_parser("setup", help="Interactive configuration")
443 setup_parser.add_argument("--server-url", help="Server URL (skips prompt)")
444 setup_parser.add_argument(
445 "--token",
446 help="Pre-issued registration key; skips server registration",
447 )
448 setup_parser.add_argument(
449 "--stream-name",
450 help="Stream name (defaults to hostname-derived)",
451 )
452 setup_parser.add_argument(
453 "--non-interactive",
454 action="store_true",
455 help="Fail instead of prompting for missing values",
456 )
457
458 # doctor
459 subparsers.add_parser(
460 "doctor",
461 help="Verify install prerequisites",
462 )
463
464 # install-service
465 subparsers.add_parser("install-service", help="Install systemd user service")
466
467 # status
468 subparsers.add_parser("status", help="Show capture and sync state")
469
470 args = parser.parse_args()
471 _setup_logging(args.verbose)
472
473 # Default to run if no subcommand
474 command = args.command or "run"
475
476 commands = {
477 "run": cmd_run,
478 "setup": cmd_setup,
479 "doctor": cmd_doctor,
480 "install-service": cmd_install_service,
481 "status": cmd_status,
482 }
483
484 handler = commands.get(command)
485 if handler:
486 sys.exit(handler(args))
487 else:
488 parser.print_help()
489 sys.exit(1)