Skip to content

fix: reconnect camera stream after a drop #36

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
deepakpopli wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix: reconnect camera stream after a drop #36

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
70 changes: 52 additions & 18 deletions cyberwave/sensor/base_video.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -512,9 +512,24 @@ async def _wait_and_publish_camera_sync_frame(
async def start(self, twin_uuid: Optional[str] = None):
"""Start streaming camera to Cyberwave.

With auto_reconnect, this also starts the connection monitor so a
dropped stream is renegotiated. stop() ends it.

Args:
twin_uuid: UUID of the digital twin (uses instance twin_uuid if not provided)
"""
await self._start_webrtc(twin_uuid)

if self.auto_reconnect:
self._is_running = True
self._should_reconnect = True
if self._monitor_task is None or self._monitor_task.done():
self._monitor_task = asyncio.create_task(
self._monitor_connection(asyncio.Event())
)

async def _start_webrtc(self, twin_uuid: Optional[str] = None):
"""Set up the peer connection and send the offer."""
self._reset_state()

if twin_uuid is not None:
Expand Down Expand Up @@ -564,7 +579,20 @@ async def start(self, twin_uuid: Optional[str] = None):
self._start_health_check()

async def stop(self):
"""Stop streaming and cleanup resources.
"""Stop streaming and cleanup resources."""
self._should_reconnect = False
monitor, self._monitor_task = self._monitor_task, None
# A cancel can get swallowed while the monitor drains the sync task,
# so repeat it until the monitor is really gone
while monitor is not None and not monitor.done():
monitor.cancel()
await asyncio.wait({monitor}, timeout=0.1)

await self._close_peer_connection()
logger.info("Camera streaming stopped")

async def _close_peer_connection(self):
"""Close the peer connection and track, leaving the monitor running.

IMPORTANT: Close peer connection BEFORE stopping tracks. aiortc can segfault
if tracks are stopped before pc.close() (see aiortc/aiortc#283).
Expand Down Expand Up @@ -602,7 +630,6 @@ async def stop(self):
finally:
self.streamer = None
self._reset_state()
logger.info("Camera streaming stopped")

async def run_with_auto_reconnect(
self,
Expand Down Expand Up @@ -636,7 +663,7 @@ async def run_with_auto_reconnect(
command_callback("connecting", "Starting camera stream")
except TypeError:
pass
await self.start()
await self._start_webrtc()
self._should_reconnect = self.auto_reconnect
if command_callback:
command_callback("ok", "Camera streaming started")
Expand All @@ -648,12 +675,16 @@ async def run_with_auto_reconnect(
except TypeError:
pass

if self.auto_reconnect:
# start() may already have one running
if self.auto_reconnect and (
self._monitor_task is None or self._monitor_task.done()
):
self._monitor_task = asyncio.create_task(self._monitor_connection(stop))

_next_retry_at: float = time.monotonic() + 15.0
_retry_backoff: float = 30.0
_initial_stream_connected: bool = False
# Once connected, drops are the monitor's job, not this loop's
_initial_stream_connected: bool = self.pc is not None

try:
while not stop.is_set() and self._is_running:
Expand All @@ -667,7 +698,7 @@ async def run_with_auto_reconnect(
logger.info(
"No active camera stream — retrying offer (pc is None)..."
)
await self.start()
await self._start_webrtc()
self._should_reconnect = self.auto_reconnect
logger.info("Camera stream retry succeeded.")
_initial_stream_connected = True
Expand Down Expand Up @@ -1014,7 +1045,7 @@ async def _handle_start_command(self, callback: Optional[Callable] = None):
return

logger.info(f"Starting video stream - Recording: {self._should_record}")
await self.start()
await self._start_webrtc()
self._should_reconnect = self.auto_reconnect
logger.info("Camera streaming started successfully!")

Expand All @@ -1029,15 +1060,15 @@ async def _handle_start_command(self, callback: Optional[Callable] = None):
async def _handle_stop_command(self, callback: Optional[Callable] = None):
"""Handle stop_video command."""
try:
self._should_reconnect = False
if self.pc is None:
logger.info("Video stream not running")
if callback:
callback("ok", "Video stream not running")
return

logger.info("Stopping video stream")
self._should_reconnect = False
await self.stop()
await self._close_peer_connection()
logger.info("Camera stream stopped successfully")

if callback:
Expand All @@ -1059,19 +1090,20 @@ async def _monitor_connection(self, stop_event: asyncio.Event):
reconnect_attempt = 0

while not stop_event.is_set() and self._is_running:
if not self._should_reconnect or self.pc is None:
# pc is also None after a failed attempt, and that one needs a retry
if not self._should_reconnect or (
self.pc is None and reconnect_attempt == 0
):
await asyncio.sleep(1.0)
continue

if self._is_connection_lost():
if self.pc is None or self._is_connection_lost():
reconnect_attempt = await self._attempt_reconnect(
stop_event,
reconnect_attempt,
reconnect_delay,
max_reconnect_attempts,
)
if reconnect_attempt < 0:
break

await asyncio.sleep(1.0)

Expand Down Expand Up @@ -1123,22 +1155,23 @@ async def _attempt_reconnect(
"""Attempt to reconnect the WebRTC connection.

Returns:
New attempt count, or -1 to signal stopping
New attempt count, 0 after a success or when the stream was stopped
"""
try:
try:
await self.stop()
# Not stop(): that would cancel this task.
await self._close_peer_connection()
except Exception as e:
logger.warning(f"Error stopping old streamer during reconnect: {e}")

await asyncio.sleep(base_delay)

if not self._should_reconnect or stop_event.is_set():
logger.info("Reconnect cancelled (stream was stopped)")
return -1
return 0

logger.info(f"Reconnecting camera stream (attempt {attempt + 1})...")
await self.start()
await self._start_webrtc()
logger.info("Camera stream reconnected successfully!")
return 0

Expand All @@ -1151,8 +1184,9 @@ async def _attempt_reconnect(
f"Max reconnection attempts ({max_attempts}) reached. "
"Stopping reconnection attempts."
)
# Stay alive so a later start_video brings reconnects back
self._should_reconnect = False
return -1
return 0

backoff_delay = min(base_delay * (2**attempt), 30.0)
await asyncio.sleep(backoff_delay)
Expand Down
Loading
Loading