877 líneas
40 KiB
Python
877 líneas
40 KiB
Python
"""
|
|
Control de cámaras Blackmagic (Pocket Cinema Camera 4K/6K) por Bluetooth LE.
|
|
|
|
Implementa el "Blackmagic Camera Control Protocol" sobre el servicio BLE de la
|
|
cámara. Toda la comunicación corre en un hilo propio con su bucle asyncio; la
|
|
interfaz gráfica habla con él mediante BleWorker.submit() y recibe eventos por
|
|
una cola (queue.Queue).
|
|
"""
|
|
import asyncio
|
|
import json
|
|
import math
|
|
import platform
|
|
import queue
|
|
import re
|
|
import struct
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
|
|
from bleak import BleakClient, BleakScanner
|
|
|
|
|
|
def _run_powershell(command: str) -> str:
|
|
"""Ejecuta un comando de PowerShell sin ventana y devuelve su salida en UTF-8."""
|
|
startupinfo = subprocess.STARTUPINFO()
|
|
startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
|
|
full = "[Console]::OutputEncoding = [System.Text.Encoding]::UTF8; " + command
|
|
res = subprocess.run(
|
|
["powershell", "-NoProfile", "-NonInteractive", "-Command", full],
|
|
capture_output=True,
|
|
timeout=20,
|
|
startupinfo=startupinfo,
|
|
)
|
|
return res.stdout.decode("utf-8", errors="replace")
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# UUIDs del servicio Blackmagic Camera Service
|
|
# ---------------------------------------------------------------------------
|
|
SERVICE_UUID = "291d567a-6d75-11e6-8b77-86f30ca893d3"
|
|
CH_OUT = "5dd3465f-1aee-4299-8493-d2eca2f8e1bb" # Outgoing Camera Control (write)
|
|
CH_IN = "b864e140-76a0-416a-bf30-5876504537d9" # Incoming Camera Control (indicate)
|
|
CH_TIMECODE = "6d8f2110-86f1-41bf-9afb-451d87e976c8" # Timecode (notify)
|
|
CH_STATUS = "7fe8691d-95dc-4fc5-8abd-ca74339b51b9" # Camera Status (read/write/notify)
|
|
CH_DEVNAME = "ffac0c52-c9fb-41a0-b063-cc76282eb89c" # Device Name (write)
|
|
CH_PROTO_PREFIX = "8f1fd018-b508-456f-8f82-" # Protocol Version (read); el sufijo cambia
|
|
|
|
# Tipos de dato del protocolo
|
|
T_VOID, T_INT8, T_INT16, T_INT32, T_INT64, T_STRING, T_FIXED16 = 0, 1, 2, 3, 4, 5, 128
|
|
# Operaciones
|
|
OP_ASSIGN, OP_OFFSET, OP_REPORT = 0, 1, 2
|
|
|
|
# Nombres legibles de (categoría, parámetro) para el registro
|
|
PARAM_NAMES = {
|
|
(0, 0): "Foco",
|
|
(0, 1): "Autofoco",
|
|
(0, 2): "Apertura (AV)",
|
|
(0, 3): "Apertura normalizada",
|
|
(0, 4): "Apertura (ordinal)",
|
|
(0, 5): "Auto iris",
|
|
(0, 6): "Estabilización óptica",
|
|
(0, 7): "Zoom (mm)",
|
|
(0, 8): "Zoom normalizado",
|
|
(0, 9): "Zoom continuo",
|
|
(1, 0): "Modo de vídeo",
|
|
(1, 1): "Ganancia (antigua)",
|
|
(1, 2): "Balance de blancos",
|
|
(1, 3): "Auto balance de blancos",
|
|
(1, 4): "Restaurar auto WB",
|
|
(1, 5): "Exposición (µs)",
|
|
(1, 6): "Exposición (ordinal)",
|
|
(1, 7): "Rango dinámico",
|
|
(1, 8): "Nitidez",
|
|
(1, 9): "Formato de grabación",
|
|
(1, 10): "Modo de autoexposición",
|
|
(1, 11): "Ángulo de obturador",
|
|
(1, 12): "Velocidad de obturador",
|
|
(1, 13): "Ganancia (dB)",
|
|
(1, 14): "ISO",
|
|
(1, 15): "LUT de monitor",
|
|
(1, 16): "Filtro ND",
|
|
(2, 0): "Nivel de micrófono",
|
|
(2, 1): "Nivel de auriculares",
|
|
(2, 3): "Nivel de altavoz",
|
|
(2, 4): "Tipo de entrada de audio",
|
|
(2, 5): "Niveles de entrada",
|
|
(2, 6): "Alimentación phantom",
|
|
(3, 0): "Overlays activos",
|
|
(3, 3): "Overlays",
|
|
(4, 0): "Brillo de pantalla",
|
|
(4, 1): "Ayudas de exposición/enfoque",
|
|
(4, 2): "Nivel de zebra",
|
|
(4, 3): "Nivel de peaking",
|
|
(4, 4): "Barras de color (s)",
|
|
(4, 5): "Focus assist",
|
|
(5, 0): "Brillo tally",
|
|
(6, 0): "Fuente de referencia",
|
|
(7, 0): "Reloj",
|
|
(7, 1): "Idioma",
|
|
(8, 0): "Color: lift",
|
|
(8, 1): "Color: gamma",
|
|
(8, 2): "Color: gain",
|
|
(8, 3): "Color: offset",
|
|
(8, 4): "Color: contraste",
|
|
(8, 5): "Color: luma mix",
|
|
(8, 6): "Color: tono/saturación",
|
|
(8, 7): "Color: reset",
|
|
(9, 0): "Alimentación (mV, %, ?)",
|
|
(9, 4): "Timecode",
|
|
(10, 0): "Códec",
|
|
(10, 1): "Transporte",
|
|
(12, 0): "Metadatos: reel",
|
|
(12, 1): "Metadatos: etiquetas de escena",
|
|
(12, 2): "Metadatos: escena",
|
|
(12, 3): "Metadatos: toma",
|
|
(12, 4): "Metadatos: toma buena",
|
|
(12, 5): "Metadatos: id de cámara",
|
|
(12, 6): "Metadatos: operador",
|
|
(12, 7): "Metadatos: director",
|
|
(12, 8): "Metadatos: proyecto",
|
|
(12, 9): "Objetivo: tipo",
|
|
(12, 10): "Objetivo: iris",
|
|
(12, 11): "Objetivo: focal",
|
|
(12, 12): "Objetivo: distancia",
|
|
(12, 13): "Objetivo: filtro",
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Codificación / decodificación
|
|
# ---------------------------------------------------------------------------
|
|
def enc_fixed16(value: float) -> bytes:
|
|
"""fixed16 = 5.11 con signo (valor * 2048), little-endian."""
|
|
v = int(round(value * 2048))
|
|
v = max(-32768, min(32767, v))
|
|
return struct.pack("<h", v)
|
|
|
|
|
|
def build_packet(category: int, parameter: int, dtype: int, op: int, data: bytes = b"") -> bytes:
|
|
"""Paquete del protocolo: cabecera de 4 bytes + comando 0 + datos, relleno a 4 bytes."""
|
|
length = 4 + len(data)
|
|
pkt = bytes([255, length, 0, 0, category, parameter, dtype, op]) + data
|
|
pkt += b"\x00" * ((-len(pkt)) % 4)
|
|
return pkt
|
|
|
|
|
|
def decode_data(dtype: int, data: bytes):
|
|
"""Convierte los bytes de datos a una lista de valores Python según el tipo."""
|
|
try:
|
|
if dtype == T_VOID:
|
|
return list(data)
|
|
if dtype == T_INT8:
|
|
return list(struct.unpack(f"<{len(data)}b", data))
|
|
if dtype == T_INT16:
|
|
n = len(data) // 2
|
|
return list(struct.unpack(f"<{n}h", data[: n * 2]))
|
|
if dtype == T_INT32:
|
|
n = len(data) // 4
|
|
return list(struct.unpack(f"<{n}i", data[: n * 4]))
|
|
if dtype == T_INT64:
|
|
n = len(data) // 8
|
|
return list(struct.unpack(f"<{n}q", data[: n * 8]))
|
|
if dtype == T_STRING:
|
|
return [data.decode("utf-8", errors="replace").rstrip("\x00")]
|
|
if dtype == T_FIXED16:
|
|
n = len(data) // 2
|
|
return [v / 2048.0 for v in struct.unpack(f"<{n}h", data[: n * 2])]
|
|
except struct.error:
|
|
pass
|
|
return list(data)
|
|
|
|
|
|
def parse_packets(buf: bytes):
|
|
"""Trocea una notificación en paquetes (cat, par, tipo, op, datos). Puede venir más de uno."""
|
|
out = []
|
|
i = 0
|
|
n = len(buf)
|
|
while i + 8 <= n:
|
|
length = buf[i + 1]
|
|
if length < 4:
|
|
break
|
|
cat, par, dtype, op = buf[i + 4], buf[i + 5], buf[i + 6], buf[i + 7]
|
|
data = bytes(buf[i + 8 : i + 4 + length])
|
|
out.append((cat, par, dtype, op, data))
|
|
i += 4 + length
|
|
i += (-i) % 4
|
|
return out
|
|
|
|
|
|
def decode_timecode(data: bytes) -> str:
|
|
"""Timecode BCD: bytes [FF, SS, MM, HH]; el bit alto de HH marca drop-frame."""
|
|
if len(data) < 4:
|
|
return "--:--:--:--"
|
|
f, s, m, h = data[0], data[1], data[2], data[3]
|
|
sep = ";" if h & 0x80 else ":"
|
|
return f"{h & 0x7F:02x}:{m:02x}:{s:02x}{sep}{f:02x}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Hilo con bucle asyncio
|
|
# ---------------------------------------------------------------------------
|
|
class BleWorker:
|
|
"""Ejecuta un bucle asyncio en un hilo aparte para que Tk no se bloquee."""
|
|
|
|
def __init__(self):
|
|
self.loop = asyncio.new_event_loop()
|
|
self._thread = threading.Thread(target=self._run, name="ble-loop", daemon=True)
|
|
self._thread.start()
|
|
|
|
def _run(self):
|
|
asyncio.set_event_loop(self.loop)
|
|
self.loop.run_forever()
|
|
|
|
def submit(self, coro):
|
|
"""Lanza una corrutina en el bucle y devuelve un concurrent.futures.Future."""
|
|
return asyncio.run_coroutine_threadsafe(coro, self.loop)
|
|
|
|
def stop(self):
|
|
self.loop.call_soon_threadsafe(self.loop.stop)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Cámara
|
|
# ---------------------------------------------------------------------------
|
|
class Camera:
|
|
"""Una cámara Blackmagic por BLE. Los métodos async deben correr en el bucle del worker."""
|
|
|
|
def __init__(self, events: queue.Queue):
|
|
self.events = events
|
|
self.client = None
|
|
self.device = None
|
|
self.devices = {} # address -> BLEDevice (de la última búsqueda)
|
|
self.state = {} # (cat, par) -> (dtype, data bytes, valores)
|
|
self.protocol_version = ""
|
|
self.power_on = None
|
|
self._send_lock = asyncio.Lock()
|
|
self._proto_uuid = None
|
|
self._payload_count = 0
|
|
self._payload_event = None
|
|
|
|
# ----- eventos hacia la interfaz -------------------------------------
|
|
def emit(self, kind: str, **payload):
|
|
payload["kind"] = kind
|
|
payload["t"] = time.time()
|
|
self.events.put(payload)
|
|
|
|
def log(self, text: str):
|
|
self.emit("log", text=text)
|
|
|
|
# ----- búsqueda y conexión ------------------------------------------
|
|
async def scan(self, timeout: float = 5.0, emit_event: bool = True):
|
|
self.log(f"Buscando cámaras ({timeout:.0f} s)...")
|
|
found = await BleakScanner.discover(timeout=timeout, return_adv=True)
|
|
cams = []
|
|
for address, (dev, adv) in found.items():
|
|
uuids = [u.lower() for u in (adv.service_uuids or [])]
|
|
name = dev.name or adv.local_name or ""
|
|
if SERVICE_UUID in uuids or name.startswith("A:"):
|
|
self.devices[address] = dev
|
|
cams.append({"address": address, "name": name or "(sin nombre)", "rssi": adv.rssi})
|
|
cams.sort(key=lambda c: -c["rssi"])
|
|
if emit_event:
|
|
self.emit("devices", devices=cams)
|
|
self.log(f"Anunciándose ahora: {len(cams)} cámara(s).")
|
|
return cams
|
|
|
|
async def paired_cameras(self):
|
|
"""Cámaras Blackmagic ya emparejadas en Windows (aunque no se estén anunciando).
|
|
|
|
Se consulta el árbol PnP de Windows con PowerShell: cada dispositivo BLE emparejado
|
|
aparece como BTHLE\\DEV_<MAC> con su nombre.
|
|
"""
|
|
cams = []
|
|
try:
|
|
cmd = (
|
|
"Get-PnpDevice -Class Bluetooth -ErrorAction SilentlyContinue | "
|
|
"Where-Object { $_.InstanceId -like 'BTHLE\\DEV_*' } | "
|
|
"Select-Object FriendlyName, InstanceId, Status | ConvertTo-Json -Compress"
|
|
)
|
|
out = await asyncio.to_thread(_run_powershell, cmd)
|
|
if out.strip():
|
|
items = json.loads(out)
|
|
if isinstance(items, dict):
|
|
items = [items]
|
|
for it in items:
|
|
name = (it.get("FriendlyName") or "").strip()
|
|
inst = it.get("InstanceId") or ""
|
|
if not (name.startswith("A:") or "blackmagic" in name.lower()):
|
|
continue
|
|
m = re.search(r"DEV_([0-9A-Fa-f]{12})", inst)
|
|
if not m:
|
|
continue
|
|
mac = m.group(1).upper()
|
|
addr = ":".join(mac[i : i + 2] for i in range(0, 12, 2))
|
|
cams.append({"address": addr, "name": name, "rssi": None, "paired": True})
|
|
except Exception as e: # noqa: BLE001
|
|
self.log(f"No se pudo consultar los dispositivos emparejados: {e}")
|
|
return cams
|
|
|
|
async def find_cameras(self, timeout: float = 5.0):
|
|
"""Une las emparejadas en Windows con las que se anuncian ahora mismo."""
|
|
paired = await self.paired_cameras()
|
|
seen = await self.scan(timeout, emit_event=False)
|
|
by_addr = {c["address"].upper(): dict(c, paired=False) for c in seen}
|
|
for c in paired:
|
|
key = c["address"].upper()
|
|
if key in by_addr:
|
|
by_addr[key]["paired"] = True
|
|
else:
|
|
by_addr[key] = c
|
|
cams = sorted(by_addr.values(), key=lambda c: (-(c["rssi"] or -999)))
|
|
self.emit("devices", devices=cams)
|
|
return cams
|
|
|
|
async def connect(self, address: str, name: str | None = None, fresh_retry: bool = False):
|
|
if self.client and self.client.is_connected:
|
|
await self.disconnect()
|
|
self.emit("connecting", address=address)
|
|
dev = self.devices.get(address)
|
|
if dev is None:
|
|
# En Windows basta la dirección: no hace falta que la cámara se esté anunciando
|
|
# (una cámara ya conectada a nivel de sistema deja de anunciarse).
|
|
from bleak.backends.device import BLEDevice
|
|
|
|
dev = BLEDevice(address, name, None)
|
|
elif name and not dev.name:
|
|
dev = type(dev)(dev.address, name, dev.details)
|
|
self.device = dev
|
|
self.state = {}
|
|
got_payload = False
|
|
for attempt in (1, 2):
|
|
self.client = BleakClient(dev, disconnected_callback=self._on_disconnected, timeout=30.0)
|
|
self.log(f"Conectando con {dev.name or address}..." if attempt == 1 else "Conectando de nuevo con enlace nuevo...")
|
|
try:
|
|
await self.client.connect()
|
|
got_payload = await self._setup_after_connect(address, dev)
|
|
except Exception:
|
|
c, self.client = self.client, None
|
|
try:
|
|
await c.disconnect()
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
raise
|
|
if got_payload or not fresh_retry or attempt == 2:
|
|
break
|
|
# Sin volcado inicial: el enlace Bluetooth venía reutilizado. Lo cerramos y esperamos a
|
|
# que Windows lo suelte para que la próxima conexión sea un enlace nuevo.
|
|
self.log("La cámara no ha enviado su configuración (enlace Bluetooth reutilizado). "
|
|
"Reinicio el enlace para forzarla...")
|
|
self.emit("connecting", address=address, text="Reiniciando el enlace Bluetooth para leer la configuración...")
|
|
c, self.client = self.client, None
|
|
try:
|
|
await c.disconnect()
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
if not await self._wait_link_down(address, 20.0):
|
|
self.log("Windows mantiene el enlace abierto: suele ser la ventana de Configuración > "
|
|
"Bluetooth. Ciérrala y vuelve a conectar para que lleguen todos los valores.")
|
|
self.emit("link_held", address=address)
|
|
self.emit(
|
|
"connected",
|
|
address=address,
|
|
name=dev.name or address,
|
|
protocol=self.protocol_version,
|
|
mtu=self.client.mtu_size,
|
|
payload=got_payload,
|
|
)
|
|
if got_payload:
|
|
self.log(f"Conectada. Protocolo {self.protocol_version or '?'}. Configuración recibida.")
|
|
else:
|
|
self.log(f"Conectada. Protocolo {self.protocol_version or '?'}. Sin configuración inicial: los "
|
|
"valores se irán rellenando según los toques.")
|
|
|
|
async def _wait_link_down(self, address: str, timeout: float) -> bool:
|
|
"""Espera a que Windows suelte el enlace BLE con la cámara (connection_status = 0)."""
|
|
try:
|
|
from winrt.windows.devices.bluetooth import BluetoothLEDevice
|
|
except Exception: # noqa: BLE001
|
|
await asyncio.sleep(3.0)
|
|
return True
|
|
addr = int(address.replace(":", ""), 16)
|
|
t0 = time.time()
|
|
while time.time() - t0 < timeout:
|
|
d = await BluetoothLEDevice.from_bluetooth_address_async(addr)
|
|
status = d.connection_status if d is not None else 0
|
|
if d is not None:
|
|
d.close()
|
|
if status == 0:
|
|
await asyncio.sleep(0.5)
|
|
return True
|
|
await asyncio.sleep(1.0)
|
|
return False
|
|
|
|
async def _setup_after_connect(self, address: str, dev) -> bool:
|
|
|
|
# Localizar la característica de versión de protocolo (el sufijo del UUID varía)
|
|
self._proto_uuid = None
|
|
found_service = False
|
|
for svc in self.client.services:
|
|
if svc.uuid.lower() != SERVICE_UUID:
|
|
continue
|
|
found_service = True
|
|
for ch in svc.characteristics:
|
|
if ch.uuid.lower().startswith(CH_PROTO_PREFIX):
|
|
self._proto_uuid = ch.uuid
|
|
if not found_service:
|
|
# Windows oculta el servicio cuando no puede cifrar el enlace: la cámara ya no
|
|
# reconoce el emparejamiento guardado (o nunca se emparejó).
|
|
self.emit("needs_pairing", address=address)
|
|
raise RuntimeError(
|
|
"Acceso denegado al servicio de control de la cámara. Causas, por orden: (1) otra "
|
|
"aplicación de este PC ya está conectada a la cámara (otra copia de Labmagic, un "
|
|
"script de prueba o la ventana de Configuración > Bluetooth): ciérrala y vuelve a "
|
|
"conectar; (2) la cámara ya no acepta el emparejamiento guardado: pulsa «Emparejar "
|
|
"(PIN)» y escribe el PIN que muestre la cámara."
|
|
)
|
|
if self._proto_uuid:
|
|
try:
|
|
raw = await self.client.read_gatt_char(self._proto_uuid)
|
|
self.protocol_version = raw.decode("utf-8", errors="replace").strip("\x00 ")
|
|
except Exception as e: # noqa: BLE001
|
|
self.protocol_version = "?"
|
|
self.log(f"No se pudo leer la versión de protocolo: {e}")
|
|
|
|
# Nombre con el que aparecemos en el menú Bluetooth de la cámara
|
|
try:
|
|
name = f"{platform.node()[:18]} (PC)"
|
|
await self.client.write_gatt_char(CH_DEVNAME, name.encode("utf-8"), response=True)
|
|
except Exception as e: # noqa: BLE001
|
|
self.log(f"No se pudo enviar el nombre del equipo: {e}")
|
|
|
|
# Estado (encendida/apagada)
|
|
try:
|
|
st = await self.client.read_gatt_char(CH_STATUS)
|
|
self._handle_status(bytes(st))
|
|
except Exception as e: # noqa: BLE001
|
|
msg = str(e).lower()
|
|
if "auth" in msg or "access" in msg or "0x05" in msg or "0x0f" in msg:
|
|
raise RuntimeError(
|
|
"La cámara exige emparejamiento. Empareja desde Configuración de Windows > "
|
|
"Bluetooth con el PIN que muestra la cámara y vuelve a conectar."
|
|
) from e
|
|
raise
|
|
|
|
self._payload_count = 0
|
|
self._payload_event = asyncio.Event()
|
|
await self.client.start_notify(CH_STATUS, lambda _c, d: self._handle_status(bytes(d)))
|
|
await self.client.start_notify(CH_TIMECODE, lambda _c, d: self._handle_timecode(bytes(d)))
|
|
await self.client.start_notify(CH_IN, lambda _c, d: self._handle_incoming(bytes(d)))
|
|
|
|
# Pedir el volcado completo de configuración. La cámara lo envía sola solo cuando el enlace
|
|
# Bluetooth es nuevo; escribir "Camera Ready" (0x20) con "Initial Payload Received" (0x10) a
|
|
# cero (y el bit de encendido) hace que lo reenvíe aunque Windows reutilice el enlace.
|
|
try:
|
|
await self.client.write_gatt_char(CH_STATUS, bytes([0x21]), response=True)
|
|
except Exception as e: # noqa: BLE001
|
|
self.log(f"No se pudo pedir la configuración a la cámara: {e}")
|
|
# El volcado llega en varios bloques a lo largo de unos segundos; esperamos a tener una
|
|
# cantidad razonable de parámetros (o 6 s) antes de dar la conexión por completa.
|
|
t0 = time.time()
|
|
while time.time() - t0 < 6.0 and self._payload_count < 8:
|
|
await asyncio.sleep(0.2)
|
|
return self._payload_count >= 8
|
|
|
|
async def disconnect(self):
|
|
c = self.client
|
|
self.client = None
|
|
if c is not None:
|
|
try:
|
|
await c.disconnect()
|
|
except Exception as e: # noqa: BLE001
|
|
self.log(f"Error al desconectar: {e}")
|
|
self.emit("disconnected", reason="manual")
|
|
|
|
def _on_disconnected(self, client):
|
|
# Solo cuenta si es el cliente actual (los avisos de clientes antiguos llegan con retraso)
|
|
if self.client is not None and client is self.client:
|
|
self.client = None
|
|
self.log("La cámara se ha desconectado.")
|
|
self.emit("disconnected", reason="remote")
|
|
|
|
@property
|
|
def connected(self) -> bool:
|
|
return bool(self.client and self.client.is_connected)
|
|
|
|
# ----- emparejamiento con PIN (sin pasar por Configuración de Windows) ----------
|
|
async def pair(self, address: str, ask_pin, name: str | None = None):
|
|
"""Quita el emparejamiento antiguo de Windows y empareja de nuevo con el PIN de la cámara.
|
|
|
|
ask_pin: función sin argumentos que se llama desde un hilo auxiliar y debe devolver el
|
|
PIN escrito por el usuario (o None para cancelar). Puede bloquear mientras espera.
|
|
Al terminar con éxito, conecta.
|
|
"""
|
|
from winrt.windows.devices.bluetooth import BluetoothLEDevice
|
|
from winrt.windows.devices.enumeration import (
|
|
DeviceInformation,
|
|
DevicePairingKinds,
|
|
DevicePairingProtectionLevel,
|
|
)
|
|
|
|
if self.connected:
|
|
await self.disconnect()
|
|
self.emit("pairing", stage="start", address=address)
|
|
dev = await BluetoothLEDevice.from_bluetooth_address_async(int(address.replace(":", ""), 16))
|
|
if dev is None:
|
|
raise RuntimeError("No se encuentra la cámara. ¿Está encendida y con el Bluetooth activado?")
|
|
info = await DeviceInformation.create_from_id_async(dev.device_information.id)
|
|
pairing = info.pairing
|
|
if pairing.is_paired:
|
|
self.log("Quitando el emparejamiento anterior de Windows...")
|
|
res = await pairing.unpair_async()
|
|
self.log(f"Emparejamiento anterior eliminado (estado {int(res.status)}).")
|
|
await asyncio.sleep(1.5)
|
|
info = await DeviceInformation.create_from_id_async(dev.device_information.id)
|
|
pairing = info.pairing
|
|
if not pairing.can_pair:
|
|
raise RuntimeError("Windows indica que este dispositivo no se puede emparejar ahora mismo.")
|
|
|
|
custom = pairing.custom
|
|
log = self.log
|
|
|
|
def handler(_sender, args):
|
|
kind = int(args.pairing_kind)
|
|
if kind == int(DevicePairingKinds.PROVIDE_PIN):
|
|
deferral = args.get_deferral()
|
|
|
|
def work():
|
|
try:
|
|
pin = ask_pin()
|
|
if pin:
|
|
args.accept_with_pin(str(pin).strip())
|
|
else:
|
|
log("Emparejamiento cancelado por el usuario.")
|
|
except Exception as e: # noqa: BLE001
|
|
log(f"Error al aceptar el PIN: {e}")
|
|
finally:
|
|
deferral.complete()
|
|
|
|
threading.Thread(target=work, name="pin-prompt", daemon=True).start()
|
|
elif kind == int(DevicePairingKinds.DISPLAY_PIN):
|
|
log(f"La cámara pide este PIN: {args.pin}")
|
|
args.accept()
|
|
else:
|
|
args.accept()
|
|
|
|
token = custom.add_pairing_requested(handler)
|
|
try:
|
|
try:
|
|
kinds = DevicePairingKinds(
|
|
int(DevicePairingKinds.PROVIDE_PIN)
|
|
| int(DevicePairingKinds.CONFIRM_ONLY)
|
|
| int(DevicePairingKinds.DISPLAY_PIN)
|
|
)
|
|
except Exception: # noqa: BLE001
|
|
kinds = DevicePairingKinds.PROVIDE_PIN
|
|
self.log("Solicitando emparejamiento: mira la pantalla de la cámara, mostrará un PIN.")
|
|
self.emit("pairing", stage="requested", address=address)
|
|
result = await custom.pair_with_protection_level_async(
|
|
kinds, DevicePairingProtectionLevel.ENCRYPTION_AND_AUTHENTICATION
|
|
)
|
|
finally:
|
|
custom.remove_pairing_requested(token)
|
|
|
|
status = int(result.status)
|
|
names = {
|
|
0: "emparejada", 1: "la cámara no está lista para emparejar", 2: "no emparejada",
|
|
3: "ya estaba emparejada", 4: "la cámara rechazó la conexión", 5: "demasiadas conexiones",
|
|
6: "fallo de hardware", 7: "tiempo de espera agotado (no llegó el PIN a tiempo)",
|
|
8: "autenticación no permitida", 9: "fallo de autenticación (PIN incorrecto)",
|
|
10: "sin perfiles compatibles", 11: "no se alcanzó el nivel de protección",
|
|
12: "acceso denegado", 13: "datos de ceremonia no válidos", 14: "emparejamiento cancelado",
|
|
15: "ya hay una operación en curso", 16: "falta el manejador de PIN",
|
|
17: "rechazado por el manejador", 18: "el dispositivo ya tiene asociación", 19: "fallo",
|
|
}
|
|
if status not in (0, 3):
|
|
self.emit("pairing", stage="failed", address=address, status=status)
|
|
raise RuntimeError(f"No se pudo emparejar: {names.get(status, status)} (código {status}).")
|
|
self.log("Cámara emparejada con Windows. Conectando...")
|
|
self.emit("pairing", stage="done", address=address)
|
|
await asyncio.sleep(1.0)
|
|
await self.connect(address, name)
|
|
|
|
# ----- notificaciones -----------------------------------------------
|
|
def _handle_status(self, data: bytes):
|
|
if not data:
|
|
return
|
|
self.power_on = bool(data[0] & 0x01)
|
|
self.emit("status", raw=data.hex(), power_on=self.power_on)
|
|
|
|
def _handle_timecode(self, data: bytes):
|
|
pk = parse_packets(data)
|
|
if pk:
|
|
tc = decode_timecode(pk[-1][4])
|
|
elif len(data) >= 4:
|
|
tc = decode_timecode(data[-4:])
|
|
else:
|
|
return
|
|
self.emit("timecode", timecode=tc)
|
|
|
|
def _handle_incoming(self, data: bytes):
|
|
for cat, par, dtype, op, raw in parse_packets(data):
|
|
if cat == 255 and par == 255:
|
|
# Fin de bloque del volcado inicial
|
|
if self._payload_event is not None:
|
|
self._payload_event.set()
|
|
continue
|
|
values = decode_data(dtype, raw)
|
|
self.state[(cat, par)] = (dtype, raw, values)
|
|
if cat != 9:
|
|
self._payload_count += 1
|
|
self.emit("param", cat=cat, par=par, dtype=dtype, op=op, raw=raw, values=values)
|
|
|
|
# ----- envío ----------------------------------------------------------
|
|
async def send(self, category: int, parameter: int, dtype: int, op: int, data: bytes = b""):
|
|
if not self.connected:
|
|
raise RuntimeError("No hay cámara conectada.")
|
|
pkt = build_packet(category, parameter, dtype, op, data)
|
|
async with self._send_lock:
|
|
await self.client.write_gatt_char(CH_OUT, pkt, response=True)
|
|
self.emit("sent", cat=category, par=parameter, dtype=dtype, op=op, raw=data)
|
|
|
|
# --- Objetivo --------------------------------------------------------
|
|
async def autofocus(self):
|
|
await self.send(0, 1, T_VOID, OP_ASSIGN)
|
|
|
|
async def set_focus(self, value: float):
|
|
await self.send(0, 0, T_FIXED16, OP_ASSIGN, enc_fixed16(max(0.0, min(1.0, value))))
|
|
|
|
async def offset_focus(self, delta: float):
|
|
await self.send(0, 0, T_FIXED16, OP_OFFSET, enc_fixed16(delta))
|
|
|
|
async def auto_iris(self):
|
|
await self.send(0, 5, T_VOID, OP_ASSIGN)
|
|
|
|
async def set_aperture_norm(self, value: float):
|
|
await self.send(0, 3, T_FIXED16, OP_ASSIGN, enc_fixed16(max(0.0, min(1.0, value))))
|
|
|
|
async def set_aperture_fstop(self, fnumber: float):
|
|
"""Apertura en f/: el protocolo usa el Aperture Value, AV = 2·log2(f)."""
|
|
av = 2.0 * math.log2(max(0.7, float(fnumber)))
|
|
await self.send(0, 2, T_FIXED16, OP_ASSIGN, enc_fixed16(av))
|
|
|
|
async def offset_aperture_norm(self, delta: float):
|
|
await self.send(0, 3, T_FIXED16, OP_OFFSET, enc_fixed16(delta))
|
|
|
|
async def set_zoom_norm(self, value: float):
|
|
await self.send(0, 8, T_FIXED16, OP_ASSIGN, enc_fixed16(max(0.0, min(1.0, value))))
|
|
|
|
async def set_ois(self, enabled: bool):
|
|
await self.send(0, 6, T_VOID, OP_ASSIGN, bytes([1 if enabled else 0]))
|
|
|
|
# --- Vídeo / exposición ---------------------------------------------
|
|
async def set_iso(self, iso: int):
|
|
await self.send(1, 14, T_INT32, OP_ASSIGN, struct.pack("<i", int(iso)))
|
|
|
|
async def set_shutter_angle(self, degrees: float):
|
|
await self.send(1, 11, T_INT32, OP_ASSIGN, struct.pack("<i", int(round(degrees * 100))))
|
|
|
|
async def set_shutter_speed(self, denominator: int):
|
|
await self.send(1, 12, T_INT32, OP_ASSIGN, struct.pack("<i", int(denominator)))
|
|
|
|
async def set_white_balance(self, kelvin: int, tint: int):
|
|
kelvin = max(2500, min(10000, int(kelvin)))
|
|
tint = max(-50, min(50, int(tint)))
|
|
await self.send(1, 2, T_INT16, OP_ASSIGN, struct.pack("<hh", kelvin, tint))
|
|
|
|
async def auto_white_balance(self):
|
|
await self.send(1, 3, T_VOID, OP_ASSIGN)
|
|
|
|
async def set_dynamic_range(self, mode: int):
|
|
await self.send(1, 7, T_INT8, OP_ASSIGN, bytes([mode & 0xFF]))
|
|
|
|
async def set_auto_exposure_mode(self, mode: int):
|
|
await self.send(1, 10, T_INT8, OP_ASSIGN, bytes([mode & 0xFF]))
|
|
|
|
async def set_display_lut(self, lut: int, enabled: bool):
|
|
await self.send(1, 15, T_INT8, OP_ASSIGN, bytes([lut & 0xFF, 1 if enabled else 0]))
|
|
|
|
async def set_gain_db(self, db: int):
|
|
await self.send(1, 13, T_INT8, OP_ASSIGN, struct.pack("<b", max(-128, min(127, int(db)))))
|
|
|
|
# --- Grabación -------------------------------------------------------
|
|
async def set_recording(self, recording: bool):
|
|
cur = self.state.get((10, 1))
|
|
if cur and len(cur[1]) >= 5:
|
|
data = bytearray(cur[1])
|
|
else:
|
|
data = bytearray([0, 0, 0, 0, 0])
|
|
data[0] = 2 if recording else 0 # 0 = preview, 1 = play, 2 = record
|
|
data[1] = 0 # velocidad: 0 = pausa / grabación normal
|
|
await self.send(10, 1, T_INT8, OP_ASSIGN, bytes(data))
|
|
|
|
# --- Monitor ---------------------------------------------------------
|
|
# Pantallas destino (segundo uint16 de 3.0 Overlay enables y 4.1 Exposure and focus tools)
|
|
OUT_LCD, OUT_HDMI, OUT_EVF, OUT_SDI_MAIN, OUT_SDI_FRONT = 1, 2, 4, 8, 16
|
|
|
|
async def set_tools(self, zebra: bool, focus_assist: bool, false_colour: bool, targets: int = 3):
|
|
"""4.1: [bits zebra 0x1, focus assist 0x2, falso color 0x4] + [pantallas destino]."""
|
|
bits = (1 if zebra else 0) | (2 if focus_assist else 0) | (4 if false_colour else 0)
|
|
await self.send(4, 1, T_INT16, OP_ASSIGN, struct.pack("<HH", bits, targets & 0xFFFF))
|
|
|
|
async def set_overlays(self, status: bool, frame_guides: bool, clean_feed: bool, targets: int = 3):
|
|
"""3.0: [bits texto de estado 0x1, guías 0x2, señal limpia 0x4] + [pantallas destino]."""
|
|
bits = (1 if status else 0) | (2 if frame_guides else 0) | (4 if clean_feed else 0)
|
|
await self.send(3, 0, T_INT16, OP_ASSIGN, struct.pack("<HH", bits, targets & 0xFFFF))
|
|
|
|
async def set_focus_assist(self, method: int, colour: int):
|
|
"""4.5 Focus assist: método 0 = peak, 1 = líneas de color; color 0 rojo, 1 verde, 2 azul, 3 blanco, 4 negro."""
|
|
await self.send(4, 5, T_INT8, OP_ASSIGN, bytes([method & 0xFF, colour & 0xFF]))
|
|
|
|
async def set_peaking(self, enabled: bool, colour: int, zebra: bool = False, false_colour: bool = False,
|
|
targets: int = 3):
|
|
"""Activa el focus assist en modo líneas de color y lo enciende/apaga en las ayudas (4.1)."""
|
|
if enabled:
|
|
await self.set_focus_assist(1, colour)
|
|
await self.set_tools(zebra, enabled, false_colour, targets)
|
|
|
|
# --- Formato de grabación -------------------------------------------
|
|
async def set_recording_format(self, file_fps: int, sensor_fps: int, width: int, height: int, *,
|
|
m_rate: bool = False, sensor_m_rate: bool = False, off_speed: bool = False,
|
|
windowed: bool = False, interlaced: bool = False):
|
|
"""1.9: [fps archivo, fps sensor (0 = sin cambio), ancho, alto, flags]."""
|
|
flags = ((1 if m_rate else 0) | (2 if sensor_m_rate else 0) | (4 if off_speed else 0)
|
|
| (8 if interlaced else 0) | (16 if windowed else 0))
|
|
await self.send(1, 9, T_INT16, OP_ASSIGN,
|
|
struct.pack("<hhhhh", int(file_fps), int(sensor_fps), int(width), int(height), flags))
|
|
|
|
async def set_codec(self, basic: int, variant: int):
|
|
"""10.0: códec básico (2 ProRes, 3 Blackmagic RAW) y variante."""
|
|
await self.send(10, 0, T_INT8, OP_ASSIGN, bytes([basic & 0xFF, variant & 0xFF]))
|
|
|
|
async def set_zebra_level(self, value: float):
|
|
await self.send(4, 2, T_FIXED16, OP_ASSIGN, enc_fixed16(value))
|
|
|
|
async def set_peaking_level(self, value: float):
|
|
await self.send(4, 3, T_FIXED16, OP_ASSIGN, enc_fixed16(value))
|
|
|
|
async def show_colour_bars(self, seconds: int):
|
|
await self.send(4, 4, T_INT8, OP_ASSIGN, bytes([max(0, min(30, int(seconds)))]))
|
|
|
|
# --- Color -----------------------------------------------------------
|
|
async def reset_colour_correction(self):
|
|
await self.send(8, 7, T_VOID, OP_ASSIGN)
|
|
|
|
# --- Objetivo (resto) --------------------------------------------------
|
|
async def set_zoom_mm(self, mm: int):
|
|
await self.send(0, 7, T_INT16, OP_ASSIGN, struct.pack("<h", int(mm)))
|
|
|
|
async def set_zoom_speed(self, speed: float):
|
|
"""0.9 zoom continuo: -1 (wide rápido) .. 0 (parar) .. +1 (tele rápido)."""
|
|
await self.send(0, 9, T_FIXED16, OP_ASSIGN, enc_fixed16(max(-1.0, min(1.0, speed))))
|
|
|
|
# --- Vídeo (resto) -----------------------------------------------------
|
|
async def set_exposure_us(self, us: int):
|
|
await self.send(1, 5, T_INT32, OP_ASSIGN, struct.pack("<i", max(1, min(42000, int(us)))))
|
|
|
|
async def set_sharpening(self, level: int):
|
|
"""1.8: 0 off, 1 bajo, 2 medio, 3 alto."""
|
|
await self.send(1, 8, T_INT8, OP_ASSIGN, bytes([max(0, min(3, int(level)))]))
|
|
|
|
# --- Audio -------------------------------------------------------------
|
|
async def set_mic_level(self, v: float):
|
|
await self.send(2, 0, T_FIXED16, OP_ASSIGN, enc_fixed16(max(0.0, min(1.0, v))))
|
|
|
|
async def set_headphone_level(self, v: float):
|
|
await self.send(2, 1, T_FIXED16, OP_ASSIGN, enc_fixed16(max(0.0, min(1.0, v))))
|
|
|
|
async def set_headphone_mix(self, v: float):
|
|
await self.send(2, 2, T_FIXED16, OP_ASSIGN, enc_fixed16(max(0.0, min(1.0, v))))
|
|
|
|
async def set_speaker_level(self, v: float):
|
|
await self.send(2, 3, T_FIXED16, OP_ASSIGN, enc_fixed16(max(0.0, min(1.0, v))))
|
|
|
|
async def set_audio_input_type(self, kind: int):
|
|
"""2.4: 0 micro interno, 1 línea, 2 micro nivel bajo, 3 micro nivel alto."""
|
|
await self.send(2, 4, T_INT8, OP_ASSIGN, bytes([max(0, min(3, int(kind)))]))
|
|
|
|
async def set_input_levels(self, ch0: float, ch1: float):
|
|
await self.send(2, 5, T_FIXED16, OP_ASSIGN,
|
|
enc_fixed16(max(0.0, min(1.0, ch0))) + enc_fixed16(max(0.0, min(1.0, ch1))))
|
|
|
|
async def set_phantom_power(self, on: bool):
|
|
await self.send(2, 6, T_VOID, OP_ASSIGN, bytes([1 if on else 0]))
|
|
|
|
# --- Salida: guías y rejilla (3.3) ------------------------------------
|
|
async def set_frame_guides(self, style: int, opacity: int, safe_area: int, grid_bits: int):
|
|
"""3.3: estilo 0 off, 1 2.4:1, 2 2.39:1, 3 2.35:1, 4 1.85:1, 5 16:9, 6 14:9, 7 4:3, 8 2:1, 9 4:5, 10 1:1;
|
|
opacidad 0-100; área segura 0-100 (0 = off); rejilla bits 1 tercios, 2 cruz, 4 punto, 8 horizonte."""
|
|
await self.send(3, 3, T_INT8, OP_ASSIGN,
|
|
bytes([style & 0xFF, max(0, min(100, int(opacity))), max(0, min(100, int(safe_area))),
|
|
grid_bits & 0x0F]))
|
|
|
|
# --- Pantalla (resto) --------------------------------------------------
|
|
async def set_display_brightness(self, v: float):
|
|
await self.send(4, 0, T_FIXED16, OP_ASSIGN, enc_fixed16(max(0.0, min(1.0, v))))
|
|
|
|
async def set_program_return(self, seconds: int):
|
|
await self.send(4, 6, T_INT8, OP_ASSIGN, bytes([max(0, min(30, int(seconds)))]))
|
|
|
|
async def set_timecode_source(self, source: int):
|
|
"""4.7: 0 = clip, 1 = timecode."""
|
|
await self.send(4, 7, T_INT8, OP_ASSIGN, bytes([1 if source else 0]))
|
|
|
|
# --- Tally -------------------------------------------------------------
|
|
async def set_tally(self, v: float, which: int = 0):
|
|
"""which: 0 ambos (5.0), 1 frontal (5.1), 2 trasero (5.2)."""
|
|
await self.send(5, which, T_FIXED16, OP_ASSIGN, enc_fixed16(max(0.0, min(1.0, v))))
|
|
|
|
# --- Configuración: reloj -----------------------------------------------
|
|
async def sync_clock(self):
|
|
"""7.0 reloj en BCD (hora UTC HHMMSSFF, fecha YYYYMMDD) y 7.2 zona horaria en minutos."""
|
|
now = time.gmtime()
|
|
t_bcd = int(time.strftime("%H%M%S", now) + "00", 16)
|
|
d_bcd = int(time.strftime("%Y%m%d", now), 16)
|
|
await self.send(7, 0, T_INT32, OP_ASSIGN, struct.pack("<ii", t_bcd, d_bcd))
|
|
offset_min = -time.timezone // 60 if not time.localtime().tm_isdst else -time.altzone // 60
|
|
await self.send(7, 2, T_INT32, OP_ASSIGN, struct.pack("<i", int(offset_min)))
|
|
|
|
# --- Corrección de color -----------------------------------------------
|
|
async def set_colour_rgby(self, parameter: int, values):
|
|
"""8.0 lift, 8.1 gamma, 8.2 gain, 8.3 offset: [rojo, verde, azul, luma]."""
|
|
data = b"".join(enc_fixed16(float(v)) for v in list(values)[:4])
|
|
await self.send(8, parameter, T_FIXED16, OP_ASSIGN, data)
|
|
|
|
async def set_contrast(self, pivot: float, adjust: float):
|
|
await self.send(8, 4, T_FIXED16, OP_ASSIGN, enc_fixed16(pivot) + enc_fixed16(adjust))
|
|
|
|
async def set_luma_mix(self, v: float):
|
|
await self.send(8, 5, T_FIXED16, OP_ASSIGN, enc_fixed16(max(0.0, min(1.0, v))))
|
|
|
|
async def set_hue_sat(self, hue: float, sat: float):
|
|
await self.send(8, 6, T_FIXED16, OP_ASSIGN, enc_fixed16(hue) + enc_fixed16(sat))
|
|
|
|
# --- Reproducción (10.1 modo 1 y 10.2) ---------------------------------
|
|
async def set_transport(self, mode: int, speed: int = 0, loop: bool = False, play_all: bool = False):
|
|
"""mode 0 preview, 1 play, 2 record; speed -n..n; conserva los bytes de ranuras de la cámara."""
|
|
cur = self.state.get((10, 1))
|
|
data = bytearray(cur[1]) if cur and len(cur[1]) >= 5 else bytearray([0, 0, 0, 0, 0])
|
|
data[0] = mode & 0xFF
|
|
data[1] = struct.pack("<b", max(-8, min(8, int(speed))))[0]
|
|
flags = data[2] & ~0x03
|
|
data[2] = flags | (1 if loop else 0) | (2 if play_all else 0)
|
|
await self.send(10, 1, T_INT8, OP_ASSIGN, bytes(data))
|
|
|
|
async def playback_clip(self, next_clip: bool):
|
|
await self.send(10, 2, T_INT8, OP_ASSIGN, bytes([1 if next_clip else 0]))
|
|
|
|
# --- Pizarra / metadatos (12) --------------------------------------------
|
|
async def set_reel(self, reel: int):
|
|
await self.send(12, 0, T_INT16, OP_ASSIGN, struct.pack("<h", max(0, min(999, int(reel)))))
|
|
|
|
async def set_scene_tags(self, tag: int, exterior: int, day: int):
|
|
"""tag -1 ninguna, 0 WS, 1 CU, 2 MS, 3 BCU, 4 MCU, 5 ECU; exterior 0 int/1 ext; day 0 noche/1 día."""
|
|
await self.send(12, 1, T_INT8, OP_ASSIGN, struct.pack("<bbb", int(tag), int(exterior), int(day)))
|
|
|
|
async def set_scene(self, text: str):
|
|
await self.send(12, 2, T_STRING, OP_ASSIGN, text.encode("utf-8")[:5])
|
|
|
|
async def set_take(self, number: int, tag: int):
|
|
"""tag -1 ninguna, 0 PU, 1 VFX, 2 SER."""
|
|
await self.send(12, 3, T_INT8, OP_ASSIGN, struct.pack("<bb", max(1, min(99, int(number))), int(tag)))
|
|
|
|
async def set_good_take(self, good: bool):
|
|
await self.send(12, 4, T_INT8, OP_ASSIGN, bytes([1 if good else 0]))
|
|
|
|
async def set_metadata_text(self, parameter: int, text: str, limit: int = 28):
|
|
"""12.5 id cámara, 12.6 operador, 12.7 director, 12.8 proyecto."""
|
|
await self.send(12, parameter, T_STRING, OP_ASSIGN, text.encode("utf-8")[:limit])
|
|
|
|
async def set_slate_mode(self, playback: bool):
|
|
await self.send(12, 14, T_INT8, OP_ASSIGN, bytes([1 if playback else 0]))
|
|
|
|
# --- Alimentación ----------------------------------------------------
|
|
async def power_off(self):
|
|
if not self.connected:
|
|
raise RuntimeError("No hay cámara conectada.")
|
|
await self.client.write_gatt_char(CH_STATUS, bytes([0x00]), response=True)
|