oscarg/sunshine-snake-glue
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Massive refactor (Zed wouldn't stop complaining about formatting) and

Browse Source 
beginning of proper GNOME implementation.
This commit is contained in:
oscarg
2026-08-15 23:56:17 +12:00
parent a4aa721a59
commit 19b7b89453
15 changed files with 346 additions and 148 deletions
Download Patch File Download Diff File Expand all files Collapse all files
__main__.py
+66 -39
View File
@@ -1,25 +1,32 @@
from pkg import sunshine
from pkg.sunshine.structs import SunshineEnvironmentVariables
from pkg import wayland_display_manager
from pkg.wayland_display_manager.structs import Modeline
from app.vars import Arguments, Instructions, CONFIG_DIR, GEN_SCRIPT_PATH, PREVIOUS_MODE_PATH
from app.errors import Errors
import sys
import os
import stat
import sys
def calculate_closest_mode(sunshine_env: SunshineEnvironmentVariables, supported_modes: list[Modeline]) -> Modeline:
from app.errors import Errors
from app.vars import (
CONFIG_DIR,
GEN_SCRIPT_PATH,
PREVIOUS_MODE_PATH,
Arguments,
Instructions,
)
from pkg import sunshine, wayland_display_manager
from pkg.sunshine.structs import SunshineEnvironmentVariables
from pkg.wayland_display_manager.structs import Modeline
def calculate_closest_mode(
sunshine_env: SunshineEnvironmentVariables, supported_modes: list[Modeline]
) -> Modeline:
def _distance(mode: Modeline):
return(
abs(mode.width - sunshine_env.specs.width) *2 +
abs(mode.height - sunshine_env.specs.height) *2 +
abs(mode.refresh_rate - sunshine_env.specs.fps)
return (
abs(mode.width - sunshine_env.specs.width) * 2
+ abs(mode.height - sunshine_env.specs.height) * 2
+ abs(mode.refresh_rate - sunshine_env.specs.fps)
)
return min(supported_modes, key = _distance)
return min(supported_modes, key=_distance)
def main():
try:
@@ -30,58 +37,78 @@ def main():
sunshine_env = sunshine.load_env_variables()
display_manager = wayland_display_manager.get_display()
if len(sys.argv) -1 < Arguments.CONNECTOR:
if len(sys.argv) - 1 < Arguments.CONNECTOR:
raise SyntaxError(Errors.INSUFFICIENT_INSTRUCTIONS)
match sys.argv[Arguments.INSTRUCTION]:
case Instructions.SET:
display_manager.save_current_mode(PREVIOUS_MODE_PATH)
for i in display_manager.get_modes():
if i.connector == sys.argv[Arguments.CONNECTOR]:
display_modes = i.modes
display_modes = next(
(
i.modes
for i in display_manager.get_modes()
if i.connector == sys.argv[Arguments.CONNECTOR]
),
None,
)
if display_modes:
modeline = calculate_closest_mode(sunshine_env, display_modes)
display_manager.set_mode(
modeline = modeline,
requested_connector = sys.argv[Arguments.CONNECTOR]
modeline=modeline, requested_connector=sys.argv[Arguments.CONNECTOR]
)
else:
raise ValueError(Errors.CONNECTOR_INVALID)
case Instructions.RESET:
modeline = display_manager.read_saved_mode(PREVIOUS_MODE_PATH)
display_manager.set_mode(modeline, sys.argv[Arguments.CONNECTOR])
case Instructions.CREATE:
if len(sys.argv) -1 < Arguments.TEMPLATE:
if len(sys.argv) - 1 < Arguments.TEMPLATE:
raise SyntaxError(Errors.INSUFFICIENT_INSTRUCTIONS)
for i in display_manager.get_modes():
if i.connector == sys.argv[Arguments.CONNECTOR]:
display_modes = i.modes
display_modes = next(
(
i.modes
for i in display_manager.get_modes()
if i.connector == sys.argv[Arguments.CONNECTOR]
),
None,
)
if not display_modes:
raise ValueError(Errors.CONNECTOR_INVALID)
modeline = calculate_closest_mode(sunshine_env, display_modes)
with open(GEN_SCRIPT_PATH, "w") as file:
file.write("#!/bin/bash\n" + sys.argv[Arguments.TEMPLATE].format(
res_x = modeline.width,
res_y = modeline.height,
rate = modeline.refresh_rate
))
_ = file.write(
"#!/bin/bash\n"
+ sys.argv[Arguments.TEMPLATE].format(
res_x=modeline.width,
res_y=modeline.height,
rate=modeline.refresh_rate,
)
)
os.chmod(
GEN_SCRIPT_PATH,
stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR |
stat.S_IRGRP | stat.S_IXGRP |
stat.S_IROTH | stat.S_IXOTH
stat.S_IRUSR
| stat.S_IWUSR
| stat.S_IXUSR
| stat.S_IRGRP
| stat.S_IXGRP
| stat.S_IROTH
| stat.S_IXOTH,
)
case _:
raise ValueError(Errors.CONNECTOR_UNSPECIFIED)
if __name__ == "__main__":
main()
main()
app/errors.py
+5 -2
View File
@@ -1,6 +1,9 @@
from enum import StrEnum
class Errors(StrEnum):
INSUFFICIENT_INSTRUCTIONS = "Not enough supplied instructions! Usage: {instruction} {connector}"
INSUFFICIENT_INSTRUCTIONS = (
"Not enough supplied instructions! Usage: {instruction} {connector}"
)
CONNECTOR_INVALID = "The requested connector is invalid!"
CONNECTOR_UNSPECIFIED = "No requested connector!"
CONNECTOR_UNSPECIFIED = "No requested connector!"
app/vars.py
+6 -2
View File
@@ -1,23 +1,27 @@
from enum import IntEnum, StrEnum
import pathlib
from enum import IntEnum, StrEnum
class Config(StrEnum):
CONFIG_DIR_NAME = ".sunshine-snake-glue"
PREVIOUS_MODE_FILE = "previous_mode"
GEN_SCRIPT_FILE = "script.sh"
CONFIG_DIR = pathlib.Path.home() / Config.CONFIG_DIR_NAME
PREVIOUS_MODE_PATH = CONFIG_DIR / Config.PREVIOUS_MODE_FILE
GEN_SCRIPT_PATH = CONFIG_DIR / Config.GEN_SCRIPT_FILE
class Arguments(IntEnum):
SCRIPT_NAME = 0
INSTRUCTION = 1
CONNECTOR = 2
TEMPLATE = 3
class Instructions(StrEnum):
SET = "set"
RESET = "reset"
CREATE = "create"
CREATE = "create"
pkg/sunshine/__init__.py
+31 -20
View File
@@ -1,29 +1,40 @@
from .env_key import SunshineEnvKey as EnvKey
from .enums import SunshineAudioConfig
from .structs import SunshineAudioConfig, SunshineEnvironmentVariables, SunshineClientSpecs, SunshineClientInfo, SunshineClientOptions
from os import environ
from .enums import SunshineAudioConfig
from .env_key import SunshineEnvKey as EnvKey
from .structs import (
SunshineAudioConfig,
SunshineClientInfo,
SunshineClientOptions,
SunshineClientSpecs,
SunshineEnvironmentVariables,
)
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"
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)
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)
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))
)
)
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
+1 -1
View File
@@ -1,7 +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
+3 -1
View File
@@ -1,5 +1,6 @@
from enum import Enum
class KeyDefaultMixin(str):
def __new__(cls, requested_key: str, default: str):
obj = str.__new__(cls, requested_key)
@@ -7,6 +8,7 @@ class KeyDefaultMixin(str):
obj.default = default
return obj
class SunshineEnvKey(KeyDefaultMixin, Enum):
APP_ID = ("SUNSHINE_APP_ID", "Unknown ID")
APP_NAME = ("SUNSHINE_APP_NAME", "Unknown App Name")
@@ -17,4 +19,4 @@ class SunshineEnvKey(KeyDefaultMixin, Enum):
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")
AUDIO_CONFIG = ("SUNSHINE_CLIENT_AUDIO_CONFIGURATION", "2.0")
pkg/sunshine/structs.py
+5 -1
View File
@@ -2,11 +2,13 @@ from dataclasses import dataclass
from .enums import SunshineAudioConfig
@dataclass
class SunshineClientInfo:
app_id: str
app_name: str
@dataclass
class SunshineClientSpecs:
width: int
@@ -15,14 +17,16 @@ class SunshineClientSpecs:
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
options: SunshineClientOptions
pkg/wayland_display_manager/__init__.py
+10 -6
View File
@@ -1,16 +1,20 @@
from .abstracts import AbstractDisplay
from .enums import DisplayManagerDriver, Error
def get_display(requested_driver: DisplayManagerDriver = DisplayManagerDriver.INFER) -> AbstractDisplay:
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()
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)
raise ValueError(Error.DISPLAY_TYPE_UNKNOWN)
pkg/wayland_display_manager/abstracts.py
+7 -6
View File
@@ -1,6 +1,8 @@
from .structs import MonitorInfo, Modeline
from abc import ABC, abstractmethod
from pathlib import Path
from .structs import Modeline, MonitorInfo
class AbstractDisplay(ABC):
@abstractmethod
@@ -12,15 +14,14 @@ class AbstractDisplay(ABC):
pass
@abstractmethod
def save_current_mode(self, location: str) -> None:
def save_current_mode(self, location: str | Path) -> None:
pass
@abstractmethod
def read_saved_mode(self, location: str) -> Modeline:
def read_saved_mode(self, location: str | Path) -> Modeline:
pass
@staticmethod
@abstractmethod
def detect(self) -> bool:
def detect() -> bool:
pass
pkg/wayland_display_manager/enums.py
+9 -5
View File
@@ -1,10 +1,10 @@
from .abstracts import AbstractDisplay
from . import gnome, plasma, wlroots
from enum import Enum, StrEnum
from typing import Optional, Type
from . import gnome, plasma, wlroots
from .abstracts import AbstractDisplay
class MetaMixin(str):
def __new__(cls, text: str, meta: AbstractDisplay):
obj = str.__new__(cls, text)
@@ -12,6 +12,7 @@ class MetaMixin(str):
obj.meta = meta
return obj
class DisplayManagerDriver(MetaMixin, Enum):
GNOME = ("GNOME (Mutter)", gnome.Display)
PLASMA = ("KDE Plasma 5/6 (KWin)", plasma.Display)
@@ -22,7 +23,10 @@ class DisplayManagerDriver(MetaMixin, Enum):
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."
DISPLAY_INFER_TYPE_FAILURE = (
"Could not infer a compatible display manager from supported list."
)
pkg/wayland_display_manager/gnome/__init__.py
+54 -37
View File
@@ -1,45 +1,67 @@
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`.
from ..abstracts import AbstractDisplay
from ..structs import Modeline, MonitorInfo
from ._gnome_types import GNOMEDisplayConfig
MODE_REGEX_PATTERN = re.compile(r'(\d+)x(\d+)@(\d+\.\d+)')
# Half replaced implementation. More refactoring needs to be done in regards to setting the display.
# Everything will be replaced with dorkbus Mutter stuff.
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)
mutter_display_state = GNOMEDisplayConfig()
mutter_display_state.get_dbus_info()
# 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)
]
primary_connectors = {
m[0]
for lm in mutter_display_state.logical_monitors
if lm.is_primary
for m in lm.monitors
}
result = []
for monitor in mutter_display_state.monitors:
modes = [
Modeline(
mode.width,
mode.height,
mode.refresh_rate,
)
for mode in monitor.modes
]
result.append(
MonitorInfo(
monitor.connector,
monitor.connector in primary_connectors,
modes
)
)
]
def set_mode(self, modeline, requested_connector = None) -> None:
subprocess.run(["gdctl", "set", "--logical-monitor", "--primary", "--monitor", requested_connector, "--mode", str(modeline)])
return result
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)
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:
@@ -49,16 +71,11 @@ class Display(AbstractDisplay):
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)
)
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()
return "gnome" in os.environ.get("XDG_CURRENT_DESKTOP", "").lower()
pkg/wayland_display_manager/gnome/_gnome_types.py
+104
View File
@@ -0,0 +1,104 @@
from dataclasses import dataclass, field
from typing import Any
from gi.repository import Gio, GLib
# This is going to need some blast zone.
@dataclass
class _GNOMEDisplayMode:
id: str
width: int
height: int
refresh_rate: float
preferred_scale: float
supported_scales: list[float]
@dataclass
class _GNOMEMonitor:
connector: str
vendor: str
product: str
serial: str
modes: list[_GNOMEDisplayMode]
properties: dict[str, Any]
@dataclass
class _GNOMELogicalMonitor:
x: int
y: int
scale: float
transform: int
is_primary: bool
monitors: list[tuple[str, str, str, str]]
properties: dict[str, Any]
@dataclass
class GNOMEDisplayConfig:
serial: int = 0
monitors: list[_GNOMEMonitor] = field(default_factory=list)
logical_monitors: list[_GNOMELogicalMonitor] = field(default_factory=list)
global_properties: dict[str, Any] = field(default_factory=dict)
def get_dbus_info(self):
bus = Gio.bus_get_sync(Gio.BusType.SESSION, None)
# Some beautiful dorkbus
res = bus.call_sync(
"org.gnome.Mutter.DisplayConfig",
"/org/gnome/Mutter/DisplayConfig",
"org.gnome.Mutter.DisplayConfig",
"GetCurrentState",
None,
None,
Gio.DBusCallFlags.NONE,
-1,
None
)
# Serial, Monitors, Logical Monitors, Properties
dbus_serial, monitors, logical_monitors, global_props = res.unpack()
self.serial = dbus_serial
self.global_properties = global_props
self.logical_monitors = [
_GNOMELogicalMonitor(
x=lm[0],
y=lm[1],
scale=lm[2],
transform=lm[3],
is_primary=lm[4],
monitors=lm[5],
properties=lm[6],
)
for lm in logical_monitors
]
self.monitors = []
for info, modes_data, monitor_props in monitors:
connector, vendor, product, monitor_serial = info
modes = [
_GNOMEDisplayMode(
id=m[0],
width=m[1],
height=m[2],
refresh_rate=m[3],
preferred_scale=m[4],
supported_scales=m[5],
)
for m in modes_data
]
monitor = _GNOMEMonitor(
connector=connector,
vendor=vendor,
product=product,
serial=monitor_serial,
modes=modes,
properties=monitor_props,
)
self.monitors.append(monitor)
pkg/wayland_display_manager/plasma/__init__.py
+3 -2
View File
@@ -1,6 +1,7 @@
import os
from ..abstracts import AbstractDisplay
import os
class Display(AbstractDisplay):
def get_modes(self):
@@ -11,4 +12,4 @@ class Display(AbstractDisplay):
@staticmethod
def detect():
return "plasma" in os.environ.get("XDG_CURRENT_DESKTOP", "").lower()
return "plasma" in os.environ.get("XDG_CURRENT_DESKTOP", "").lower()
pkg/wayland_display_manager/structs.py
+4 -3
View File
@@ -1,20 +1,21 @@
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
+38 -23
View File
@@ -1,46 +1,58 @@
from ..structs import MonitorInfo, Modeline
from ..abstracts import AbstractDisplay
import json
import os
import re
import subprocess
import json
from ..abstracts import AbstractDisplay
from ..structs import Modeline, MonitorInfo
# Matches "1920x1080 @ 60.000 Hz" or "1920x1080@60.000"
MODE_REGEX = re.compile(r'(\d+)x(\d+)@(\d+\.\d+)')
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)
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
))
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)
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
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']}")
@@ -53,4 +65,7 @@ class Display(AbstractDisplay):
@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()
return (
os.environ.get("WLR_BACKENDS") == "headless"
or "cage" in os.environ.get("XDG_CURRENT_DESKTOP", "").lower()
)