Projekt

Obecné

Profil

Stáhnout (3.15 KB) Statistiky
| Větev: | Tag: | Revize:
1
import logging
2
import logging.config
3
import argparse
4
from os.path import exists
5
from threading import Thread
6
from tendo import singleton
7
from sys import exit
8

    
9
from config_manager import Config
10
from usb_detector.detector import register_listener, usb_detector_run, usb_detector_set_config
11
from usb_detector.event_listener import usb_connected_callback, usb_disconnected_callback
12
from usb_detector.api_client import api_client_run, api_client_set_config
13

    
14

    
15
def init_logging(app_config: Config):
16
    """Initializes logging, api client, and usb detector.
17

    
18
    The function checks whether the path to the logger configuration
19
    file is valid or not. The path is defined in the logger section of the
20
    main configuration file. It also calls api_client_set_config and
21
    usb_detector_set_config to fully initialize the application.
22

    
23
    :param app_config: instance of Config (config manager)
24
    """
25
    # If the logger configuration file exists.
26
    if exists(app_config.logger_config_file):
27
        # Initialize logging according to the logger config file.
28
        logging.config.fileConfig(fname=app_config.logger_config_file)
29

    
30
        # Initialize the rest of the application.
31
        api_client_set_config(app_config)
32
        usb_detector_set_config(app_config)
33
    else:
34
        # If the file does not exist, terminate the application.
35
        print(f"Cannot find logger configuration \"{app_config.logger_config_file}\"! Please specify valid a path or define a new one.")
36
        exit(1)
37

    
38

    
39
if __name__ == "__main__":
40
    """Main entry point of the application.
41
    
42
    The application expects one parameter to be passed in -
43
    the path to the configuration file. The user can print out help
44
    using the '-h' option.
45
    """
46
    # Make sure that there is only one running instance (process) of this application.
47
    try:
48
        app_instance = singleton.SingleInstance()
49
    except singleton.SingleInstanceException:
50
        exit(1)
51

    
52
    # Parse the arguments passed in from the command line.
53
    arg_parser = argparse.ArgumentParser(description="ZF USB License Detector")
54
    arg_parser.add_argument("-c", "--config", dest="config", required=True, help="Path to the configuration file")
55
    args = arg_parser.parse_args()
56

    
57
    # Read the configuration file and initialize the application (logging).
58
    config = Config(args.config)
59
    init_logging(config)
60

    
61
    # Register callbacks (connected/disconnected USB device).
62
    register_listener(callback=usb_connected_callback, connected=True)
63
    register_listener(callback=usb_disconnected_callback, connected=False)
64

    
65
    # Create a thread for the USB detector.
66
    usb_detector_thread = Thread(target=usb_detector_run)
67
    usb_detector_thread.setDaemon(True)
68

    
69
    # Create a thread for resending failed payloads to the server.
70
    api_thread = Thread(target=api_client_run)
71
    api_thread.setDaemon(True)
72

    
73
    # Start the USB detector thread.
74
    logging.info("Starting USB detector.")
75
    usb_detector_thread.start()
76

    
77
    # Start the API client thread.
78
    logging.info("Starting API communication manager.")
79
    api_thread.start()
80

    
81
    # The execution should never get here as both
82
    # threads are infinite loops.
83
    usb_detector_thread.join()
84
    api_thread.join()
(2-2/2)