refactored

This commit is contained in:
Dawson Matthews
2026-03-08 18:12:19 -06:00
parent acd351f559
commit 41047adac3
+167 -138
View File
@@ -2,201 +2,230 @@
import argparse
import json
import logging
import os
import subprocess
import sys
import os
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict, List, Optional, Union
VERBOSE = True
# Configuration
LOG_FORMAT = "%(message)s"
logging.basicConfig(level=logging.INFO, format=LOG_FORMAT)
logger = logging.getLogger(__name__)
def log(*args):
if VERBOSE:
print(*args)
# Constants for attribute mappings
BOOL_ENABLE_MAP = {"true": "enable", "false": "disable"}
BOOL_ALLOW_MAP = {"true": "allow", "false": "disallow"}
RGB_RANGE_MAP = {
"0": "automatic", "1": "full", "2": "limited",
"automatic": "automatic", "full": "full", "limited": "limited"
}
ROTATION_MAP = {"1": "normal", "2": "left", "4": "inverted", "8": "right"}
VRR_POLICY_MAP = {
"0": "never", "1": "always", "2": "automatic",
"never": "never", "always": "always", "automatic": "automatic"
}
def run_command(cmd):
log(f"[CMD] {cmd}")
try:
subprocess.run(cmd, shell=True, check=True)
except subprocess.CalledProcessError as e:
log(f"Error executing command: {e}")
@dataclass
class OutputConfig:
"""Represents the configuration for a single display output."""
data: Dict[str, Any]
def get_mode_string(output, mode_id):
for mode in output.get('modes', []):
if mode['id'] == mode_id:
width = mode['size']['width']
height = mode['size']['height']
refresh_rate = round(mode['refreshRate'])
return f"{width}x{height}@{refresh_rate}"
return None
@property
def name(self) -> str:
return self.data.get("name", "Unknown")
def save_profile(profile_path):
log(f"Saving current display profile to {profile_path}...")
try:
with open(profile_path, 'w') as f:
subprocess.run(['kscreen-doctor', '--json'], stdout=f, check=True)
log("Profile saved successfully.")
except Exception as e:
print(f"Error saving profile: {e}", file=sys.stderr)
sys.exit(1)
@property
def is_enabled(self) -> bool:
return bool(self.data.get("enabled", False))
def apply_attribute(output_name, attribute, value, value_map=None):
if value is None:
log(f"Output {output_name} has no attribute {attribute}, skipping...")
return
def get(self, key: str, default: Any = None) -> Any:
return self.data.get(key, default)
# If it's a list or dict, something might be wrong or it's a complex structure we don't handle this way
if isinstance(value, (list, dict)):
log(f"Output {output_name} attribute {attribute} has complex value {value}, skipping...")
return
class KDEProfileManager:
"""Manages KDE display profiles using kscreen-doctor."""
str_value = str(value).lower()
if value_map:
if str_value in value_map:
value = value_map[str_value]
else:
log(f"Warning: Value {str_value} not found in map for {attribute}, using as is.")
def __init__(self, dry_run: bool = False):
self.dry_run = dry_run
cmd = f"kscreen-doctor output.{output_name}.{attribute}.{value}"
run_command(cmd)
def _run_command(self, cmd: Union[str, List[str]]) -> None:
"""Executes a shell command."""
cmd_str = cmd if isinstance(cmd, str) else " ".join(cmd)
logger.info(f"[CMD] {cmd_str}")
def load_profile(profile_path):
log(f"Loading display profile from {profile_path}...")
if not os.path.exists(profile_path):
print(f"Profile file not found: {profile_path}", file=sys.stderr)
sys.exit(1)
if self.dry_run:
return
try:
with open(profile_path, 'r') as f:
profile = json.load(f)
except Exception as e:
print(f"Error reading profile JSON: {e}", file=sys.stderr)
sys.exit(1)
try:
subprocess.run(cmd, shell=isinstance(cmd, str), check=True, capture_output=False)
except subprocess.CalledProcessError as e:
logger.error(f"Error executing command: {e}")
outputs = profile.get('outputs', [])
def save_profile(self, profile_path: Path) -> None:
"""Saves the current display configuration to a JSON file."""
logger.info(f"Saving current display profile to {profile_path}...")
try:
profile_path.parent.mkdir(parents=True, exist_ok=True)
with open(profile_path, "w") as f:
subprocess.run(["kscreen-doctor", "--json"], stdout=f, check=True)
logger.info("Profile saved successfully.")
except Exception as e:
logger.error(f"Failed to save profile: {e}")
sys.exit(1)
# Maps from JSON values to kscreen-doctor options
bool_enable_map = {"true": "enable", "false": "disable"}
bool_allow_map = {"true": "allow", "false": "disallow"}
rgb_range_map = {
"0": "automatic", "1": "full", "2": "limited",
"automatic": "automatic", "full": "full", "limited": "limited"
}
rotation_map = {"1": "normal", "2": "left", "4": "inverted", "8": "right"}
vrr_policy_map = {
"0": "never", "1": "always", "2": "automatic",
"never": "never", "always": "always", "automatic": "automatic"
}
def _apply_attribute(self, output_name: str, attribute: str, value: Any, value_map: Optional[Dict[str, str]] = None) -> None:
"""Applies a single attribute to an output."""
if value is None:
logger.debug(f"Output {output_name} has no attribute {attribute}, skipping...")
return
# 1. Handle Enable/Disable (Enable first to avoid no-output state)
log("Enabling enabled outputs...")
for out in outputs:
if out.get('enabled'):
run_command(f"kscreen-doctor output.{out['name']}.enable")
if isinstance(value, (list, dict)):
logger.warning(f"Output {output_name} attribute {attribute} has complex value {value}, skipping...")
return
log("Disabling disabled outputs...")
for out in outputs:
if not out.get('enabled'):
run_command(f"kscreen-doctor output.{out['name']}.disable")
str_value = str(value).lower()
if value_map:
if str_value in value_map:
value = value_map[str_value]
else:
logger.warning(f"Value '{str_value}' not found in map for '{attribute}', using as is.")
# 2. Apply other attributes
for out in outputs:
name = out['name']
log(f"Configuring output: {name}")
self._run_command(f"kscreen-doctor output.{output_name}.{attribute}.{value}")
# WCG
apply_attribute(name, "wcg", out.get('wcg'), bool_enable_map)
def _get_mode_string(self, output: OutputConfig, mode_id: str) -> Optional[str]:
"""Finds the mode string (WxH@R) for a given mode ID."""
for mode in output.get("modes", []):
if str(mode.get("id")) == str(mode_id):
size = mode.get("size", {})
width = size.get("width")
height = size.get("height")
refresh = round(mode.get("refreshRate", 0))
if width and height:
return f"{width}x{height}@{refresh}"
return None
# SDR Brightness
apply_attribute(name, "sdr-brightness", out.get('sdr-brightness'))
def _configure_output(self, output: OutputConfig, all_outputs: List[OutputConfig]) -> None:
"""Configures all attributes for a single output."""
name = output.name
logger.info(f"Configuring output: {name}")
# VRR Policy
apply_attribute(name, "vrrpolicy", out.get('vrrPolicy'), vrr_policy_map)
# Direct mappings
self._apply_attribute(name, "wcg", output.get("wcg"), BOOL_ENABLE_MAP)
self._apply_attribute(name, "sdr-brightness", output.get("sdr-brightness"))
self._apply_attribute(name, "vrrpolicy", output.get("vrrPolicy"), VRR_POLICY_MAP)
self._apply_attribute(name, "rgbrange", output.get("rgbRange"), RGB_RANGE_MAP)
self._apply_attribute(name, "overscan", output.get("overscan"))
self._apply_attribute(name, "hdr", output.get("hdr"), BOOL_ENABLE_MAP)
# RGB Range
apply_attribute(name, "rgbrange", out.get('rgbRange'), rgb_range_map)
# Overscan
apply_attribute(name, "overscan", out.get('overscan'))
# HDR
apply_attribute(name, "hdr", out.get('hdr'), bool_enable_map)
# Brightness (0.0-1.0 to 0-100)
brightness = out.get('brightness')
# Brightness (normalized 0.0-1.0 to 0-100)
brightness = output.get("brightness")
if brightness is not None:
apply_attribute(name, "brightness", int(float(brightness) * 100))
self._apply_attribute(name, "brightness", int(float(brightness) * 100))
# Max BPC
max_bpc = out.get('maxBpc')
max_bpc = output.get("maxBpc")
if max_bpc == 0:
max_bpc = "automatic"
apply_attribute(name, "maxbpc", max_bpc)
self._apply_attribute(name, "maxbpc", max_bpc)
# DDC/CI
apply_attribute(name, "ddcCi", out.get('ddcCiAllowed'), bool_allow_map)
self._apply_attribute(name, "ddcCi", output.get("ddcCiAllowed"), BOOL_ALLOW_MAP)
# Mirroring
replication_source_id = out.get('replicationSource', 0)
if replication_source_id != 0:
# Find the name of the replication source
source_name = "none"
for other_out in outputs:
if other_out.get('id') == replication_source_id:
source_name = other_out['name']
break
apply_attribute(name, "mirror", source_name)
mirror_source_id = output.get("replicationSource", 0)
if mirror_source_id != 0:
source_name = next((o.name for o in all_outputs if o.get("id") == mirror_source_id), "none")
self._apply_attribute(name, "mirror", source_name)
else:
apply_attribute(name, "mirror", "none")
self._apply_attribute(name, "mirror", "none")
# ICC Profile
icc_path = out.get('iccProfilePath')
if icc_path:
apply_attribute(name, "iccProfilePath", icc_path)
self._apply_attribute(name, "iccProfilePath", output.get("iccProfilePath"))
# Mode
mode_id = out.get('currentModeId')
mode_id = output.get("currentModeId")
if mode_id:
mode_str = get_mode_string(out, mode_id)
mode_str = self._get_mode_string(output, mode_id)
if mode_str:
apply_attribute(name, "mode", mode_str)
self._apply_attribute(name, "mode", mode_str)
# Position
pos = out.get('pos')
pos = output.get("pos")
if pos:
apply_attribute(name, "position", f"{pos['x']},{pos['y']}")
self._apply_attribute(name, "position", f"{pos.get('x', 0)},{pos.get('y', 0)}")
# Scale
apply_attribute(name, "scale", out.get('scale'))
# Scale and Rotation
self._apply_attribute(name, "scale", output.get("scale"))
self._apply_attribute(name, "rotation", output.get("rotation"), ROTATION_MAP)
# Rotation
apply_attribute(name, "rotation", out.get('rotation'), rotation_map)
# Priority and Primary
priority = out.get('priority')
# Priority and Primary status
priority = output.get("priority")
if priority is not None:
if priority == 1:
run_command(f"kscreen-doctor output.{name}.primary")
run_command(f"kscreen-doctor output.{name}.priority.{priority}")
self._run_command(f"kscreen-doctor output.{name}.primary")
self._run_command(f"kscreen-doctor output.{name}.priority.{priority}")
log("Display configuration restored.")
def load_profile(self, profile_path: Path) -> None:
"""Loads and applies a display configuration from a JSON file."""
logger.info(f"Loading display profile from {profile_path}...")
if not profile_path.exists():
logger.error(f"Profile file not found: {profile_path}")
sys.exit(1)
try:
with open(profile_path, "r") as f:
data = json.load(f)
except Exception as e:
logger.error(f"Error reading profile JSON: {e}")
sys.exit(1)
outputs = [OutputConfig(out) for out in data.get("outputs", [])]
# 1. Handle Enable/Disable (Enable first to ensure we always have an active display)
logger.info("Enabling requested outputs...")
for out in outputs:
if out.is_enabled:
self._run_command(f"kscreen-doctor output.{out.name}.enable")
logger.info("Disabling requested outputs...")
for out in outputs:
if not out.is_enabled:
self._run_command(f"kscreen-doctor output.{out.name}.disable")
# 2. Configure each output
for out in outputs:
if out.is_enabled:
self._configure_output(out, outputs)
logger.info("Display configuration restored.")
def main():
parser = argparse.ArgumentParser(description="KDE Display Profile Manager")
parser.add_argument("--dry-run", action="store_true", help="Print commands without executing them")
parser.add_argument("--verbose", action="store_true", help="Enable verbose logging")
subparsers = parser.add_subparsers(dest="command", required=True)
# Save command
save_parser = subparsers.add_parser("save", help="Save current display configuration to a profile")
save_parser.add_argument("profile", help="Path to the profile file (e.g., profiles/myprofile.json)")
save_parser = subparsers.add_parser("save", help="Save current configuration")
save_parser.add_argument("profile", type=Path, help="Path to save the profile")
# Load command
load_parser = subparsers.add_parser("load", help="Load a display configuration from a profile")
load_parser.add_argument("profile", help="Path to the profile file")
load_parser = subparsers.add_parser("load", help="Load a configuration")
load_parser.add_argument("profile", type=Path, help="Path to the profile to load")
args = parser.parse_args()
if args.verbose:
logger.setLevel(logging.DEBUG)
manager = KDEProfileManager(dry_run=args.dry_run)
if args.command == "save":
save_profile(args.profile)
manager.save_profile(args.profile)
elif args.command == "load":
load_profile(args.profile)
manager.load_profile(args.profile)
if __name__ == "__main__":
main()