Add a Command Line Interface

setup.py

@@ -65,5 +65,7 @@
       tests_require=['pytest', 'mock', 'rasterio'],
       python_requires='>=3.9',
       test_suite='trollflow2.tests.suite',
-      use_scm_version=True
+      use_scm_version=True,
+      entry_points={
+          'console_scripts': ['satpy_cli = trollflow2.cli:cli',]}
       )

trollflow2/cli.py

@@ -0,0 +1,77 @@
+"""Trollflow2 command line interface."""
+
+import argparse
+import json
+import logging
+from datetime import datetime
+from queue import Queue
+
+import yaml
+
+from trollflow2.launcher import logging_on, process_files
+
+logger = logging.getLogger(__name__)
+
+
+def parse_args(args=None):
+    """Parse commandline arguments."""
+    parser = argparse.ArgumentParser(
+        description='Launch trollflow2 processing with Satpy on the provides files then quit.')
+    parser.add_argument("files", nargs='*',
+                        help="Data files to run on",
+                        type=str)
+    parser.add_argument("-p", "--product-list",
+                        help="The yaml file with the product list",
+                        type=str,
+                        required=True)
+    parser.add_argument("-m", "--metadata",
+                        help="Metadata (json) to pass on",
+                        type=str, required=False, default="{}")
+    parser.add_argument("-c", "--log-config",
+                        help="Log config file (yaml) to use",
+                        type=str, required=False, default=None)
+    return parser.parse_args(args)
+
+
+def cli(args=None):
+    """Command line interface."""
+    args = parse_args(args)
+
+    log_config = _read_log_config(args)
+
+    with logging_on(log_config):
+        logger.info("Starting Satpy.")
+        produced_files = Queue()
+        process_files(args.files, json.loads(args.metadata, object_hook=datetime_decoder),
+                      args.product_list, produced_files)
+
+
+def _read_log_config(args):
+    """Read the config."""
+    log_config = args.log_config
+    if log_config is not None:
+        with open(log_config) as fd:
+            log_config = yaml.safe_load(fd.read())
+    return log_config
+
+
+def datetime_decoder(dct):
+    """Decode datetimes to python objects."""
+    if isinstance(dct, list):
+        pairs = enumerate(dct)
+    elif isinstance(dct, dict):
+        pairs = dct.items()
+    result = []
+    for key, val in pairs:
+        if isinstance(val, str):
+            try:
+                val = datetime.fromisoformat(val)
+            except ValueError:
+                pass
+        elif isinstance(val, (dict, list)):
+            val = datetime_decoder(val)
+        result.append((key, val))
+    if isinstance(dct, list):
+        return [x[1] for x in result]
+    elif isinstance(dct, dict):
+        return dict(result)

trollflow2/launcher.py

@@ -257,18 +257,24 @@
 
 def message_to_jobs(msg, product_list):
     """Convert a posttroll message *msg* to a list of jobs given a *product_list*."""
+    input_filenames = _extract_filenames(msg)
+    input_mda = msg.data
+    return file_list_to_jobs(input_filenames, product_list, input_mda)
+
+
+def file_list_to_jobs(input_filenames, product_list, input_mda):
+    """Convert a file list to jobs."""
     formats = product_list['product_list'].get('formats', None)
     for _product, pconfig in plist_iter(product_list['product_list'], level='product'):
         if 'formats' not in pconfig and formats is not None:
             pconfig['formats'] = copy.deepcopy(formats)
     jobs = OrderedDict()
     priorities = get_area_priorities(product_list)
     # TODO: check the uri is accessible from the current host.
-    input_filenames = _extract_filenames(msg)
     for prio, areas in priorities.items():
         jobs[prio] = OrderedDict()
         jobs[prio]['input_filenames'] = input_filenames.copy()
-        jobs[prio]['input_mda'] = msg.data.copy()
+        jobs[prio]['input_mda'] = input_mda.copy()
         jobs[prio]['product_list'] = {}
         for section in product_list:
             if section == 'product_list':
@@ -322,7 +328,7 @@
     return yml
 
 
-def get_dask_client(config):
+def get_dask_distributed_client(config):
     """Create Dask client if configured."""
     client = None
 
@@ -374,38 +380,20 @@
 
 def process(msg, prod_list, produced_files):
     """Process a message."""
-    config = read_config(prod_list, Loader=UnsafeLoader)
+    """Convert a posttroll message *msg* to a list of jobs given a *product_list*."""
+    input_filenames = _extract_filenames(msg)
+    input_mda = msg.data
+    process_files(input_filenames, input_mda, prod_list, produced_files)
 
-    # Get distributed client
-    client = get_dask_client(config)
 
+def process_files(input_filenames, input_mda, prod_list, produced_files):
+    """Process files."""
+    config = read_config(prod_list, Loader=UnsafeLoader)
+    client = get_dask_distributed_client(config)
     try:
         config = expand(config)
-        jobs = message_to_jobs(msg, config)
-        for prio in sorted(jobs.keys()):
-            job = jobs[prio]
-            job['processing_priority'] = prio
-            job['produced_files'] = produced_files
-            try:
-                for wrk in config['workers']:
-                    cwrk = wrk.copy()
-                    if "timeout" in cwrk:
-
-                        def _timeout_handler(signum, frame, wrk=wrk):
-                            raise TimeoutError(
-                                f"Timeout for {wrk['fun']!s} expired "
-                                f"after {wrk['timeout']:.1f} seconds, "
-                                "giving up")
-                        signal.signal(signal.SIGALRM, _timeout_handler)
-                        # using setitimer because it accepts floats,
-                        # unlike signal.alarm
-                        signal.setitimer(signal.ITIMER_REAL,
-                                         cwrk.pop("timeout"))
-                    cwrk.pop('fun')(job, **cwrk)
-                    if "timeout" in cwrk:
-                        signal.alarm(0)  # cancel the alarm
-            except AbortProcessing as err:
-                logger.warning(str(err))
+        jobs = file_list_to_jobs(input_filenames, config, input_mda)
+        process_jobs(config["workers"], jobs, produced_files)
     except Exception:
         logger.exception("Process crashed")
         if "crash_handlers" in config:
@@ -423,13 +411,39 @@
             except AttributeError:
                 continue
         del config
-        try:
+        with suppress(AttributeError):
             client.close()
-        except AttributeError:
-            pass
         gc.collect()
 
 
+def process_jobs(workers, jobs, produced_files):
+    """Process the jobs."""
+    for prio in sorted(jobs.keys()):
+        job = jobs[prio]
+        job['processing_priority'] = prio
+        job['produced_files'] = produced_files
+        try:
+            for wrk in workers:
+                cwrk = wrk.copy()
+                if "timeout" in cwrk:
+                    def _timeout_handler(signum, frame, wrk=wrk):
+                        raise TimeoutError(
+                            f"Timeout for {wrk['fun']!s} expired "
+                            f"after {wrk['timeout']:.1f} seconds, "
+                            "giving up")
+
+                    signal.signal(signal.SIGALRM, _timeout_handler)
+                    # using setitimer because it accepts floats,
+                    # unlike signal.alarm
+                    signal.setitimer(signal.ITIMER_REAL,
+                                     cwrk.pop("timeout"))
+                cwrk.pop('fun')(job, **cwrk)
+                if "timeout" in cwrk:
+                    signal.alarm(0)  # cancel the alarm
+        except AbortProcessing as err:
+            logger.warning(str(err))
+
+
 def read_config(fname=None, raw_string=None, Loader=SafeLoader):
     """Read the configuration file."""
     try:
@@ -439,6 +453,8 @@
                 raw_config = fid.read()
         elif raw_string:
             raw_config = raw_string
+        if not raw_config:
+            raise IOError
         config = yaml.load(_remove_null_keys(raw_config), Loader=Loader)
     except (IOError, yaml.YAMLError):
         # Either open() or yaml.load() failed