libervia-web: libervia/web/pages/calls/_browser/webrtc.py comparison

comparison libervia/web/pages/calls/_browser/webrtc.py @ 1564:bd3c880f4a47

browser (calls): add camera switching: - devices are listed on startup, and a `has_multiple_cameras` boolean is set - based on this data, switch camera button is hidden or not - a click/touch on the button will switch to the next available camera, and loop back to first one at the end

author	Goffi <goffi@goffi.org>
date	Fri, 18 Aug 2023 17:43:01 +0200
parents	e3449beac8d8
children	d282dbdd5ffd

comparison

equal deleted inserted replaced

-:e3449beac8d8
+:bd3c880f4a47
 bridge = Bridge()
 GATHER_TIMEOUT = 10000
 class WebRTC:
-def __init__(self):
+def __init__(
+self,
+screen_sharing_cb=None,
+on_connection_established_cb=None,
+on_reconnect_cb=None,
+on_connection_lost_cb=None,
+on_video_devices=None,
+):
 self.reset_instance()
+# ICE events
 bridge.register_signal("ice_candidates_new", self._on_ice_candidates_new)
 bridge.register_signal("ice_restart", self._on_ice_restart)
-self.on_connection_established_cb = None
-self.on_reconnect_cb = None
+# connection events callbacks
-self.on_connection_lost_cb = None
+self.on_connection_established_cb = on_connection_established_cb
+self.on_reconnect_cb = on_reconnect_cb
+self.on_connection_lost_cb = on_connection_lost_cb
+# video devices
+self.on_video_devices = on_video_devices
+self.video_devices = []
+self.has_multiple_cameras = False
+self.current_camera = None
+# Initially populate the video devices list
+aio.run(self._populate_video_devices())
+# muting
 self.is_audio_muted = None
 self.is_video_muted = None
+# screen sharing
 self._is_sharing_screen = False
-self.screen_sharing_cb = None
+self.screen_sharing_cb = screen_sharing_cb
+# video elements
 self.local_video_elt = document["local_video"]
 self.remote_video_elt = document["remote_video"]
 @property
 def is_sharing_screen(self) -> bool:
 "video": {"candidates": []},
 }
 self.media_candidates = {}
 self.candidates_gathered = aio.Future()
+async def _populate_video_devices(self):
+devices = await window.navigator.mediaDevices.enumerateDevices()
+devices_ids = set()
+self.video_devices.clear()
+for device in devices:
+if device.kind != "videoinput":
+continue
+# we can have multiple devices with same IDs in some corner cases (e.g.
+# infrared camera)
+device_id = device.deviceId
+if device_id in devices_ids:
+continue
+devices_ids.add(device_id)
+self.video_devices.append(device)
+self.has_multiple_cameras = len(self.video_devices) > 1
+if self.on_video_devices is not None:
+self.on_video_devices(self.has_multiple_cameras)
+# Set the initial camera to the default (usually front on mobile)
+if self.video_devices:
+self.current_camera = self.video_devices[0].deviceId
+log.debug(
+f"devices populated: {self.video_devices=} {self.has_multiple_cameras=}"
+)
 @property
 def media_types(self):
 if self._media_types is None:
 raise Exception("self._media_types should not be None!")
 return self._media_types
 else:
 if self.local_video_elt.srcObject:
 for track in self.local_video_elt.srcObject.getTracks():
 if track.kind == "video":
 track.stop()
-media_constraints = {"video": True}
+media_constraints = {
+"video": {"deviceId": self.current_camera}
+if self.current_camera
+else True
+}
 new_stream = await window.navigator.mediaDevices.getUserMedia(
 media_constraints
 )
 if not new_stream:
 new_video_tracks[0].bind("ended", on_track_ended)
 self.is_sharing_screen = screen
 return local_stream
+async def switch_camera(self) -> None:
+"""Switches to the next available camera.
+This method cycles through the list of available video devices, replaces the
+current video track with the next one in the user's local video element, and then
+updates the sender's track in the peer connection. If there's only one camera or
+if an error occurs while switching, the method logs the error and does nothing.
+"""
+log.info("switching camera")
+if not self.has_multiple_cameras:
+log.error("No multiple cameras to switch.")
+return
+current_camera_index = -1
+for i, device_info in enumerate(self.video_devices):
+if device_info.deviceId == self.current_camera:
+current_camera_index = i
+break
+if current_camera_index == -1:
+log.error("Current camera not found in available devices.")
+return
+# Switch to the next camera in the list
+next_camera_index = (current_camera_index + 1) % len(self.video_devices)
+self.current_camera = self.video_devices[next_camera_index].deviceId
+log.debug(f"{next_camera_index=} {self.current_camera=}")
+new_stream = await window.navigator.mediaDevices.getUserMedia(
+{"video": {"deviceId": self.current_camera}}
+)
+new_video_tracks = [
+track for track in new_stream.getTracks() if track.kind == "video"
+]
+if not new_video_tracks:
+log.error("Failed to retrieve the video track from the new stream.")
+return
+# Update local video element's stream
+local_stream = self.local_video_elt.srcObject
+if local_stream:
+local_video_tracks = [
+track for track in local_stream.getTracks() if track.kind == "video"
+]
+if local_video_tracks:
+local_video_tracks[0].stop()
+local_stream.removeTrack(local_video_tracks[0])
+local_stream.addTrack(new_video_tracks[0])
+self.local_video_elt.srcObject = local_stream
+# update remove video stream
+video_sender = next(
+(
+sender
+for sender in self._peer_connection.getSenders()
+if sender.track and sender.track.kind == "video"
+),
+None,
+)
+if video_sender:
+await video_sender.replaceTrack(new_video_tracks[0])
 async def _gather_ice_candidates(self, is_initiator: bool, remote_candidates=None):
 """Get ICE candidates and wait to have them all before returning them
 @param is_initiator: Boolean indicating if the user is the initiator of the connection

Mercurial > libervia-web

comparison libervia/web/pages/calls/_browser/webrtc.py @ 1564:bd3c880f4a47