Tyrell/tyrell.py

from os import name as _OS_NAME, environ as _env
from os import listdir as _listdir, mkdir as _mkdir, chown as _chown
from os.path import exists as _exists, join as _join, isdir as _isdir
from copy import copy as _copy
from time import sleep as _sleep

from typing import Dict as _Dict, List as _List, Optional as _Option, Any as _Any

VERSION: str = '1.1-rel'

try:
    from click import echo as _echo, style as _style, command as _command, argument as _arg
except ImportError:
    print("ERROR Please activate a virtual environment and install requirements.txt")
    exit()

def _print_err(msg: str):
    _echo(_style("ERROR", fg="bright_red") + " " + msg)

def _print_warn(msg: str):
    _echo(_style("WARN ", fg="bright_yellow") + " " + msg)

def _print_ok(msg: str):
    _echo(_style(" OK  ", fg="bright_green") + " " + msg)

def _print_on(msg: str):
    _echo(_style(" ON  ", fg="bright_green") + " " + msg)

def _print_off(msg: str):
    _echo(_style(" OFF ", fg="bright_red") + " " + msg)

def _print_info(msg: str):
    print("      " + msg)

from keyboard import press as _key_down, release as _key_up, KeyboardEvent as _KeyboardEvent, hook as _key_hook, unhook as _key_unhook

platform: str = ""
os_name = _OS_NAME.lower()
if "linux" in os_name or "posix" in os_name:
    platform = "LINUX"
elif "windows" in os_name or "nt" in os_name:
    platform = "WINDOWS"
elif "darwin" in os_name:
    platform = "MACOS"
else:
    _print_err("Platform 'Linux', 'Windows' or 'MacOS' expected")
    _print_info(f"Platform: {_OS_NAME}")
    exit()

if platform == 'LINUX':
    # It appears those under LINUX, either don't work or work but play with mouse position
    from Xlib.display import Display as _Display
    _display = _Display(_env['DISPLAY'])
    from Xlib.X import ButtonPress as _ButtonPress, ButtonRelease as _ButtonRelease
    from Xlib.ext.xtest import fake_input as _fake_input
    from pyautogui import LEFT as _LEFT, RIGHT as _RIGHT, MIDDLE as _MIDDLE
    _BUTTON_MAP: _Dict[str, int] = {_LEFT: 1, _MIDDLE:3, _RIGHT: 3}
    def _mouse_down(button: str=_LEFT):
        assert button in _BUTTON_MAP.keys(), "button argument not in ('left', 'middle', 'right')"
        btn = _BUTTON_MAP[button]
        _fake_input(_display, _ButtonPress, btn)
        _display.sync()
    def _mouse_up(button: str=_LEFT):
        assert button in _BUTTON_MAP.keys(), "button argument not in ('left', 'middle', 'right')"
        btn = _BUTTON_MAP[button]
        _fake_input(_display, _ButtonRelease, btn)
        _display.sync()
else:
    # We can simply use the mouse module under WINDOWS and MACOS
    from mouse import press as _mouse_down, release as _mouse_up

from toml import load as _toml_load, TomlDecodeError as _TomlDecodeError

from asyncio import run as _run, sleep as _asleep, TaskGroup as _TaskGroup

def _profile_pre_check(profile_name: str):
    if profile_name.endswith(".toml"):
        profile_name = profile_name.removesuffix(".toml")
    if not _exists(profile_name):
        _mkdir(profile_name)
        if "SUDO_USER" in _env:
            _chown(profile_name, int(_env['SUDO_UID']), int(_env['SUDO_GID']))
    if not _exists(_join(profile_name, "keys")):
        _mkdir(_join(profile_name, "keys"))
        if "SUDO_USER" in _env:
            _chown(_join(profile_name, "keys"), int(_env['SUDO_UID']), int(_env['SUDO_GID']))

def _write_profile(profile_name: str):
    if profile_name.endswith(".toml"):
        profile_name = profile_name.removesuffix(".toml")
    _profile_pre_check(profile_name)
    with open(_join(profile_name, profile_name+".toml"), "w") as f:
        f.writelines("\n".join([
            f'# Tyrell :: v{VERSION} :: More human than human',
            '',
            '# Placeholders',
            '',
            "# Enable '{name}', emitting profile name",
            "placeholder_name = 'true'",
            '',
            '# Tick speed',
            '# Value in milliseconds',
            "# Defaults to 20 ticks per second",
            "# Can't be overridden",
            'tick = 50',
            '',
            '# What is the key to activate/deactivate Tyrell',
            "# This hotkey must be in the format, 'ctrl+shift+a' (user holds ctrl, shift and 'a' at once)",
            "# Can't be overridden",
            "activator = '`'",
            '',
            '# What is the key to display help',
            "# This hotkey must be in the format, 'ctrl+shift+a' (user holds ctrl, shift and 'a' at once)",
            "# Defaults to '?'",
            "# Can't be overridden",
            "helper = 'shift+/'",
            '',
            '# These can be overridden per key-bind',
            '',
            '# Delay between each keys',
            '# Value in milliseconds',
            'delay = 50',
            '',
            '# Hold delay for each key',
            '# Keys are sent like below:',
            '# 1. key down',
            '# 2. hold delay',
            '# 3. key up',
            '# 4. hold delay',
            '# 5. delay (see above)',
            '# 6. Repeat for keys',
            '# Value in milliseconds',
            'hold = 20',
        ]))
    if "SUDO_USER" in _env:
        _chown(_join(profile_name, profile_name+".toml"), int(_env['SUDO_UID']), int(_env['SUDO_GID']))

def _write_example(profile_name: str):
    if profile_name.endswith(".toml"):
        profile_name = profile_name.removesuffix(".toml")
    _profile_pre_check(profile_name)
    with open(_join(profile_name, "keys", "example.toml"), "w") as f:
        f.writelines("\n".join([
            '# Example Key-bind',
            '',
            "# When 'h' is pressed, write 'Hello World!'",
            '',
            '# Do we want this key-bind active on start-up',
            "# This is a simple example, it won't use this",
            '#startup = true',
            '',
            '# Key to trigger on',
            "# This hotkey must be in the format, 'ctrl+shift+a' (user holds ctrl, shift and 'a' at once)",
            "key = 'h'",
            '',
            '# Kind of action to perform',
            '# click       -> Click a mouse button',
            '# click down  -> Click a mouse button down, but never release up',
            '# key         -> Type the following keys, pressing each key',
            '# key down    -> Press the following keys down, never releasing them',
            '# mirror      -> When the mirror key is pressed down so is key or button, on release so are key or button',
            '#                See also example_mirror.toml',
            "kind = 'key'",
            '',
            '# Type this',
            "# This is required for 'key' and 'key down' kinds (Optional for 'mirror')",
            "# This hotkey must be in the format, 'ctrl+shift+a' (user holds ctrl, shift and 'a' at once)",
            "# This example will write 'Hello World!'",
            "write = 'shift+h, e, l, l, o, space, shift+w, o, r, l, d, shift+1'",
            '',
            '# If this should repeat X number of ticks',
            "# This is a simple example, it won't use this",
            '# Assuming tick speed is 20ms, this would occur every second',
            '#duration = 50',
            '',
            '# Use this mouse button',
            "# This is a simple example, it won't use this",
            "# This is required for 'click' and 'click down' kinds (Optional for 'mirror')",
            "# There currently is 'left' or 'right'",
            "#button = 'left'",
            '',
            '# Mirror key',
            "# This is a simple example, it won't use this",
            '# This is the key whose state will be mirrored',
            '# See also example_mirror.toml',
            '#',
            "# This hotkey must be in the format, 'ctrl+shift+a' (user holds ctrl, shift and 'a' at once)",
            "# This example will press 'g' when 'h' is down, and release 'g' when 'h' is up",
            "#mirror = 'g'",
            '',
            '# Optional Overrides',
            '',
            '# Override delay between each keys',
            '# Value in milliseconds',
            'delay = 50',
            '',
            '# Override hold delay for each key',
            '# Keys are sent like below:',
            '# 1. key down',
            '# 2. hold delay',
            '# 3. key up',
            '# 4. hold delay',
            '# 5. delay (see above)',
            '# 6. Repeat for keys',
            '# Value in milliseconds',
            'hold = 20',
        ]))
    if "SUDO_USER" in _env:
        _chown(_join(profile_name, "keys", "example.toml"), int(_env['SUDO_UID']), int(_env['SUDO_GID']))

def _write_example_mirror(profile_name: str):
    if profile_name.endswith(".toml"):
        profile_name = profile_name.removesuffix(".toml")
    _profile_pre_check(profile_name)
    with open(_join(profile_name, "keys", "example_mirror.toml"), "w") as f:
        f.writelines("\n".join([
            '# Example Mirror Key-bind',
            '',
            "# Press alt when 'w' is down, and release when 'w' is up",
            "# Use 'z' to toggle on/off",
            '',
            '# Do we want this key-bind active on start-up',
            "# This is a simple example, it won't use this",
            '#startup = true',
            '',
            '# Key to toggle',
            "key = 'z'",
            '',
            '# Kind of event',
            "kind = 'mirror'",
            '',
            '# Mirror what key',
            "mirror = 'w'",
            '',
            '# When mirror is down, so will this',
            '# When mirror is up, so will this',
            "write = 'alt'",
            '',
            '# Or a mouse button could be configured',
            "# But this example doesn't",
            "#button = 'right'",
            '',
            '# You could override delay and hold',
            '# Tyrell will turn them off for Mirror kinds automatically',
        ]))
    if "SUDO_USER" in _env:
        _chown(_join(profile_name, "keys", "example_mirror.toml"), int(_env['SUDO_UID']), int(_env['SUDO_GID']))

def _match_key(key: str, event: _KeyboardEvent) -> bool:
    # Process modifiers
    key_mods: _List[str] = []
    k = _copy(key)
    if 'shift' in k:
        k = k.replace('shift+', '')
        key_mods.append('shift')
    elif 'ctrl' in k:
        k = k.replace('ctrl+', '')
        key_mods.append('ctrl')
    elif 'alt' in k:
        k = k.replace('alt+', '')
        key_mods.append('alt')
    # Check Modifiers
    if len(key_mods) == 0 and event.modifiers is not None and len(event.modifiers) != 0:
        # We register no modifiers
        # Given modifiers (not the same)
        return False
    if len(key_mods) != 0:
        mods_ok: _Dict[str, bool] = {}
        for mod in key_mods:
            mods_ok[mod] = False
        if len(key_mods) != 0 and event.modifiers is None:
            # We register modifiers
            # Given no modifiers (not the same)
            return False
        for mod in key_mods:
            for m in event.modifiers:
                if mod == m:
                    mods_ok[mod] = True
        for mod in mods_ok:
            val = mods_ok[mod]
            if val is False:
                return False
        if len(k) == 0:
            # It was only a modifier
            return True
    return event.name == k

def _key_write(write: str, delay:float=50.0, hold:float=20.0):
    if ', ' in write:
        for part in write.split(', '):
            if '+' in part:
                mod, key = part.split('+', maxsplit=1)
                _key_down(mod)
                _key_down(key)
                _sleep(hold / 1000)
                _key_up(key)
                _key_up(mod)
            else:
                _key_down(part)
                _sleep(hold / 1000)
                _key_up(part)
            _sleep(delay / 1000)
    else:
        if '+' in write:
            mod, key = write.split('+', maxsplit=1)
            _key_down(mod)
            _key_down(key)
            _sleep(hold / 1000)
            _key_up(key)
            _key_up(mod)
        else:
            _key_down(write)
            _sleep(hold / 1000)
            _key_up(write)
        _sleep(delay / 1000)

class Action:
    # Toggle to determine if this action is 'on' or 'off'
    _state: bool
    # Key to trigger on (toggle on mirror)
    key: str
    # Kind of action to perform ('click', 'click down', 'key', 'key down', 'mirror')
    kind: str
    # Optional, keys to send
    write: _Option[str]
    # Optional, Repeat X number of ticks
    duration: _Option[int]
    # Internal record of where the duration is
    elapsed: _Option[int]
    # Optional, mouse button to send
    button: _Option[str]
    # Optional, Key to mirror
    mirror: _Option[str]
    # Optional, Override, Delay between each key (in milliseconds)
    delay: _Option[int]
    # Optional, Override, Hold delay for each key (in milliseconds)
    hold: _Option[int]
    def __init__(self, data: _Dict[str, _Option[_Any]]):
        # Validate data!
        assert 'key' in data, "Action missing 'key'"
        assert 'kind' in data, "Action missing 'kind'"
        assert data['kind'] in ('click', 'click down', 'key', 'key down', 'mirror'), f"Action 'kind' is unsupported, got '{data['kind']}'"
        assert 'write' in data or 'button' in data, "Action missing 'write' or 'button' (one needed)"
        # Assign class data
        if 'startup' in data:
            self._state = True
        else:
            self._state = False
        self.key = str(data['key'])
        self.kind = str(data['kind'])
        if 'write' in data:
            self.write = str(data['write'])
        else:
            self.write = None
        if 'button' in data:
            self.button = str(data['button'])
        else:
            self.button = None
        if 'mirror' in data:
            self.mirror = str(data['mirror'])
        else:
            self.mirror = None
        if 'duration' in data and data['duration'] is not None:
            self.duration = int(data['duration'])
            self.elapsed = None
        else:
            self.duration = None
            self.elapsed = None
        if 'delay' in data and data['delay'] is not None:
            self.delay = int(data['delay'])
        else:
            self.delay = None
        if 'hold' in data and data['hold'] is not None:
            self.hold = int(data['hold'])
        else:
            self.hold = None
    
    def is_on(self) -> bool:
        return self._state
    
    def toggle(self, on: bool = False, off: bool = False) -> bool:
        if (not on and not off) or (on and off):
            self._state = not self._state
        elif on and not off:
            self._state = True
        elif off and not on:
            self._state = False
        if self._state is False and self.is_ticker():
            # Reset elapsed time when set to off
            if self.elapsed is not None:
                self.elapsed = None
        elif self._state is True and self.is_ticker():
            # Start elapsed time when set to on
            if self.elapsed is None:
                self.elapsed = self.duration
        return self._state
    
    def is_ticker(self) -> bool:
        return self.duration is not None
    
    def tick(self) -> bool:
        if not self.is_ticker():
            return False
        if self.elapsed is None:
            return False
        self.elapsed -= 1
        if self.elapsed <= 0:
            self.elapsed = self.duration
            return True
        return False
    
    def trigger_key(self, event: _KeyboardEvent) -> bool:
        return _match_key(self.key, event)
    
    def trigger_mirror(self, event: _KeyboardEvent) -> bool:
        if self.kind != 'mirror' or self.mirror is None:
            return False
        return _match_key(self.key, event)
    
    def do(self, global_delay: int=50, global_hold: int=20, mirror: str=""):
        if self.kind == 'key down' and self.write is not None:
            _key_down(self.write)
        elif self.kind == 'click down' and self.button is not None:
            _mouse_down(self.button)
        elif self.kind == 'key' and self.write is not None:
            if self.hold is not None:
                if self.delay is not None:
                    _key_write(self.write, delay=self.delay, hold=self.hold)
                else:
                    _key_write(self.write, delay=global_delay, hold=self.hold)
            else:
                if self.delay is not None:
                    _key_write(self.write, delay=self.delay, hold=global_hold)
                else:
                    _key_write(self.write, delay=global_delay, hold=global_hold)
        elif self.kind == 'click' and self.button is not None:
            _mouse_down(self.button)
            if self.hold is not None:
                _sleep(self.hold / 1000)
            else:
                _sleep(global_hold / 1000)
            _mouse_up(self.button)
            if self.delay is not None:
                _sleep(self.delay / 1000)
            else:
                _sleep(global_delay / 1000)
        elif self.kind == 'mirror':
            if self.write is not None:
                if mirror == 'down':
                    _key_down(self.write)
                elif mirror == 'up':
                    _key_up(self.write)
            elif self.button is not None:
                if mirror == 'down':
                    _mouse_down(self.button)
                elif mirror == 'up':
                    _mouse_up(self.button)

class Profile:
    name: str
    placeholders: _Dict[str, bool]
    tick: int
    delay: int
    hold: int
    activator: str
    helper: str
    _actions: _Dict[str, Action]
    _state: bool
    _help_screen: _List[str]
    def __init__(self, name: str):
        self._state = False
        # Unify name without extension
        if name.endswith(".toml"):
            self.name = name.removesuffix(".toml")
        else:
            self.name = name
        # Pre-Check
        if not _exists(self.name):
            # Does not exist, new everything
            _write_profile(self.name)
            _write_example(self.name)
            _write_example_mirror(self.name)
        else:
            # Profile config?
            if not _exists(_join(self.name, self.name+".toml")):
                _write_profile(self.name)
            # Do we have keys?
            if not _exists(_join(self.name, "keys")):
                _write_example(self.name)
                _write_example_mirror(self.name)
        if self.name == "_example":
            _print_warn("Got example profile, creation only")
            return # Done with example code
        # Load toml config
        with open(_join(self.name, self.name+".toml"), "r") as f:
            dat = _toml_load(f)
            self.placeholders = {
                "name": False,
            }
            self.tick = 50
            self.activator = "`"
            self.helper = "shift+/" # ?
            self.delay = 50
            self.hold = 20
            for key in dat:
                val = dat[key]
                if key.startswith("placeholder_"):
                    if str(val) in ('true', 'TRUE', 'True', 'T', 't', 'yes', 'YES', 'Yes', 'Y', 'y', '1'):
                        self.placeholders[key.removeprefix("placeholder_")] = True
                    else:
                        self.placeholders[key.removeprefix("placeholder_")] = False
                elif key == "tick":
                    self.tick = int(val)
                elif key == "delay":
                    self.delay = int(val)
                elif key == "hold":
                    self.hold = int(val)
                elif key == "activator":
                    self.activator = str(val)
                elif key == "helper":
                    self.helper = str(val)
        # Locate and "load" actions
        self._actions = {}
        self._help_screen = [
            _style(f"{self.activator:15}", fg='bright_magenta') + " Activator",
            _style(f"{self.helper:15}", fg='bright_magenta') + " Help",
            _style(f"{'ctrl+c':15}", fg='bright_red') + " Quit",
            "",
        ]
        for entry in _listdir(_join(self.name, "keys")):
            if entry.startswith(".") or _isdir(entry) or not entry.endswith(".toml"):
                _print_info(f"{_join(self.name, "keys", entry)} => skipped")
                continue
            # Must be a toml file, probably an action
            file = _join(self.name, "keys", entry)
            try:
                data = _toml_load(file)
                a = Action(data)
                if a.kind != 'mirror':
                    if a.duration is not None:
                        if a.duration > 0:
                            d = int(a.duration*self.tick)
                            if d < 1000:
                                self._help_screen.append(_style(f"{a.key:15}", fg='bright_cyan') + f" executes {entry.removesuffix(".toml")} every {a.duration} ticks ({d}ms)")
                            else:
                                df = float(d / 1000.0)
                                self._help_screen.append(_style(f"{a.key:15}", fg='bright_cyan') + f" executes {entry.removesuffix(".toml")} every {a.duration} ticks ({df:0.3}s)")
                        else:
                            self._help_screen.append(_style(f"{a.key:15}", fg='bright_cyan') + f" executes {entry.removesuffix(".toml")} every tick ({int(self.tick)}ms)")
                    else:
                        self._help_screen.append(_style(f"{a.key:15}", fg='bright_cyan') + f" executes {entry.removesuffix(".toml")}")
                else:
                    self._help_screen.append(_style(f"{a.key:15}", fg='bright_cyan') + f" toggles {entry.removesuffix(".toml")} (Mirrors {_style(f'{a.mirror}', fg='bright_cyan')})")
                if a.write is not None:
                    a.write = self.fill_in(a.write)
                self._actions[entry.removesuffix(".toml")] = a
            except _TomlDecodeError:
                _print_warn(f"{file} => invalid toml")

    def fill_in(self, text: str) -> str:
        if self.placeholders['name']:
            if "{name}" in text:
                return text.replace("{name}", self.name)
        return text

    def is_on(self) -> bool:
        return self._state

    def toggle(self, on: bool=False, off: bool=False) -> bool:
        if (not on and not off) or (on and off):
            self._state = not self._state
        elif on and not off:
            self._state = True
        elif off and not on:
            self._state = False
        if self._state:
            for act in self._actions:
                a = self._actions[act]
                if a.is_ticker() and a.is_on():
                    a.toggle(False)
                    _print_off(f"{act} (auto)")
        return self._state
    
    def show_help(self):
        _echo(_style("Tyr", fg='bright_red') + _style("ell", fg='bright_green') + " v" + _style(f"{VERSION}", fg='bright_cyan'))
        _echo("Profile: " + _style(f"{self.name}", fg='bright_yellow'))
        for line in self._help_screen:
            _echo(line)

    def _trigger_activator(self, event: _KeyboardEvent) -> bool:
        return _match_key(self.activator, event)
    
    def _trigger_helper(self, event: _KeyboardEvent) -> bool:
        return _match_key(self.helper, event)
        
    def _callback(self, event: _KeyboardEvent):
        # Quickly process CTRL+C as quit
        if event.modifiers is not None:
            if 'ctrl' in event.modifiers:
                if event.name == 'c':
                    return
        # Process activator and helper (These occur before actions defined)
        if event.event_type == 'up':
            if self._trigger_activator(event):
                if self.toggle():
                    _print_on("Tyrell")
                else:
                    _print_off("Tyrell")
                return
            elif self._trigger_helper(event) and self.is_on():
                _print_ok("Help")
                self.show_help()
                self.toggle(off=True)
                _print_off("Tyrell")
                return
        # Process actions defined
        for act in self._actions:
            a = self._actions[act]
            if a.trigger_key(event) and event.event_type == 'up' and self.is_on():
                if a.kind != 'mirror' and not a.is_ticker():
                    #_print_ok(f"{act} => '{a.kind}' action, trigger on '{a.key}'")
                    _print_ok(f"{act}")
                    a.do(self.delay, self.hold)
                elif a.kind == 'mirror' or a.is_ticker():
                    if a.toggle():
                        _print_on(f"{act}")
                    else:
                        _print_off(f"{act}")
                    #_print_ok(f"{act} => '{a.kind}' action, trigger on '{a.mirror}' (toggle with '{a.key}')")
                self.toggle(off=True)
                _print_off("Tyrell")
                return
            elif a.trigger_mirror(event):
                if a.is_on() and event.event_type is not None:
                    #_print_warn(f"{act} => mirror '{a.mirror}' is {event.event_type}")
                    a.do(self.delay, self.hold, mirror=event.event_type)

    async def _ticker(self):
        while True:
            for act in self._actions:
                a = self._actions[act]
                if a.tick():
                    # Print a status if the duration should exceed 5 seconds
                    # This is so things like afk are more noticeable
                    if a.duration is not None:
                        d: float = (a.duration * self.tick) / 1000.0
                        if d >= 5.0:
                            _print_info(f"* {act} *")
                    a.do(self.delay, self.hold)
            await _asleep(self.tick / 1000)

    async def run(self):
        self.show_help()
        _h = _key_hook(self._callback)
        # I'm not sure why if I don't have any async TaskGroup
        # We get some ugly errors
        """
        Task was destroyed but it is pending!
        task: <Task pending name='Task-3' coro=<BaseEventLoop.shutdown_default_executor() running at /usr/lib/python3.12/asyncio/base_events.py:586>>
        tyrell.py:607: RuntimeWarning: coroutine 'BaseEventLoop.shutdown_default_executor' was never awaited
        RuntimeWarning: Enable tracemalloc to get the object allocation traceback

        OR

        tyrell.py:571: RuntimeWarning: coroutine 'BaseEventLoop.shutdown_default_executor' was never awaited
        RuntimeWarning: Enable tracemalloc to get the object allocation traceback
        """
        async with _TaskGroup() as background:
            background.create_task(self._ticker())
        _key_unhook(_h)


@_command()
@_arg('profile', required=True)
def _main(profile: str):
    p = Profile(profile)
    try:
        _run(p.run())
    except KeyboardInterrupt:
        print()

if __name__ == "__main__":
    if not _exists("_example"):
        _print_warn("Missing '_example' profile, creating...")
        Profile("_example")
        _print_ok("")
    _main()