Source code for

#!/usr/bin/env python
# coding: utf8
# Copyright (c) 2020 Centre National d'Etudes Spatiales (CNES).
# This file is part of CARS
# (see
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.
Main CARS Command Line Interface
user main argparse wrapper to CARS 3D pipelines submodules

# Standard imports
# TODO refactor but keep local functions for performance and remove pylint
# pylint: disable=import-outside-toplevel
import argparse
import json
import logging
import os
import re
import sys

# CARS imports
from cars import __version__

[docs]class StreamCapture: """Filter stream (for stdout) with a re pattern From """ def __init__(self, stream, re_pattern): """StreamCapture constructor: add pattern, triggered parameters""" = stream self.pattern = ( re.compile(re_pattern) if isinstance(re_pattern, str) else re_pattern ) self.triggered = False
[docs] def __getattr__(self, attr_name): """Redefine assignment""" return getattr(, attr_name)
[docs] def write(self, data): """Change write function of stream and deals \n for loops""" if data == "\n" and self.triggered: self.triggered = False else: if is None: # Pattern not found, write normally else: # caught pattern to filter, no writing. self.triggered = True
[docs] def flush(self):
[docs]class CarsArgumentParser(argparse.ArgumentParser): """ ArgumentParser class adaptation for CARS """
[docs] def convert_arg_line_to_args(self, arg_line): """ Redefine herited function to accept one line argument parser in @file from fromfile_prefix_chars file argument """ return arg_line.split()
[docs]def cars_parser() -> CarsArgumentParser: """ Main CLI argparse parser function It builds argparse objects and constructs CLI interfaces parameters. :return: CARS arparse CLI interface object """ # Create cars cli parser from argparse # use @file to use a file containing parameters parser = CarsArgumentParser( "cars", description="CARS: CNES Algorithms to Reconstruct Surface", fromfile_prefix_chars="@", ) parser.add_argument("conf", type=str, help="Inputs Configuration File") parser.add_argument( "--loglevel", default="WARNING", choices=("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"), help="Logger level (default: WARNING. Should be one of " "(DEBUG, INFO, WARNING, ERROR, CRITICAL)", ) # General arguments at first level parser.add_argument( "--version", "-v", action="version", version="%(prog)s {version}".format(version=__version__), ) return parser
[docs]def main_cli(args, dry_run=False): # noqa: C901 """ Main for command line management :param dry_run: activate only arguments checking """ # TODO : refactor in order to avoid a slow argparse # Don't move the local function imports for now # CARS imports from cars.conf.log_conf import setup_log # Change stdout to clean (Os) OTB output from image_envelope app. original_stdout = sys.stdout sys.stdout = StreamCapture(sys.stdout, r"(0s)") # Logging configuration with args Loglevel setup_log(getattr(args, "loglevel", "WARNING").upper()) # Debug argparse show args logging.debug("Show argparse arguments: {}".format(args)) # Force the use of OpenMP in numba os.environ["NUMBA_THREADING_LAYER"] = "omp" # Main try/except to catch all program exceptions from cars.pipelines.pipeline import Pipeline try: # main(s) for each command if args.conf is not None: # Transform conf file to dict with open(args.conf, "r", encoding="utf8") as fstream: config = json.load(fstream) config_json_dir = os.path.abspath(os.path.dirname(args.conf)) pipeline_name = config.get( "pipeline", "sensor_to_full_resolution_dsm" ) # Generate pipeline and check conf used_pipeline = Pipeline(pipeline_name, config, config_json_dir) if not dry_run: # run pipeline else: raise SystemExit("CARS wrong subcommand. Use cars --help") except BaseException: # Catch all exceptions, show debug traceback and exit logging.exception("CARS terminated with following error") sys.exit(1) finally: # Go back to original stdout sys.stdout = original_stdout
[docs]def main(): """ Main initial cars cli entry point Configure and launch parser before main_cli function """ # CARS parser parser = cars_parser() args = parser.parse_args(args=None if sys.argv[1:] else ["--help"]) main_cli(args)
if __name__ == "__main__": main()