plugin_api_python.py

#!/usr/bin/env python3
"""Dependency-free Sony Camera Control Plugin API example."""

import argparse
import json
import os
import pathlib
import urllib.parse
import urllib.request
import time


def validated_base_url(record: dict) -> str:
    try:
        with urllib.request.urlopen(record["healthUrl"], timeout=2) as response:
            identity = json.loads(response.read().decode("utf-8"))
        if (
            identity.get("name") == "Sony Camera Control Plugin API"
            and str(identity.get("apiVersion", "")).split(".")[0] == "1"
        ):
            return str(record["apiBaseUrl"])
    except (OSError, KeyError, ValueError):
        pass
    return ""


def discover_base_url(discovery_file: str = "") -> str:
    candidates = []
    requested_file = discovery_file or os.environ.get("CAMREMOTE_DISCOVERY_FILE", "")
    if requested_file:
        candidates.append(pathlib.Path(requested_file))
    local_app_data = os.environ.get("LOCALAPPDATA", "")
    if local_app_data and not requested_file:
        candidates.append(
            pathlib.Path(local_app_data) / "SonyCameraControl" / "service.json"
        )
    for discovery in candidates:
        try:
            record = json.loads(discovery.read_text(encoding="utf-8"))
            base_url = validated_base_url(record)
            if base_url:
                return base_url
        except (OSError, KeyError, ValueError):
            pass

    if os.name == "nt":
        try:
            import winreg

            with winreg.OpenKey(
                winreg.HKEY_CURRENT_USER,
                r"Software\SonyCameraControl\Service",
            ) as key:
                record = {
                    "apiBaseUrl": winreg.QueryValueEx(key, "ApiBaseUrl")[0],
                    "healthUrl": winreg.QueryValueEx(key, "HealthUrl")[0],
                }
            base_url = validated_base_url(record)
            if base_url:
                return base_url
        except (OSError, ImportError):
            pass
    return "http://127.0.0.1:8080/api/v1"


class CamRemoteClient:
    def __init__(self, base_url: str, token: str = "") -> None:
        self.base_url = base_url.rstrip("/")
        self.token = token

    def request(self, method: str, path: str, body=None):
        data = None
        headers = {"Accept": "application/json"}
        if self.token:
            headers["Authorization"] = f"Bearer {self.token}"
        if body is not None:
            data = json.dumps(body).encode("utf-8")
            headers["Content-Type"] = "application/json"
        request = urllib.request.Request(
            self.base_url + path, data=data, headers=headers, method=method
        )
        with urllib.request.urlopen(request, timeout=35) as response:
            content_type = response.headers.get("Content-Type", "")
            payload = response.read()
            if "application/json" in content_type:
                return json.loads(payload.decode("utf-8"))
            return payload

    def download(self, path: str, destination: pathlib.Path) -> None:
        payload = self.request("GET", path)
        destination.write_bytes(payload)


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "command",
        choices=("cameras", "discover", "connect", "reconnect", "properties", "frame", "capture"),
    )
    parser.add_argument("--base", default="")
    parser.add_argument("--discovery-file", default="")
    parser.add_argument("--camera", default="active")
    parser.add_argument("--index", type=int)
    parser.add_argument("--ip", default="")
    parser.add_argument("--user", default="")
    parser.add_argument("--password", default="")
    parser.add_argument("--transport", default="")
    parser.add_argument("--token", default=os.environ.get("CAMREMOTE_TOKEN", ""))
    parser.add_argument("--output", default="frame.jpg")
    args = parser.parse_args()

    client = CamRemoteClient(
        args.base or discover_base_url(args.discovery_file), args.token
    )
    camera = urllib.parse.quote(args.camera, safe="")
    if args.command == "cameras":
        result = client.request("GET", "/cameras")
    elif args.command == "discover":
        result = client.request(
            "POST", "/discovery", {"timeoutSeconds": 2, "preferCache": True}
        )
        while not result["discovery"]["complete"]:
            time.sleep(0.25)
            result = client.request("GET", result["statusUrl"])
    elif args.command == "connect":
        body = {
            "user": args.user,
            "password": args.password,
            "transport": args.transport,
        }
        if args.ip:
            body["ip"] = args.ip
        elif args.index is not None:
            body["index"] = args.index
        else:
            parser.error("connect requires --index or --ip")
        result = client.request(
            "POST", f"/cameras/{camera}/actions/connect", body
        )
    elif args.command == "reconnect":
        result = client.request(
            "POST", f"/cameras/{camera}/actions/reconnect", {}
        )
    elif args.command == "properties":
        result = client.request("GET", f"/cameras/{camera}/properties")
    elif args.command == "capture":
        result = client.request(
            "POST", f"/cameras/{camera}/actions/capture", {"waitMs": 15000}
        )
    else:
        client.download(
            f"/cameras/{camera}/live-view/frame", pathlib.Path(args.output)
        )
        result = {"ok": True, "saved": args.output}
    print(json.dumps(result, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()