diff --git a/computer_vision/default_camera.py b/computer_vision/default_camera.py new file mode 100644 index 000000000000..195c58b91243 --- /dev/null +++ b/computer_vision/default_camera.py @@ -0,0 +1,705 @@ +""" +Discover cameras, persist a default, and optionally apply it on the computer. + +Windows and Linux do not share one "default camera" API. Applications usually +open the first available capture device, so this module: + +1. Lists installed cameras. +2. Selects one by index, name, or device path. +3. Saves that choice to a JSON preference file. +4. Optionally makes it the only enabled camera (Windows) or creates a + ``/dev/video-default`` symlink (Linux). + +Command line:: + + python computer_vision/default_camera.py list + python computer_vision/default_camera.py set 1 + python computer_vision/default_camera.py set "usb" --system + python computer_vision/default_camera.py get + +https://en.wikipedia.org/wiki/Webcam +https://www.kernel.org/doc/html/latest/userspace-api/media/v4l/v4l2.html +https://learn.microsoft.com/windows-hardware/drivers/stream/camera-settings-page +""" + +from __future__ import annotations + +import argparse +import json +import os +import platform +import re +import shutil +import subprocess # noqa: S404 +from collections.abc import Callable +from dataclasses import asdict, dataclass +from pathlib import Path + +CAMERA_NAME_PATTERN = re.compile(r"camera|webcam|imaging", re.IGNORECASE) +MACOS_CAMERA_FIELDS = { + "camera", + "model id", + "unique id", + "manufacturer", + "model identifier", +} + + +@dataclass(frozen=True) +class CameraDevice: + """A capture device that can be chosen as the default camera.""" + + index: int + name: str + path: str = "" + instance_id: str = "" + enabled: bool = True + backend: str = "" + + def matches(self, identifier: str | int) -> bool: + """ + Return whether *identifier* refers to this camera. + + >>> camera = CameraDevice(1, "Logitech HD Webcam", path="/dev/video2") + >>> camera.matches(1) + True + >>> camera.matches("logitech") + True + >>> camera.matches("/dev/video2") + True + >>> camera.matches("video2") + True + >>> camera.matches(0) + False + """ + text = str(identifier).strip() + if not text: + return False + if text.isdigit() and int(text) == self.index: + return True + lowered = text.lower() + if lowered in self.name.lower(): + return True + path_name = Path(self.path).name if self.path else "" + if self.path and text in {self.path, path_name}: + return True + if self.path and self.path.endswith(text): + return True + return bool(self.instance_id) and lowered == self.instance_id.lower() + + +def video_device_index(device_name: str) -> int: + """ + Return the integer index encoded in a Video4Linux node name. + + >>> video_device_index("video0") + 0 + >>> video_device_index("video12") + 12 + >>> video_device_index("camera") + Traceback (most recent call last): + ... + ValueError: camera is not a video device name + """ + digits = "".join(character for character in device_name if character.isdigit()) + if not digits: + message = f"{device_name} is not a video device name" + raise ValueError(message) + return int(digits) + + +def default_config_path() -> Path: + """Return the user-level JSON file that stores the preferred camera.""" + if os.name == "nt": + roaming = os.environ.get("APPDATA") + base = Path(roaming) if roaming else Path.home() / "AppData" / "Roaming" + return base / "default_camera.json" + return Path.home() / ".config" / "default_camera.json" + + +def camera_from_preference(payload: dict[str, object]) -> CameraDevice: + """ + Rebuild a camera from saved preference data. + + >>> camera_from_preference({"index": 2, "name": "USB Webcam"}).name + 'USB Webcam' + """ + return CameraDevice( + index=int(str(payload.get("index", 0))), + name=str(payload.get("name", "")), + path=str(payload.get("path", "")), + instance_id=str(payload.get("instance_id", "")), + enabled=bool(payload.get("enabled", True)), + backend=str(payload.get("backend", "")), + ) + + +def save_default_camera(camera: CameraDevice, config_path: Path | None = None) -> Path: + """ + Write the selected camera to *config_path* and return that path. + + >>> import tempfile + >>> camera = CameraDevice(1, "USB Webcam", path="/dev/video2") + >>> with tempfile.TemporaryDirectory() as folder: + ... path = Path(folder) / "default_camera.json" + ... saved = save_default_camera(camera, path) + ... saved == path and load_default_camera(path).name == "USB Webcam" + True + """ + path = config_path or default_config_path() + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(asdict(camera), indent=2) + "\n", encoding="utf-8") + return path + + +def load_default_camera(config_path: Path | None = None) -> CameraDevice: + """Load the saved default camera or raise FileNotFoundError / ValueError.""" + path = config_path or default_config_path() + payload = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(payload, dict): + message = f"{path} does not contain a camera preference object" + raise ValueError(message) + return camera_from_preference(payload) + + +def select_camera(cameras: list[CameraDevice], identifier: str | int) -> CameraDevice: + """ + Return the single camera that matches *identifier*. + + >>> cameras = [ + ... CameraDevice(0, "Integrated Camera", path="/dev/video0"), + ... CameraDevice(2, "USB Webcam", path="/dev/video2"), + ... ] + >>> select_camera(cameras, "usb").name + 'USB Webcam' + >>> select_camera(cameras, 0).path + '/dev/video0' + >>> select_camera(cameras, "missing") + Traceback (most recent call last): + ... + ValueError: no camera matched 'missing' + """ + if not cameras: + message = "no cameras were found" + raise ValueError(message) + matches = [camera for camera in cameras if camera.matches(identifier)] + text = str(identifier).strip() + if text.isdigit(): + exact_index = [camera for camera in matches if camera.index == int(text)] + if len(exact_index) == 1: + return exact_index[0] + if len(matches) == 1: + return matches[0] + if len(matches) > 1: + names = ", ".join(camera.name for camera in matches) + message = f"identifier {identifier!r} matched {len(matches)} cameras: {names}" + raise ValueError(message) + message = f"no camera matched {identifier!r}" + raise ValueError(message) + + +def format_camera_list( + cameras: list[CameraDevice], default: CameraDevice | None = None +) -> str: + """ + Format cameras for display. + + >>> cameras = [CameraDevice(0, "Integrated Camera")] + >>> print(format_camera_list(cameras, cameras[0])) + [0] Integrated Camera (enabled, default) + """ + if not cameras: + return "No cameras were found." + lines: list[str] = [] + for camera in cameras: + flags = ["enabled" if camera.enabled else "disabled"] + if default is not None and ( + camera == default + or (camera.instance_id and camera.instance_id == default.instance_id) + or (camera.path and camera.path == default.path) + or (camera.index == default.index and camera.name == default.name) + ): + flags.append("default") + extra = f" {camera.path}" if camera.path else "" + lines.append(f"[{camera.index}] {camera.name}{extra} ({', '.join(flags)})") + return "\n".join(lines) + + +def parse_windows_pnp_json(payload: str) -> list[CameraDevice]: + """ + Parse ``Get-PnpDevice ... | ConvertTo-Json`` output. + + >>> payload = ( + ... '[{"Status":"OK","Class":"Camera",' + ... '"FriendlyName":"Integrated Camera","InstanceId":"USBVID1111"}]' + ... ) + >>> devices = parse_windows_pnp_json(payload) + >>> devices[0].name + 'Integrated Camera' + >>> devices[0].enabled + True + >>> parse_windows_pnp_json("") + [] + """ + text = payload.strip() + if not text: + return [] + data = json.loads(text) + if data is None: + return [] + items = [data] if isinstance(data, dict) else list(data) + cameras: list[CameraDevice] = [] + index = 0 + for item in items: + if not isinstance(item, dict): + continue + name = str(item.get("FriendlyName") or item.get("Name") or "").strip() + device_class = str(item.get("Class") or "") + if device_class.lower() != "camera" and not CAMERA_NAME_PATTERN.search(name): + continue + cameras.append( + CameraDevice( + index=index, + name=name or f"Camera {index}", + instance_id=str(item.get("InstanceId") or ""), + enabled=str(item.get("Status") or "").upper() == "OK", + backend="windows-pnp", + ) + ) + index += 1 + return cameras + + +def parse_macos_profiler(text: str) -> list[CameraDevice]: + """ + Parse ``system_profiler SPCameraDataType`` text output. + + >>> sample = "Camera:\\n\\n FaceTime HD Camera:\\n\\n Model ID: UVC" + >>> parse_macos_profiler(sample)[0].name + 'FaceTime HD Camera' + """ + cameras: list[CameraDevice] = [] + for match in re.finditer(r"^\s{4}([^:\n]+):\s*$", text, flags=re.MULTILINE): + name = match.group(1).strip() + if name.lower() in MACOS_CAMERA_FIELDS: + continue + cameras.append( + CameraDevice(index=len(cameras), name=name, backend="avfoundation") + ) + return cameras + + +def list_linux_cameras( + device_dir: Path = Path("/dev"), + sysfs_dir: Path = Path("/sys/class/video4linux"), +) -> list[CameraDevice]: + """ + List Video4Linux capture nodes. + + >>> import tempfile + >>> with tempfile.TemporaryDirectory() as folder: + ... root = Path(folder) + ... dev = root / "dev" + ... sysfs = root / "sys" + ... dev.mkdir() + ... (dev / "video0").touch() + ... (dev / "video1").touch() + ... capture = sysfs / "video0" + ... metadata = sysfs / "video1" + ... capture.mkdir(parents=True) + ... metadata.mkdir(parents=True) + ... _ = (capture / "name").write_text("Integrated Camera\\n") + ... _ = (metadata / "name").write_text("Integrated Camera: Metadata\\n") + ... cameras = list_linux_cameras(dev, sysfs) + ... [(camera.index, camera.name) for camera in cameras] + [(0, 'Integrated Camera')] + """ + cameras: list[CameraDevice] = [] + nodes = sorted( + device_dir.glob("video[0-9]*"), + key=lambda path: video_device_index(path.name), + ) + for node in nodes: + index = video_device_index(node.name) + name_file = sysfs_dir / node.name / "name" + name = ( + name_file.read_text(encoding="utf-8").strip() + if name_file.is_file() + else node.name + ) + if "metadata" in name.lower(): + continue + cameras.append( + CameraDevice( + index=index, + name=name, + path=str(node), + enabled=True, + backend="v4l2", + ) + ) + return cameras + + +def list_opencv_cameras( + max_index: int = 8, + opener: Callable[[int], object] | None = None, +) -> list[CameraDevice]: + """ + Probe sequential OpenCV capture indexes when OS listing is unavailable. + + >>> list_opencv_cameras(max_index=3, opener=lambda index: index == 1)[0].name + 'OpenCV camera 1' + """ + probe = opener + if probe is None: + try: + import cv2 + except ImportError: + return [] + + def default_opener(index: int) -> object: + capture = cv2.VideoCapture(index) + opened = bool(capture.isOpened()) + capture.release() + return opened + + probe = default_opener + cameras: list[CameraDevice] = [] + for index in range(max_index): + if not probe(index): + continue + cameras.append( + CameraDevice( + index=index, + name=f"OpenCV camera {index}", + backend="opencv", + ) + ) + return cameras + + +def _run_command(command: list[str]) -> str: + result = subprocess.run( + command, # noqa: S603 + check=False, + capture_output=True, + text=True, + timeout=30, + ) + if result.returncode != 0 and not result.stdout.strip(): + message = result.stderr.strip() or f"command failed: {' '.join(command)}" + raise OSError(message) + return result.stdout + + +def list_windows_cameras() -> list[CameraDevice]: + """List cameras through Windows PnP (PowerShell).""" + powershell = shutil.which("powershell") or shutil.which("pwsh") + if not powershell: + return [] + script = ( + "Get-PnpDevice -ErrorAction SilentlyContinue | " + "Where-Object { $_.Class -in @('Camera','Image') " + "-or $_.FriendlyName -match 'Camera|Webcam' } | " + "Select-Object Status, Class, FriendlyName, InstanceId | " + "ConvertTo-Json -Compress" + ) + return parse_windows_pnp_json( + _run_command([powershell, "-NoProfile", "-NonInteractive", "-Command", script]) + ) + + +def list_macos_cameras() -> list[CameraDevice]: + """List cameras through macOS system_profiler.""" + profiler = shutil.which("system_profiler") + if not profiler: + return [] + return parse_macos_profiler(_run_command([profiler, "SPCameraDataType"])) + + +def list_cameras() -> list[CameraDevice]: + """Discover cameras using the best backend for the current operating system.""" + system = platform.system() + discovered: list[CameraDevice] = [] + if system == "Windows": + discovered = list_windows_cameras() + elif system == "Linux": + discovered = list_linux_cameras() + elif system == "Darwin": + discovered = list_macos_cameras() + return discovered or list_opencv_cameras() + + +def windows_default_script(cameras: list[CameraDevice], selected: CameraDevice) -> str: + """ + Build a PowerShell script that enables *selected* and disables the others. + + >>> cameras = [ + ... CameraDevice(0, "Integrated Camera", instance_id=r"USB\\VID_1111"), + ... CameraDevice(1, "USB Webcam", instance_id=r"USB\\VID_2222"), + ... ] + >>> script = windows_default_script(cameras, cameras[1]) + >>> "Enable-PnpDevice" in script and "Disable-PnpDevice" in script + True + """ + if not selected.instance_id: + message = f"{selected.name} has no Windows instance id" + raise ValueError(message) + + def quote(value: str) -> str: + return "'" + value.replace("'", "''") + "'" + + lines = [ + f"Enable-PnpDevice -InstanceId {quote(selected.instance_id)} -Confirm:$false" + ] + for camera in cameras: + if camera.instance_id and camera.instance_id != selected.instance_id: + lines.append( + "Disable-PnpDevice -InstanceId " + f"{quote(camera.instance_id)} -Confirm:$false" + ) + return "\n".join(lines) + + +def linux_udev_rule(camera: CameraDevice) -> str: + """ + Return a udev rule that symlinks the selected node to ``video-default``. + + >>> "video2" in linux_udev_rule(CameraDevice(2, "USB Camera", path="/dev/video2")) + True + """ + kernel = Path(camera.path).name if camera.path else f"video{camera.index}" + return ( + "# Default camera selected by computer_vision/default_camera.py\n" + f'SUBSYSTEM=="video4linux", KERNEL=="{kernel}", SYMLINK+="video-default"\n' + ) + + +def apply_linux_default( + selected: CameraDevice, + *, + udev_path: Path | None = None, + symlink_path: Path | None = None, +) -> str: + """ + Write a udev rule and a user-level symlink for the selected Linux camera. + + >>> import tempfile + >>> with tempfile.TemporaryDirectory() as folder: + ... root = Path(folder) + ... device = root / "video2" + ... device.touch() + ... udev = root / "99-default-camera.rules" + ... link = root / "video-default" + ... camera = CameraDevice(2, "USB Camera", path=str(device)) + ... _ = apply_linux_default(camera, udev_path=udev, symlink_path=link) + ... udev.is_file() and link.resolve() == device.resolve() + True + """ + rule_file = udev_path or Path("/etc/udev/rules.d/99-default-camera.rules") + link = symlink_path or Path.home() / ".config" / "video-default" + messages: list[str] = [] + try: + rule_file.parent.mkdir(parents=True, exist_ok=True) + rule_file.write_text(linux_udev_rule(selected), encoding="utf-8") + messages.append(f"wrote udev rule {rule_file}") + except OSError as error: + messages.append(f"could not write udev rule ({error})") + if selected.path: + target = Path(selected.path) + try: + link.parent.mkdir(parents=True, exist_ok=True) + if link.is_symlink() or link.exists(): + link.unlink() + link.symlink_to(target) + messages.append(f"linked {link} -> {target}") + except OSError as error: + messages.append(f"could not create symlink ({error})") + return "; ".join(messages) + + +def apply_windows_default(selected: CameraDevice, cameras: list[CameraDevice]) -> str: + """Enable the selected PnP camera and disable the other listed cameras.""" + powershell = shutil.which("powershell") or shutil.which("pwsh") + if not powershell: + message = "PowerShell is required to change the Windows default camera" + raise OSError(message) + script = windows_default_script(cameras, selected) + _run_command([powershell, "-NoProfile", "-NonInteractive", "-Command", script]) + return ( + f"enabled {selected.name} and disabled {len(cameras) - 1} other camera(s); " + "run this program as Administrator if devices did not change" + ) + + +def apply_system_default(selected: CameraDevice, cameras: list[CameraDevice]) -> str: + """Apply the OS-level default using the current platform.""" + system = platform.system() + if system == "Windows": + return apply_windows_default(selected, cameras) + if system == "Linux": + return apply_linux_default(selected) + if system == "Darwin": + return ( + "macOS has no system-wide default camera API; " + "the preference was saved for this program" + ) + return f"system default is not supported on {system}" + + +def set_default_camera( + identifier: str | int, + cameras: list[CameraDevice] | None = None, + *, + config_path: Path | None = None, + apply_system: bool = False, +) -> CameraDevice: + """ + Select, save, and optionally apply a default camera. + + >>> import tempfile + >>> cameras = [ + ... CameraDevice(0, "Integrated Camera"), + ... CameraDevice(1, "USB Webcam"), + ... ] + >>> with tempfile.TemporaryDirectory() as folder: + ... path = Path(folder) / "default_camera.json" + ... chosen = set_default_camera("usb", cameras, config_path=path) + ... chosen.name, load_default_camera(path).index + ('USB Webcam', 1) + """ + discovered = list_cameras() if cameras is None else cameras + selected = select_camera(discovered, identifier) + save_default_camera(selected, config_path) + if apply_system: + apply_system_default(selected, discovered) + return selected + + +def get_default_camera( + cameras: list[CameraDevice] | None = None, + *, + config_path: Path | None = None, +) -> CameraDevice: + """ + Return the saved default if it still exists, otherwise the first enabled camera. + + >>> cameras = [ + ... CameraDevice(0, "Integrated Camera"), + ... CameraDevice(1, "USB Webcam"), + ... ] + >>> import tempfile + >>> with tempfile.TemporaryDirectory() as folder: + ... path = Path(folder) / "default_camera.json" + ... _ = set_default_camera(1, cameras, config_path=path) + ... get_default_camera(cameras, config_path=path).name + 'USB Webcam' + """ + discovered = list_cameras() if cameras is None else cameras + path = config_path or default_config_path() + if path.is_file(): + saved = load_default_camera(path) + identifiers: list[str | int] = [] + if saved.instance_id: + identifiers.append(saved.instance_id) + if saved.path: + identifiers.append(saved.path) + if saved.name: + identifiers.append(saved.name) + identifiers.append(saved.index) + for candidate in identifiers: + try: + return select_camera(discovered, candidate) + except ValueError: + continue + return saved + enabled = [camera for camera in discovered if camera.enabled] + if enabled: + return enabled[0] + if discovered: + return discovered[0] + message = "no cameras were found" + raise ValueError(message) + + +def _build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="List cameras and set the computer default camera." + ) + parser.add_argument( + "command", + nargs="?", + choices=("list", "get", "set"), + default="list", + help="list cameras, show the default, or set the default", + ) + parser.add_argument( + "identifier", + nargs="?", + help="camera index, name fragment, or device path (required for set)", + ) + parser.add_argument( + "--config", + type=Path, + default=None, + help="preference file (default: user config directory)", + ) + parser.add_argument( + "--system", + action="store_true", + help="also apply an OS-level default (Windows: disable other cameras; " + "Linux: udev symlink). Windows requires Administrator.", + ) + return parser + + +def main(argv: list[str] | None = None) -> int: + """Command-line entry point. Returns a process exit code.""" + args = _build_parser().parse_args(argv) + config_path: Path | None = args.config + try: + cameras = list_cameras() + saved: CameraDevice | None = None + path = config_path or default_config_path() + if path.is_file(): + saved = load_default_camera(path) + + if args.command == "list": + default = saved + if default is None and cameras: + default = get_default_camera(cameras, config_path=path) + print(format_camera_list(cameras, default)) + print(f"Preference file: {path}") + return 0 + + if args.command == "get": + camera = get_default_camera(cameras, config_path=path) + print(format_camera_list([camera], camera)) + print(f"Preference file: {path}") + return 0 + + if args.identifier is None: + message = "set requires a camera index, name, or device path" + raise ValueError(message) + selected = set_default_camera( + args.identifier, + cameras, + config_path=path, + apply_system=False, + ) + print(f"Default camera set to [{selected.index}] {selected.name}") + print(f"Preference file: {path}") + if args.system: + print(apply_system_default(selected, cameras)) + else: + print("Saved as the preferred camera for this program.") + print("Re-run with --system to change the OS default (may need admin).") + return 0 + except (OSError, ValueError, json.JSONDecodeError) as error: + print(f"error: {error}") + return 1 + + +if __name__ == "__main__": + raise SystemExit(main())