sessionctl

#!/usr/bin/python3

# sessiond - standalone X session manager
# Copyright (C) 2020 James Reed
#
# This program is free software: you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free Software
# Foundation, either version 3 of the License, or (at your option) any later
# version.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along with
# this program. If not, see <https://www.gnu.org/licenses/>.

import logging
import sys
from argparse import ArgumentParser
from pathlib import PurePath
from subprocess import run
from time import sleep

from dbus.exceptions import DBusException
from sessiond import Session, Backlight, AudioSink

SESSIOND_SESSION = "sessiond-session.target"
GRAPHICAL_SESSION = "graphical-session.target"
WINDOW_MANAGER_ALIAS = "window-manager.service"


def log_error(msg):
    sys.stderr.write(msg)
    logging.error(msg)


def _run(cmd, **kw):
    return run(cmd.split(), **kw)


def consists_of(unit=GRAPHICAL_SESSION):
    cmd = "systemctl --user show -p ConsistsOf --value {}".format(unit)
    return _run(cmd, capture_output=True, text=True, check=True).stdout.split()


def stop_session(step=0.1, timeout=5):
    def check_active():
        for u in consists_of():
            if Unit(u).active():
                return True
        return False

    Unit(SESSIOND_SESSION).stop()

    while check_active():
        if timeout == 0:
            for n in consists_of():
                u = Unit(n)
                if u.active():
                    u.kill()
            break
        sleep(step)
        timeout -= step


def get_window_manager():
    cmd = "systemctl --user show -p Id --value {}".format(WINDOW_MANAGER_ALIAS)
    wm = _run(cmd, capture_output=True, text=True).stdout.rstrip()
    if wm == WINDOW_MANAGER_ALIAS:
        raise ValueError("No window manager is enabled")
    return wm


class Unit:
    def __init__(self, name):
        self.name = name

    def active(self):
        cmd = "systemctl --user -q is-active {}".format(self.name)
        return _run(cmd).returncode == 0

    def failed(self):
        cmd = "systemctl --user -q is-failed {}".format(self.name)
        return _run(cmd).returncode == 0

    def status(self):
        cmd = "systemctl --user is-active {}".format(self.name)
        return _run(cmd, capture_output=True, text=True).stdout.rstrip()

    def stop(self):
        _run("systemctl --user stop {}".format(self.name), check=True)

    def reset(self):
        if not self.failed():
            return
        _run("systemctl --user reset-failed {}".format(self.name), check=True)

    def kill(self):
        _run("systemctl --user kill {}".format(self.name), check=True)

    def start_wait(self):
        _run("systemctl --user --wait restart {}".format(self.name))
        stop_session()


def status():
    g = Unit(GRAPHICAL_SESSION)
    if not g.active():
        sys.stderr.write("Session is not active\n")
        sys.exit(1)
    units = consists_of()
    units.sort()
    n = len(max(units, key=len))
    status = {}
    for u in units:
        status[u] = Unit(u).status()
    for u, s in status.items():
        print("{u: <{n}} {s}".format(u=u + ":", n=n + 1, s=s))


def get_session():
    try:
        return Session()
    except DBusException as e:
        log_error(
            "Failed to connect to sessiond DBus service: {}".format(
                e.get_dbus_message()
            )
        )
        sys.exit(1)


def add_audiosinks_parser(sp):
    p = sp.add_parser("audiosink", help="Interact with audio sinks")
    p.add_argument("id", nargs="?", metavar="ID")
    p.add_argument(
        "-s",
        "--set",
        type=float,
        metavar="VALUE",
        help="Set audio sink volume",
    )
    p.add_argument(
        "-i",
        "--inc",
        type=float,
        metavar="VALUE",
        help="Increment audio sink volume",
    )
    p.add_argument(
        "-m",
        "--mute",
        action="store_true",
        help="Mute audio sink",
    )
    p.add_argument(
        "-u",
        "--unmute",
        action="store_true",
        help="Unmute audio sink",
    )
    p.add_argument(
        "-t",
        "--toggle-mute",
        action="store_true",
        help="Toggle audio sink mute state",
    )

    return p


if __name__ == "__main__":
    logging.basicConfig(
        filename=".sessionctl.log", filemode="w", encoding="utf-8", level=logging.ERROR
    )

    p = ArgumentParser(description="With no arguments, show session status.")
    subp = p.add_subparsers(dest="cmd")

    r = subp.add_parser("run", help="Run session")
    r.add_argument("service", nargs="?", help="Window manager service to run")

    subp.add_parser("stop", help="Stop the running session")
    subp.add_parser("status", help="Show session status")
    subp.add_parser("lock", help="Lock the session")
    subp.add_parser("unlock", help="Unlock the session")
    subp.add_parser("properties", help="List session properties")

    blp = subp.add_parser("backlight", help="Interact with backlights")
    blp.add_argument("name", nargs="?", metavar="NAME")
    blp.add_argument(
        "-s",
        "--set",
        type=int,
        metavar="VALUE",
        help="Set backlight brightness",
    )
    blp.add_argument(
        "-i",
        "--inc",
        type=int,
        metavar="VALUE",
        help="Increment backlight brightness",
    )

    asp = add_audiosinks_parser(subp)

    subp.add_parser("version", help="Show sessiond version")

    args = p.parse_args()

    if not args.cmd:
        status()
        sys.exit(0)

    if args.cmd == "run":
        if Unit(SESSIOND_SESSION).active():
            log_error("A session is already running")
            sys.exit(1)
        if not args.service:
            try:
                args.service = get_window_manager()
            except ValueError as e:
                log_error(str(e))
                sys.exit(1)
        _run("systemctl --user import-environment", check=True)
        for u in consists_of():
            Unit(u).reset()
        Unit(args.service).start_wait()
    elif args.cmd == "stop":
        stop_session()
    elif args.cmd == "status":
        status()
    elif args.cmd == "lock":
        get_session().lock()
    elif args.cmd == "unlock":
        get_session().unlock()
    elif args.cmd == "properties":
        for k, v in get_session().get_properties().items():
            print("{}={}".format(k, v))
    elif args.cmd == "backlight":

        def check_name():
            if not args.name:
                sys.stderr.write("Backlight NAME required\n\n")
                blp.print_help()
                sys.exit(2)

        if args.set:
            check_name()
            Backlight(args.name).set_brightness(args.set)
        elif args.inc:
            check_name()
            print(Backlight(args.name).inc_brightness(args.inc))
        elif args.name:
            print(Backlight(args.name).brightness)
        else:
            bls = get_session().backlights
            if len(bls) == 0:
                sys.stderr.write("No backlights\n")
                sys.exit(1)
            for o in bls:
                print(PurePath(o).name)
    elif args.cmd == "audiosink":
        try:
            s = get_session()
            assert s.audiosinks
        except DBusException:
            sys.stderr.write(
                "sessiond {} does not support audio sinks\n".format(
                    s.version
                )
            )
            sys.exit(1)

        def check_id():
            if not args.id:
                sys.stderr.write("Audio sink ID required\n\n")
                asp.print_help()
                sys.exit(2)

        def muted(m):
            return "MUTED" if m else "UNMUTED"

        if args.set:
            check_id()
            AudioSink(args.id).set_volume(args.set)
        elif args.inc:
            check_id()
            print(round(AudioSink(args.id).inc_volume(args.inc), 2))
        elif args.mute:
            check_id()
            AudioSink(args.id).set_mute(True)
        elif args.unmute:
            check_id()
            AudioSink(args.id).set_mute(False)
        elif args.toggle_mute:
            check_id()
            print(muted(AudioSink(args.id).toggle_mute()))
        elif args.id:
            a = AudioSink(args.id)
            print(round(a.volume, 2), muted(a.mute))
        else:
            s = get_session()
            if len(s.audiosinks) == 0:
                sys.stderr.write("No audio sinks\n")
                sys.exit(1)
            def_id = 0
            try:
                def_id = int(PurePath(s.default_audiosink).name)
            except ValueError:
                pass
            for o in s.audiosinks:
                a = AudioSink(PurePath(o).name)
                print(
                    "{}: {}{}".format(
                        a.id, a.name, " [DEFAULT]" if a.id == def_id else ""
                    )
                )
    elif args.cmd == "version":
        print(get_session().version)