Source code for LandSurveyCodesImport.toolbelt.qlsc_reader

#! python3  # noqa: E265

"""
    Read QGIS Land Survey Codification (QLSC) files, which are YAML.
"""


# #############################################################################
# ########## Libraries #############
# ##################################

# standard library
import logging
from io import BufferedIOBase
from os import R_OK, access
from pathlib import Path
from typing import Union

# 3rd party
import yaml

# project
# from LandSurveyCodesImport.available_codes import AVAILABLE_CODE

# #############################################################################
# ########## Globals ###############
# ##################################

logger = logging.getLogger(__name__)

# ##############################################################################
# ########## Exceptions ############
# ##################################


[docs]class BadQlscStructure(Exception): """Custom exception.""" pass
# ############################################################################## # ########## Classes ############### # ##################################
[docs]class QlscYamlReader: """Read and validate a QLSC (YAML) file. :param in_yaml: path to the yaml file to read. """ QLSC_KEYS = [ "AllPoints", "BoundingGeometry", "Codification", "CodeSeparator", "ErrorPoints", "ParameterSeparator", ] def __init__(self, in_yaml: Union[str, Path, BufferedIOBase]): """Instanciating YAML Reader.""" # check and get YAML path if isinstance(in_yaml, (str, Path)): self.input_yaml = self.check_yaml_file(in_yaml) # extract data from input file with self.input_yaml.open(mode="r") as bytes_data: self.check_yaml_structure(yaml.full_load(bytes_data)) elif isinstance(in_yaml, BufferedIOBase): self.input_yaml = self.check_yaml_buffer(in_yaml) # extract data from input file self.check_yaml_structure(yaml.full_load(self.input_yaml)) else: raise TypeError # PROPERTIES @property def as_dict(self) -> dict: """Returns the YAML loaded into a Python dictionary. Returns: dict: YAML document as Python dictionary object """ with self.input_yaml.open(mode="r") as stream_data: yaml_loaded = yaml.full_load(stream_data) return yaml_loaded # CHECKS
[docs] def check_yaml_file(self, yaml_path: Union[str, Path]) -> Path: """Perform some checks on passed yaml file and load it as Path object. :param yaml_path: path to the yaml file to check :returns: sanitized yaml path :rtype: Path """ # if path as string load it in Path object if isinstance(yaml_path, str): try: yaml_path = Path(yaml_path) except Exception as exc: raise TypeError(f"Converting yaml path failed: {exc}") # check if file exists if not yaml_path.exists(): raise FileExistsError( f"YAML file to check doesn't exist: {yaml_path.resolve()}" ) # check if it's a file if not yaml_path.is_file(): raise IOError(f"YAML file is not a file: {yaml_path.resolve()}") # check if file is readable if not access(yaml_path, R_OK): raise IOError(f"YAML file isn't readable: {yaml_path.resolve()}") # check integrity and structure with yaml_path.open(mode="r") as in_yaml_file: try: yaml.safe_load_all(in_yaml_file) except yaml.YAMLError as exc: logger.error(msg=f"YAML file is invalid: {yaml_path.resolve()}") raise exc except Exception as exc: logger.error(msg=f"Structure of YAML file is incorrect: {exc}") raise exc # return sanitized path return yaml_path
[docs] def check_yaml_buffer(self, yaml_buffer: BufferedIOBase) -> BufferedIOBase: """Perform some checks on passed yaml file. :param yaml_buffer: bytes reader of the yaml file to check :returns: checked bytes object :rtype: BufferedIOBase """ # check integrity try: yaml.safe_load_all(yaml_buffer) except yaml.YAMLError as exc: logger.error(f"Invalid YAML {yaml_buffer}. Trace: {exc}") raise exc # return sanitized path return yaml_buffer
[docs] def check_yaml_structure(self, in_yaml_data: dict) -> bool: """Look into the YAML structure and check everything it's OK. :param dict in_yaml_data: YAML file data loaded as dict :return: check result :rtype: bool :example: .. code-block:: python # here comes an example in Python my_yaml = Path("sample.yml") with my_yaml.open(mode="r") as op: yaml_data = yaml.full_load_all(bytes_data) if not check_yaml_structure(yaml_data): print("Bad YAML spotted!") """ # check root structure if not isinstance(in_yaml_data, dict): raise BadQlscStructure( f"QLSC is not a valid dictionary ({type(in_yaml_data)}): {self.input_yaml}" ) # store into var qlsc = in_yaml_data # check doc structure if not all(k in self.QLSC_KEYS for k in qlsc): raise BadQlscStructure(f"Keys are not matching the model: {self.QLSC_KEYS}") # check separators for c in ["CodeSeparator", "ParameterSeparator"]: sep = qlsc.get(c) # TODO: Code and Parameter check in a predefined list? if not isinstance(sep, str) and not len(sep) == 1: raise BadQlscStructure( f"QLSC is not valid. Error in separators: {self.input_yaml}" ) for c in ["AllPoints", "ErrorPoints"]: dict_points = qlsc.get(c) if not ( isinstance(dict_points, dict) and "Layer" in dict_points and isinstance(dict_points.get("Layer"), str) and "isChecked" in dict_points and isinstance(dict_points.get("isChecked"), bool) ): raise BadQlscStructure( f"QLSC is not valid. Error in points: {self.input_yaml}" ) # check codification section codif = qlsc.get("Codification") if not (isinstance(codif, dict) and len(codif)): raise BadQlscStructure( f"QLSC is not valid. Error in codification: {self.input_yaml}" ) for value in codif.values(): if not ( all( [ x in value for x in ["Attributes", "Description", "GeometryType", "Layer"] ] ) and isinstance(value.get("Attributes"), list) and isinstance(value.get("Description"), str) and isinstance(value.get("GeometryType"), str) and isinstance(value.get("Layer"), str) ): raise BadQlscStructure( f"QLSC is not valid. Error in codification values: {self.input_yaml}" ) return True
# UTILS
[docs] def replace_parent_paths(self, original_path: str, new_path: str) -> dict: """Helper to replace string parts within QLSC layers paths. :param str original_path: string to replace :param str new_path: replacement string :raises BadQlscStructure: [description] :return: modified YAML content ready to be dumped :rtype: dict :example: .. code-block:: python qlsc_rdr = QlscYamlReader("codification.qlsc") with open("codification_replaced.qlsc", "w") as stream: yaml.dump( qlsc_rdr.replace_parent_paths("/tmp/lsci", "/tmp/youpi"), stream, default_flow_style=False, ) """ # because of Python limitations on trailing slash... original_path.rstrip("/") new_path.rstrip("/") # load input yaml with self.input_yaml.open(mode="r") as stream_data: yaml_loaded = yaml.full_load(stream_data) if not isinstance(yaml_loaded, dict): raise BadQlscStructure(f"QLSC file is invalid: {self.input_yaml}") # change points paths yaml_loaded.get("AllPoints")["Layer"] = ( yaml_loaded.get("AllPoints") .get("Layer") .replace(original_path, str(new_path)) ) # change error points path yaml_loaded.get("ErrorPoints")["Layer"] = ( yaml_loaded.get("AllPoints") .get("Layer") .replace(original_path, str(new_path)) ) for code in yaml_loaded.get("Codification").values(): code["Layer"] = code["Layer"].replace(original_path, str(new_path)) return yaml_loaded
# ############################################################################# # ##### Main ####################### # ################################## if __name__ == "__main__": """Quick and dirty tests""" logging.basicConfig(level=logging.DEBUG) # fixtures fixtures_dir = Path("./tests/fixtures/") for i in fixtures_dir.glob("**/*.qlsc"): # testing with a Path logger.debug(f"Processing: {i} as {type(i)}") t = QlscYamlReader(i) # testing with a bytes object with i.open("rb") as in_yaml: logger.debug(f"Processing: {i} as {type(in_yaml)}") t = QlscYamlReader(in_yaml) logger.info("YAML processing succeeded.") # demo codifiction qlsc_rdr = QlscYamlReader("LandSurveyCodesImport/resources/demo/codification.qlsc") modified_yaml = qlsc_rdr.replace_parent_paths("/tmp/lsci", "/tmp/youpi") assert isinstance(modified_yaml, dict) with open("codification_replaced.qlsc", "w") as stream: yaml.dump(modified_yaml, stream) qlsc_reader_replaced = QlscYamlReader("codification_replaced.qlsc")