feat: add toml config and arg parsing for poll time

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()

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

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}")