#!/usr/bin/env python3 """Capture one webcam frame through the Camera portal, then v4l2.""" import json import os import sys try: import gi gi.require_version('Gio', '2.0') from gi.repository import Gio, GLib except Exception as exc: print(json.dumps({'ok': False, 'error': f'PyGObject unavailable: {exc}'}), flush=True) raise SystemExit(1) ACTION = sys.argv[1] if len(sys.argv) > 1 else 'capture' TARGET = sys.argv[2] if len(sys.argv) > 2 else '' DEVICE = os.environ.get('JARVIS_WEBCAM_DEVICE', '').strip() bus = Gio.bus_get_sync(Gio.BusType.SESSION, None) def fail(reason): print(json.dumps({'ok': False, 'error': str(reason)}), flush=True) raise SystemExit(1) def portal_proxy(): return Gio.DBusProxy.new_sync( bus, Gio.DBusProxyFlags.NONE, None, 'org.freedesktop.portal.Desktop', '/org/freedesktop/portal/desktop', 'org.freedesktop.portal.Camera', None, ) def request_access(proxy, timeout_ms=120000): token = f'jarviscam{GLib.get_real_time()}' sender_name = bus.get_unique_name()[1:].replace('.', '_') request_path = f'/org/freedesktop/portal/desktop/request/{sender_name}/{token}' options = { 'handle_token': GLib.Variant('s', token), 'parent_window': GLib.Variant('s', ''), } loop = GLib.MainLoop() result = {'code': None} def response(_conn, _sender, _path, _interface, _member, params): result['code'] = params.unpack()[0] loop.quit() def expired(): result['code'] = 'timeout' loop.quit() return False sub = bus.signal_subscribe(None, 'org.freedesktop.portal.Request', 'Response', request_path, None, Gio.DBusSignalFlags.NONE, response) timer = GLib.timeout_add(timeout_ms, expired) try: proxy.call_sync('Access', GLib.Variant('(a{sv})', (options,)), Gio.DBusCallFlags.NONE, timeout_ms, None) if result['code'] is None: loop.run() finally: if result['code'] != 'timeout': GLib.source_remove(timer) bus.signal_unsubscribe(sub) if result['code'] not in (0, None): raise RuntimeError(f'Camera portal Access response {result["code"]}') def open_pipewire_fd(proxy): incoming = Gio.UnixFDList.new() variant, outgoing = proxy.call_with_unix_fd_list_sync( 'OpenPipeWireRemote', GLib.Variant('(a{sv})', ({},)), Gio.DBusCallFlags.NONE, 15000, incoming, None, ) if outgoing is not None and outgoing.get_length() > 0: unpacked = variant.unpack() if variant is not None else 0 index = unpacked[0] if isinstance(unpacked, (tuple, list)) else unpacked if isinstance(index, int) and 0 <= index < outgoing.get_length(): return outgoing.get(index) return outgoing.get(0) unpacked = variant.unpack() if variant is not None else None fd = unpacked[0] if isinstance(unpacked, (tuple, list)) else unpacked if isinstance(fd, int) and fd >= 0: return fd raise RuntimeError('Camera portal returned no PipeWire file descriptor') def grab_rgb(pipeline_desc, path, timeout_ms=4000): gi.require_version('Gst', '1.0') from gi.repository import Gst from PIL import Image Gst.init(None) pipeline = Gst.parse_launch(pipeline_desc) sink = pipeline.get_by_name('sink') if not sink: raise RuntimeError('GStreamer appsink missing') pipeline.set_state(Gst.State.PLAYING) change, state, _pending = pipeline.get_state(5 * Gst.SECOND) try: if change == Gst.StateChangeReturn.FAILURE or state != Gst.State.PLAYING: raise RuntimeError('webcam pipeline failed to play') sample = sink.emit('try-pull-sample', timeout_ms * Gst.MSECOND) if sample is None: sample = sink.get_property('last-sample') if sample is None: raise RuntimeError('no webcam frame yet') buf = sample.get_buffer() caps = sample.get_caps().get_structure(0) width = int(caps.get_value('width')) height = int(caps.get_value('height')) ok, mapped = buf.map(Gst.MapFlags.READ) if not ok: raise RuntimeError('could not map webcam frame') try: data = bytes(mapped.data) finally: buf.unmap(mapped) stride = max(width * 3, len(data) // max(1, height)) if stride != width * 3: rows = [data[i * stride:i * stride + width * 3] for i in range(height)] data = b''.join(rows) os.makedirs(os.path.dirname(path) or '.', exist_ok=True) Image.frombytes('RGB', (width, height), data[:width * height * 3]).save(path) return path finally: pipeline.set_state(Gst.State.NULL) def v4l2_device(): if DEVICE.startswith('/dev/video'): return DEVICE for index in range(0, 8): candidate = f'/dev/video{index}' if os.path.exists(candidate): return candidate return '/dev/video0' def pipewire_target(): if DEVICE.startswith('/dev/'): return '' safe = ''.join(ch for ch in DEVICE if ch.isalnum() or ch in '._:-') return f' target-object={safe}' if safe else '' def capture_portal(path): proxy = portal_proxy() request_access(proxy) fd = open_pipewire_fd(proxy) desc = ( f'pipewiresrc fd={int(fd)}{pipewire_target()} always-copy=true do-timestamp=true client-name=jarvis-webcam ! ' 'videoconvert ! video/x-raw,format=RGB ! appsink name=sink max-buffers=1 drop=true sync=false enable-last-sample=true' ) grab_rgb(desc, path) return 'portal' def capture_v4l2(path): device = v4l2_device() desc = ( f'v4l2src device={device} ! videoconvert ! video/x-raw,format=RGB ! ' 'appsink name=sink max-buffers=1 drop=true sync=false enable-last-sample=true' ) grab_rgb(desc, path) return 'v4l2' if ACTION == 'access': try: request_access(portal_proxy()) print(json.dumps({'ok': True, 'via': 'portal'}), flush=True) raise SystemExit(0) except Exception as exc: print(json.dumps({'ok': True, 'via': 'grant', 'warning': str(exc)}), flush=True) raise SystemExit(0) if ACTION != 'capture' or not TARGET: fail('usage: portal_camera.py access | capture ') errors = [] for grab, via_name in ((capture_portal, 'portal'), (capture_v4l2, 'v4l2')): try: via = grab(TARGET) print(json.dumps({'ok': True, 'path': TARGET, 'via': via or via_name}), flush=True) raise SystemExit(0) except Exception as exc: errors.append(f'{via_name}: {exc}') fail('; '.join(errors) or 'webcam unavailable')