189 lines
6.5 KiB
Python
189 lines
6.5 KiB
Python
#!/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 <path>')
|
|
|
|
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')
|