15876623240

Committed 25 Jun 2025 12:41PM UTC coverage: 69.899% (+0.03%) from 69.872%

Build # 15876623240

Build Type

Pull #2830

github

Committed by

web-flow

Commit Message

Merge 412cf3942 into 001cc76b1

Pull Request Pull Request #2830: added a special handling for path strings

Run Details

8 of 12 new or added lines in 1 file covered. (66.67%)

15 existing lines in 2 files now uncovered.

14483 of 20720 relevant lines covered (69.9%)

0.7 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

92.04

/mslib/utils/config.py

# -*- coding: utf-8 -*-
"""

    mslib.utils.config
    ~~~~~~~~~~~~~~~~

    Collection of functions all around config handling.

    This file is part of MSS.

    :copyright: Copyright 2008-2014 Deutsches Zentrum fuer Luft- und Raumfahrt e.V.
    :copyright: Copyright 2011-2014 Marc Rautenhaus (mr)
    :copyright: Copyright 2016-2025 by the MSS team, see AUTHORS.
    :license: APACHE-2.0, see LICENSE for details.

    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

       http://www.apache.org/licenses/LICENSE-2.0

    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.
"""
from PyQt5 import QtCore

import copy
import json
import logging
import fs
import os

from mslib.utils import FatalUserError
from mslib.msui import constants
from mslib.support.qt_json_view.datatypes import match_type, UrlType, StrType


class MSUIDefaultConfig:
    """Central configuration for the Mission Support System User Interface
       Application (msui).

    DESCRIPTION:
    ============

    This file includes configuration settings central to the entire
    Mission Support User Interface (msui). Among others, define
     -- available map projections
     -- vertical section interpolation options
     -- the lists of predefined web service URLs
     -- predefined waypoints for the table view
    in this file.

    Do not change any value for good reasons.
    Your values can be set in your personal msui_settings.json file
    """
    # this skips the verification of the user token on each mscolab request
    mscolab_skip_verify_user_token = True

    # Default for general filepicker. Pick "default", "qt", or "fs"
    filepicker_default = "default"

    # dir where msui output files are stored
    data_dir = "~/mssdata"

    # layout of different views, with immutable they can't resized
    layout = {"topview": [963, 702],
              "sideview": [913, 557],
              "linearview": [913, 557],
              "tableview": [1236, 424],
              "immutable": False}

    # Predefined map regions to be listed in the corresponding topview combobox
    predefined_map_sections = {
        "00 global (cyl)": {
            "CRS": "EPSG:4326",
            "map": {
                "llcrnrlon": -180.0,
                "llcrnrlat": -90.0,
                "urcrnrlon": 180.0,
                "urcrnrlat": 90.0
            }
        },
        "01 SADPAP (stereo)": {
            "CRS": "EPSG:77890290",
            "map": {
                "llcrnrlon": -150.0,
                "llcrnrlat": -45.0,
                "urcrnrlon": -25.0,
                "urcrnrlat": -20.0
            }
        },
        "02 SADPAP zoom (stereo)": {
            "CRS": "EPSG:77890290",
            "map": {
                "llcrnrlon": -120.0,
                "llcrnrlat": -65.0,
                "urcrnrlon": -45.0,
                "urcrnrlat": -28.0
            }
        },
        "03 SADPAP (cyl)": {
            "CRS": "EPSG:4326",
            "map": {
                "llcrnrlon": -100.0,
                "llcrnrlat": -75.0,
                "urcrnrlon": -30.0,
                "urcrnrlat": -30.0
            }
        },
        "04 Southern Hemisphere (stereo)": {
            "CRS": "EPSG:77889270",
            "map": {
                "llcrnrlon": 135.0,
                "llcrnrlat": 0.0,
                "urcrnrlon": -45.0,
                "urcrnrlat": 0.0
            }
        },
        "05 EDMO-SAL (cyl)": {
            "CRS": "EPSG:4326",
            "map": {
                "llcrnrlon": -40,
                "llcrnrlat": 10,
                "urcrnrlon": 30,
                "urcrnrlat": 60
            }
        },
        "06 SAL-BA (cyl)": {
            "CRS": "EPSG:4326",
            "map": {
                "llcrnrlon": -80,
                "llcrnrlat": -40,
                "urcrnrlon": -10,
                "urcrnrlat": 30
            }
        },
        "07 Europe (cyl)": {
            "CRS": "EPSG:4326",
            "map": {
                "llcrnrlon": -15.0,
                "llcrnrlat": 35.0,
                "urcrnrlon": 30.0,
                "urcrnrlat": 65.0
            }
        },
        "08 Germany (cyl)": {
            "CRS": "EPSG:4326",
            "map": {
                "llcrnrlon": 5.0,
                "llcrnrlat": 45.0,
                "urcrnrlon": 15.0,
                "urcrnrlat": 57.0
            }
        },
        "09 Northern Hemisphere (stereo)": {
            "CRS": "MSS:stere,0,90,90",
            "map": {
                "llcrnrlon": -45.0,
                "llcrnrlat": 0.0,
                "urcrnrlon": 135.0,
                "urcrnrlat": 0.0
            }
        }
    }

    # Side View.
    # The following two parameters are passed to the WMS in the BBOX
    # argument when a vertical cross section is requested.

    # Number of interpolation points used to interpolate the flight track
    # to a great circle.
    num_interpolation_points = 201

    # Number of x-axis labels in the side view.
    num_labels = 10

    # Web Map Service Client.
    # Settings for the WMS client. Set the URLs of WMS servers that appear
    # by default in the WMS control (for examples, see
    # http://external.opengis.org/twiki_public/bin/view/MetOceanDWG/MetocWMS_Servers).
    # Also set the location of the image file cache and its size.

    # URLs of default WMS servers.
    default_WMS = [
        "http://localhost:8081/",
        "https://view.eumetsat.int/geoserver/wms",
        "http://eccharts.ecmwf.int/wms/?token=public",
        "https://neo.gsfc.nasa.gov/wms/wms"
    ]

    default_VSEC_WMS = [
        "http://localhost:8081/"
    ]

    default_LSEC_WMS = [
        "http://localhost:8081/"
    ]

    # URLs of default mscolab servers
    default_MSCOLAB = [
        "http://localhost:8083",
    ]

    # Username used for http auth
    MSCOLAB_auth_user_name = "mscolab"

    # category for MSC operations
    MSCOLAB_category = "default"

    # timeout for MSColab in seconds. First value is for connection, second for reply
    MSCOLAB_timeout = [2, 10]

    # don't query for archived operations
    MSCOLAB_skip_archived_operations = False

    # list of MSC servers {"http://www.your-mscolab-server.de": "authuser",
    # "http://www.your-wms-server.de": "authuser"}
    MSS_auth = {}

    # timeout of Url request
    WMS_request_timeout = 30

    WMS_preload = []

    # WMS image cache settings:
    wms_cache = str(constants.MSUI_CACHE_PATH / "wms_cache")

    # Maximum size of the cache in bytes.
    wms_cache_max_size_bytes = 20 * 1024 * 1024

    # Maximum age of a cached file in seconds.
    wms_cache_max_age_seconds = 5 * 86400

    wms_prefetch = {
        "validtime_fwd": 0,
        "validtime_bck": 0,
        "level_up": 0,
        "level_down": 0
    }

    locations = {
        "EDMO": [48.08, 11.28],
        "Hannover": [52.37, 9.74],
        "Hamburg": [53.55, 9.99],
        "Juelich": [50.92, 6.36],
        "Leipzig": [51.34, 12.37],
        "Muenchen": [48.14, 11.57],
        "Stuttgart": [48.78, 9.18],
        "Wien": [48.20833, 16.373064],
        "Zugspitze": [47.42, 10.98],
        "Kiruna": [67.821, 20.336],
        "Ny-Alesund": [78.928, 11.986],
        "Zhukovsky": [55.6, 38.116],
        "Paphos": [34.775, 32.425],
        "Sharjah": [25.35, 55.65],
        "Brindisi": [40.658, 17.947],
        "Nagpur": [21.15, 79.083],
        "Mumbai": [19.089, 72.868],
        "Delhi": [28.566, 77.103],
    }

    # Main application: Template for new flight tracks
    # Flight track template that is used when a new flight track is
    # created. Specify a list of place names that can be found in the
    # "locations" dictionary defined above.
    new_flighttrack_template = ["Nagpur", "Delhi"]

    # This configures the flight level for waypoints inserted by the
    # flighttrack template
    new_flighttrack_flightlevel = 0

    # None is not wanted here
    proxies = {}

    # ToDo configurable later
    # mscolab server
    mscolab_server_url = "http://localhost:8083"
    # ToDo refactor to rename this to data_dir/mss_data_dir
    # mss dir
    mss_dir = "~/mss"

    # list of gravatar email ids to automatically fetch
    gravatar_ids = []

    # dictionary for export plugins, e.g.  {"Text": ["txt", "mslib.plugins.io.text", "save_to_txt"] }
    export_plugins = {}

    # dictionary for import plugins, e.g. { "FliteStar": ["txt", "mslib.plugins.io.flitestar", "load_from_flitestar"] }
    import_plugins = {}

    # dictionary to make title, label and ticklabel sizes for topview and sideview configurable.
    # You can put your default value here, whatever you want to give,it should be a number.
    topview = {"plot_title_size": 10,
               "axes_label_size": 10}

    sideview = {"plot_title_size": 10,
                "axes_label_size": 10}

    linearview = {"plot_title_size": 10,
                  "axes_label_size": 10}

    automated_plotting_flights = [[]]
    automated_plotting_hsecs = [[]]
    automated_plotting_vsecs = [[]]
    automated_plotting_lsecs = [[]]

    # Dictionary options with fixed key/value pairs
    fixed_dict_options = ["layout", "wms_prefetch", "topview", "sideview", "linearview"]
    # List options with fixed length
    fixed_list_options = ["MSCOLAB_timeout", ]

    # Fixed key/value pair options
    key_value_options = [
        'mscolab_skip_verify_user_token',
        'filepicker_default',
        'mss_dir',
        'data_dir',
        'num_labels',
        'num_interpolation_points',
        'new_flighttrack_flightlevel',
        'MSCOLAB_category',
        'MSCOLAB_skip_archived_operations',
        'mscolab_server_url',
        'MSCOLAB_auth_user_name',
        'wms_cache',
        'wms_cache_max_size_bytes',
        'wms_cache_max_age_seconds',
        'WMS_request_timeout',
    ]

    # Dictionary options with predefined structure
    dict_option_structure = {
        "MSS_auth": {"http://www.your-wms-server.de": "authusername"},
        "predefined_map_sections": {
            "new_map_section": {
                "CRS": "crs_value",
                "map": {
                    "llcrnrlon": 0.0,
                    "llcrnrlat": 0.0,
                    "urcrnrlon": 0.0,
                    "urcrnrlat": 0.0,
                },
            }
        },
        "locations": {
            "new-location": [0.0, 0.0],
        },
        "export_plugins": {
            "plugin-name": ["extension", "module", "function", "default"],
        },
        "import_plugins": {
            "plugin-name": ["extension", "module", "function", "default"],
        },
        "proxies": {
            "https": "https://proxy.com",
        },
    }

    # List options with predefined structure
    list_option_structure = {
        "default_WMS": ["https://wms-server-url.com"],
        "default_VSEC_WMS": ["https://vsec-wms-server-url.com"],
        "default_LSEC_WMS": ["https://lsec-wms-server-url.com"],
        "default_MSCOLAB": ["https://mscolab-server-url.com"],
        "new_flighttrack_template": ["new-location"],
        "gravatar_ids": ["example@email.com"],
        "WMS_preload": ["https://wms-preload-url.com"],
        "MSCOLAB_timeout": [0, 0],
        "automated_plotting_flights": [["", "", "", "", "", ""]],
        "automated_plotting_hsecs": [["http://www.your-wms-server.de", "", "", ""]],
        "automated_plotting_vsecs": [["http://www.your-wms-server.de", "", "", ""]],
        "automated_plotting_lsecs": [["http://www.your-wms-server.de", "", "", "pressure"]]
    }

    config_descriptions = {
        "filepicker_default": "Defines the type of file-picker to be used. Can be 'default', 'qt', or 'fs'",
        "data_dir": "Directory where MSUI output files are stored",
        "predefined_map_sections": "Dictionary containing predefined map sections with their settings",
        "num_interpolation_points": "Number of interpolation points used for vertical cross section requests",
        "num_labels": "Number of x-axis labels in the side view",
        "default_WMS": "List of the URLs of default WMS servers",
        "default_VSEC_WMS": "List of the URLs of default Vertical Section WMS servers",
        "default_LSEC_WMS": "List of the URLs of default Linear Section WMS servers",
        "default_MSCOLAB": "List of the URLs of default MSColab servers",
        "MSS_auth": "Dictionary containing credentials for http auth",
        "MSCOLAB_auth_user_name": "Username used for http auth",
        "MSCOLAB_timeout": "Tuple specifying timeout for MSColab in seconds. First value is for connection,"
                           " second for reply",
        "WMS_request_timeout": "Timeout of WMS Url request",
        "WMS_preload": "List of WMS URLs to preload",
        "wms_cache": "Path to WMS image cache directory",
        "wms_cache_max_size_bytes": "Maximum size of the cache in bytes",
        "wms_cache_max_age_seconds": "Maximum age of a cached file in seconds",
        "wms_prefetch": "Dictionary configuring the prefetch behaviour of WMS",
        "locations": "Dictionary providing geo-coordinates for predefined locations",
        "new_flighttrack_template": "List of predefined location names used as a flight track template"
                                    " when creating a new flight track",
        "new_flighttrack_flightlevel": "Default flight level for waypoints inserted by the flighttrack template",
        "proxies": "Proxy settings for network requests",
        "mscolab_server_url": "URL of the MSColab server",
        "mss_dir": "Directory path used for MSS",  # ToDo is this needed or can be replaced by data_dir?
        "gravatar_ids": "List of gravatar email ids to automatically fetch",
        "export_plugins": "Dictionary of export plugins",
        "import_plugins": "Dictionary of import plugins",
        "layout": "Dictionary for layout of different views",
        "topview": "Dictionary to make title, label, and ticklabel sizes for topview configurable",
        "sideview": "Dictionary to make title, label, and ticklabel sizes for sideview configurable",
        "linearview": "Dictionary to make title, label, and ticklabel sizes for linearview configurable",
    }


# default options as dictionary
default_options = dict(MSUIDefaultConfig.__dict__)
for key in [
    "__module__",
    "__doc__",
    "__dict__",
    "__weakref__",
    "fixed_dict_options",
    "fixed_list_options",
    "dict_option_structure",
    "list_option_structure",
    "key_value_options",
    "config_descriptions",
]:
    del default_options[key]


# user options as dictionary
user_options = copy.deepcopy(default_options)


def read_config_file(path=constants.MSUI_SETTINGS):
    """
    reads a config file and updates global user_options

    Args:
        path: path of config file

    Note:
        sole purpose of the path argument is to be able to test with example config files
    """
    path = path.replace("\\", "/")
    dir_name, file_name = fs.path.split(path)
    json_file_data = {}
    with fs.open_fs(dir_name) as _fs:
        if _fs.exists(file_name):
            file_content = _fs.readtext(file_name)
            try:
                json_file_data = json.loads(file_content, object_pairs_hook=dict_raise_on_duplicates_empty)
            except json.JSONDecodeError as e:
                logging.error("Error while loading json file %s", e)
                error_message = f"Unexpected error while loading config\n{e}"
                raise FatalUserError(error_message)
            except ValueError as e:
                logging.error("Error while loading json file %s", e)
                error_message = f"Invalid keys detected in config\n{e}"
                raise FatalUserError(error_message)
        else:
            error_message = f"MSS config File '{path}' not found"
            raise FileNotFoundError(error_message)

    global user_options
    if json_file_data:
        user_options = merge_dict(copy.deepcopy(default_options), json_file_data)
        logging.debug("Merged default and user settings")
    else:
        user_options = copy.deepcopy(default_options)
        logging.debug("No user settings found, using default settings")


def modify_config_file(data, path=constants.MSUI_SETTINGS):
    """
    modifies a config file

    Args:
        data: data to be modified/written
        path: path of config file

    Note:
        sole purpose of the path argument is to be able to test with example config files
    """
    path = path.replace("\\", "/")
    dir_name, file_name = fs.path.split(path)
    json_file_data = {}
    with fs.open_fs(dir_name) as _fs:
        if _fs.exists(file_name):
            try:
                file_content = _fs.readtext(file_name)
                json_file_data = json.loads(file_content, object_pairs_hook=dict_raise_on_duplicates_empty)
                json_file_data_copy = copy.deepcopy(json_file_data)
                for key in data:
                    if key not in json_file_data:
                        json_file_data_copy[key] = config_loader(dataset=key, default=True)
                modified_data = merge_dict(json_file_data_copy, data)
                logging.debug("Merged default and user settings")
                _fs.writetext(file_name, json.dumps(modified_data, indent=4))
                read_config_file()
            except json.JSONDecodeError as e:
                logging.error("Error while loading json file %s", e)
                error_message = f"Unexpected error while loading config\n{e}"
                raise FatalUserError(error_message)
            except ValueError as e:
                logging.error("Error while loading json file %s", e)
                error_message = f"Invalid keys detected in config\n{e}"
                raise FatalUserError(error_message)
        else:
            error_message = f"MSS config File '{path}' not found"
            raise FileNotFoundError(error_message)


def config_loader(dataset=None, default=False):
    """
    Function for returning config value

    Args:
        dataset: section to pull from json file
        default: option to return default config for the dataset

    Returns: a the dataset value or the config as dictionary

    """
    if dataset is not None and dataset not in user_options:
        raise KeyError(f"requested dataset '{dataset}' not in defaults!")

    if dataset is not None:
        if default:
            return default_options[dataset]
        return user_options[dataset]
    else:
        if default:
            return default_options
        return user_options


def save_settings_qsettings(tag, settings):
    """
    Saves a dictionary settings to disk.

    :param tag: string specifying the settings
    :param settings: dictionary of settings
    :return: None
    """
    assert isinstance(tag, str)
    assert isinstance(settings, dict)
    # ToDo we have to verify if we can all switch to this definition, not having 3 different
    q_settings = QtCore.QSettings(os.path.join(constants.MSUI_CONFIG_SYSPATH, "msui-core.conf"),
                                  QtCore.QSettings.IniFormat)

    file_path = q_settings.fileName()
    logging.debug("storing settings for %s to %s", tag, file_path)
    try:
        q_settings.setValue(tag, QtCore.QVariant(settings))
    except (OSError, IOError) as ex:
        logging.warning("Problems storing %s settings (%s: %s).", tag, type(ex), ex)
    return settings


def load_settings_qsettings(tag, default_settings=None):
    """
    Loads a dictionary of settings from disk. May supply a dictionary of default settings
    to return in case the settings file is not present or damaged. The default_settings one will
    be updated by the restored one so one may rely on all keys of the default_settings dictionary
    being present in the returned dictionary.

    :param tag: string specifying the settings
    :param default_settings: dictionary of settings or None
    :return: dictionary of settings
    """
    if default_settings is None:
        default_settings = {}
    assert isinstance(default_settings, dict)

    settings = {}

    q_settings = QtCore.QSettings(os.path.join(constants.MSUI_CONFIG_SYSPATH, "msui-core.conf"),
                                  QtCore.QSettings.IniFormat)
    file_path = q_settings.fileName()
    logging.debug("loading settings for %s from %s", tag, file_path)
    try:
        settings = q_settings.value(tag)
    except Exception as ex:
        logging.warning("Problems reloading stored %s settings (%s: %s). Switching to default",
                        tag, type(ex), ex)
    if isinstance(settings, dict):
        default_settings.update(settings)
    return default_settings


def merge_dict(existing_dict, new_dict):
    """
    Merge two dictionaries by comparing all the options from
    the MSUIDefaultConfig class

    Arguments:
    existing_dict -- Dict to merge new_dict into
    new_dict -- Dict with new values
    """
    # Check if dictionary options with fixed key/value pairs match data types from default
    for key in MSUIDefaultConfig.fixed_dict_options + MSUIDefaultConfig.fixed_list_options:
        if key in new_dict:
            existing_dict[key] = compare_data(
                existing_dict[key], new_dict[key]
            )[0]

    # Check if dictionary options with predefined structure match data types from default
    dos = copy.deepcopy(MSUIDefaultConfig.dict_option_structure)
    # adding plugin structure : ["extension", "module", "function"] to
    # recognize user plugin options that don't have the optional filepicker option set
    dos["import_plugins"]["plugin-name-a"] = dos["import_plugins"]["plugin-name"][:3]
    dos["export_plugins"]["plugin-name-a"] = dos["export_plugins"]["plugin-name"][:3]
    for key in dos:
        if key in new_dict:
            temp_data = {}
            for option_key in new_dict[key]:
                for dos_key_key in dos[key]:
                    data, match = compare_data(dos[key][dos_key_key], new_dict[key][option_key])
                    if key in MSUIDefaultConfig.fixed_list_options:
                        match = len(dos[key][dos_key_key]) == len(new_dict[key][option_key])
                    if match:
                        temp_data[option_key] = new_dict[key][option_key]
                        break
            if temp_data != {}:
                existing_dict[key] = temp_data

    # Check if list options with predefined structure match data types from default
    los = copy.deepcopy(MSUIDefaultConfig.list_option_structure)
    for key in los:
        if key in new_dict:
            temp_data = []
            if key not in MSUIDefaultConfig.fixed_list_options:
                for i in range(len(new_dict[key])):
                    for los_key_item in los[key]:
                        data, match = compare_data(los_key_item, new_dict[key][i])
                        if match:
                            temp_data.append(data)
                            break
                if temp_data != []:
                    existing_dict[key] = temp_data

    # Check if options with fixed key/value pair structure match data types from default
    for key in MSUIDefaultConfig.key_value_options:
        if key in new_dict:
            data, match = compare_data(existing_dict[key], new_dict[key])
            if match:
                existing_dict[key] = data

    # add filepicker default to import and export plugins if missing
    for plugin_type in ["import_plugins", "export_plugins"]:
        if plugin_type in existing_dict:
            for plugin in existing_dict[plugin_type]:
                if len(existing_dict[plugin_type][plugin]) == 3:
                    existing_dict[plugin_type][plugin].append(
                        existing_dict.get("filepicker_default", "default")
                    )

    return existing_dict


def compare_data(default, user_data):
    """
    Recursively compares two dictionaries based on qt_json_view datatypes
    and returns default or user_data appropriately.

    Arguments:
    default -- Dict to return if datatype not matching
    user_data -- Dict to return if datatype is matching
    """
    # If data is neither list not dict type, compare individual type
    if not isinstance(default, dict) and not isinstance(default, list):
        if isinstance(default, float) and isinstance(user_data, int):
            user_data = float(default)
        if isinstance(match_type(default), UrlType) and isinstance(match_type(user_data), StrType):
            return user_data, True
        if isinstance(match_type(default), type(match_type(user_data))):
            return user_data, True
        # Special handling for path strings - both absolute and relative paths should be treated as valid strings
        elif isinstance(default, str) and isinstance(user_data, str):
            return user_data, True
        else:
            return default, False

    data = copy.deepcopy(default)
    matches = []
    # If data is list type, compare all values in list
    if isinstance(default, list) and isinstance(user_data, list):
        if len(default) == len(user_data):
            for i in range(len(default)):
                data[i], match = compare_data(default[i], user_data[i])
                matches.append(match)
        else:
            return default, False

    # If data is dict type, goes through the dict and update
    elif isinstance(default, dict) and isinstance(user_data, dict):
        if default.keys() == user_data.keys():
            for key in default:
                if key in user_data:
                    data[key], match = compare_data(default[key], user_data[key])
                    matches.append(match)
                else:
                    matches.append(False)
        else:
            return default, False

    return data, all(matches)


def dict_raise_on_duplicates_empty(ordered_pairs):
    """Reject duplicate and empty keys."""
    accepted = {}
    for key, value in ordered_pairs:
        if key in accepted:
            raise ValueError(f"duplicate key found: {key}")
        elif key == "":
            raise ValueError("empty key found")
        else:
            accepted[key] = value
    return accepted

Open-MSS / MSS / 15876623240

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous