Revize 6b5427e6
Přidáno uživatelem Jakub Šilhavý před asi 3 roky(ů)
client/src/config_manager.py | ||
---|---|---|
3 | 3 |
|
4 | 4 |
|
5 | 5 |
class Config: |
6 |
"""This class holds the configuration values of the application. |
|
7 |
|
|
8 |
It reads the configuration file passed in through the constructor |
|
9 |
and stores the values into class variables. |
|
10 |
""" |
|
6 | 11 |
|
7 | 12 |
def __init__(self, filepath): |
13 |
"""Constructor of the class. |
|
14 |
|
|
15 |
It instantiates the class, reads the configuration file, |
|
16 |
and parses all sections defined in it. |
|
17 |
|
|
18 |
:param filepath: path to the configuration file |
|
19 |
""" |
|
20 |
# Create a new ConfigParser |
|
8 | 21 |
self.config = RawConfigParser() |
22 |
|
|
23 |
# Try to parse the configuration file. If it fails, |
|
24 |
# terminate the application. |
|
9 | 25 |
if not self.config.read(filepath): |
10 | 26 |
print(f"Failed to parse the config file {filepath}. Make sure you entered a valid path.") |
11 | 27 |
exit(1) |
12 | 28 |
|
29 |
# Parse the 'usb detector' section. |
|
13 | 30 |
self._parse_usb_detector_section() |
31 |
|
|
32 |
# Parse the 'server' section (API). |
|
14 | 33 |
self._parse_server_section() |
34 |
|
|
35 |
# Parse the 'logger' section. |
|
15 | 36 |
self._parse_logger_section() |
37 |
|
|
38 |
# Parse the 'cache' section. |
|
16 | 39 |
self._parse_cache_section() |
17 | 40 |
|
18 | 41 |
def _parse_usb_detector_section(self): |
42 |
"""Parse the 'usb detector' section of the configuration file. |
|
43 |
""" |
|
19 | 44 |
section_name = "usb_detector" |
20 | 45 |
self.scan_period_seconds = float(self.config[section_name]["scan_period_seconds"]) |
21 | 46 |
self.connected_devices_filename = self.config[section_name]["connected_devices_filename"] |
22 | 47 |
|
23 | 48 |
def _parse_server_section(self): |
49 |
"""Parse the 'server' section of the configuration file. |
|
50 |
""" |
|
24 | 51 |
section_name = "server" |
25 | 52 |
self.server_url = self.config[section_name]["url"] |
26 | 53 |
self.server_port = self.config[section_name]["port"] |
27 | 54 |
self.server_endpoint = self.config[section_name]["end_point"] |
28 | 55 |
|
29 | 56 |
def _parse_logger_section(self): |
57 |
"""Parse the 'logger' section of the configuration file. |
|
58 |
""" |
|
30 | 59 |
section_name = "logger" |
31 | 60 |
self.logger_config_file = self.config[section_name]["config_file"] |
32 | 61 |
|
33 | 62 |
def _parse_cache_section(self): |
63 |
"""Parse the 'cache' section of the configuration file. |
|
64 |
""" |
|
34 | 65 |
section_name = "cache" |
35 | 66 |
self.cache_dir = self.config[section_name]["directory"] |
36 | 67 |
self.cache_max_entries = int(self.config[section_name]["max_entries"]) |
client/src/main.py | ||
---|---|---|
13 | 13 |
|
14 | 14 |
|
15 | 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 exists. 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 initialize the application. |
|
22 |
|
|
23 |
:param app_config: instance of Config (config manager) |
|
24 |
""" |
|
25 |
# If the logger configuration file exists. |
|
16 | 26 |
if exists(app_config.logger_config_file): |
27 |
# Initialize logging according to the logger config file. |
|
17 | 28 |
logging.config.fileConfig(fname=app_config.logger_config_file) |
29 |
|
|
30 |
# Initialize the rest of the application. |
|
18 | 31 |
api_client_set_config(app_config) |
19 | 32 |
usb_detector_set_config(app_config) |
20 | 33 |
else: |
34 |
# If the file does not exist, terminate the application. |
|
21 | 35 |
print(f"Cannot find logger configuration \"{app_config.logger_config_file}\"! Please specify valid a path or define a new one.") |
22 | 36 |
exit(1) |
23 | 37 |
|
24 | 38 |
|
25 | 39 |
if __name__ == "__main__": |
40 |
"""Main entry point of the application. |
|
41 |
|
|
42 |
The application expects one parameter to be passed it - |
|
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 of this application. |
|
26 | 47 |
try: |
27 | 48 |
app_instance = singleton.SingleInstance() |
28 | 49 |
except singleton.SingleInstanceException: |
29 | 50 |
exit(1) |
30 | 51 |
|
52 |
# Parse the arguments passed in from the command line. |
|
31 | 53 |
arg_parser = argparse.ArgumentParser(description="ZF USB License Detector") |
32 | 54 |
arg_parser.add_argument("-c", "--config", dest="config", required=True, help="Path to the configuration file") |
33 | 55 |
args = arg_parser.parse_args() |
34 | 56 |
|
57 |
# Read the configuration file and initialize the application (logging). |
|
35 | 58 |
config = Config(args.config) |
36 | 59 |
init_logging(config) |
37 | 60 |
|
61 |
# Register callbacks (connected/disconnected USB device). |
|
38 | 62 |
register_listener(callback=usb_connected_callback, connected=True) |
39 | 63 |
register_listener(callback=usb_disconnected_callback, connected=False) |
40 | 64 |
|
65 |
# Create a thread for the USB detector. |
|
41 | 66 |
usb_detector_thread = Thread(target=usb_detector_run) |
42 | 67 |
usb_detector_thread.setDaemon(True) |
43 | 68 |
|
69 |
# Create a thread for resending failed payloads to the server. |
|
44 | 70 |
api_thread = Thread(target=api_client_run) |
45 | 71 |
api_thread.setDaemon(True) |
46 | 72 |
|
73 |
# Start the USB detector thread. |
|
47 | 74 |
logging.info("Starting USB detector.") |
48 | 75 |
usb_detector_thread.start() |
49 | 76 |
|
77 |
# Start the API client thread. |
|
50 | 78 |
logging.info("Starting API communication manager.") |
51 | 79 |
api_thread.start() |
52 | 80 |
|
81 |
# The execution should never get here as both |
|
82 |
# threads are infinite loops. |
|
53 | 83 |
usb_detector_thread.join() |
54 | 84 |
api_thread.join() |
Také k dispozici: Unified diff
re #9422 Commented config_manager.py and main.py