rfidacd/src/rfidacd/daemon.py

"""
"""

import argparse
import logging
import os
import subprocess
import sys
from datetime import datetime as dt

from evdev import InputDevice, ecodes

from rfidacd import __version__

__author__ = "Thomas Verchow"
__copyright__ = "Thomas Verchow"
__license__ = "MIT"

_logger = logging.getLogger(__name__)

SHELL = os.getenv("SHELL", "/bin/sh")


def parse_args(args):
    """
    """

    def dir_path(path: str):
        if os.path.isdir(path):
            return path
        else:
            parser.error(f"{path} is not a readable directory")

    parser = argparse.ArgumentParser(description="RFID action commander daemon")
    parser.add_argument(
        "--version", action="version", version="rfidacd {ver}".format(ver=__version__),
    )
    parser.add_argument(
        "-d", "--device", help="device to listen", type=argparse.FileType("r")
    )
    parser.add_argument(
        "-p", "--path", default=".", help="path to action scripts", type=dir_path
    )
    parser.add_argument(
        "-s", "--seconds", default=5, help="seconds between same rfid", type=int
    )
    parser.add_argument(
        "-v",
        "--verbose",
        dest="loglevel",
        help="set loglevel to INFO",
        action="store_const",
        const=logging.INFO,
    )
    parser.add_argument(
        "-vv",
        "--very-verbose",
        dest="loglevel",
        help="set loglevel to DEBUG",
        action="store_const",
        const=logging.DEBUG,
    )
    return parser.parse_args(args)


def read_rfid(device):
    combined_string = ""
    for event in InputDevice(device.name).read_loop():
        if event.type == ecodes.EV_KEY and event.value == 0:  # value 0 = release key
            if event.code == 28:  # code 28 = KEY_ENTER
                return combined_string
            # [4:5]? .. KEY[] starts with 'KEY_' and we expect one char
            combined_string += ecodes.KEY[event.code][4:5]


def run_action(script: str):
    _logger.info(f"RFID action: run '{SHELL} {script}'")
    try:
        r = subprocess.call([SHELL, script])
    except Exception as err:
        _logger.info(err)
        last_rfid = None


def main(args):
    args = parse_args(args)
    logging.basicConfig(level=args.loglevel)
    _logger.debug(f"Device to read from: {args.device}")
    _logger.debug(f"Shell to execute commands: {SHELL}")
    _logger.debug(f"Expect scripts to run in {args.path}")

    last_rfid = None
    last_ts = dt.now()
    while True:
        rfid = read_rfid(args.device)
        seconds_gone = (dt.now() - last_ts).total_seconds()

        script = os.path.join(args.path, rfid)

        if os.path.islink(script):  # special RFID - always run!
            _logger.info(f"Action script {script} is special/link.")
            run_action(script)
            continue

        if not os.path.isfile(script):
            _logger.info(f"Action script {script} not found.")
            continue

        if rfid == last_rfid and seconds_gone <= args.seconds:
            _logger.info(f"Same RFID after {seconds_gone:.1f} seconds. Do nothing.")
            continue

        last_ts = dt.now()
        last_rfid = rfid
        run_action(script)


def run():
    main(sys.argv[1:])


if __name__ == "__main__":
    run()