from __future__ import annotations from dataclasses import dataclass from app.config.models import InfinityMirrorConfig, TileConfig from app.network.wled import DiscoveredWledDevice, normalize_mac_address @dataclass(frozen=True) class TileAssignmentSnapshot: tile_id: str controller_ip: str = "" controller_host: str = "" controller_name: str = "" controller_mac: str = "" @property def is_assigned(self) -> bool: return any((self.controller_ip, self.controller_host, self.controller_name, self.controller_mac)) @dataclass(frozen=True) class DeviceAssignmentResult: target_tile_id: str previous_tile_id: str | None = None displaced_tile_assignment: TileAssignmentSnapshot | None = None def capture_tile_assignment(tile: TileConfig) -> TileAssignmentSnapshot: return TileAssignmentSnapshot( tile_id=tile.tile_id, controller_ip=tile.controller_ip.strip(), controller_host=tile.controller_host.strip(), controller_name=tile.controller_name.strip(), controller_mac=normalize_mac_address(tile.controller_mac), ) def clear_tile_assignment(tile: TileConfig) -> None: tile.controller_ip = "" tile.controller_host = "" tile.controller_name = "" tile.controller_mac = "" def tile_matches_device(tile: TileConfig, device: DiscoveredWledDevice) -> bool: tile_mac = normalize_mac_address(tile.controller_mac) device_mac = normalize_mac_address(device.mac_address) if tile_mac and device_mac: return tile_mac == device_mac return bool(tile.controller_ip.strip() and tile.controller_ip.strip() == device.ip_address) def find_tile_for_device(config: InfinityMirrorConfig, device: DiscoveredWledDevice) -> TileConfig | None: for tile in config.sorted_tiles(): if tile_matches_device(tile, device): return tile return None def assign_device_to_tile( config: InfinityMirrorConfig, device: DiscoveredWledDevice, target_tile_id: str, ) -> DeviceAssignmentResult: tile_lookup = config.tile_lookup() target_tile = tile_lookup.get(target_tile_id) if target_tile is None: raise KeyError(f"Unknown tile id: {target_tile_id}") previous_tile = find_tile_for_device(config, device) displaced_assignment = capture_tile_assignment(target_tile) displaced_snapshot = displaced_assignment if displaced_assignment.is_assigned else None if previous_tile is not None and previous_tile.tile_id != target_tile.tile_id: clear_tile_assignment(previous_tile) if displaced_snapshot is not None: target_mac = normalize_mac_address(target_tile.controller_mac) if previous_tile is None or previous_tile.tile_id != target_tile.tile_id: if not (target_mac and target_mac == device.mac_address) and target_tile.controller_ip.strip() != device.ip_address: clear_tile_assignment(target_tile) target_tile.controller_ip = device.ip_address target_tile.controller_host = device.hostname target_tile.controller_name = device.instance_name target_tile.controller_mac = device.mac_address return DeviceAssignmentResult( target_tile_id=target_tile.tile_id, previous_tile_id=previous_tile.tile_id if previous_tile is not None and previous_tile.tile_id != target_tile.tile_id else None, displaced_tile_assignment=displaced_snapshot, )