Lago
124 lines
3.9 KiB
Python
124 lines
3.9 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from threading import Lock
|
|
from typing import Protocol
|
|
|
|
from monitorcontrol import get_monitors
|
|
|
|
from app.config import SUPPORTED_TARGET_PORTS
|
|
|
|
|
|
ALIENWARE_MODEL_TOKEN = "AW3423DWF"
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class DDMCommandResult:
|
|
success: bool
|
|
message: str
|
|
raw_output: str = ""
|
|
|
|
|
|
class DDMBackend(Protocol):
|
|
def is_available(self) -> bool: ...
|
|
|
|
def supports_monitor_targeting(self) -> bool: ...
|
|
|
|
def resolve_alienware_slot(self, force: bool = False) -> int | None: ...
|
|
|
|
def invalidate_slot(self) -> None: ...
|
|
|
|
def switch_to_port(self, slot: int, port_name: str) -> DDMCommandResult: ...
|
|
|
|
|
|
class RealDDMBackend:
|
|
def __init__(
|
|
self,
|
|
alienware_model_token: str = ALIENWARE_MODEL_TOKEN,
|
|
):
|
|
self.alienware_model_token = alienware_model_token
|
|
self._cached_slot: int | None = None
|
|
self._lock = Lock()
|
|
|
|
def is_available(self) -> bool:
|
|
_, error = self._resolve_alienware_monitor()
|
|
return error is None
|
|
|
|
def supports_monitor_targeting(self) -> bool:
|
|
return True
|
|
|
|
def invalidate_slot(self) -> None:
|
|
with self._lock:
|
|
self._cached_slot = None
|
|
|
|
def resolve_alienware_slot(self, force: bool = False) -> int | None:
|
|
with self._lock:
|
|
if self._cached_slot is not None and not force:
|
|
return self._cached_slot
|
|
|
|
monitor, error = self._resolve_alienware_monitor()
|
|
del monitor
|
|
if error is not None:
|
|
self._cached_slot = None
|
|
return None
|
|
# Service still expects a slot-like value. With monitorcontrol we
|
|
# target by model, so we expose a synthetic stable slot.
|
|
self._cached_slot = 1
|
|
return self._cached_slot
|
|
|
|
def switch_to_port(self, slot: int, port_name: str) -> DDMCommandResult:
|
|
del slot
|
|
port_spec = SUPPORTED_TARGET_PORTS.get(port_name.upper())
|
|
if not port_spec:
|
|
return DDMCommandResult(False, f"Port {port_name} is not supported.")
|
|
|
|
target_input = _to_monitor_input_value(str(port_spec["ddm_input"]))
|
|
monitor, error = self._resolve_alienware_monitor()
|
|
if error is not None:
|
|
return DDMCommandResult(False, error)
|
|
assert monitor is not None
|
|
|
|
try:
|
|
with monitor:
|
|
monitor.set_input_source(target_input)
|
|
except Exception as exc:
|
|
return DDMCommandResult(
|
|
False,
|
|
f"monitorcontrol failed to switch Alienware input to {target_input}: {exc}",
|
|
str(exc),
|
|
)
|
|
|
|
return DDMCommandResult(
|
|
True,
|
|
f"Alienware {self.alienware_model_token} was instructed to switch to {target_input} via monitorcontrol.",
|
|
)
|
|
|
|
def _resolve_alienware_monitor(self) -> tuple[object | None, str | None]:
|
|
try:
|
|
monitors = list(get_monitors())
|
|
except Exception as exc:
|
|
return None, f"monitorcontrol could not enumerate monitors: {exc}"
|
|
|
|
candidates: list[object] = []
|
|
token = self.alienware_model_token.upper()
|
|
for monitor in monitors:
|
|
description = str(getattr(getattr(monitor, "vcp", None), "description", "") or str(monitor))
|
|
if token in description.upper():
|
|
candidates.append(monitor)
|
|
|
|
if not candidates:
|
|
return None, f"Alienware {self.alienware_model_token} was not found via monitorcontrol."
|
|
if len(candidates) > 1:
|
|
return None, f"Multiple Alienware {self.alienware_model_token} monitors were detected; refusing to switch."
|
|
return candidates[0], None
|
|
|
|
|
|
def _to_monitor_input_value(port_name: str) -> int | str:
|
|
normalized = port_name.strip().upper()
|
|
if normalized == "DP2":
|
|
# Alienware AW3423DWF uses vendor-specific DP2 code.
|
|
return 19
|
|
if normalized == "HDMI":
|
|
return "HDMI1"
|
|
return normalized
|