working state
This commit is contained in:
Dawson Matthews
+138
-167
@@ -2,230 +2,201 @@
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional, Union
|
||||
import os
|
||||
|
||||
# Configuration
|
||||
LOG_FORMAT = "%(message)s"
|
||||
logging.basicConfig(level=logging.INFO, format=LOG_FORMAT)
|
||||
logger = logging.getLogger(__name__)
|
||||
VERBOSE = True
|
||||
|
||||
# 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 log(*args):
|
||||
if VERBOSE:
|
||||
print(*args)
|
||||
|
||||
@dataclass
|
||||
class OutputConfig:
|
||||
"""Represents the configuration for a single display output."""
|
||||
data: Dict[str, Any]
|
||||
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}")
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return self.data.get("name", "Unknown")
|
||||
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 is_enabled(self) -> bool:
|
||||
return bool(self.data.get("enabled", False))
|
||||
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)
|
||||
|
||||
def get(self, key: str, default: Any = None) -> Any:
|
||||
return self.data.get(key, default)
|
||||
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
|
||||
|
||||
class KDEProfileManager:
|
||||
"""Manages KDE display profiles using kscreen-doctor."""
|
||||
# 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
|
||||
|
||||
def __init__(self, dry_run: bool = False):
|
||||
self.dry_run = dry_run
|
||||
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 _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}")
|
||||
cmd = f"kscreen-doctor output.{output_name}.{attribute}.{value}"
|
||||
run_command(cmd)
|
||||
|
||||
if self.dry_run:
|
||||
return
|
||||
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)
|
||||
|
||||
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}")
|
||||
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)
|
||||
|
||||
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)
|
||||
outputs = profile.get('outputs', [])
|
||||
|
||||
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
|
||||
# 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"
|
||||
}
|
||||
|
||||
if isinstance(value, (list, dict)):
|
||||
logger.warning(f"Output {output_name} attribute {attribute} has complex value {value}, 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")
|
||||
|
||||
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.")
|
||||
log("Disabling disabled outputs...")
|
||||
for out in outputs:
|
||||
if not out.get('enabled'):
|
||||
run_command(f"kscreen-doctor output.{out['name']}.disable")
|
||||
|
||||
self._run_command(f"kscreen-doctor output.{output_name}.{attribute}.{value}")
|
||||
# 2. Apply other attributes
|
||||
for out in outputs:
|
||||
name = out['name']
|
||||
log(f"Configuring output: {name}")
|
||||
|
||||
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
|
||||
# WCG
|
||||
apply_attribute(name, "wcg", out.get('wcg'), bool_enable_map)
|
||||
|
||||
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}")
|
||||
# SDR Brightness
|
||||
apply_attribute(name, "sdr-brightness", out.get('sdr-brightness'))
|
||||
|
||||
# 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)
|
||||
# VRR Policy
|
||||
apply_attribute(name, "vrrpolicy", out.get('vrrPolicy'), vrr_policy_map)
|
||||
|
||||
# Brightness (normalized 0.0-1.0 to 0-100)
|
||||
brightness = output.get("brightness")
|
||||
# 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')
|
||||
if brightness is not None:
|
||||
self._apply_attribute(name, "brightness", int(float(brightness) * 100))
|
||||
apply_attribute(name, "brightness", int(float(brightness) * 100))
|
||||
|
||||
# Max BPC
|
||||
max_bpc = output.get("maxBpc")
|
||||
max_bpc = out.get('maxBpc')
|
||||
if max_bpc == 0:
|
||||
max_bpc = "automatic"
|
||||
self._apply_attribute(name, "maxbpc", max_bpc)
|
||||
apply_attribute(name, "maxbpc", max_bpc)
|
||||
|
||||
# DDC/CI
|
||||
self._apply_attribute(name, "ddcCi", output.get("ddcCiAllowed"), BOOL_ALLOW_MAP)
|
||||
apply_attribute(name, "ddcCi", out.get('ddcCiAllowed'), bool_allow_map)
|
||||
|
||||
# Mirroring
|
||||
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)
|
||||
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)
|
||||
else:
|
||||
self._apply_attribute(name, "mirror", "none")
|
||||
apply_attribute(name, "mirror", "none")
|
||||
|
||||
# ICC Profile
|
||||
self._apply_attribute(name, "iccProfilePath", output.get("iccProfilePath"))
|
||||
icc_path = out.get('iccProfilePath')
|
||||
if icc_path:
|
||||
apply_attribute(name, "iccProfilePath", icc_path)
|
||||
|
||||
# Mode
|
||||
mode_id = output.get("currentModeId")
|
||||
mode_id = out.get('currentModeId')
|
||||
if mode_id:
|
||||
mode_str = self._get_mode_string(output, mode_id)
|
||||
mode_str = get_mode_string(out, mode_id)
|
||||
if mode_str:
|
||||
self._apply_attribute(name, "mode", mode_str)
|
||||
apply_attribute(name, "mode", mode_str)
|
||||
|
||||
# Position
|
||||
pos = output.get("pos")
|
||||
pos = out.get('pos')
|
||||
if pos:
|
||||
self._apply_attribute(name, "position", f"{pos.get('x', 0)},{pos.get('y', 0)}")
|
||||
apply_attribute(name, "position", f"{pos['x']},{pos['y']}")
|
||||
|
||||
# Scale and Rotation
|
||||
self._apply_attribute(name, "scale", output.get("scale"))
|
||||
self._apply_attribute(name, "rotation", output.get("rotation"), ROTATION_MAP)
|
||||
# Scale
|
||||
apply_attribute(name, "scale", out.get('scale'))
|
||||
|
||||
# Priority and Primary status
|
||||
priority = output.get("priority")
|
||||
# Rotation
|
||||
apply_attribute(name, "rotation", out.get('rotation'), rotation_map)
|
||||
|
||||
# Priority and Primary
|
||||
priority = out.get('priority')
|
||||
if priority is not None:
|
||||
if priority == 1:
|
||||
self._run_command(f"kscreen-doctor output.{name}.primary")
|
||||
self._run_command(f"kscreen-doctor output.{name}.priority.{priority}")
|
||||
run_command(f"kscreen-doctor output.{name}.primary")
|
||||
run_command(f"kscreen-doctor output.{name}.priority.{priority}")
|
||||
|
||||
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.")
|
||||
log("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_parser = subparsers.add_parser("save", help="Save current configuration")
|
||||
save_parser.add_argument("profile", type=Path, help="Path to save the profile")
|
||||
# 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)")
|
||||
|
||||
load_parser = subparsers.add_parser("load", help="Load a configuration")
|
||||
load_parser.add_argument("profile", type=Path, help="Path to the profile to load")
|
||||
# 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")
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.verbose:
|
||||
logger.setLevel(logging.DEBUG)
|
||||
|
||||
manager = KDEProfileManager(dry_run=args.dry_run)
|
||||
|
||||
if args.command == "save":
|
||||
manager.save_profile(args.profile)
|
||||
save_profile(args.profile)
|
||||
elif args.command == "load":
|
||||
manager.load_profile(args.profile)
|
||||
load_profile(args.profile)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
|
||||
VERBOSE=1
|
||||
|
||||
|
||||
function log {
|
||||
if [ $VERBOSE -eq 1 ]; then
|
||||
echo "$@"
|
||||
|
||||
Reference in New Issue
Block a user