Labmagic 1.0: control de la Blackmagic Pocket 4K por Bluetooth, panel para OBS y licencia MIT
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Este commit está contenido en:
876
bmpcc_ble.py
Archivo normal
876
bmpcc_ble.py
Archivo normal
@@ -0,0 +1,876 @@
|
||||
"""
|
||||
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)
|
||||
Referencia en una nueva incidencia
Block a user