diff --git a/src/aw_watcher_input/__main__.py b/src/aw_watcher_input/__main__.py index 96bb905..316b6c9 100644 --- a/src/aw_watcher_input/__main__.py +++ b/src/aw_watcher_input/__main__.py @@ -1,4 +1,11 @@ -from aw_watcher_input.main import main +from aw_watcher_input.main import main as _main +from aw_watcher_input.config import parse_args + +def main() -> None: + args = parse_args() + + # Start watcher + _main(args, testing=args.testing) if __name__ == "__main__": main() diff --git a/src/aw_watcher_input/config.py b/src/aw_watcher_input/config.py new file mode 100644 index 0000000..17da7e5 --- /dev/null +++ b/src/aw_watcher_input/config.py @@ -0,0 +1,37 @@ +import argparse +import sys + +from aw_core.config import load_config_toml + +default_config = """ +[aw-watcher-input] +poll_time = 5 + +[aw-watcher-input-testing] +poll_time = 1 +""".strip() + + +def load_config(testing: bool): + section = "aw-watcher-input" + ("-testing" if testing else "") + return load_config_toml("aw-watcher-input", default_config)[section] + + +def parse_args(): + # get testing in a dirty way, because we need it for the config lookup + testing = "--testing" in sys.argv + config = load_config(testing) + + default_poll_time = config["poll_time"] + + parser = argparse.ArgumentParser( + description="A watcher for keyboard and mouse input." + ) + parser.add_argument( + "--testing", dest="testing", action="store_true", help="run in testing mode" + ) + parser.add_argument( + "--poll-time", dest="poll_time", type=float, default=default_poll_time + ) + parsed_args = parser.parse_args() + return parsed_args diff --git a/src/aw_watcher_input/main.py b/src/aw_watcher_input/main.py index 0e4b76d..9ba2b66 100644 --- a/src/aw_watcher_input/main.py +++ b/src/aw_watcher_input/main.py @@ -3,26 +3,21 @@ from time import sleep import aw_client -import click from aw_core import Event from aw_watcher_afk.listeners import KeyboardListener, MouseListener logger = logging.getLogger(__name__) - -@click.command() -@click.option("--testing", is_flag=True) -def main(testing: bool): +def main(args, testing: bool): logging.basicConfig(level=logging.INFO) - logger.info("Starting watcher...") + logger.info("aw_watcher_input started" + (" in testing mode" if testing else "")) client = aw_client.ActivityWatchClient("aw-watcher-input", testing=testing) client.connect() - # Create bucjet + # Create bucket bucket_name = "{}_{}".format(client.client_name, client.client_hostname) eventtype = "os.hid.input" - client.create_bucket(bucket_name, eventtype, queued=False) - poll_time = 5 + client.create_bucket(bucket_name, eventtype) keyboard = KeyboardListener() keyboard.start() @@ -35,9 +30,9 @@ def main(testing: bool): last_run = now # we want to ensure that the polling happens with a predictable cadence - time_to_sleep = poll_time - datetime.now().timestamp() % poll_time + time_to_sleep = args.poll_time - datetime.now().timestamp() % args.poll_time # ensure that the sleep time is between 0 and poll_time (if system time is changed, this might be negative) - time_to_sleep = max(min(time_to_sleep, poll_time), 0) + time_to_sleep = max(min(time_to_sleep, args.poll_time), 0) sleep(time_to_sleep) now = datetime.now(tz=timezone.utc) @@ -53,7 +48,7 @@ def main(testing: bool): pulsetime = 0.0 if all(map(lambda v: v == 0, merged_data.values())): - pulsetime = poll_time + 0.1 + pulsetime = args.poll_time + 0.1 logger.info("No new input") else: logger.info(f"New input: {e}")