Publish LumaOps source
This commit is contained in:
LumaOps release export
@@ -0,0 +1,549 @@
|
||||
"""Bounded parser and serializer for OpenRGB SDK protocol version 5."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import struct
|
||||
from dataclasses import dataclass, field
|
||||
from enum import IntEnum, IntFlag
|
||||
from hashlib import sha256
|
||||
|
||||
from ...errors import ConnectorError
|
||||
from ..base import DeviceCapabilities, RGBColor
|
||||
|
||||
MAGIC = b"ORGB"
|
||||
HEADER = struct.Struct("<4sIII")
|
||||
PROTOCOL_VERSION = 5
|
||||
DEFAULT_MAX_PACKET_SIZE = 16 * 1024 * 1024
|
||||
MAX_COLLECTION_ITEMS = 65_535
|
||||
MAX_STRING_BYTES = 16_384
|
||||
MAX_MATRIX_CELLS = 1_000_000
|
||||
|
||||
|
||||
class PacketId(IntEnum):
|
||||
REQUEST_CONTROLLER_COUNT = 0
|
||||
REQUEST_CONTROLLER_DATA = 1
|
||||
REQUEST_PROTOCOL_VERSION = 40
|
||||
SET_CLIENT_NAME = 50
|
||||
DEVICE_LIST_UPDATED = 100
|
||||
REQUEST_RESCAN_DEVICES = 140
|
||||
REQUEST_PROFILE_LIST = 150
|
||||
REQUEST_SAVE_PROFILE = 151
|
||||
REQUEST_LOAD_PROFILE = 152
|
||||
REQUEST_DELETE_PROFILE = 153
|
||||
REQUEST_PLUGIN_LIST = 200
|
||||
PLUGIN_SPECIFIC = 201
|
||||
RESIZE_ZONE = 1000
|
||||
CLEAR_SEGMENTS = 1001
|
||||
ADD_SEGMENT = 1002
|
||||
UPDATE_LEDS = 1050
|
||||
UPDATE_ZONE_LEDS = 1051
|
||||
UPDATE_SINGLE_LED = 1052
|
||||
SET_CUSTOM_MODE = 1100
|
||||
UPDATE_MODE = 1101
|
||||
SAVE_MODE = 1102
|
||||
|
||||
|
||||
class ModeFlag(IntFlag):
|
||||
HAS_SPEED = 1 << 0
|
||||
HAS_DIRECTION_LR = 1 << 1
|
||||
HAS_DIRECTION_UD = 1 << 2
|
||||
HAS_DIRECTION_HV = 1 << 3
|
||||
HAS_BRIGHTNESS = 1 << 4
|
||||
HAS_PER_LED_COLOR = 1 << 5
|
||||
HAS_MODE_SPECIFIC_COLOR = 1 << 6
|
||||
HAS_RANDOM_COLOR = 1 << 7
|
||||
MANUAL_SAVE = 1 << 8
|
||||
AUTOMATIC_SAVE = 1 << 9
|
||||
|
||||
|
||||
class ZoneFlag(IntFlag):
|
||||
RESIZE_EFFECTS_ONLY = 1 << 0
|
||||
|
||||
|
||||
DEVICE_TYPES = {
|
||||
0: "motherboard",
|
||||
1: "dram",
|
||||
2: "gpu",
|
||||
3: "cooler",
|
||||
4: "ledstrip",
|
||||
5: "keyboard",
|
||||
6: "mouse",
|
||||
7: "mousemat",
|
||||
8: "headset",
|
||||
9: "headset_stand",
|
||||
10: "gamepad",
|
||||
11: "light",
|
||||
12: "speaker",
|
||||
13: "virtual",
|
||||
14: "storage",
|
||||
15: "case",
|
||||
16: "microphone",
|
||||
17: "accessory",
|
||||
18: "keypad",
|
||||
19: "unknown",
|
||||
}
|
||||
|
||||
|
||||
class ProtocolError(ConnectorError):
|
||||
def __init__(self, message: str, **details: object) -> None:
|
||||
super().__init__(
|
||||
"openrgb_protocol_error",
|
||||
message,
|
||||
status_code=502,
|
||||
retryable=False,
|
||||
details={str(key): value for key, value in details.items()},
|
||||
)
|
||||
|
||||
|
||||
@dataclass(slots=True, frozen=True)
|
||||
class PacketHeader:
|
||||
device_index: int
|
||||
packet_id: int
|
||||
payload_size: int
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class Mode:
|
||||
index: int
|
||||
name: str
|
||||
value: int
|
||||
flags: int
|
||||
speed_min: int
|
||||
speed_max: int
|
||||
brightness_min: int
|
||||
brightness_max: int
|
||||
colors_min: int
|
||||
colors_max: int
|
||||
speed: int
|
||||
brightness: int
|
||||
direction: int
|
||||
color_mode: int
|
||||
colors: list[RGBColor] = field(default_factory=list)
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class Segment:
|
||||
name: str
|
||||
zone_type: int
|
||||
start_index: int
|
||||
led_count: int
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class Zone:
|
||||
index: int
|
||||
name: str
|
||||
zone_type: int
|
||||
leds_min: int
|
||||
leds_max: int
|
||||
led_count: int
|
||||
matrix_height: int | None
|
||||
matrix_width: int | None
|
||||
matrix_map: list[int | None]
|
||||
segments: list[Segment]
|
||||
flags: int
|
||||
start_index: int = 0
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class Led:
|
||||
name: str
|
||||
value: int
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class Controller:
|
||||
index: int
|
||||
device_type: int
|
||||
name: str
|
||||
vendor: str
|
||||
description: str
|
||||
firmware_version: str
|
||||
serial: str
|
||||
location: str
|
||||
modes: list[Mode]
|
||||
active_mode: int
|
||||
zones: list[Zone]
|
||||
leds: list[Led]
|
||||
colors: list[RGBColor]
|
||||
led_alt_names: list[str]
|
||||
flags: int
|
||||
|
||||
@property
|
||||
def fingerprint(self) -> str:
|
||||
parts = [self.vendor, self.description, self.serial, self.location, self.name]
|
||||
durable = "\x1f".join(part.strip().casefold() for part in parts)
|
||||
return sha256(durable.encode("utf-8")).hexdigest()
|
||||
|
||||
@property
|
||||
def device_type_name(self) -> str:
|
||||
return DEVICE_TYPES.get(self.device_type, "unknown")
|
||||
|
||||
@property
|
||||
def capabilities(self) -> DeviceCapabilities:
|
||||
flags = [ModeFlag(mode.flags) for mode in self.modes]
|
||||
speeds = [mode for mode in self.modes if ModeFlag.HAS_SPEED in ModeFlag(mode.flags)]
|
||||
brightness = [
|
||||
mode for mode in self.modes if ModeFlag.HAS_BRIGHTNESS in ModeFlag(mode.flags)
|
||||
]
|
||||
return DeviceCapabilities(
|
||||
power=bool(self.leds),
|
||||
restore=bool(self.colors),
|
||||
rgb=bool(self.colors or self.leds),
|
||||
brightness=bool(brightness),
|
||||
effect=bool(self.modes),
|
||||
speed=bool(speeds),
|
||||
direction=any(
|
||||
f & (ModeFlag.HAS_DIRECTION_LR | ModeFlag.HAS_DIRECTION_UD | ModeFlag.HAS_DIRECTION_HV)
|
||||
for f in flags
|
||||
),
|
||||
multiple_colors=any(
|
||||
f & (ModeFlag.HAS_PER_LED_COLOR | ModeFlag.HAS_MODE_SPECIFIC_COLOR) for f in flags
|
||||
),
|
||||
per_zone=bool(self.zones),
|
||||
per_segment=any(zone.segments for zone in self.zones),
|
||||
per_led=bool(self.leds),
|
||||
profiles=True,
|
||||
max_leds=len(self.leds),
|
||||
min_speed=min((m.speed_min for m in speeds), default=None),
|
||||
max_speed=max((m.speed_max for m in speeds), default=None),
|
||||
)
|
||||
|
||||
|
||||
class Cursor:
|
||||
def __init__(self, data: bytes) -> None:
|
||||
self.data = memoryview(data)
|
||||
self.offset = 0
|
||||
|
||||
@property
|
||||
def remaining(self) -> int:
|
||||
return len(self.data) - self.offset
|
||||
|
||||
def take(self, size: int) -> bytes:
|
||||
if size < 0 or size > self.remaining:
|
||||
raise ProtocolError(
|
||||
"OpenRGB-pakket eindigt onverwacht.",
|
||||
requested=size,
|
||||
remaining=self.remaining,
|
||||
offset=self.offset,
|
||||
)
|
||||
result = self.data[self.offset : self.offset + size].tobytes()
|
||||
self.offset += size
|
||||
return result
|
||||
|
||||
def u16(self) -> int:
|
||||
return int(struct.unpack("<H", self.take(2))[0])
|
||||
|
||||
def u32(self) -> int:
|
||||
return int(struct.unpack("<I", self.take(4))[0])
|
||||
|
||||
def i32(self) -> int:
|
||||
return int(struct.unpack("<i", self.take(4))[0])
|
||||
|
||||
def string(self) -> str:
|
||||
length = self.u16()
|
||||
if length < 1 or length > MAX_STRING_BYTES:
|
||||
raise ProtocolError("Ongeldige OpenRGB-stringlengte.", length=length)
|
||||
raw = self.take(length)
|
||||
if raw[-1:] != b"\0":
|
||||
raise ProtocolError("OpenRGB-string is niet NUL-afgesloten.")
|
||||
try:
|
||||
return raw[:-1].decode("utf-8", errors="strict")
|
||||
except UnicodeDecodeError as exc:
|
||||
raise ProtocolError("OpenRGB-string bevat ongeldige UTF-8.") from exc
|
||||
|
||||
def count(self, label: str, maximum: int = MAX_COLLECTION_ITEMS) -> int:
|
||||
count = self.u16()
|
||||
if count > maximum:
|
||||
raise ProtocolError("OpenRGB-verzameling is te groot.", field=label, count=count)
|
||||
return count
|
||||
|
||||
|
||||
def pack_header(device_index: int, packet_id: int | PacketId, payload_size: int) -> bytes:
|
||||
if not 0 <= device_index <= 0xFFFFFFFF:
|
||||
raise ProtocolError("Ongeldige OpenRGB-controllerindex.", index=device_index)
|
||||
if not 0 <= payload_size <= DEFAULT_MAX_PACKET_SIZE:
|
||||
raise ProtocolError("Ongeldige OpenRGB-pakketgrootte.", size=payload_size)
|
||||
return HEADER.pack(MAGIC, device_index, int(packet_id), payload_size)
|
||||
|
||||
|
||||
def parse_header(data: bytes, max_packet_size: int = DEFAULT_MAX_PACKET_SIZE) -> PacketHeader:
|
||||
if len(data) != HEADER.size:
|
||||
raise ProtocolError("Ongeldige OpenRGB-headerlengte.", length=len(data))
|
||||
magic, device_index, packet_id, payload_size = HEADER.unpack(data)
|
||||
if magic != MAGIC:
|
||||
raise ProtocolError("Ongeldige OpenRGB-pakketmagic.")
|
||||
if payload_size > max_packet_size:
|
||||
raise ProtocolError(
|
||||
"OpenRGB-pakket overschrijdt de ingestelde limiet.",
|
||||
size=payload_size,
|
||||
maximum=max_packet_size,
|
||||
)
|
||||
return PacketHeader(device_index, packet_id, payload_size)
|
||||
|
||||
|
||||
def pack_string(value: str) -> bytes:
|
||||
raw = value.encode("utf-8")
|
||||
if len(raw) + 1 > min(MAX_STRING_BYTES, 0xFFFF):
|
||||
raise ProtocolError("OpenRGB-string is te lang.", length=len(raw))
|
||||
return struct.pack("<H", len(raw) + 1) + raw + b"\0"
|
||||
|
||||
|
||||
def pack_color(color: RGBColor) -> bytes:
|
||||
return struct.pack("<BBBx", color.red, color.green, color.blue)
|
||||
|
||||
|
||||
def parse_color(cursor: Cursor) -> RGBColor:
|
||||
red, green, blue, _padding = struct.unpack("<BBBB", cursor.take(4))
|
||||
return RGBColor(red=red, green=green, blue=blue)
|
||||
|
||||
|
||||
def parse_mode(cursor: Cursor, index: int) -> Mode:
|
||||
name = cursor.string()
|
||||
value = cursor.i32()
|
||||
flags = cursor.u32()
|
||||
speed_min = cursor.u32()
|
||||
speed_max = cursor.u32()
|
||||
brightness_min = cursor.u32()
|
||||
brightness_max = cursor.u32()
|
||||
colors_min = cursor.u32()
|
||||
colors_max = cursor.u32()
|
||||
speed = cursor.u32()
|
||||
brightness = cursor.u32()
|
||||
direction = cursor.u32()
|
||||
color_mode = cursor.u32()
|
||||
color_count = cursor.count("mode.colors")
|
||||
colors = [parse_color(cursor) for _ in range(color_count)]
|
||||
return Mode(
|
||||
index,
|
||||
name,
|
||||
value,
|
||||
flags,
|
||||
speed_min,
|
||||
speed_max,
|
||||
brightness_min,
|
||||
brightness_max,
|
||||
colors_min,
|
||||
colors_max,
|
||||
speed,
|
||||
brightness,
|
||||
direction,
|
||||
color_mode,
|
||||
colors,
|
||||
)
|
||||
|
||||
|
||||
def parse_zone(cursor: Cursor, index: int, start_index: int) -> Zone:
|
||||
name = cursor.string()
|
||||
zone_type = cursor.i32()
|
||||
leds_min = cursor.u32()
|
||||
leds_max = cursor.u32()
|
||||
led_count = cursor.u32()
|
||||
if led_count > MAX_COLLECTION_ITEMS:
|
||||
raise ProtocolError("OpenRGB-zone bevat te veel leds.", count=led_count)
|
||||
matrix_len = cursor.u16()
|
||||
matrix_height: int | None = None
|
||||
matrix_width: int | None = None
|
||||
matrix: list[int | None] = []
|
||||
if matrix_len:
|
||||
if matrix_len < 8 or (matrix_len - 8) % 4:
|
||||
raise ProtocolError("Ongeldige OpenRGB-matrixlengte.", length=matrix_len)
|
||||
matrix_height = cursor.u32()
|
||||
matrix_width = cursor.u32()
|
||||
cells = (matrix_len - 8) // 4
|
||||
if cells > MAX_MATRIX_CELLS or matrix_height * matrix_width != cells:
|
||||
raise ProtocolError(
|
||||
"OpenRGB-matrixafmetingen komen niet overeen.",
|
||||
height=matrix_height,
|
||||
width=matrix_width,
|
||||
cells=cells,
|
||||
)
|
||||
matrix = [None if (cell := cursor.u32()) == 0xFFFFFFFF else cell for _ in range(cells)]
|
||||
segment_count = cursor.count("zone.segments")
|
||||
segments = [
|
||||
Segment(cursor.string(), cursor.i32(), cursor.u32(), cursor.u32())
|
||||
for _ in range(segment_count)
|
||||
]
|
||||
flags = cursor.u32()
|
||||
return Zone(
|
||||
index,
|
||||
name,
|
||||
zone_type,
|
||||
leds_min,
|
||||
leds_max,
|
||||
led_count,
|
||||
matrix_height,
|
||||
matrix_width,
|
||||
matrix,
|
||||
segments,
|
||||
flags,
|
||||
start_index,
|
||||
)
|
||||
|
||||
|
||||
def parse_controller(payload: bytes, index: int) -> Controller:
|
||||
cursor = Cursor(payload)
|
||||
declared_size = cursor.u32()
|
||||
if declared_size != len(payload):
|
||||
raise ProtocolError(
|
||||
"OpenRGB-controllerblok heeft een afwijkende lengte.",
|
||||
declared=declared_size,
|
||||
actual=len(payload),
|
||||
)
|
||||
device_type = cursor.i32()
|
||||
name = cursor.string()
|
||||
vendor = cursor.string()
|
||||
description = cursor.string()
|
||||
firmware_version = cursor.string()
|
||||
serial = cursor.string()
|
||||
location = cursor.string()
|
||||
mode_count = cursor.count("controller.modes")
|
||||
active_mode = cursor.i32()
|
||||
modes = [parse_mode(cursor, mode_index) for mode_index in range(mode_count)]
|
||||
zone_count = cursor.count("controller.zones")
|
||||
zones: list[Zone] = []
|
||||
start = 0
|
||||
for zone_index in range(zone_count):
|
||||
zone = parse_zone(cursor, zone_index, start)
|
||||
zones.append(zone)
|
||||
start += zone.led_count
|
||||
led_count = cursor.count("controller.leds")
|
||||
leds = [Led(cursor.string(), cursor.u32()) for _ in range(led_count)]
|
||||
color_count = cursor.count("controller.colors")
|
||||
colors = [parse_color(cursor) for _ in range(color_count)]
|
||||
alt_count = cursor.count("controller.led_alt_names")
|
||||
led_alt_names = [cursor.string() for _ in range(alt_count)]
|
||||
flags = cursor.u32()
|
||||
if cursor.remaining:
|
||||
raise ProtocolError("Onverwachte bytes na OpenRGB-controllerblok.", remaining=cursor.remaining)
|
||||
if len(colors) not in (0, len(leds)):
|
||||
raise ProtocolError(
|
||||
"Aantal OpenRGB-kleuren komt niet overeen met het aantal leds.",
|
||||
colors=len(colors),
|
||||
leds=len(leds),
|
||||
)
|
||||
if sum(zone.led_count for zone in zones) > len(leds):
|
||||
raise ProtocolError("OpenRGB-zones verwijzen buiten de ledlijst.")
|
||||
return Controller(
|
||||
index,
|
||||
device_type,
|
||||
name,
|
||||
vendor,
|
||||
description,
|
||||
firmware_version,
|
||||
serial,
|
||||
location,
|
||||
modes,
|
||||
active_mode,
|
||||
zones,
|
||||
leds,
|
||||
colors,
|
||||
led_alt_names,
|
||||
flags,
|
||||
)
|
||||
|
||||
|
||||
def pack_update_leds(colors: list[RGBColor], expected_count: int) -> bytes:
|
||||
if len(colors) != expected_count or len(colors) > 0xFFFF:
|
||||
raise ProtocolError(
|
||||
"Aantal kleuren moet exact overeenkomen met het aantal leds.",
|
||||
colors=len(colors),
|
||||
expected=expected_count,
|
||||
)
|
||||
body = struct.pack("<H", len(colors)) + b"".join(pack_color(color) for color in colors)
|
||||
return struct.pack("<I", len(body) + 4) + body
|
||||
|
||||
|
||||
def pack_update_zone(zone_index: int, colors: list[RGBColor], expected_count: int) -> bytes:
|
||||
if zone_index < 0:
|
||||
raise ProtocolError("Ongeldige OpenRGB-zoneindex.", index=zone_index)
|
||||
if len(colors) != expected_count or len(colors) > 0xFFFF:
|
||||
raise ProtocolError(
|
||||
"Aantal zonekleuren moet exact overeenkomen met het aantal zoneleds.",
|
||||
colors=len(colors),
|
||||
expected=expected_count,
|
||||
)
|
||||
body = struct.pack("<IH", zone_index, len(colors)) + b"".join(
|
||||
pack_color(color) for color in colors
|
||||
)
|
||||
return struct.pack("<I", len(body) + 4) + body
|
||||
|
||||
|
||||
def pack_update_single_led(led_index: int, color: RGBColor, led_count: int) -> bytes:
|
||||
if not 0 <= led_index < led_count:
|
||||
raise ProtocolError("OpenRGB-ledindex valt buiten bereik.", index=led_index, count=led_count)
|
||||
return struct.pack("<i", led_index) + pack_color(color)
|
||||
|
||||
|
||||
def pack_mode(
|
||||
mode: Mode,
|
||||
mode_index: int,
|
||||
*,
|
||||
brightness_percent: int | None = None,
|
||||
speed: int | None = None,
|
||||
direction: int | None = None,
|
||||
colors: list[RGBColor] | None = None,
|
||||
) -> bytes:
|
||||
if mode_index < 0:
|
||||
raise ProtocolError("Ongeldige OpenRGB-modusindex.")
|
||||
selected_speed = mode.speed if speed is None else speed
|
||||
if ModeFlag.HAS_SPEED in ModeFlag(mode.flags) and not min(mode.speed_min, mode.speed_max) <= selected_speed <= max(mode.speed_min, mode.speed_max):
|
||||
raise ProtocolError("OpenRGB-effectsnelheid valt buiten bereik.")
|
||||
selected_brightness = mode.brightness
|
||||
if brightness_percent is not None:
|
||||
if not 0 <= brightness_percent <= 100:
|
||||
raise ProtocolError("Helderheid moet tussen 0 en 100 liggen.")
|
||||
if ModeFlag.HAS_BRIGHTNESS not in ModeFlag(mode.flags):
|
||||
raise ProtocolError("De geselecteerde OpenRGB-modus ondersteunt geen helderheid.")
|
||||
low, high = mode.brightness_min, mode.brightness_max
|
||||
selected_brightness = round(low + (high - low) * brightness_percent / 100)
|
||||
selected_direction = mode.direction if direction is None else direction
|
||||
direction_flags = (
|
||||
ModeFlag.HAS_DIRECTION_LR | ModeFlag.HAS_DIRECTION_UD | ModeFlag.HAS_DIRECTION_HV
|
||||
)
|
||||
if direction is not None and not ModeFlag(mode.flags) & direction_flags:
|
||||
raise ProtocolError("De geselecteerde OpenRGB-modus ondersteunt geen richting.")
|
||||
selected_colors = list(mode.colors if colors is None else colors)
|
||||
if colors is not None:
|
||||
if ModeFlag.HAS_MODE_SPECIFIC_COLOR not in ModeFlag(mode.flags):
|
||||
raise ProtocolError("De geselecteerde OpenRGB-modus ondersteunt geen effectkleuren.")
|
||||
if not mode.colors_min <= len(selected_colors) <= mode.colors_max:
|
||||
raise ProtocolError(
|
||||
"Aantal OpenRGB-effectkleuren valt buiten bereik.",
|
||||
colors=len(selected_colors),
|
||||
minimum=mode.colors_min,
|
||||
maximum=mode.colors_max,
|
||||
)
|
||||
mode_body = (
|
||||
pack_string(mode.name)
|
||||
+ struct.pack(
|
||||
"<iIIIIIIIIIIIH",
|
||||
mode.value,
|
||||
mode.flags,
|
||||
mode.speed_min,
|
||||
mode.speed_max,
|
||||
mode.brightness_min,
|
||||
mode.brightness_max,
|
||||
mode.colors_min,
|
||||
mode.colors_max,
|
||||
selected_speed,
|
||||
selected_brightness,
|
||||
selected_direction,
|
||||
mode.color_mode,
|
||||
len(selected_colors),
|
||||
)
|
||||
+ b"".join(pack_color(color) for color in selected_colors)
|
||||
)
|
||||
body = struct.pack("<i", mode_index) + mode_body
|
||||
return struct.pack("<I", len(body) + 4) + body
|
||||
|
||||
|
||||
def parse_profile_list(payload: bytes) -> list[str]:
|
||||
cursor = Cursor(payload)
|
||||
declared = cursor.u32()
|
||||
if declared != len(payload):
|
||||
raise ProtocolError("OpenRGB-profiellijst heeft een afwijkende lengte.")
|
||||
profiles = [cursor.string() for _ in range(cursor.count("profiles"))]
|
||||
if cursor.remaining:
|
||||
raise ProtocolError("Onverwachte bytes na OpenRGB-profiellijst.")
|
||||
return profiles
|
||||
Reference in New Issue
Block a user