irl4you-pip/dji_daemon.py at v1.0.71 · IRL4YOU/irl4you-pip

Repository navigation

10BC8

#!/usr/bin/env python3

import argparse

import asyncio

import hmac

import json

import logging

import os

import re

import secrets

import signal

import struct

import subprocess

import time

import urllib.request

try:

from bleak import BleakClient, BleakScanner

except ImportError:

BleakClient = BleakScanner = None

import dji

import controllers

import phone_battery

LISTEN_HOST, LISTEN_PORT = "127.0.0.1", 9101

RTMP_PORT = 1935

RTMP_APP = "publish"

STAT_URL = "http://127.0.0.1:1936/"

ONBOARD_USB_ID = "13d3:3572"

log = logging.getLogger("pipbox-dji")

def _bleak_major():

try:

from importlib.metadata import version

return int(version("bleak").split(".")[0])

except Exception:

return 0

BLEAK_MAJOR = _bleak_major()

FIRST_BYTE = 0x55

VERSION = 0x04

def crc_generic(data, width, poly, init, ref_in=True, ref_out=True, xor=0):

mask = (1 << width) - 1

top = 1 << (width - 1)

crc = init

if ref_in:

rpoly = int("{:0{w}b}".format(poly, w=width)[::-1], 2)

rinit = int("{:0{w}b}".format(init, w=width)[::-1], 2)

crc = rinit

for b in data:

crc ^= b

for _ in range(8):

crc = (crc >> 1) ^ rpoly if crc & 1 else crc >> 1

if not ref_out:

crc = int("{:0{w}b}".format(crc, w=width)[::-1], 2)

return (crc ^ xor) & mask

for b in data:

crc ^= b << (width - 8)

for _ in range(8):

crc = ((crc << 1) ^ poly) & mask if crc & top else (crc << 1) & mask

if ref_out:

crc = int("{:0{w}b}".format(crc, w=width)[::-1], 2)

return (crc ^ xor) & mask

def crc8(data):

return crc_generic(data, 8, 0x31, 0xEE, True, True, 0x00)

def crc16(data):

return crc_generic(data, 16, 0x1021, 0x496C, True, True, 0x0000)

def pack_string(value):

b = value.encode("utf-8")

return bytes([len(b) & 0xFF]) + b

def pack_url(url):

b = url.encode("utf-8")

return bytes([len(b) & 0xFF, 0]) + b

class Message:

def __init__(self, target, mid, mtype, payload=b""):

self.target = target

self.id = mid

self.type = mtype

self.payload = bytes(payload)

def encode(self):

out = bytearray([FIRST_BYTE, (13 + len(self.payload)) & 0xFF, VERSION])

out.append(crc8(bytes(out)))

out += self.target.to_bytes(2, "little")

out += self.id.to_bytes(2, "little")

out += self.type.to_bytes(3, "little")

out += self.payload

out += crc16(bytes(out)).to_bytes(2, "little")

return bytes(out)

@staticmethod

def decode(data):

data = bytes(data)

if len(data) < 13 or data[0] != FIRST_BYTE:

raise ValueError("falsches erstes Byte oder zu kurz")

if data[1] != len(data):

raise ValueError("falsche Länge")

if data[2] != VERSION:

raise ValueError("falsche Version")

if data[3] != crc8(data[0:3]):

raise ValueError("Kopf-Prüfsumme falsch")

if int.from_bytes(data[-2:], "little") != crc16(data[:-2]):

raise ValueError("Prüfsumme falsch")

return Message(int.from_bytes(data[4:6], "little"),

int.from_bytes(data[6:8], "little"),

int.from_bytes(data[8:11], "little"),

data[11:-2])

def __repr__(self):

return "Message(target=0x%04x id=0x%04x type=0x%06x payload=%s)" % (

self.target, self.id, self.type, self.payload.hex())

ID_PAIR, ID_STOP, ID_PREPARE, ID_WIFI, ID_START, ID_CONFIGURE = 0x8092, 0xEAC8, 0x8C12, 0x8C19, 0x8C2C, 0x8C2D

T_PAIR, T_STOP, T_PREPARE, T_WIFI, T_CONFIGURE, T_START = 0x0702, 0x0802, 0x0802, 0x0702, 0x0102, 0x0802

TY_PAIR, TY_STOP, TY_PREPARE, TY_WIFI, TY_CONFIGURE, TY_START, TY_STATUS = \

0x450740, 0x8E0240, 0xE10240, 0x470740, 0x8E0240, 0x780840, 0x020D00

PAIR_PAYLOAD = bytes([0x20]) + b"284ae5b8d76b3375a04a6417ad71bea3"

PAIR_PIN = "mbln"

STOP_PAYLOAD = bytes([0x01, 0x01, 0x1A, 0x00, 0x01, 0x02])

CONFIRM_PAYLOAD = bytes([0x01, 0x01, 0x1A, 0x00, 0x01, 0x01])

MODELS = {

0x0010: ("Osmo Action 2", "action23"),

0x0012: ("Osmo Action 3", "action23"),

0x0014: ("Osmo Action 4", "action4"),

0x0015: ("Osmo Action 5 Pro", "action5"),

0x0017: ("Osmo 360", "action5"),

0x0018: ("Osmo Action 6", "action6"),

0x0020: ("Osmo Pocket 3", "pocket3"),

0x0021: ("Osmo Pocket 4", "pocket4"),

}

POCKET4_PRO = ("Osmo Pocket 4 Pro", "pocket4")

POCKET4_PRO_TYPE = b"\xda\x00"

NEW_PROTOCOL = ("action5", "action6", "pocket4")

CONFIGURE_KINDS = {"action4": 0x08, "action6": 0x08, "action5": 0x1A}

ACTION2_NAME = "Osmo Action 2"

POWER_FROM_CURRENT = {"action4": (5, -100), "action5": (5, -100), "action6": (5, -100), "pocket3": (5, -100)}

POWER_NOT_KNOWN_MODELS = ("360",)

def power_spec(cfg):

if any(m in str(cfg.get("model", "")) for m in POWER_NOT_KNOWN_MODELS):

return None

return POWER_FROM_CURRENT.get(cfg_kind(cfg))

RESOLUTIONS = {"480p": 0x47, "720p": 0x04, "1080p": 0x0A}

FPS = {25: 2, 30: 3}

STABILIZATION = {"off": 0, "rocksteady": 1, "horizonsteady": 2, "rocksteadyplus": 3, "horizonbalancing": 4}

DJI_COMPANY_IDS = (0x08AA, 0xF7AA)

def model_from_manufacturer_data(manufacturer_data):

for cid, payload in manufacturer_data.items():

if cid in DJI_COMPANY_IDS:

mid = int.from_bytes(payload[0:2], "little") if len(payload) >= 2 else -1

if mid == 0 and len(payload) >= 12 and bytes(payload[10:12]) == POCKET4_PRO_TYPE:

return (mid,) + POCKET4_PRO

name, kind = MODELS.get(mid, ("DJI-Gerät (unbekanntes Modell)", "unknown"))

return mid, name, kind

return None

def build_start_payload(kind, rtmp_url, resolution, bitrate_kbps, fps):

res = RESOLUTIONS[resolution]

fpsb = FPS.get(fps, 3)

if kind in ("action6", "pocket4"):

header, middle = ((b"\x01\x9c\x00", b"\xfe\x00") if kind == "action6"

else (b"\x01\xb5\x00", b"\x02\x01"))

js = ('{"codec":"AVC","EnhancedRTMP":false,"supportStopLive":false,"watermark":0,'

'"rtmpAddress":%s,"orientation":"landscape"}' % json.dumps(rtmp_url).replace("/", "\\/")).encode()

return (header + bytes([res]) + (bitrate_kbps & 0xFFFF).to_bytes(2, "little") + middle +

bytes([fpsb]) + b"\x00\x00\x00" + len(js).to_bytes(2, "little") + js)

oa5 = 0x2A if kind in NEW_PROTOCOL else 0x2E

return (b"\x00" + bytes([oa5]) + b"\x00" + bytes([res]) +

(bitrate_kbps & 0xFFFF).to_bytes(2, "little") + b"\x02\x00" + bytes([fpsb]) +

b"\x00\x00\x00" + pack_url(rtmp_url))

def camera_rtmp_url(kind, rtmp_url):

return rtmp_url + "?c=6" if kind == "action6" and "?" not in rtmp_url else rtmp_url

def build_configure_payload(kind, stabilization):

return b"\x01\x01" + bytes([CONFIGURE_KINDS[kind]]) + b"\x00\x01" + bytes([STABILIZATION.get(stabilization, 0)])

def run(cmd):

try:

return subprocess.run(cmd, capture_output=True, text=True, timeout=10).stdout.strip()

except Exception:

return ""

SYSFS_NET = "/sys/class/net"

IFNAME_RE = re.compile(r"^[A-Za-z0-9_.@-]{1,15}$")

def iface_mac(name, root=None):

if not isinstance(name, str) or not IFNAME_RE.match(name):

return ""

try:

with open(os.path.join(root or SYSFS_NET, name, "address")) as f:

mac = f.read().strip().lower()

except OSError:

return ""

return mac if re.fullmatch(r"([0-9a-f]{2}:){5}[0-9a-f]{2}", mac) and mac != "00:00:00:00:00:00" else ""

def iface_by_mac(mac, root=None):

if not mac:

return None

try:

names = sorted(os.listdir(root or SYSFS_NET))

except OSError:

return None

for n in names:

if iface_mac(n, root) == mac:

return n

return None

def other_connections(skip):

found = []

for line in run(["ip", "-4", "-o", "addr", "show"]).splitlines():

parts = line.split()

if len(parts) < 4:

continue

ifname, ip = parts[1], parts[3].split("/")[0]

continue

if ip.startswith("169.254."):

continue

found.append({"ifname": ifname, "ssid": "", "password": "", "ip": ip, "type": "other"})

return found

HOTSPOT_FILE = "/var/lib/pipbox/hotspot.json"

def hotspot_password(ifname, ssid):

try:

with open(HOTSPOT_FILE) as f:

h = json.load(f).get(ifname)

except (OSError, ValueError, AttributeError):

return ""

if isinstance(h, dict) and h.get("ssid") == ssid and isinstance(h.get("password"), str):

return h["password"]

return ""

def nm_wifi_options():

options = []

for line in run(["nmcli", "-t", "-f", "DEVICE,TYPE,STATE,CONNECTION", "dev"]).splitlines():

parts = line.replace("\\:", "\x00").split(":")

if len(parts) < 4 or parts[1] != "wifi" or parts[2] != "connected":

continue

ifname, conn = parts[0], parts[3].replace("\x00", ":")

ssid = run(["nmcli", "-g", "802-11-wireless.ssid", "connection", "show", conn])

mode = run(["nmcli", "-g", "802-11-wireless.mode", "connection", "show", conn])

psk = run(["nmcli", "-s", "-g", "802-11-wireless-security.psk", "connection", "show", conn])

secured = bool(run(["nmcli", "-g", "802-11-wireless-security.key-mgmt", "connection", "show", conn]))

if mode == "ap" and secured and not psk and conn.startswith("pipbox-hotspot-"):

psk = hotspot_password(ifname, ssid)

ip = run(["nmcli", "-g", "IP4.ADDRESS", "dev", "show", ifname]).split("/")[0].split("\n")[0]

if not ssid or not ip:

continue

options.append({"ifname": ifname, "ssid": ssid, "password": psk, "ip": ip,

"type": "hotspot" if mode == "ap" else "client",

"secret_missing": bool(secured and not psk)})

options += other_connections({o["ifname"] for o in options})

return options

def public_option(o):

return {k: o[k] for k in ("ifname", "ssid", "ip", "type") if k in o} | {"secret_missing": bool(o.get("secret_missing"))}

def rtmp_publishing(key, stat_url=None):

try:

with urllib.request.urlopen(stat_url or STAT_URL, timeout=2) as r:

body = r.read().decode("utf-8", "replace")

except Exception:

return None

m = re.search(r"<stream>\s*<name>\s*%s\s*</name>[\s\S]*?</stream>" % re.escape(key), body)

return bool(m and "<publishing/>" in m.group(0))

def rtmp_video_kbps(key, stat_url=None):

try:

with urllib.request.urlopen(stat_url or STAT_URL, timeout=2) as r:

body = r.read().decode("utf-8", "replace")

except Exception:

return None

m = re.search(r"<stream>\s*<name>\s*%s\s*</name>[\s\S]*?</stream>" % re.escape(key), body)

v = re.search(r"<bw_video>(\d+)</bw_video>", m.group(0)) if m else None

return int(v.group(1)) // 1000 if v else None

def preferred_adapter():

try:

names = [n for n in os.listdir(dji.SYSFS_BT) if re.fullmatch(r"hci\d+", n)]

except OSError:

return None

if not names:

return None

names.sort(key=lambda n: (dji.usb_id_for_hci(n) == ONBOARD_USB_ID, int(n[3:])))

return names[0]

def adapter_address(hci):

if not hci:

return None

try:

with open(os.path.join(dji.SYSFS_BT, hci, "address")) as f:

a = f.read().strip().upper()

return a if MAC_RE.match(a) else None

except OSError:

return None

async def bluez_cleanup(addr, remove=True):

for args in ((["disconnect", addr], ["remove", addr]) if remove else (["disconnect", addr],)):

try:

p = await asyncio.create_subprocess_exec("bluetoothctl", *args,

stdout=asyncio.subprocess.DEVNULL,

stderr=asyncio.subprocess.DEVNULL)

await asyncio.wait_for(p.wait(), 8)

except Exception:

pass

def make_ble_device(state, addr):

from bleak.backends.device import BLEDevice

details = {"path": state["path"], "props": {"Address": addr, "Name": state.get("name", "")}}

try:

return BLEDevice(addr, state.get("name") or None, details)

except TypeError:

return BLEDevice(addr, state.get("name") or None, details, -60)

def cfg_kind(cfg):

return cfg.get("kind", "unknown")

def friendly_error(e):

text = str(e).lower()

if isinstance(e, (asyncio.TimeoutError, TimeoutError)) and not text:

return "Die Kamera hat auf den Bluetooth-Verbindungsaufbau nicht geantwortet (Zeitüberschreitung)"

if "not found" in tex 10000 t and ("device" in text or "dev_" in text):

return "Die Kamera ist für Bluetooth nicht mehr sichtbar (aus, im Ruhemodus oder mit dem Handy verbunden?)"

if "failed to discover services" in text or "device disconnected" in text:

return "Die Kamera hat die Bluetooth-Verbindung beim Einrichten beendet"

if "le-connection-abort-by-local" in text:

return "Der Bluetooth-Verbindungsaufbau wurde abgebrochen (Funk gestört?)"

return None

STATUS_ONLY_KEY = "hdmi"

def camera_key(addr, taken=()):

hexa = re.sub(r"[^0-9a-f]", "", addr.lower())

for n in (6, 8, 10, 12):

key = "dji-" + hexa[-n:]

if key not in taken:

return key

return "dji-" + hexa

class CameraError(Exception):

pass

class Camera:

AFTER_LINK_SECONDS = 90

OLD_STREAM_WAIT_S = 25

RETRY_SCHEDULE = (8, 8, 15, 30)

SEARCH_SECONDS = 15

STREAM_CHECK_SECONDS = 5

STREAM_LOST_SECONDS = 30

NO_VIDEO_SECONDS = 25

CONNECT_SETTLE = 1.0

@property

def manual_off(self):

return bool(self.cfg.get("manual_off"))

@manual_off.setter

def manual_off(self, value):

if value:

self.cfg["manual_off"] = True

else:

self.cfg.pop("manual_off", None)

def __init__(self, daemon, addr, cfg):

self.daemon = daemon

self.addr = addr

self.cfg = cfg

self.state = "idle"

self.detail = ""

self.battery = None

self.battery_at = 0.0

self.charging = None

self._status_prev = None

self._status_flips = {}

self.task = None

self.stop_requested = False

self.last_seen = 0

self.last_rx = 0

self.fail_count = 0

self.retry_at = 0

self.wake = asyncio.Event()

self.switch_until = 0.0

self.publishing = False

self.force_cleanup = False

self.typed_network = True

self._scanner = None

self._search_started = 0.0

self._client = None

self.link_ended = 0.0

self._reusing = False

self.reuse_failed = False

self._resume_sig = None

self.leaving = False

def locked(self):

return self.state not in ("idle", "error", "searching", "connecting") or self.publishing

def mode_locked(self):

return self.state not in ("idle", "error") or self.publishing

def switch_wait(self):

left = self.switch_until - time.monotonic()

return 0 if left <= 0 or self.state == "error" else int(left) + 1

def set_state(self, state, detail=""):

if state != self.state or detail != self.detail:

log.info("%s: %s %s", self.addr, state, detail)

self.state, self.detail = state, detail

if state == "streaming":

key = self.cfg.get("rtmp_key")

if key:

self.daemon.feed_hold(key, 0)

def public(self):

c = dict(self.cfg)

c.pop("password", None)

mac = c.pop("wifi_mac", "")

if mac and iface_by_mac(mac):

c["wifi_ifname"] = iface_by_mac(mac)

c["saved"] = [n["ssid"] for n in self.cfg.get("saved", [])]

retry_in = max(0, int(self.retry_at - time.time())) if self.retry_at else 0

c.update({"addr": self.addr, "state": self.state, "detail": self.detail,

"battery": self.battery, "charging": self.charging,

"battery_age": int(time.time() - self.battery_at) if self.battery_at else None,

"in_range": time.time() - self.last_seen < 30, "manual_off": bool(self.manual_off),

"retry_in": retry_in, "publishing": self.publishing, "locked": self.locked(), "mode_locked": self.mode_locked(),

"switch_wait": self.switch_wait()})

return c

def network_for_unreadable(self, o):

cfg, ifname, want = self.cfg, o["ifname"], o["ssid"]

typed = cfg.get("ssid") and cfg.get("password")

if typed and cfg["ssid"] == want:

cfg["ssid_if"] = ifname

return cfg["ssid"], cfg["password"]

if typed and cfg.get("ssid_if") == ifname:

return cfg["ssid"], cfg["password"]

cands = [n for n in cfg.get("saved", []) if isinstance(n, dict) and n.get("ssid") == want and n.get("password")]

m = self.daemon.by_conn.get(ifname)

if not cands and isinstance(m, dict) and m.get("ssid") == want and m.get("password"):

cands = [m]

if cands:

cfg["ssid"], cfg["password"], cfg["ssid_if"] = cands[-1]["ssid"], cands[-1]["password"], ifname

return cfg["ssid"], cfg["password"]

if typed and not cfg.get("ssid_if"):

return cfg["ssid"], cfg["password"]

raise CameraError('Die Box ist über %s im WLAN "%s", dessen Passwort sich nicht aus NetworkManager lesen lässt. Die Kamera muss diesem Netz beitreten: '

'Bitte in der DJI-Karte WLAN-Name und Passwort von Hand eintragen (oder ein gespeichertes Netz wählen).' % (ifname, want))

def resolve_target(self):

cfg = self.cfg

self.typed_network = True

ifname = cfg.get("wifi_ifname")

mac = cfg.get("wifi_mac")

if mac and ifname != "manual":

found = iface_by_mac(mac)

if not found:

raise CameraError("Die gewählte Verbindung (Router) ist nicht da. Router prüfen: eingeschaltet, per Kabel oder USB verbunden, "

"im USB-Modus? Die Kamera bleibt bei dieser Verbindung und wechselt nicht ins Heimnetz.")

if found != ifname:

log.info("%s: die gewählte Verbindung heißt jetzt %s (vorher %s)", self.addr, found, ifname)

cfg["wifi_ifname"] = ifname = found

if ifname and ifname != "manual":

for o in self.daemon.wifi_options():

if o["ifname"] == ifname:

if o["type"] == "other":

ssid, pw, ip = cfg.get("ssid") or o["ssid"], cfg.get("password", ""), o["ip"]

elif o.get("secret_missing"):

ssid, pw = self.network_for_unreadable(o)

ip = o["ip"]

else:

self.typed_network = False

ssid, pw, ip = o["ssid"], o["password"], o["ip"]

break

else:

raise CameraError("Die Verbindung %s ist nicht aktiv. Router prüfen: eingeschaltet, per USB verbunden, "

"im USB-Modus?" % ifname)

else:

ssid, pw, ip = cfg.get("ssid", ""), cfg.get("password", ""), cfg.get("ip", "")

if not ssid or not ip:

raise CameraError("Für die Kamera ist keine Verbindung gewählt (WLAN-Name oder Adresse der Box fehlt)")

key = cfg.get("rtmp_key") or "cam1"

return ssid, pw, "rtmp://%s:%d/%s/%s" % (ip, self.daemon.rtmp_port, self.daemon.rtmp_app, key)

async def find_device(self):

self._reusing = False

if not self.reuse_failed and not self.stop_requested:

try:

st = await asyncio.get_event_loop().run_in_executor(None, dji.bluez_device_state, self.addr)

except Exception:

st = None

if st and st.get("connected"):

log.info("%s: BlueZ hält die Kamera noch verbunden, die Verbindung wird weiterverwendet", self.addr)

self._reusing = True

self.force_cleanup = False

self._search_started = time.time()

return make_ble_device(st, self.addr)

if self.fail_count >= 2 or self.force_cleanup:

self.force_cleanup = False

await bluez_cleanup(self.addr)

self._search_started = time.time()

scanner = BleakScanner(**self.daemon.scan_kwargs())

try:

await scanner.start()

except Exception as e:

raise CameraError("Die Bluetooth-Suche konnte nicht gestartet werden: %s" % e)

self._scanner = scanner

end = time.time() + self.SEARCH_SECONDS

while time.time() < end and not self.stop_requested:

try:

for addr, (dev, _adv) in scanner.discovered_devices_and_advertisement_data.items():

if str(addr).upper() == self.addr.upper():

self.last_seen = time.time()

return dev

except Exception as e:

log.info("%s: Suchergebnis nicht lesbar: %s", self.addr, e)

await asyncio.sleep(0.2)

await self.stop_scan()

return None

async def stop_scan(self):

s, self._scanner = self._scanner, None

if s is not None:

try:

await s.stop()

except Exception as e:

log.info("%s: Suche beenden fehlgeschlagen: %s", self.addr, e)

async def session(self):

cfg = self.cfg

model_kind = cfg.get("kind", "unknown")

if BleakClient is None:

raise CameraError("Die Bluetooth-Bibliothek (bleak) ist nicht installiert. install.sh erneut ausführen.")

if cfg.get("model") == ACTION2_NAME:

raise CameraError("Osmo Action 2: Die Fernsteuerung funktioniert laut Moblin nicht.")

loop = asyncio.get_event_loop()

if cfg.get("status_only"):

ssid = password = rtmp_url = ""

else:

ssid, password, rtmp_url = await loop.run_in_executor(None, self.resolve_target)

key = cfg.get("rtmp_key") or "cam1"

if self.daemon.adapter_missing():

prob = dji.adapter_problems([], dji.usb_bluetooth_devices())

raise CameraError(prob[0]["hint"] if prob else "Kein Bluetooth-Adapter gefunden")

if self.daemon.conn_lock.locked():

self.set_state("connecting", "Wartet, bis eine andere Kamera fertig verbunden ist")

await self.daemon.conn_lock.acquire()

held = [True]

loop2 = asyncio.get_event_loop()

async def release():

await self.stop_scan()

if held[0]:

held[0] = False

loop2.call_later(self.CONNECT_SETTLE, self.daemon.conn_lock.release)

try:

await self._connect_and_stream(loop, ssid, password, rtmp_url, key, release)

< 10BC8 div id="LC612" class="react-file-line html-div" data-testid="code-cell" data-line-number="612" style="position:relative"> finally:

if self._client is not None:

self.link_ended = time.time()

self._client = None

await release()

async def _connect_and_stream(self, loop, ssid, password, rtmp_url, key, release):

cfg = self.cfg

model_kind = cfg.get("kind", "unknown")

self.set_state("searching", "Kamera wird gesucht")

device = await self.find_device()

if device is None:

raise CameraError("Kamera nicht gefunden. Ist sie an, Bluetooth aktiv und nicht mit dem Handy verbunden?")

queue = asyncio.Queue()

disconnected = asyncio.Event()

def on_notify(_char, data):

try:

msg = Message.decode(data)

except ValueError as e:

log.debug("Nachricht verworfen %s: %s", bytes(data).hex(), e)

return

self.last_rx = time.time()

if msg.type == TY_STATUS and len(msg.payload) >= 21:

self.battery = msg.payload[20]

self.battery_at = time.time()

spec = power_spec(self.cfg)

if spec and len(msg.payload) >= spec[0] + 4:

self.charging = struct.unpack_from("<i", bytes(msg.payload), spec[0])[0] > spec[1]

cur = bytes(msg.payload)

if self._status_prev is None:

log.info("%s: Statusnachricht (%d Byte, Akku %d %%): %s", self.addr, len(cur), self.battery, cur.hex())

else:

changed = [i for i in range(min(len(cur), len(self._status_prev))) if i != 20 and cur[i] != self._status_prev[i]]

for i in changed:

self._status_flips[i] = self._status_flips.get(i, 0) + 1

if any(self._status_flips[i] <= 3 for i in changed):

log.info("%s: Statusnachricht (%d Byte, Akku %d %%): %s", self.addr, len(cur), self.battery, cur.hex())

self._status_prev = cur

return

queue.put_nowait(msg)

async def wait_for(mid, timeout=15):

end = time.time() + timeout

while True:

left = end - time.time()

if left <= 0:

raise CameraError("Keine Antwort der Kamera (0x%04x)" % mid)

try:

msg = await asyncio.wait_for(queue.get(), left)

except asyncio.TimeoutError:

raise CameraError("Keine Antwort der Kamera (0x%04x)" % mid)

if msg.id == mid:

return msg

self.set_state("connecting", "Verbinde per Bluetooth")

self.last_rx = time.time()

t_connect = time.time()

async with BleakClient(device, timeout=20, disconnected_callback=lambda c: disconnected.set()) as client:

self._client = client

t_connected = time.time()

write_char = None

for service in client.services:

for ch in service.characteristics:

if ch.uuid.startswith("0000fff5"):

write_char = ch

if "notify" in ch.properties or "indicate" in ch.properties:

try:

await client.start_notify(ch, on_notify)

except Exception as e:

log.debug("Benachrichtigung auf %s fehlgeschlagen: %s", ch.uuid, e)

if write_char is None:

raise CameraError("Kein DJI-Gerät (Schreib-Kanal FFF5 fehlt)")

log.info("%s: Zeiten: Verbinden und Dienste %.1f s, Benachrichtigungen %.1f s (Suche vorher %.1f s)", self.addr,

t_connected - t_connect, time.time() - t_connected, t_connect - self._search_started)

await release()

async def send(msg):

await client.write_gatt_char(write_char, msg.encode(), response=False)

self.set_state("pairing", "Falls die Kamera fragt, bitte dort bestätigen")

await send(Message(T_PAIR, ID_PAIR, TY_PAIR, PAIR_PAYLOAD + pack_string(PAIR_PIN)))

resp = await wait_for(ID_PAIR)

if resp.payload != bytes([0, 1]):

try:

await asyncio.wait_for(queue.get(), 60)

except asyncio.TimeoutError:

raise CameraError("Die Kopplung wurde an der Kamera nicht bestätigt")

if self.stop_requested:

return

if cfg.get("status_only"):

await self._status_only_loop(client, disconnected)

return

async def setup_stream():

nonlocal ssid, password, rtmp_url

self.set_state("preparing", "Stream wird vorbereitet")

await send(Message(T_STOP, ID_STOP, TY_STOP, STOP_PAYLOAD))

await wait_for(ID_STOP)

await send(Message(T_PREPARE, ID_PREPARE, TY_PREPARE, b"\x1a"))

await wait_for(ID_PREPARE)

if self.stop_requested:

return False

if not cfg.get("status_only"):

ssid, password, rtmp_url = await loop.run_in_executor(None, self.resolve_target)

self.set_state("wifi", ssid)

await send(Message(T_WIFI, ID_WIFI, TY_WIFI, pack_string(ssid) + pack_string(password)))

resp = await wait_for(ID_WIFI, 30)

if resp.payload != bytes([0, 0]):

raise CameraError('Die Kamera konnte dem WLAN "%s" nicht beitreten (Name oder Passwort?)' % ssid)

if self.typed_network and ssid:

self.daemon.remember_network(self, ssid, password)

if self.stop_requested:

return False

if model_kind in CONFIGURE_KINDS:

self.set_state("configuring", "Bildstabilisierung wird eingestellt")

await send(Message(T_CONFIGURE, ID_CONFIGURE, TY_CONFIGURE,

build_configure_payload(model_kind, cfg.get("stabilization", "off"))))

await wait_for(ID_CONFIGURE)

if self.stop_requested:

return False

end = time.monotonic() + self.OLD_STREAM_WAIT_S

while time.monotonic() < end and not self.stop_requested:

if (await loop.run_in_executor(None, rtmp_publishing, key, self.daemon.stat_url)) is not True:

break

self.set_state("starting", "Wartet, bis der alte Stream beendet ist")

await asyncio.sleep(1)

self.set_state("starting", "Stream wird gestartet")

send_url = camera_rtmp_url(model_kind, rtmp_url)

payload = build_start_payload(model_kind, send_url, cfg.get("resolution", "1080p"),

int(cfg.get("bitrate", 6000)), int(cfg.get("fps", 30)))

await send(Message(T_START, ID_START, TY_START, payload))

if model_kind in NEW_PROTOCOL:

await send(Message(T_STOP, ID_STOP, TY_STOP, CONFIRM_PAYLOAD))

await wait_for(ID_START, 30)

self.set_state("streaming", rtmp_url)

self.fail_count = 0

self.reuse_failed = False

return True

sig = (model_kind, rtmp_url, ssid, cfg.get("resolution", "1080p"), int(cfg.get("bitrate", 6000)), int(cfg.get("fps", 30)), cfg.get("stabilization", "off"))

resume = False

if self._resume_sig == sig:

resume = (await loop.run_in_executor(None, rtmp_publishing, key, self.daemon.stat_url)) is True

if resume:

log.info("%s: Der Stream der Kamera läuft noch: Einrichtung übersprungen, nur die Verbindung wird wieder aufgenommen", self.addr)

self.set_state("streaming", rtmp_url)

self.fail_count = 0

self.reuse_failed = False

else:

self._resume_sig = None

if not await setup_stream():

return

self._resume_sig = sig

last_ok = time.time()

last_check = 0

bt_lost = False

low_since = None

while not self.stop_requested:

if bt_lost:

await asyncio.sleep(1)

else:

try:

await asyncio.wait_for(disconnected.wait(), 1)

except asyncio.TimeoutError:

pass

while not queue.empty():

queue.get_nowait()

now = time.time()

if disconnected.is_set() and not bt_lost:

bt_lost = True

log.info("%s: Bluetooth-Verbindung getrennt, prüfe den Stream", self.addr)

if now - last_check >= self.STREAM_CHECK_SECONDS or (bt_lost and last_check == 0):

last_check = now

publishing = await loop.run_in_executor(None, rtmp_publishing, key, self.daemon.stat_url)

if publishing is None and bt_lost:

if now - last_ok > self.STREAM_LOST_SECONDS:

raise CameraError("Die Bluetooth-Verbindung ging verloren")

elif publishing is None or publishing:

last_ok = now

if publishing:

kb = await loop.run_in_executor(None, rtmp_video_kbps, key, self.daemon.stat_url)

if kb is not None and kb < max(300, int(cfg.get("bitrate", 6000)) // 10):

low_since = low_since or now

if now - low_since > self.NO_VIDEO_SECONDS:

self._resume_sig = None

raise CameraError("Der Stream liefert kein Bild (nur Ton)")

else:

low_since = None

if bt_lost:

if time.time() - self.last_rx < 20:

self.set_state("streaming", "Der Stream läuft; die Steuerung per Bluetooth ist beendet, Statusmeldungen (Akku) kommen weiter")

else:

self.set_state("streaming", "Der Stream läuft, die Bluetooth-Verbindung ist getrennt")

elif now - last_ok > self.STREAM_LOST_SECONDS:

raise CameraError("Der Stream ist ausgefallen (Kamera außer Reichweite des WLANs?)")

if not bt_lost and now - self.last_rx > 180 and (await loop.run_in_executor(None, rtmp_publishing, key, self.daemon.stat_url)) is False:

raise CameraError("Die Bluetooth-Verbindung ging verloren")

if self.stop_requested and client.is_connected and not bt_lost and not self.leaving:

self.set_state("stopping", "Stream wird beendet")

try:

await send(Message(T_STOP, ID_STOP, TY_STOP, STOP_PAYLOAD))

await wait_for(ID_STOP, 10)

except Exception:

pass

STATUS_SILENCE_SECONDS = 180

async def _status_only_loop(self, client, disconnected):

self.set_state("status", "Nur Akkustand per Bluetooth")

self.fail_count = 0

self.reuse_failed = False

self.last_rx = time.time()

while not self.stop_requested:

try:

await asyncio.wait_for(disconnected.wait(), 1)

except asyncio.TimeoutError:

pass

if disconnected.is_set() or not client.is_connected:

raise CameraError("Die Bluetooth-Verbindung ging verloren")

if time.time() - self.last_rx > self.STATUS_SILENCE_SECONDS:

raise CameraError("Die Kamera sendet keine Statusmeldungen mehr")

async def run_loop(self):

while True:

self.wake.clear()

try:

await self.session()

self.fail_count = 0

self._resume_sig = None

self.set_state("idle", "Getrennt")

except asyncio.CancelledError:

self._resume_sig = None

self.set_state("idle")

raise

except CameraError as e:

self.fail_count += 1

self._reuse_failed_now()

self.set_state("error", str(e))

except Exception as e:

self.fail_count += 1

self._reuse_failed_now()

msg = friendly_error(e)

if msg is None:

log.exception("Sitzung fehlgeschlagen")

msg = "%s: %s" % (type(e).__name__, e)

else:

log.info("%s: %s (%s)", self.addr, msg, type(e).__name__)

if self.recovering(e):

self.set_state("connecting", "Die Kamera meldet sich nach dem Trennen oft erst nach etwa einer Minute wieder. Der Dienst versucht es weiter.")

else:

self.set_state("error", msg)

if self.stop_requested or not self.cfg.get("autoconnect"):

self.retry_at = 0

return

delay = self.retry_delay()

self.retry_at = time.time() + delay

try:

await asyncio.wait_for(self.wake.wait(), delay)

except asyncio.TimeoutError:

pass

self.retry_at = 0

if self.stop_requested:

return

def recovering(self, exc):

connect_failure = isinstance(exc, (asyncio.TimeoutError, TimeoutError)) or "disconnected" in str(exc).lower() or "failed to discover" in str(exc).lower()

return connect_failure and 0 < time.time() - self.link_ended < self.AFTER_LINK_SECONDS

def _reuse_failed_now(self):

if self._reusing:

log.info("%s: Die weiterverwendete Verbindung hat nicht funktioniert, der nächste Versuch verbindet neu", self.addr)

self._reusing = False

self.reuse_failed = True

self.force_cleanup = True

def retry_delay(self):

return self.RETRY_SCHEDULE[min(max(self.fail_count, 1) - 1, len(self.RETRY_SCHEDULE) - 1)]

def running(self):

return self.task is not None and not self.task.done()

def start(self):

if self.running():

self.wake.set()

return

self.fail_count = 0

self.stop_requested = False

self.manual_off = False

self.force_cleanup = True

self.task = asyncio.ensure_future(self.run_loop())

async def release_link(self):

client = self._client

if client is not None:

try:

await asyncio.wait_for(client.disconnect(), 5)

except Exception as e:

log.info("%s: Trennen beim Beenden fehlgeschlagen: %s", self.addr, e)

async def release_feed(self):

key = self.cfg.get("rtmp_key")

if not key or not (self.publishing or self.state in ("starting", "configuring", "streaming")):

return

if self.daemon.feed_hold(key, FEED_HOLD_S):

await asyncio.sleep(FEED_HOLD_WAIT_S)

async def stop_session(self):

await self.release_feed()

self.stop_requested = True

self._resume_sig = None

self.wake.set()

if self.running():

try:

await asyncio.wait_for(asyncio.shield(self.task), 15)

except Exception:

self.task.cancel()

self.retry_at = 0

async def stop(self):

self.manual_off = True

await self.stop_session()

self.set_state("idle")

async def restart(self):

await self.stop_session()

self.set_state("idle")

self.start()

FEED_HOLD_FILE = "feed-hold.json"

FEED_HOLD_S = 45

SWITCH_LOCK_S = 120

FEED_HOLD_WAIT_S = 4.5

SETTINGS_ALLOWED = {"name": str, "wifi_ifname": str, "ssid": str, "password": str, "ip": str,

"resolution": str, "fps": int, "bitrate": int, "stabilization": str, "autoconnect": bool, "status_only": bool}

NETWORK_FIELDS = ("wifi_ifname", "ssid", "password", "ip")

LEGACY_MODEL_IDS = {"osmoAction2": 0x0010, "osmoAction3": 0x0012, "osmoAction4": 0x0014, "osmoAction5Pro": 0x0015,

"osmo360": 0x0017, "osmoAction6": 0x0018, "osmoPocket3": 0x0020, "osmoPocket4": 0x0021}

LEGACY_STAB = {"rockSteady": "rocksteady", "rockSteadyPlus": "rocksteadyplus",

"horizonBalancing": "horizonbalancing", "horizonSteady": "horizonsteady"}

MAC_RE = re.compile(r"^([0-9A-Fa-f]{2}:){5}[0-9A-Fa-f]{2}$")

DEFAULT_SETTINGS = {"resolution": "1080p", "fps": 30, "bitrate": 6000, "stabilization": "off"}

def _read_json(path):

try:

with open(path) as f:

d = json.load(f)

return d if isinstance(d, dict) else {}

except (OSError, ValueError):

return {}

def iface_ip(name):

import socket

import struct

try:

import fcntl

sk = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

try:

return socket.inet_ntoa(fcntl.ioctl(sk.fileno(), 0x8915,

struct.pack("256s", name[:15].encode()))[20:24])

finally:

sk.close()

except (ImportError, OSError):

return None

def migrate_legacy(state_dir, wifi_options=None):

known = _read_json(os.path.join(state_dir, "dji-known.json"))

settings = _read_json(os.path.join(state_dir, "dji-settings.json"))

wifi = _read_json(os.path.join(state_dir, "dji-wifi.json"))

active = _read_json(os.path.join(state_dir, "dji-active.json"))

net = _read_json(os.path.join(state_dir, "camera-net.json")).get("iface")

names = {}

try:

with open(os.path.join(state_dir, "cameras.json")) as f:

cams = json.load(f)

View remainder of file in raw view

FilesExpand file tree

dji_daemon.py

Latest commit

History

dji_daemon.py

File metadata and controls