31
31
from functools import partial
32
32
from glob import glob
33
33
from multiprocessing .pool import ThreadPool
34
+ from pathlib import Path
34
35
from subprocess import PIPE , Popen
35
36
36
37
import yaml
41
42
logger = logging .getLogger ("pytroll-runner" )
42
43
43
44
44
- def main (args = None ):
45
+ def main (args : list [ str ] | None = None ):
45
46
"""Main script."""
46
47
parsed_args = parse_args (args = args )
47
48
setup_logging (parsed_args .log_config )
@@ -50,7 +51,7 @@ def main(args=None):
50
51
return run_and_publish (parsed_args .config_file , parsed_args .message_file )
51
52
52
53
53
- def setup_logging (config_file ):
54
+ def setup_logging (config_file : str | None ):
54
55
"""Setup the logging from a log config yaml file."""
55
56
if config_file is not None :
56
57
with open (config_file ) as fd :
@@ -59,12 +60,13 @@ def setup_logging(config_file):
59
60
return
60
61
61
62
62
- def parse_args (args = None ):
63
+ def parse_args (args : list [ str ] | None = None ):
63
64
"""Parse command line arguments."""
64
65
parser = argparse .ArgumentParser ("Pytroll Runner" ,
65
66
description = "Automate third party software in a pytroll environment" )
66
67
parser .add_argument ("config_file" ,
67
- help = "The configuration file to run on." )
68
+ help = "The configuration file to run on." ,
69
+ type = Path )
68
70
parser .add_argument ("-l" , "--log_config" ,
69
71
help = "The log configuration yaml file." ,
70
72
default = None )
@@ -74,7 +76,7 @@ def parse_args(args=None):
74
76
return parser .parse_args (args )
75
77
76
78
77
- def run_and_publish (config_file , message_file = None ):
79
+ def run_and_publish (config_file : Path , message_file : str | None = None ):
78
80
"""Run the command and publish the expected files."""
79
81
command_to_call , subscriber_config , publisher_config = read_config (config_file )
80
82
with suppress (KeyError ):
@@ -109,20 +111,20 @@ def check_existing_files(publisher_config):
109
111
return set (glob (filepattern ))
110
112
111
113
112
- def read_config (config_file ):
114
+ def read_config (config_file : Path ):
113
115
"""Read the configuration file."""
114
116
with open (config_file ) as fd :
115
- config = yaml .safe_load (fd .read ())
116
- return validate_config (config )
117
+ config : dict [ str , object ] = yaml .safe_load (fd .read ())
118
+ return curate_config (config )
117
119
118
120
119
- def validate_config (config ):
121
+ def curate_config (config ):
120
122
"""Validate the configuration file."""
121
123
publisher_config = config ["publisher_config" ]
122
124
if "output_files_log_regex" not in publisher_config and "expected_files" not in publisher_config :
123
125
raise KeyError ("Missing ways to identify output files. "
124
- "Either provide 'expected_files' or "
125
- "'output_files_log_regex' in the config file." )
126
+ "Either provide 'expected_files' or "
127
+ "'output_files_log_regex' in the config file." )
126
128
127
129
subscriber_config = config ["subscriber_config" ]
128
130
logger .debug ("Subscriber config settings: " )
0 commit comments