94 lines
3.3 KiB
Python
94 lines
3.3 KiB
Python
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,
|
|
)
|