WIP

.gitignore

@@ -288,3 +288,5 @@ fabric.properties
 # Android studio 3.1+ serialized cache file
 .idea/caches/build_file_checksums.ser
 
+
+.idea/

auto_huami_token/__init__.py

@@ -0,0 +1 @@
+

auto_huami_token/args_handler.py

@@ -0,0 +1,30 @@
+#!/usr/bin/env python3
+
+import argparse
+from argparse import Namespace
+
+class ArgsHandler:
+    def __init__(self):
+        parser = argparse.ArgumentParser(
+            prog = 'auto-huami-token',
+            description = 'Can acquire an amazfit or xiaomi key for your gadget and downloads amazfit and huami agps firmware updates',
+            prefix_chars='+-',
+            exit_on_error=False)
+        parser.add_argument('-s', '--save', action = 'store_true', help = 'Save argument values to the config file')
+        parser.add_argument('-t', '--target-dir', help = 'Move results to this directory after download')
+        parser.add_argument('-c', '--config-path', help = 'Use the specified filepath to load or save config values')
+        parser.add_argument('-l', '--with-logging', dest='log_to', help = "Write log messages to this file or folder, or the systemd journal with 'journal'. If a folder is specified the logfile will be called hf-cli.log. The target folder must exist.")
+        parser.add_argument('+l', '-q', '--without-logging', dest='log_to', action='store_const', const='disabled', help = "Write log messages to this file or folder, or the systemd journal with 'journal'. If a folder is specified the logfile will be called hf-cli.log. The target folder must exist.")
+        parser.add_argument('-v', '--verbose', action='store_true', default=None, help = 'Write each operation to the console')
+        targets_group = parser.add_mutually_exclusive_group()
+        targets_group.add_argument('-C', '--use-config', action='store_true', help = 'Load targets from the specified config file or ./config.json if not specified')
+        targets_group.add_argument('-k', '--keys', action = 'store_true', help = 'Fetch keys')
+        targets_group.add_argument('-i', '--setup', action = 'store_true', help = 'Interactively setup config.json. You will be prompted for values')
+        targets_group.add_argument('-g', '--gps', action = 'store_true', help = 'Fetch agps files',)
+        self.parser = parser
+
+    def parse_args(self, args: list[str]) -> Namespace:
+        return self.parser.parse_args(args)
+
+    def print_help(self):
+        self.parser.print_help()

auto_huami_token/cli.py

@@ -0,0 +1,64 @@
+#!/usr/bin/env python3
+"""Main entrypoint for auto-huami-token."""
+
+import sys
+from argparse import ArgumentError
+
+from huami_token.zepp import LogoutError, Path, ZeppClient, ZeppSession
+from huami_token.helpers import build_gps_uihh
+
+from .args_handler import ArgsHandler
+from .config_error import ConfigError
+from .config_handler import ConfigHandler
+
+
+def main() -> int:
+    handler = ArgsHandler()
+    try:
+        args = handler.parse_args(sys.argv[1:])
+    except ArgumentError as e:
+        print(e.message)
+        print(handler.print_help())
+        exit(4)
+
+    try:
+        config = ConfigHandler(args, "auto-huami-token")
+    except ConfigError:
+        print(
+            "Neither -k, -a or targets where provided and no config file containing targets could be found"
+        )
+        exit(3)
+
+    if args.setup:
+        exit()
+
+    config.log.debug("Initiating run")
+
+    session = ZeppSession(username=args.email, password=args.password)
+    session.login()
+    client = ZeppClient(session)
+
+    if args.keys:
+        devices = client.get_devices()
+        for i, device in enumerate(devices):
+            active = "Yes" if device.active else "No"
+            print(f"Device {i}:")
+            print(f"  MAC: {device.mac}, Active: {active}")
+            print(f"  Key: 0x{device.auth_key}")
+
+    if args.gps:
+        output_dir = Path(config.target_dir)
+        client.download_gps_data(output_dir)
+        build_gps_uihh(base_folder=output_dir)
+
+    try:
+        session.logout()
+        config.log.info("\nLogged out.")
+        return 0
+    except LogoutError:
+        print("\nError logging out.")
+        return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

auto_huami_token/config_error.py

@@ -0,0 +1,4 @@
+#!/usr/bin/env python3
+
+class ConfigError(Exception):
+    pass

auto_huami_token/config_handler.py

@@ -0,0 +1,259 @@
+#!/usr/bin/env python3
+
+import json
+import logging
+import os
+import sys
+import typing
+from argparse import Namespace
+from collections import namedtuple
+from dataclasses import dataclass
+from getpass import getpass
+from logging import StreamHandler
+from types import SimpleNamespace
+
+import psutil
+from xdg import BaseDirectory as Xdg
+
+class ConfigHandler:
+    def __init__(self, args: Namespace, log_name: str):
+        self.config = Config()
+        self.config_path = ConfigHandler.get_config_path(args)
+        new_config = self.load_config()
+
+        self.log = logging.getLogger(log_name)
+        config_change = (
+            self.configure_credentials(args)
+            | self.configure_target_dir(args)
+            | self.configure_logging(args)
+        )
+
+        if config_change and (
+            new_config or args.setup or (getattr(args, "save", False))
+        ):
+            self.log.debug(
+                "saving changed config "
+                + ("because of new config" if new_config else "from args")
+            )
+            self.save_config()
+
+    @staticmethod
+    def get_config_path(args: Namespace) -> str:
+        config_file_path: str = ""
+        if getattr(args, "config_path", False):
+            config_file_path = os.path.abspath(os.path.expanduser(args.config_path))
+        elif Xdg.load_first_config("auto-huami-token") and os.path.exists(
+            Xdg.load_first_config("auto-huami-token") + "/config.json"
+        ):
+            config_file_path = (
+                Xdg.load_first_config("auto-huami-token") + "/config.json"
+            )
+        elif os.path.exists("./config.json"):
+            config_file_path = "./config.json"
+        elif (
+            os.getuid() == 0
+            or psutil.Process(os.getpid()).ppid() == 1
+            and os.path.exists("/etc/auto-huami-token/config.json")
+        ):
+            config_file_path = "/etc/auto-huami-token/config.json"
+        if args.setup:
+            new_config_file_path = ConfigHandler.request_text_input_with_fallback(
+                "Where should the config file be saved?", config_file_path
+            )
+            if new_config_file_path:
+                config_file_path = new_config_file_path
+        return (
+            config_file_path
+            or Xdg.save_config_path("auto-huami-token") + "/config.json"
+        )
+
+    def configure_credentials(self, args: Namespace) -> bool:
+        config_change = args.setup
+        self.config.credentials = credentials = getattr(
+            self.config, "credentials", SimpleNamespace()
+        )
+        if not hasattr(credentials, "username") or args.setup:
+            current_username = getattr(credentials, "username", "")
+            credentials.username = ConfigHandler.request_text_input_with_fallback(
+                "Please enter your username.", current_username
+            )
+            self.log.debug("config.credentials.email has changed")
+            config_change = True
+        if not hasattr(self.config.credentials, "password") or args.setup:
+            credentials.password = ConfigHandler.request_password()
+            self.log.debug("config.credentials.password has changed")
+            config_change = True
+        return config_change
+
+    def configure_target_dir(self, args: Namespace) -> bool:
+        config_change = args.setup
+        if getattr(args, "target_dir", False):
+            abs_target_dir = os.path.abspath(os.path.expanduser(args.target_dir))
+            config_change = not hasattr(self.config, "target_dir") or (
+                self.config.target_dir != abs_target_dir
+            )
+            self.config.target_dir = abs_target_dir
+            self.log.debug("config.target_dir has changed")
+        elif not getattr(self.config, "target_dir", False):
+            self.config.target_dir = "."
+        if args.setup:
+            self.config.target_dir = ConfigHandler.request_text_input_with_fallback(
+                "Where should the downloaded file(s) be saved?", self.config.target_dir
+            )
+            self.config.target_dir = (
+                self.config.target_dir
+                if os.path.isabs(self.config.target_dir)
+                else os.path.abspath(os.path.expanduser(self.config.target_dir))
+            )
+        return config_change
+
+    def configure_logging(self, args: Namespace) -> bool:
+        config_change = False
+        if hasattr(args, "log_to") and args.log_to is not None:
+            self.config.log_to = args.log_to
+        if hasattr(args, "verbose") and args.verbose is not None:
+            self.config.verbose = args.verbose
+        if args.setup:
+            self.config.log_to = ConfigHandler.request_text_input_with_fallback(
+                "Where should logs be written, when not running as a systemd service? A path or 'disabled', which disables logging or '' to use the default location '~/.local/state/hf_cli'",
+                getattr(self.config, "log_to", ""),
+            )
+            self.config.verbose = ConfigHandler.request_boolean_with_fallback(
+                "Would you like to always log verbosely?",
+                getattr(self.config, "verbose", False),
+            )
+        if (
+            hasattr(self.config, "log_to")
+            and self.config.log_to is not None
+            and self.config.log_to != "disabled"
+        ):
+            if "JOURNAL_STREAM" in os.environ:
+                from systemd.journal import JournalHandler
+
+                self.log.addHandler(JournalHandler())
+            else:
+                if (
+                    self.config.log_to
+                    and not os.path.isabs(self.config.log_to)
+                    and (
+                        os.path.exists(self.config.log_to)
+                        or os.path.exists(os.path.dirname(self.config.log_to))
+                    )
+                ):
+                    self.config.log_to = os.path.abspath(
+                        os.path.expanduser(self.config.log_to)
+                    )
+                    self.config.log_to = (
+                        self.config.log_to + "/auto-huami-token.log"
+                        if os.path.isdir(self.config.log_to)
+                        else self.config.log_to
+                    )
+                if os.path.isabs(self.config.log_to):
+                    from logging import FileHandler
+
+                    self.config.log_to = self.config.log_to + (
+                        "/auto-huami-token.log"
+                        if os.path.isdir(self.config.log_to)
+                        else ""
+                    )
+                    if not os.path.exists(os.path.dirname(self.config.log_to)):
+                        os.makedirs(os.path.dirname(self.config.log_to), exist_ok=True)
+                    self.log.addHandler(FileHandler(self.config.log_to))
+                    config_change = True
+                elif self.config.log_to != "console":
+                    from logging import FileHandler
+
+                    self.config.log_to = (
+                        Xdg.save_state_path("auto-huami-token")
+                        + "/auto-huami-token.log"
+                    )
+                    self.log.addHandler(FileHandler(self.config.log_to))
+                    config_change = True
+                self.log.addHandler(StreamHandler(sys.stdout))
+            self.log.setLevel(
+                logging.DEBUG
+                if getattr(self.config, "verbose", False)
+                else logging.INFO
+            )
+        else:
+            self.log.propagate = False
+            self.log.disabled = True
+            config_change = (
+                args.setup
+                or hasattr(args, "log_to")
+                and args.log_to is not None
+                and self.config.log_to == "disabled"
+            )
+        return config_change
+
+    @property
+    def target_dir(self) -> str:
+        return self.config.target_dir
+
+    def load_config(self) -> bool:
+        self.config = self.load_config_json(self.config_path)
+        if self.config is None:
+            self.config = SimpleNamespace()
+            return True
+        return False
+
+    @staticmethod
+    def load_config_json(file_path: str):
+        try:
+            with open(file_path) as config_file:
+                return json.load(
+                    config_file,
+                    object_hook=lambda d: namedtuple("X", d.keys())(*d.values()),
+                )
+        except IOError:
+            return None
+
+    def save_config(self):
+        self.log.debug("Saving config")
+        if typing.TYPE_CHECKING:
+            from _typeshed import SupportsWrite
+
+            config_file: SupportsWrite[str]
+        with open(self.config_path, "w") as config_file:
+            json.dump(self.config, config_file, default=lambda x: vars(x))
+
+    @staticmethod
+    def request_password() -> str:
+        return getpass("Please enter your password: ")
+
+    @staticmethod
+    def request_boolean_with_fallback(question: str, current: bool = False) -> bool:
+        result = ConfigHandler.request_text_input_with_fallback(
+            f"{question} (yes/no) (currently {'yes' if current else 'no'}): ",
+            "yes" if current else "no",
+        ).lower()
+        if not result.startswith(("y", "n")):
+            raise ValueError(
+                f"'{result}' is invalid. Only yes and no are valid choices"
+            )
+        return True if result.startswith("y") else False
+
+    @staticmethod
+    def request_text_input_with_fallback(question: str, current: str = "") -> str:
+        response = ConfigHandler.request_input(question + f" (currently '{current}'): ")
+        if not response:
+            return "" if current is None else current
+        return response
+
+    @staticmethod
+    def request_input(message: str):
+        return input(message)
+
+
+@dataclass
+class Credentials:
+    email: str
+    password: str
+
+
+@dataclass
+class Config:
+    target_dir: str
+    log_to: str
+    verbose: bool
+    credentials: Credentials

project.toml

@@ -0,0 +1,57 @@
+[project]
+name = "auto_huami_token"
+version = "0.0.1"
+description = "A wrapper to automate calling huami-token on a schedule."
+readme = "README.md"
+requires-python = ">=3.14"
+authors = [
+  { name = "Daniel Demus", email = "daniel-git@demus.dk" },
+]
+classifiers = [
+  "Programming Language :: Python :: 3",
+  "License :: OSI Approved :: MIT License",
+  "Operating System :: OS Independent",
+]
+
+dependencies = [
+    "argcomplete",
+    "requests",
+    "psutil",
+    "systemd-python",
+    "pyxdg >= 0.28",
+    "huami_token >= 0.8.0"
+]
+
+[project.scripts]
+auto-huami-token = "auto_huami_token.cli:main"
+
+[project.optional-dependencies]
+dev = [
+  "pytest>=8,<9",
+  "types-requests>=2.31.0.20231231",
+  "ruff>=0.5,<1",
+  "mypy>=1.10,<2",
+]
+
+[project.urls]
+"Homepage" = "https://git.demus.dk/demus/auto-huami-token"
+"Bug Tracker" = "https://git.demus.dk/demus/auto-huami-token/issues"
+
+[build-system]
+requires = ["hatchling>=1.25"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["auto_huami_token"]
+
+[tool.pytest.ini_options]
+markers = ["integration: real API integration tests (require --email and --password)"]
+
+[tool.ruff]
+line-length = 100
+target-version = "py314"
+
+[tool.mypy]
+python_version = "3.14"
+warn_unused_ignores = true
+warn_redundant_casts = true

tests/__init__.py

@@ -0,0 +1 @@
+