Init

pkg/sunshine/__init__.py

@@ -0,0 +1,29 @@
+from .env_key import SunshineEnvKey as EnvKey
+from .enums import SunshineAudioConfig
+from .structs import SunshineAudioConfig, SunshineEnvironmentVariables, SunshineClientSpecs, SunshineClientInfo, SunshineClientOptions
+
+from os import environ
+
+def load_env_variables() -> SunshineEnvironmentVariables:
+    def ld_str(var: EnvKey): return environ.get(var.requested_key, var.default)
+    def ld_int(var: EnvKey): return int(ld_str(var))
+    def ld_bool(var: EnvKey): return ld_str(var).lower() == "true" 
+    
+    return SunshineEnvironmentVariables(
+        SunshineClientInfo(
+            app_id = ld_str(EnvKey.APP_ID),
+            app_name = ld_str(EnvKey.APP_NAME)
+        ),
+        SunshineClientSpecs(
+            width = ld_int(EnvKey.WIDTH),
+            height = ld_int(EnvKey.HEIGHT),
+            fps = ld_int(EnvKey.FPS),
+            hdr = ld_bool(EnvKey.HDR),
+            gcmap = ld_int(EnvKey.GCMAP)
+        ),
+        SunshineClientOptions(
+            host_audio = ld_bool(EnvKey.HOST_AUDIO),
+            enable_sops = ld_bool(EnvKey.ENABLE_SOPS),
+            audio_config = SunshineAudioConfig(ld_str(EnvKey.AUDIO_CONFIG))
+        )
+    )

pkg/sunshine/enums.py

@@ -0,0 +1,7 @@
+from enum import StrEnum
+
+class SunshineAudioConfig(StrEnum):
+    STEREO = "2.0"
+    SURROUND_5_1 = "5.1"
+    SURROUND_7_1 = "7.1"
+

pkg/sunshine/env_key.py

@@ -0,0 +1,20 @@
+from enum import Enum
+
+class KeyDefaultMixin(str):
+    def __new__(cls, requested_key: str, default: str):
+        obj = str.__new__(cls, requested_key)
+        obj.requested_key = requested_key
+        obj.default = default
+        return obj
+
+class SunshineEnvKey(KeyDefaultMixin, Enum):
+    APP_ID = ("SUNSHINE_APP_ID", "Unknown ID")
+    APP_NAME = ("SUNSHINE_APP_NAME", "Unknown App Name")
+    WIDTH = ("SUNSHINE_CLIENT_WIDTH", "1920")
+    HEIGHT = ("SUNSHINE_CLIENT_HEIGHT", "1080")
+    FPS = ("SUNSHINE_CLIENT_FPS", "60")
+    HDR = ("SUNSHINE_CLIENT_HDR", "0")
+    GCMAP = ("SUNSHINE_CLIENT_GCMAP", "0")
+    HOST_AUDIO = ("SUNSHINE_CLIENT_HOST_AUDIO", "0")
+    ENABLE_SOPS = ("SUNSHINE_CLIENT_ENABLE_SOPS", "0")
+    AUDIO_CONFIG = ("SUNSHINE_CLIENT_AUDIO_CONFIGURATION", "2.0")

pkg/sunshine/structs.py

@@ -0,0 +1,28 @@
+from dataclasses import dataclass
+
+from .enums import SunshineAudioConfig
+
+@dataclass
+class SunshineClientInfo:
+    app_id: str
+    app_name: str
+
+@dataclass
+class SunshineClientSpecs:
+    width: int
+    height: int
+    fps: int
+    hdr: bool
+    gcmap: int
+
+@dataclass
+class SunshineClientOptions:
+    host_audio: bool
+    enable_sops: bool
+    audio_config: SunshineAudioConfig
+
+@dataclass
+class SunshineEnvironmentVariables:
+    info: SunshineClientInfo
+    specs: SunshineClientSpecs
+    options: SunshineClientOptions

pkg/wayland_display_manager/__init__.py

@@ -0,0 +1,16 @@
+from .abstracts import AbstractDisplay
+from .enums import DisplayManagerDriver, Error
+
+def get_display(requested_driver: DisplayManagerDriver = DisplayManagerDriver.INFER) -> AbstractDisplay:
+    match requested_driver:
+        case DisplayManagerDriver.INFER:
+            for display in DisplayManagerDriver.get_display_list():
+                if display.meta.detect(): return display.meta()
+                
+            raise RuntimeError(Error.DISPLAY_INFER_TYPE_FAILURE)
+                
+        case _ if requested_driver in DisplayManagerDriver.get_display_list():
+            return requested_driver.meta()
+        
+        case _:
+            raise ValueError(Error.DISPLAY_TYPE_UNKNOWN)

pkg/wayland_display_manager/abstracts.py

@@ -0,0 +1,26 @@
+from .structs import MonitorInfo, Modeline
+
+from abc import ABC, abstractmethod
+
+class AbstractDisplay(ABC):
+    @abstractmethod
+    def get_modes(self) -> list[MonitorInfo]:
+        pass
+
+    @abstractmethod
+    def set_mode(self, modeline: Modeline, requested_connector: str) -> None:
+        pass
+
+    @abstractmethod
+    def save_current_mode(self, location: str) -> None:
+        pass
+
+    @abstractmethod
+    def read_saved_mode(self, location: str) -> Modeline:
+        pass
+
+    @staticmethod
+    @abstractmethod
+    def detect(self) -> bool:
+        pass
+

pkg/wayland_display_manager/enums.py

@@ -0,0 +1,28 @@
+from .abstracts import AbstractDisplay
+
+from . import gnome, plasma, wlroots
+
+from enum import Enum, StrEnum
+from typing import Optional, Type
+
+class MetaMixin(str):
+    def __new__(cls, text: str, meta: AbstractDisplay):
+        obj = str.__new__(cls, text)
+        obj.text = text
+        obj.meta = meta
+        return obj
+
+class DisplayManagerDriver(MetaMixin, Enum):
+    GNOME = ("GNOME (Mutter)", gnome.Display)
+    PLASMA = ("KDE Plasma 5/6 (KWin)", plasma.Display)
+    SWAY = ("Sway", wlroots.Display)
+    INFER = ("Inferring Wayland Compositor", None)
+
+    @classmethod
+    def get_display_list(cls):
+        return [display for display in cls if display.meta is not None]
+
+class Error(StrEnum):
+    DISPLAY_TYPE_NOT_IMPL = "Requested manager type has not been implemented!"
+    DISPLAY_TYPE_UNKNOWN = "Requested manager type does not exist."
+    DISPLAY_INFER_TYPE_FAILURE = "Could not infer a compatible display manager from supported list."

pkg/wayland_display_manager/gnome/__init__.py

@@ -0,0 +1,64 @@
+from ..structs import MonitorInfo, Modeline
+from ..abstracts import AbstractDisplay
+
+import os
+import re
+import subprocess
+
+### Dirty Implementation ###
+# Needs to be replaced with a proper dbus solution. Preferably by taking reference from gdctl. `cat /usr/bin/gdctl` #
+# Issues:
+#   - Only works with one monitor (will lead to strange results if multiple are present).
+#   - Possible has 1 quadrillion issues that aren't known.
+#   - Sucks dog eggs and unironically uses the unholy trinity `os`, `re`, and `subprocess`.
+
+MODE_REGEX_PATTERN = re.compile(r'(\d+)x(\d+)@(\d+\.\d+)')
+
+class Display(AbstractDisplay):
+    def get_modes(self):
+        gdctl_output = subprocess.run(["gdctl", "show", "--modes"], capture_output=True, text=True, check=True)
+        modes = MODE_REGEX_PATTERN.findall(gdctl_output.stdout)
+
+        # Hard coding my beloved
+        return [ 
+            MonitorInfo(
+                connector = "HDMI-2",
+                primary = True,
+                modes = [
+                    Modeline(
+                        width = int(w),
+                        height = int(h),
+                        refresh_rate = float(r)
+                    )
+                    for _, (w, h, r) in enumerate(modes)
+                ]
+            )
+        ]
+    
+    def set_mode(self, modeline, requested_connector = None) -> None:
+        subprocess.run(["gdctl", "set", "--logical-monitor", "--primary", "--monitor", requested_connector, "--mode", str(modeline)])
+
+    def save_current_mode(self, location):
+        gdctl_output = subprocess.run(["gdctl", "show"], capture_output=True, text=True, check=True)
+        match = MODE_REGEX_PATTERN.search(gdctl_output.stdout)
+        w, h, r = match.groups()
+        with open(location, "w") as file:
+            file.write(f"{w}x{h}@{r}")
+
+    def read_saved_mode(self, location):
+        with open(location, "r") as file:
+            config_file_content = file.read()
+
+
+        match = MODE_REGEX_PATTERN.search(config_file_content)
+
+        w, h, r = match.groups()
+        return Modeline(
+            width = int(w),
+            height = int(h),
+            refresh_rate = float(r)
+        )
+
+    @staticmethod
+    def detect():
+        return "gnome" in os.environ.get("XDG_CURRENT_DESKTOP", "").lower()

pkg/wayland_display_manager/plasma/__init__.py

@@ -0,0 +1,14 @@
+from ..abstracts import AbstractDisplay
+
+import os
+
+class Display(AbstractDisplay):
+    def get_modes(self):
+        pass
+
+    def set_mode(self, modeline: str):
+        pass
+
+    @staticmethod
+    def detect():
+        return "plasma" in os.environ.get("XDG_CURRENT_DESKTOP", "").lower()

pkg/wayland_display_manager/structs.py

@@ -0,0 +1,20 @@
+from dataclasses import dataclass
+
+@dataclass
+class Modeline:
+    width: int
+    height: int
+    refresh_rate: float
+    
+    def __str__(self):
+        return f"{self.width}x{self.height}@{self.refresh_rate:.3f}"
+    
+    def __repr__(self):
+        return self.__str__()
+
+@dataclass
+class MonitorInfo:
+    connector: str
+    primary: bool
+    modes: list[Modeline]
+

pkg/wayland_display_manager/wlroots/__init__.py

@@ -0,0 +1,56 @@
+from ..structs import MonitorInfo, Modeline
+from ..abstracts import AbstractDisplay
+
+import os
+import re
+import subprocess
+import json
+
+# Matches "1920x1080 @ 60.000 Hz" or "1920x1080@60.000"
+MODE_REGEX = re.compile(r'(\d+)x(\d+)@(\d+\.\d+)')
+
+class Display(AbstractDisplay):
+    def get_modes(self):
+        # wlr-randr --json is the 'civilized' way to do this on Trixie
+        result = subprocess.run(["wlr-randr", "--json"], capture_output=True, text=True, check=True)
+        data = json.loads(result.stdout)
+        
+        monitors = []
+        for output in data:
+            modes = []
+            for m in output.get("modes", []):
+                modes.append(Modeline(
+                    width=int(m["width"]),
+                    height=int(m["height"]),
+                    refresh_rate=float(m["refresh"])
+                ))
+            
+            monitors.append(MonitorInfo(
+                connector=output["name"], # Usually 'HEADLESS-1' in Cage
+                primary=output.get("focused", False),
+                modes=modes
+            ))
+        return monitors
+    
+    def set_mode(self, modeline, requested_connector="HEADLESS-1") -> None:
+        # Format: 1920x1080@60Hz
+        mode_str = f"{modeline.width}x{modeline.height}@{modeline.refresh_rate}Hz"
+        subprocess.run(["wlr-randr", "--output", requested_connector, "--mode", mode_str], check=True)
+
+    def save_current_mode(self, location):
+        # Implementation similar to your GNOME one but using wlr-randr
+        result = subprocess.run(["wlr-randr", "--json"], capture_output=True, text=True, check=True)
+        data = json.loads(result.stdout)[0] # Grab first monitor
+        curr = data["current_mode"]
+        with open(location, "w") as f:
+            f.write(f"{curr['width']}x{curr['height']}@{curr['refresh']}")
+
+    def read_saved_mode(self, location):
+        with open(location, "r") as f:
+            w, h, r = MODE_REGEX.search(f.read()).groups()
+        return Modeline(width=int(w), height=int(h), refresh_rate=float(r))
+
+    @staticmethod
+    def detect():
+        # Cage doesn't always set XDG_CURRENT_DESKTOP, but it's a wlroots compositor
+        return os.environ.get("WLR_BACKENDS") == "headless" or "cage" in os.environ.get("XDG_CURRENT_DESKTOP", "").lower()