Add D-Bus service interface with pause/resume for observer

+159

src/solstone_linux/dbus_service.py

··· 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 + # Copyright (c) 2026 sol pbc 3 + # ruff: noqa: F722, F821 4 + 5 + import logging 6 + import time 7 + from datetime import datetime 8 + 9 + from dbus_next import PropertyAccess, Variant 10 + from dbus_next.service import ( 11 + ServiceInterface, 12 + dbus_property, 13 + method, 14 + signal as dbus_signal, 15 + ) 16 + 17 + logger = logging.getLogger(__name__) 18 + 19 + BUS_NAME = "org.solpbc.solstone.Observer1" 20 + OBJECT_PATH = "/org/solpbc/solstone/Observer1" 21 + 22 + 23 + class ObserverService(ServiceInterface): 24 + """D-Bus service interface for the observer.""" 25 + 26 + def __init__(self, observer): 27 + super().__init__("org.solpbc.solstone.Observer1") 28 + self._observer = observer 29 + 30 + @dbus_property(access=PropertyAccess.READ) 31 + def Status(self) -> "s": 32 + if self._observer._paused: 33 + return "paused" 34 + if self._observer.current_mode == "screencast": 35 + return "recording" 36 + return "idle" 37 + 38 + @dbus_property(access=PropertyAccess.READ) 39 + def SyncStatus(self) -> "s": 40 + if self._observer._sync: 41 + return self._observer._sync.sync_status 42 + return "synced" 43 + 44 + @dbus_property(access=PropertyAccess.READ) 45 + def SyncProgress(self) -> "s": 46 + if self._observer._sync: 47 + return self._observer._sync.sync_progress 48 + return "" 49 + 50 + @dbus_property(access=PropertyAccess.READ) 51 + def CaptureDir(self) -> "s": 52 + return str(self._observer.config.captures_dir) 53 + 54 + @dbus_property(access=PropertyAccess.READ) 55 + def SegmentTimer(self) -> "i": 56 + if self._observer._paused or self._observer.segment_dir is None: 57 + return 0 58 + remaining = self._observer.interval - ( 59 + time.monotonic() - self._observer.start_at_mono 60 + ) 61 + return max(0, int(remaining)) 62 + 63 + @dbus_property(access=PropertyAccess.READ) 64 + def PauseRemaining(self) -> "i": 65 + if not self._observer._paused or self._observer._pause_until <= 0: 66 + return 0 67 + return max(0, int(self._observer._pause_until - time.monotonic())) 68 + 69 + @dbus_property(access=PropertyAccess.READ) 70 + def Error(self) -> "s": 71 + return "" 72 + 73 + @dbus_property(access=PropertyAccess.READ) 74 + def ServerUrl(self) -> "s": 75 + return self._observer.config.server_url or "" 76 + 77 + @dbus_property(access=PropertyAccess.READ) 78 + def Stream(self) -> "s": 79 + return self._observer.stream 80 + 81 + @dbus_property(access=PropertyAccess.READ) 82 + def SegmentInterval(self) -> "i": 83 + return self._observer.interval 84 + 85 + @method() 86 + def Pause(self, duration_seconds: "i") -> "s": 87 + self._observer._paused = True 88 + if duration_seconds > 0: 89 + self._observer._pause_until = time.monotonic() + duration_seconds 90 + else: 91 + self._observer._pause_until = 0.0 92 + self.StatusChanged("paused") 93 + logger.info("Pause requested: %ss", duration_seconds) 94 + return "ok" 95 + 96 + @method() 97 + def Resume(self) -> "s": 98 + self._observer._paused = False 99 + self._observer._pause_until = 0.0 100 + self.StatusChanged( 101 + "recording" if self._observer.current_mode == "screencast" else "idle" 102 + ) 103 + return "ok" 104 + 105 + @method() 106 + def GetStats(self) -> "a{sv}": 107 + captures_today = 0 108 + total_size = 0 109 + today = datetime.now().strftime("%Y%m%d") 110 + captures_dir = self._observer.config.captures_dir 111 + 112 + try: 113 + if captures_dir.exists(): 114 + for day_dir in captures_dir.iterdir(): 115 + if not day_dir.is_dir(): 116 + continue 117 + for stream_dir in day_dir.iterdir(): 118 + if not stream_dir.is_dir(): 119 + continue 120 + for seg_dir in stream_dir.iterdir(): 121 + if not seg_dir.is_dir(): 122 + continue 123 + if seg_dir.name.endswith(".incomplete"): 124 + continue 125 + if seg_dir.name.endswith(".failed"): 126 + continue 127 + if day_dir.name == today: 128 + captures_today += 1 129 + for file_path in seg_dir.iterdir(): 130 + if file_path.is_file(): 131 + total_size += file_path.stat().st_size 132 + except OSError: 133 + pass 134 + 135 + synced_days = 0 136 + if self._observer._sync: 137 + synced_days = len(self._observer._sync._synced_days) 138 + 139 + total_size_mb = int(total_size / (1024 * 1024)) 140 + uptime_seconds = int(time.monotonic() - self._observer.start_at_mono) 141 + 142 + return { 143 + "captures_today": Variant("i", captures_today), 144 + "total_size_mb": Variant("i", total_size_mb), 145 + "synced_days": Variant("i", synced_days), 146 + "uptime_seconds": Variant("i", uptime_seconds), 147 + } 148 + 149 + @dbus_signal() 150 + def StatusChanged(self, status) -> "s": 151 + return status 152 + 153 + @dbus_signal() 154 + def SyncProgressChanged(self, progress) -> "s": 155 + return progress 156 + 157 + @dbus_signal() 158 + def ErrorOccurred(self, message) -> "s": 159 + return message

+74

src/solstone_linux/observer.py

··· 112 112 # Mute state at segment start (determines save format) 113 113 self.segment_is_muted = False 114 114 115 + # Pause state 116 + self._paused = False 117 + self._pause_until = 0.0 118 + 119 + # D-Bus service interface 120 + self._dbus_service = None 121 + 115 122 async def setup(self) -> bool: 116 123 """Initialize audio devices, DBus connection, and sync service.""" 117 124 # Detect audio devices with retry (devices may still be initializing) ··· 153 160 if self.config.server_url: 154 161 self._client.ensure_registered(self.config) 155 162 self._sync = SyncService(self.config, self._client) 163 + 164 + from .dbus_service import BUS_NAME, OBJECT_PATH, ObserverService 165 + 166 + self._dbus_service = ObserverService(self) 167 + self.bus.export(OBJECT_PATH, self._dbus_service) 168 + await self.bus.request_name(BUS_NAME) 169 + self._sync._dbus_service = self._dbus_service 170 + logger.info("D-Bus service exported as %s", BUS_NAME) 156 171 logger.info("Sync service initialized") 157 172 158 173 return True ··· 451 466 while self.running: 452 467 await asyncio.sleep(CHUNK_DURATION) 453 468 469 + # Check auto-resume from timed pause 470 + if ( 471 + self._paused 472 + and self._pause_until > 0 473 + and time.monotonic() >= self._pause_until 474 + ): 475 + self._paused = False 476 + self._pause_until = 0.0 477 + if self._dbus_service: 478 + self._dbus_service.StatusChanged( 479 + "recording" 480 + if self.current_mode == MODE_SCREENCAST 481 + else "idle" 482 + ) 483 + logger.info("Auto-resumed from timed pause") 484 + 485 + # Handle paused state 486 + if self._paused: 487 + if self.segment_dir: 488 + if self.current_mode == MODE_SCREENCAST: 489 + await self.screencaster.stop() 490 + self.current_streams = [] 491 + if self.threshold_hits >= MIN_HITS_FOR_SAVE: 492 + self._save_audio_segment( 493 + self.segment_dir, self.segment_is_muted 494 + ) 495 + self.accumulated_audio_buffer = np.array( 496 + [], dtype=np.float32 497 + ).reshape(0, 2) 498 + self.threshold_hits = 0 499 + segment_key = self._finalize_segment() 500 + self.segment_dir = None 501 + if segment_key and self._sync: 502 + self._sync.trigger() 503 + self.audio_recorder.get_buffers() 504 + self.emit_status() 505 + continue 506 + 507 + # Resume: start new segment if needed (segment_dir is None after pause) 508 + if self.segment_dir is None: 509 + try: 510 + new_mode = await self.check_activity_status() 511 + except Exception: 512 + new_mode = self.current_mode 513 + self.segment_is_muted = self.cached_is_muted 514 + self.current_mode = new_mode 515 + if new_mode == MODE_SCREENCAST and not self.cached_screen_locked: 516 + try: 517 + await self.initialize_screencast() 518 + except RuntimeError: 519 + self._start_segment() 520 + else: 521 + self._start_segment() 522 + self.emit_status() 523 + continue 524 + 454 525 # Check activity status and determine new mode 455 526 try: 456 527 new_mode = await self.check_activity_status() ··· 522 593 f"hits={self.threshold_hits}/{MIN_HITS_FOR_SAVE}" 523 594 ) 524 595 await self.handle_boundary(new_mode) 596 + if mode_changed and self._dbus_service: 597 + status = "recording" if new_mode == MODE_SCREENCAST else "idle" 598 + self._dbus_service.StatusChanged(status) 525 599 526 600 # Emit status event 527 601 self.emit_status()

+26

src/solstone_linux/sync.py

··· 60 60 self._last_full_sync: float = 0 61 61 self._running = True 62 62 self._trigger = asyncio.Event() 63 + self.sync_status = "synced" 64 + self.sync_progress = "" 65 + self._dbus_service = None 63 66 64 67 # Load synced days cache 65 68 self._load_synced_days() ··· 121 124 self._running = False 122 125 self._trigger.set() 123 126 127 + def _set_sync_status(self, status: str, progress: str = "") -> None: 128 + """Update sync status and emit D-Bus signal if changed.""" 129 + changed = self.sync_status != status or self.sync_progress != progress 130 + self.sync_status = status 131 + self.sync_progress = progress 132 + if changed and self._dbus_service: 133 + self._dbus_service.SyncProgressChanged(f"{status}:{progress}") 134 + 124 135 async def run(self) -> None: 125 136 """Main sync loop — waits for triggers, then syncs.""" 126 137 # Prune on startup ··· 141 152 142 153 if self._circuit_open: 143 154 if self._circuit_open_permanent: 155 + self._set_sync_status("offline") 144 156 logger.warning( 145 157 "Circuit breaker open (permanent) — skipping sync" 146 158 ) ··· 149 161 elapsed = time.monotonic() - self._circuit_open_since 150 162 if elapsed < self._circuit_cooldown: 151 163 remaining = self._circuit_cooldown - elapsed 164 + self._set_sync_status( 165 + "retrying", f"{remaining:.0f}s until probe" 166 + ) 152 167 logger.warning( 153 168 f"Circuit breaker open — {remaining:.0f}s until probe" 154 169 ) 155 170 continue 156 171 172 + self._set_sync_status("retrying", "probing server...") 157 173 logger.info("Circuit breaker half-open — probing server") 158 174 today = datetime.now().strftime("%Y%m%d") 159 175 probe_result = await asyncio.to_thread( ··· 167 183 self._circuit_cooldown = CIRCUIT_COOLDOWN_INITIAL 168 184 self._consecutive_failures = 0 169 185 self._last_error_type = None 186 + self._set_sync_status("syncing") 170 187 else: 171 188 self._circuit_cooldown = min( 172 189 self._circuit_cooldown * CIRCUIT_COOLDOWN_FACTOR, 173 190 CIRCUIT_COOLDOWN_MAX, 174 191 ) 175 192 self._circuit_open_since = time.monotonic() 193 + self._set_sync_status( 194 + "retrying", 195 + f"probe failed, next in {self._circuit_cooldown:.0f}s", 196 + ) 176 197 logger.warning( 177 198 f"Circuit breaker probe failed — next probe in {self._circuit_cooldown:.0f}s" 178 199 ) ··· 182 203 now = time.time() 183 204 force_full = (now - self._last_full_sync) > 86400 184 205 206 + self._set_sync_status("syncing") 185 207 await self._sync(force_full=force_full) 208 + self._set_sync_status("synced") 186 209 187 210 if force_full: 188 211 self._last_full_sync = now ··· 218 241 local_segments = segments_by_day[day] 219 242 220 243 # Query server for existing segments 244 + self._set_sync_status("syncing", f"checking {day}...") 221 245 server_segments = await asyncio.to_thread( 222 246 self._client.get_server_segments, day 223 247 ) ··· 243 267 continue 244 268 245 269 any_needed_upload = True 270 + self._set_sync_status("uploading", f"uploading {segment_key}") 246 271 success = await self._upload_segment(day, segment_dir) 247 272 248 273 if not success: ··· 257 282 f"{self._last_error_type.value if self._last_error_type else 'unknown'} " 258 283 f"failures (threshold: {threshold})" 259 284 ) 285 + self._set_sync_status("retrying") 260 286 break 261 287 else: 262 288 self._consecutive_failures = 0

+297

tests/test_dbus_service.py

··· 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 + # Copyright (c) 2026 sol pbc 3 + 4 + import time 5 + from datetime import datetime, timedelta 6 + from pathlib import Path 7 + from unittest.mock import MagicMock 8 + 9 + from dbus_next.service import ServiceInterface 10 + 11 + from solstone_linux.config import Config 12 + from solstone_linux.dbus_service import ObserverService 13 + from solstone_linux.sync import SyncService 14 + from solstone_linux.upload import UploadClient 15 + 16 + 17 + def _get_prop(service, name): 18 + for prop in ServiceInterface._get_properties(service): 19 + if prop.name == name: 20 + return prop.prop_getter(service) 21 + raise KeyError(name) 22 + 23 + 24 + def _call_method(service, name, *args): 25 + for method in ServiceInterface._get_methods(service): 26 + if method.name == name: 27 + return method.fn(service, *args) 28 + raise KeyError(name) 29 + 30 + 31 + def _make_observer(captures_dir: Path | None = None): 32 + observer = MagicMock() 33 + observer._paused = False 34 + observer._pause_until = 0.0 35 + observer.current_mode = "screencast" 36 + observer.config = MagicMock() 37 + observer.config.captures_dir = captures_dir or Path("/tmp/test-captures") 38 + observer.config.server_url = "https://test.example.com" 39 + observer.interval = 300 40 + observer.segment_dir = None 41 + observer.start_at_mono = time.monotonic() 42 + observer.stream = "test-stream" 43 + observer._sync = None 44 + observer._dbus_service = None 45 + return observer 46 + 47 + 48 + class TestObserverServiceStatus: 49 + def test_status_recording(self): 50 + observer = _make_observer() 51 + observer.current_mode = "screencast" 52 + 53 + service = ObserverService(observer) 54 + 55 + assert _get_prop(service, "Status") == "recording" 56 + 57 + def test_status_idle(self): 58 + observer = _make_observer() 59 + observer.current_mode = "idle" 60 + 61 + service = ObserverService(observer) 62 + 63 + assert _get_prop(service, "Status") == "idle" 64 + 65 + def test_status_paused(self): 66 + observer = _make_observer() 67 + observer._paused = True 68 + 69 + service = ObserverService(observer) 70 + 71 + assert _get_prop(service, "Status") == "paused" 72 + 73 + 74 + class TestPauseResume: 75 + def test_pause_sets_state(self): 76 + observer = _make_observer() 77 + service = ObserverService(observer) 78 + before = time.monotonic() 79 + 80 + result = _call_method(service, "Pause", 30) 81 + 82 + assert result == "ok" 83 + assert observer._paused is True 84 + assert before + 29 <= observer._pause_until <= before + 31 85 + 86 + def test_pause_indefinite(self): 87 + observer = _make_observer() 88 + service = ObserverService(observer) 89 + 90 + _call_method(service, "Pause", 0) 91 + 92 + assert observer._paused is True 93 + assert observer._pause_until == 0.0 94 + 95 + def test_resume_clears_state(self): 96 + observer = _make_observer() 97 + service = ObserverService(observer) 98 + _call_method(service, "Pause", 30) 99 + 100 + result = _call_method(service, "Resume") 101 + 102 + assert result == "ok" 103 + assert observer._paused is False 104 + assert observer._pause_until == 0.0 105 + 106 + def test_resume_returns_mode(self): 107 + observer = _make_observer() 108 + observer.current_mode = "idle" 109 + observer._paused = True 110 + observer._pause_until = time.monotonic() + 30 111 + service = ObserverService(observer) 112 + 113 + result = _call_method(service, "Resume") 114 + 115 + assert result == "ok" 116 + assert _get_prop(service, "Status") == "idle" 117 + 118 + 119 + class TestAutoResume: 120 + def test_auto_resume_expiry(self): 121 + observer = _make_observer() 122 + observer._paused = True 123 + observer._pause_until = time.monotonic() - 1 124 + 125 + if ( 126 + observer._paused 127 + and observer._pause_until > 0 128 + and time.monotonic() >= observer._pause_until 129 + ): 130 + observer._paused = False 131 + observer._pause_until = 0.0 132 + 133 + assert observer._paused is False 134 + assert observer._pause_until == 0.0 135 + 136 + 137 + class TestSegmentTimerAndPauseRemaining: 138 + def test_segment_timer_while_recording(self): 139 + observer = _make_observer() 140 + observer.segment_dir = Path("/tmp/test.incomplete") 141 + observer.start_at_mono = time.monotonic() - 60 142 + service = ObserverService(observer) 143 + 144 + timer = _get_prop(service, "SegmentTimer") 145 + 146 + assert 238 <= timer <= 242 147 + 148 + def test_segment_timer_zero_when_paused(self): 149 + observer = _make_observer() 150 + observer._paused = True 151 + service = ObserverService(observer) 152 + 153 + assert _get_prop(service, "SegmentTimer") == 0 154 + 155 + def test_segment_timer_zero_when_no_segment(self): 156 + observer = _make_observer() 157 + observer.segment_dir = None 158 + service = ObserverService(observer) 159 + 160 + assert _get_prop(service, "SegmentTimer") == 0 161 + 162 + def test_pause_remaining_during_timed_pause(self): 163 + observer = _make_observer() 164 + observer._paused = True 165 + observer._pause_until = time.monotonic() + 120 166 + service = ObserverService(observer) 167 + 168 + remaining = _get_prop(service, "PauseRemaining") 169 + 170 + assert 118 <= remaining <= 122 171 + 172 + def test_pause_remaining_zero_when_not_paused(self): 173 + observer = _make_observer() 174 + observer._paused = False 175 + service = ObserverService(observer) 176 + 177 + assert _get_prop(service, "PauseRemaining") == 0 178 + 179 + def test_pause_remaining_zero_for_indefinite_pause(self): 180 + observer = _make_observer() 181 + observer._paused = True 182 + observer._pause_until = 0.0 183 + service = ObserverService(observer) 184 + 185 + assert _get_prop(service, "PauseRemaining") == 0 186 + 187 + 188 + class TestGetStats: 189 + def test_returns_stats_dict(self, tmp_path: Path): 190 + captures_dir = tmp_path / "captures" 191 + today = datetime.now().strftime("%Y%m%d") 192 + segment_dir = captures_dir / today / "stream-a" / "120000_300" 193 + segment_dir.mkdir(parents=True) 194 + (segment_dir / "audio.flac").write_bytes(b"x" * (1024 * 1024)) 195 + 196 + observer = _make_observer(captures_dir) 197 + observer._sync = MagicMock() 198 + observer._sync._synced_days = {"20260410", "20260411"} 199 + service = ObserverService(observer) 200 + 201 + stats = _call_method(service, "GetStats") 202 + 203 + assert stats["captures_today"].value == 1 204 + assert stats["total_size_mb"].value == 1 205 + assert stats["synced_days"].value == 2 206 + assert stats["uptime_seconds"].value >= 0 207 + 208 + def test_empty_captures(self, tmp_path: Path): 209 + observer = _make_observer(tmp_path / "captures") 210 + service = ObserverService(observer) 211 + 212 + stats = _call_method(service, "GetStats") 213 + 214 + assert stats["captures_today"].value == 0 215 + assert stats["total_size_mb"].value == 0 216 + assert stats["synced_days"].value == 0 217 + 218 + def test_counts_today_only(self, tmp_path: Path): 219 + captures_dir = tmp_path / "captures" 220 + today = datetime.now().strftime("%Y%m%d") 221 + yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y%m%d") 222 + today_segment = captures_dir / today / "stream-a" / "120000_300" 223 + old_segment = captures_dir / yesterday / "stream-a" / "130000_300" 224 + today_segment.mkdir(parents=True) 225 + old_segment.mkdir(parents=True) 226 + (today_segment / "audio.flac").write_bytes(b"x") 227 + (old_segment / "audio.flac").write_bytes(b"x") 228 + 229 + observer = _make_observer(captures_dir) 230 + service = ObserverService(observer) 231 + 232 + stats = _call_method(service, "GetStats") 233 + 234 + assert stats["captures_today"].value == 1 235 + 236 + 237 + class TestSyncStatusTracking: 238 + def test_initial_status(self, tmp_path: Path): 239 + config = Config(base_dir=tmp_path) 240 + config.ensure_dirs() 241 + client = UploadClient(config) 242 + 243 + sync = SyncService(config, client) 244 + 245 + assert sync.sync_status == "synced" 246 + assert sync.sync_progress == "" 247 + 248 + def test_set_sync_status(self, tmp_path: Path): 249 + config = Config(base_dir=tmp_path) 250 + config.ensure_dirs() 251 + client = UploadClient(config) 252 + 253 + sync = SyncService(config, client) 254 + sync._set_sync_status("uploading", "uploading 120000_300") 255 + 256 + assert sync.sync_status == "uploading" 257 + assert sync.sync_progress == "uploading 120000_300" 258 + 259 + def test_set_sync_status_emits_signal(self, tmp_path: Path): 260 + config = Config(base_dir=tmp_path) 261 + config.ensure_dirs() 262 + client = UploadClient(config) 263 + 264 + sync = SyncService(config, client) 265 + sync._dbus_service = MagicMock() 266 + 267 + sync._set_sync_status("retrying", "30s until probe") 268 + 269 + sync._dbus_service.SyncProgressChanged.assert_called_once_with( 270 + "retrying:30s until probe" 271 + ) 272 + 273 + 274 + class TestObserverServiceConfig: 275 + def test_capture_dir(self, tmp_path: Path): 276 + observer = _make_observer(tmp_path / "captures") 277 + service = ObserverService(observer) 278 + 279 + assert _get_prop(service, "CaptureDir") == str(observer.config.captures_dir) 280 + 281 + def test_server_url(self): 282 + observer = _make_observer() 283 + service = ObserverService(observer) 284 + 285 + assert _get_prop(service, "ServerUrl") == "https://test.example.com" 286 + 287 + def test_stream(self): 288 + observer = _make_observer() 289 + service = ObserverService(observer) 290 + 291 + assert _get_prop(service, "Stream") == "test-stream" 292 + 293 + def test_segment_interval(self): 294 + observer = _make_observer() 295 + service = ObserverService(observer) 296 + 297 + assert _get_prop(service, "SegmentInterval") == 300

+19

tests/test_observer.py

··· 6 6 from pathlib import Path 7 7 8 8 from solstone_linux.config import Config 9 + from solstone_linux.observer import Observer 9 10 from solstone_linux.recovery import write_segment_metadata 10 11 11 12 ··· 37 38 config = Config(base_dir=tmp_path) 38 39 assert str(config.restore_token_path).endswith("restore_token") 39 40 assert "config" in str(config.restore_token_path) 41 + 42 + 43 + class TestPauseResumeState: 44 + def test_observer_init_not_paused(self, tmp_path: Path): 45 + config = Config(base_dir=tmp_path) 46 + 47 + observer = Observer(config) 48 + 49 + assert observer._paused is False 50 + assert observer._pause_until == 0.0 51 + 52 + def test_pause_state_fields_exist(self, tmp_path: Path): 53 + config = Config(base_dir=tmp_path) 54 + 55 + observer = Observer(config) 56 + 57 + assert hasattr(observer, "_paused") 58 + assert hasattr(observer, "_pause_until")

Configure Feed

Configure Feed