Skip to content

Commit 029ec87

Browse files
bennyzbennyz
committed
refactor: move uStreamer onto the video driver interface
UStreamerClient was already source-agnostic apart from state(): its MJPEG proxy and CLI only spoke HTTP over the connect tunnel. Implement VideoInterface on the UStreamer driver and inherit VideoClient, so that code lives once in jumpstarter-driver-video rather than being duplicated by every video source. UStreamerState now extends VideoState, filling online/width/height/fps from ustreamer's own status document. Generic consumers can therefore read the common fields from a uStreamer source, while its richer detail stays available through the unchanged result field, and `j video state` keeps printing what it printed before. Signed-off-by: Benny Zlotnik <bzlotnik@redhat.com>
1 parent a3408cc commit 029ec87

6 files changed

Lines changed: 40 additions & 423 deletions

File tree

‎python/packages/jumpstarter-driver-ustreamer/jumpstarter_driver_ustreamer/client.py‎

Lines changed: 5 additions & 154 deletions
Original file line numberDiff line numberDiff line change
@@ -1,136 +1,22 @@
1-
import io
2-
import webbrowser
3-
from base64 import b64decode
4-
51
import click
6-
from aiohttp import web
7-
from anyio import EndOfStream, get_cancelled_exc_class, move_on_after
8-
from PIL import Image
2+
from jumpstarter_driver_video.client import VideoClient
93

104
from .common import UStreamerState
11-
from jumpstarter.client import DriverClient
12-
from jumpstarter.client.decorators import driver_click_group
13-
14-
LANDING_PAGE = """\
15-
<!DOCTYPE html>
16-
<html>
17-
<head>
18-
<title>Video</title>
19-
<style>
20-
body { background: #1a1a1a; color: #eee; font-family: system-ui; margin: 0;
21-
display: flex; flex-direction: column; align-items: center; padding: 20px; }
22-
img { max-width: 100%; border: 1px solid #333; }
23-
a { color: #6cf; }
24-
.info { margin: 10px 0; font-size: 14px; color: #aaa; }
25-
</style>
26-
</head>
27-
<body>
28-
<h2>Jumpstarter Video Stream</h2>
29-
<img src="/stream" alt="Live video stream" />
30-
<p class="info"><a href="/snapshot">Single snapshot (JPEG)</a></p>
31-
</body>
32-
</html>
33-
"""
34-
35-
36-
def _parse_content_type(header_bytes: bytes) -> str:
37-
"""Extract Content-Type from raw HTTP response headers."""
38-
for line in header_bytes.decode("ascii", errors="replace").split("\r\n"):
39-
if line.lower().startswith("content-type:"):
40-
return line.split(":", 1)[1].strip()
41-
return "multipart/x-mixed-replace; boundary=--"
42-
43-
44-
def _run_server(client, app, port, open_browser):
45-
"""Run an aiohttp app, opening the browser and blocking until Ctrl+C."""
46-
runner = web.AppRunner(app)
47-
48-
async def serve():
49-
await runner.setup()
50-
try:
51-
site = web.TCPSite(runner, "127.0.0.1", port)
52-
await site.start()
53-
54-
addresses = runner.addresses
55-
if not addresses:
56-
raise RuntimeError("Video server started without a bound address")
57-
actual_port = int(addresses[0][1])
58-
url = f"http://127.0.0.1:{actual_port}"
59-
click.echo(f"Video stream available at: {url}")
60-
click.echo(f"Snapshot endpoint: {url}/snapshot")
61-
click.echo("Press Ctrl+C to stop.")
62-
63-
if open_browser:
64-
webbrowser.open(url)
65-
66-
from anyio import sleep_forever
67-
await sleep_forever()
68-
finally:
69-
with move_on_after(2, shield=True):
70-
await runner.cleanup()
71-
72-
try:
73-
client.portal.call(serve)
74-
except KeyboardInterrupt:
75-
click.echo("\nStopping video server.")
76-
77-
78-
async def _proxy_mjpeg_stream(client, request):
79-
"""Proxy ustreamer's native MJPEG stream through the jumpstarter tunnel."""
80-
async with client.stream_async("connect") as tunnel:
81-
await tunnel.send(b"GET /stream HTTP/1.1\r\nHost: localhost\r\n\r\n")
82-
83-
buf = b""
84-
while b"\r\n\r\n" not in buf:
85-
buf += await tunnel.receive()
86-
87-
header_part, _, body_start = buf.partition(b"\r\n\r\n")
88-
89-
response = web.StreamResponse()
90-
response.content_type = _parse_content_type(header_part)
91-
await response.prepare(request)
925

93-
if body_start:
94-
await response.write(body_start)
956

96-
try:
97-
while True:
98-
chunk = await tunnel.receive()
99-
await response.write(chunk)
100-
except (EndOfStream, ConnectionResetError, ConnectionAbortedError, get_cancelled_exc_class()):
101-
pass
102-
103-
return response
104-
105-
106-
class UStreamerClient(DriverClient):
7+
class UStreamerClient(VideoClient):
1078
"""UStreamer client class
1089
109-
Client methods for the UStreamer driver.
10+
Client methods for the UStreamer driver. Inherits snapshot and
11+
streaming functionality from VideoClient.
11012
"""
11113

11214
def state(self):
11315
"""Get state of ustreamer service"""
11416
return UStreamerState.model_validate(self.call("state"))
11517

116-
def snapshot(self):
117-
"""Get a snapshot image from the video input
118-
119-
:return: PIL Image object of the snapshot image
120-
:rtype: PIL.Image
121-
"""
122-
input_jpg_data = b64decode(self.call("snapshot"))
123-
return Image.open(io.BytesIO(input_jpg_data))
124-
125-
def snapshot_bytes(self):
126-
"""Get raw JPEG bytes from the video input"""
127-
return b64decode(self.call("snapshot"))
128-
12918
def cli(self):
130-
@driver_click_group(self)
131-
def video():
132-
"""Video capture and streaming"""
133-
pass
19+
video = super().cli()
13420

13521
@video.command()
13622
def state():
@@ -143,39 +29,4 @@ def state():
14329
click.echo(f"FPS: {src.captured_fps}/{src.desired_fps}")
14430
click.echo(f"Encoder: {enc.type} (quality: {enc.quality})")
14531

146-
@video.command()
147-
@click.option("-o", "--output", default="snapshot.jpg", help="Output file path")
148-
def snapshot(output):
149-
"""Save a single snapshot to file"""
150-
img = self.snapshot()
151-
img.save(output)
152-
click.echo(f"Saved snapshot to {output}")
153-
154-
@video.command()
155-
@click.option("-p", "--port", default=0, type=int, help="Local server port (0 = auto)")
156-
@click.option("--browser/--no-browser", default=True, help="Open in web browser")
157-
def stream(port, browser):
158-
"""Start local MJPEG streaming server
159-
160-
Proxies ustreamer's native MJPEG stream through the jumpstarter
161-
tunnel. Frame rate is controlled by ustreamer's configuration.
162-
"""
163-
164-
async def handle_index(request):
165-
return web.Response(text=LANDING_PAGE, content_type="text/html")
166-
167-
async def handle_snapshot(request):
168-
data = b64decode(await self.call_async("snapshot"))
169-
return web.Response(body=data, content_type="image/jpeg")
170-
171-
async def handle_stream(request):
172-
return await _proxy_mjpeg_stream(self, request)
173-
174-
app = web.Application()
175-
app.router.add_get("/", handle_index)
176-
app.router.add_get("/snapshot", handle_snapshot)
177-
app.router.add_get("/stream", handle_stream)
178-
179-
_run_server(self, app, port, browser)
180-
18132
return video

0 commit comments

Comments
 (0)