ilaandClaude Opus 4.8
ab574f1a27
Config accepts structured host/port/channel/username/password (or the existing rtsp_url_env) and builds the RTSP URL with RFC 3986 percent- encoded credentials, so passwords containing @/:/ no longer break parsing. Credentials are excluded from config_version; a guard test keeps the public example credential-free. probe_stream does a bounded connection test. 59 tests pass. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
58 lines
1.8 KiB
Python
58 lines
1.8 KiB
Python
"""RTSP URL building and a bounded connection probe for camera setup.
|
|
|
|
Kept Qt-free so URL construction (including credential percent-encoding) and the
|
|
connection test are unit-testable without a display or a real camera. Credentials
|
|
are only ever held in memory here; callers must not log or persist the built URL.
|
|
"""
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Optional
|
|
|
|
import numpy as np
|
|
|
|
from v1.config import build_rtsp_url # re-exported for the settings UI
|
|
from v1.video_source import SourceMode, SourceStatus, VideoSource
|
|
|
|
__all__ = ["build_rtsp_url", "ProbeResult", "probe_stream"]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ProbeResult:
|
|
ok: bool
|
|
message: str
|
|
frame: Optional[np.ndarray]
|
|
|
|
|
|
def probe_stream(
|
|
url: str,
|
|
attempts: int = 15,
|
|
capture_factory=None,
|
|
) -> ProbeResult:
|
|
"""Try to open the stream and grab one frame; never raise on a bad source.
|
|
|
|
Returns a decoded frame on success so the caller can show a preview. The
|
|
message is safe to display: it never contains the URL or credentials.
|
|
"""
|
|
|
|
source = VideoSource(
|
|
url,
|
|
SourceMode.STREAM,
|
|
retry_initial_seconds=0.1,
|
|
retry_max_seconds=0.2,
|
|
capture_factory=capture_factory,
|
|
)
|
|
last_error = None
|
|
try:
|
|
for _ in range(max(1, attempts)):
|
|
packet = source.read()
|
|
if packet.image is not None:
|
|
height, width = packet.image.shape[:2]
|
|
return ProbeResult(True, "连接成功 {0}x{1}".format(width, height), packet.image)
|
|
if packet.error:
|
|
last_error = packet.error
|
|
if packet.status is SourceStatus.ERROR:
|
|
return ProbeResult(False, "连接失败：{0}".format(last_error or "无法打开来源"), None)
|
|
return ProbeResult(False, "连接超时，未获取到画面", None)
|
|
finally:
|
|
source.close()
|