Mercurial > libervia-web
changeset 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 |
files | libervia/web/pages/calls/_browser/__init__.py libervia/web/pages/calls/_browser/webrtc.py |
diffstat | 2 files changed, 152 insertions(+), 17 deletions(-) [+] |
line wrap: on
line diff
--- a/libervia/web/pages/calls/_browser/__init__.py Thu Aug 17 16:23:01 2023 +0200 +++ b/libervia/web/pages/calls/_browser/__init__.py Fri Aug 18 17:43:01 2023 +0200 @@ -34,11 +34,13 @@ class CallUI: def __init__(self): - self.webrtc = WebRTC() - self.webrtc.screen_sharing_cb = self.on_sharing_screen - self.webrtc.on_connection_established_cb = self.on_connection_established - self.webrtc.on_reconnect_cb = self.on_reconnect - self.webrtc.on_connection_lost_cb = self.on_connection_lost + self.webrtc = WebRTC( + screen_sharing_cb=self.on_sharing_screen, + on_connection_established_cb=self.on_connection_established, + on_reconnect_cb=self.on_reconnect, + on_connection_lost_cb=self.on_connection_lost, + on_video_devices=self.on_video_devices, + ) self.mode = "search" self._status = None self._callee = None @@ -71,13 +73,14 @@ ) document["mute_audio_btn"].bind("click", self.toggle_audio_mute) document["mute_video_btn"].bind("click", self.toggle_video_mute) - self.share_desktop_btn_elt = document["share_desktop_btn"] + self.share_desktop_col_elt = document["share_desktop_column"] if hasattr(window.navigator.mediaDevices, "getDisplayMedia"): - self.share_desktop_btn_elt.classList.remove("is-hidden-touch") + self.share_desktop_col_elt.classList.remove("is-hidden-touch") # screen sharing is supported - self.share_desktop_btn_elt.bind("click", self.toggle_screen_sharing) + document["share_desktop_btn"].bind("click", self.toggle_screen_sharing) else: - self.share_desktop_btn_elt.classList.add("is-hidden") + self.share_desktop_col_elt.classList.add("is-hidden") + document["switch_camera_btn"].bind("click", self.on_switch_camera) # search self.search_elt = document["search"] @@ -309,6 +312,13 @@ def on_connection_lost(self): self.status = "connection-lost" + def on_video_devices(self, has_multiple_cameras: bool) -> None: + switch_camera_col_elt = document["switch_camera_column"] + if has_multiple_cameras: + switch_camera_col_elt.classList.remove("is-hidden", "is-hidden-desktop") + else: + switch_camera_col_elt.classList.add("is-hidden") + async def make_call(self, audio: bool = True, video: bool = True) -> None: """Start a WebRTC call @@ -444,7 +454,7 @@ if not self.search_elt.value: return # clear the search field - self.search_elt.value = '' + self.search_elt.value = "" # and dispatch the input event so items are updated self.search_elt.dispatchEvent(window.Event.new("input")) @@ -510,7 +520,7 @@ def on_sharing_screen(self, sharing: bool) -> None: """Called when screen sharing state changes""" - share_desktop_btn_elt = self.share_desktop_btn_elt + share_desktop_btn_elt = document["share_desktop_btn"] if sharing: share_desktop_btn_elt.classList.add("is-danger") share_desktop_btn_elt.classList.remove(INACTIVE_CLASS, SCREEN_OFF_CLASS) @@ -518,6 +528,9 @@ share_desktop_btn_elt.classList.remove("is-danger") share_desktop_btn_elt.classList.add(INACTIVE_CLASS, SCREEN_OFF_CLASS) + def on_switch_camera(self, __) -> None: + aio.run(self.webrtc.switch_camera()) + def _on_entity_click(self, item: dict) -> None: aio.run(self.on_entity_click(item))
--- a/libervia/web/pages/calls/_browser/webrtc.py Thu Aug 17 16:23:01 2023 +0200 +++ b/libervia/web/pages/calls/_browser/webrtc.py Fri Aug 18 17:43:01 2023 +0200 @@ -15,17 +15,43 @@ 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 - self.on_connection_lost_cb = None + + # connection events callbacks + 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"] @@ -56,6 +82,30 @@ 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: @@ -318,7 +368,13 @@ 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 ) @@ -369,6 +425,72 @@ 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