Merge pull request #93 from asam-ev/2024-09-11

Add schema and basic checks

pyproject.toml

@@ -18,6 +18,7 @@ numpy = "^1.26.0"
 scipy = "^1.14.0"
 pyclothoids = "^0.1.5"
 transforms3d = "^0.4.2"
+xmlschema = "^3.3.2"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^8.2.2"

qc_opendrive/checks/basic/__init__.py

@@ -0,0 +1,6 @@
+from . import basic_constants as basic_constants
+from . import basic_checker as basic_checker
+from . import valid_xml_document as valid_xml_document
+from . import root_tag_is_opendrive as root_tag_is_opendrive
+from . import fileheader_is_present as fileheader_is_present
+from . import version_is_defined as version_is_defined

qc_opendrive/checks/basic/basic_checker.py

@@ -0,0 +1,67 @@
+import logging
+import os
+
+from lxml import etree
+
+from qc_baselib import Configuration, Result, StatusType
+
+from qc_opendrive import constants
+from qc_opendrive.base import models, utils
+
+from qc_opendrive.checks.basic import (
+    basic_constants,
+    valid_xml_document,
+    root_tag_is_opendrive,
+    fileheader_is_present,
+    version_is_defined,
+)
+
+
+def run_checks(config: Configuration, result: Result) -> bool:
+    logging.info("Executing basic checks")
+
+    result.register_checker(
+        checker_bundle_name=constants.BUNDLE_NAME,
+        checker_id=basic_constants.CHECKER_ID,
+        description="Check if basic properties of input file are properly set",
+        summary="",
+    )
+
+    xml_file_path = config.get_config_param("InputFile")
+
+    is_xml = valid_xml_document.check_rule(xml_file_path, result)
+
+    basic_rule_list = [
+        root_tag_is_opendrive.check_rule,
+        fileheader_is_present.check_rule,
+        version_is_defined.check_rule,
+    ]
+
+    validation_result = is_xml
+    if validation_result:
+        root = utils.get_root_without_default_namespace(xml_file_path)
+
+        for rule in basic_rule_list:
+            validation_result = validation_result and rule(root, result)
+            if not validation_result:
+                break
+
+    if not validation_result:
+        logging.warning(
+            "There are problems with input file. Error found in basic rules!"
+        )
+    else:
+        logging.info("Basic rules check successfull. Input file is valid")
+
+    logging.info(
+        f"Issues found - {result.get_checker_issue_count(checker_bundle_name=constants.BUNDLE_NAME, checker_id=basic_constants.CHECKER_ID)}"
+    )
+
+    # TODO: Add logic to deal with error or to skip it
+    result.set_checker_status(
+        checker_bundle_name=constants.BUNDLE_NAME,
+        checker_id=basic_constants.CHECKER_ID,
+        status=StatusType.COMPLETED,
+    )
+
+    return validation_result

qc_opendrive/checks/basic/basic_constants.py

@@ -0,0 +1 @@
+CHECKER_ID = "basic_xodr"

qc_opendrive/checks/basic/fileheader_is_present.py

@@ -0,0 +1,58 @@
+import logging
+from lxml import etree
+from qc_baselib import IssueSeverity, Result
+from qc_opendrive import constants
+from qc_opendrive.checks.basic import basic_constants
+
+
+def check_rule(tree: etree._ElementTree, result: Result) -> bool:
+    """
+    Below the root element a tag with FileHeader must be defined
+
+    More info at
+        - https://github.com/asam-ev/qc-opendrive/issues/89
+    """
+    logging.info("Executing fileheader_is_present check")
+
+    rule_uid = result.register_rule(
+        checker_bundle_name=constants.BUNDLE_NAME,
+        checker_id=basic_constants.CHECKER_ID,
+        emanating_entity="asam.net",
+        standard="xodr",
+        definition_setting="1.0.0",
+        rule_full_name="xml.fileheader_is_present",
+    )
+
+    root = tree.getroot()
+
+    is_valid = False
+    # Check if root contains a tag 'header'
+    file_header_tag = root.find("header")
+    if file_header_tag is not None:
+        logging.info("- Root tag contains header -> OK")
+        is_valid = True
+    else:
+        logging.error("- header not found under root element")
+        is_valid = False
+
+    if not is_valid:
+
+        issue_id = result.register_issue(
+            checker_bundle_name=constants.BUNDLE_NAME,
+            checker_id=basic_constants.CHECKER_ID,
+            description="Issue flagging when no header is found under root element",
+            level=IssueSeverity.ERROR,
+            rule_uid=rule_uid,
+        )
+
+        result.add_xml_location(
+            checker_bundle_name=constants.BUNDLE_NAME,
+            checker_id=basic_constants.CHECKER_ID,
+            issue_id=issue_id,
+            xpath=tree.getpath(root),
+            description=f"No child element header",
+        )
+
+        return False
+
+    return True

qc_opendrive/checks/basic/root_tag_is_opendrive.py

@@ -0,0 +1,56 @@
+import logging
+from lxml import etree
+from qc_baselib import IssueSeverity, Result
+from qc_opendrive import constants
+from qc_opendrive.checks.basic import basic_constants
+
+
+def check_rule(tree: etree._ElementTree, result: Result) -> bool:
+    """
+    The root element of a valid XML document must be OpenDRIVE
+
+    More info at
+        - https://github.com/asam-ev/qc-opendrive/issues/88
+    """
+    logging.info("Executing root_tag_is_opendrive check")
+
+    rule_uid = result.register_rule(
+        checker_bundle_name=constants.BUNDLE_NAME,
+        checker_id=basic_constants.CHECKER_ID,
+        emanating_entity="asam.net",
+        standard="xodr",
+        definition_setting="1.0.0",
+        rule_full_name="xml.root_tag_is_opendrive",
+    )
+
+    root = tree.getroot()
+
+    is_valid = False
+    if root.tag == "OpenDRIVE":
+        logging.info("- Root tag is 'OpenDRIVE'")
+        is_valid = True
+    else:
+        logging.error("- Root tag is not 'OpenDRIVE'")
+        is_valid = False
+
+    if not is_valid:
+
+        issue_id = result.register_issue(
+            checker_bundle_name=constants.BUNDLE_NAME,
+            checker_id=basic_constants.CHECKER_ID,
+            description="Issue flagging when root tag is not OpenDRIVE",
+            level=IssueSeverity.ERROR,
+            rule_uid=rule_uid,
+        )
+
+        result.add_xml_location(
+            checker_bundle_name=constants.BUNDLE_NAME,
+            checker_id=basic_constants.CHECKER_ID,
+            issue_id=issue_id,
+            xpath=tree.getpath(root),
+            description=f"Root is not OpenDRIVE",
+        )
+
+        return False
+
+    return True

qc_opendrive/checks/basic/valid_xml_document.py

@@ -0,0 +1,62 @@
+import logging
+from lxml import etree
+from qc_baselib import Result, IssueSeverity
+from qc_opendrive import constants
+from qc_opendrive.checks.basic import basic_constants
+
+
+def _is_xml_doc(file_path: str) -> tuple[bool, tuple[int, int]]:
+    try:
+        with open(file_path, "rb") as file:
+            xml_content = file.read()
+            etree.fromstring(xml_content)
+            logging.info("- It is an xml document.")
+        return True, None
+    except etree.XMLSyntaxError as e:
+        logging.error(f"- Error: {e}")
+        logging.error(f"- Error occurred at line {e.lineno}, column {e.offset}")
+        return False, (e.lineno, e.offset)
+
+
+def check_rule(input_xml_file_path: str, result: Result) -> bool:
+    """
+    Implements a rule to check if input file is a valid xml document
+
+    More info at
+        - https://github.com/asam-ev/qc-opendrive/issues/88
+    """
+    logging.info("Executing valid_xml_document check")
+
+    rule_uid = result.register_rule(
+        checker_bundle_name=constants.BUNDLE_NAME,
+        checker_id=basic_constants.CHECKER_ID,
+        emanating_entity="asam.net",
+        standard="xodr",
+        definition_setting="1.0.0",
+        rule_full_name="xml.valid_xml_document",
+    )
+
+    is_valid, error_location = _is_xml_doc(input_xml_file_path)
+
+    if not is_valid:
+
+        issue_id = result.register_issue(
+            checker_bundle_name=constants.BUNDLE_NAME,
+            checker_id=basic_constants.CHECKER_ID,
+            description="Issue flagging when input file is not a valid xml document",
+            level=IssueSeverity.ERROR,
+            rule_uid=rule_uid,
+        )
+
+        result.add_file_location(
+            checker_bundle_name=constants.BUNDLE_NAME,
+            checker_id=basic_constants.CHECKER_ID,
+            issue_id=issue_id,
+            row=error_location[0],
+            column=error_location[1],
+            description=f"Invalid xml detected",
+        )
+
+        return False
+
+    return True

qc_opendrive/checks/basic/version_is_defined.py

@@ -0,0 +1,85 @@
+import logging
+from lxml import etree
+from qc_baselib import IssueSeverity, Result
+from qc_opendrive import constants
+from qc_opendrive.checks.basic import basic_constants
+
+
+def is_unsigned_short(value: int) -> bool:
+    """Helper function to check if a value is within the xsd:unsignedShort range (0-65535)."""
+    try:
+        num = int(value)
+        return 0 <= num <= 65535
+    except ValueError:
+        return False
+
+
+def check_rule(tree: etree._ElementTree, result: Result) -> bool:
+    """
+    The header tag must have the attributes revMajor and revMinor and of type unsignedShort.
+
+    More info at
+        - https://github.com/asam-ev/qc-opendrive/issues/90
+    """
+    logging.info("Executing version_is_defined check")
+
+    rule_uid = result.register_rule(
+        checker_bundle_name=constants.BUNDLE_NAME,
+        checker_id=basic_constants.CHECKER_ID,
+        emanating_entity="asam.net",
+        standard="xodr",
+        definition_setting="1.0.0",
+        rule_full_name="xml.version_is_defined",
+    )
+
+    root = tree.getroot()
+
+    is_valid = True
+    # Check if root contains a tag 'header'
+    file_header_tag = root.find("header")
+
+    if file_header_tag is None:
+        logging.error("- No header found, cannot check version. Skipping...")
+        return True
+
+    # Check if 'header' has the attributes 'revMajor' and 'revMinor'
+    if (
+        "revMajor" not in file_header_tag.attrib
+        or "revMinor" not in file_header_tag.attrib
+    ):
+        logging.error("- 'header' tag does not have both 'revMajor' and 'revMinor'")
+        is_valid = False
+
+    if is_valid:
+        # Check if 'attr1' and 'attr2' are xsd:unsignedShort (i.e., in the range 0-65535)
+        rev_major = file_header_tag.attrib["revMajor"]
+        rev_minor = file_header_tag.attrib["revMinor"]
+
+        if not is_unsigned_short(rev_major) or not is_unsigned_short(rev_minor):
+            logging.error(
+                "- 'revMajor' and/or 'revMinor' are not xsd:unsignedShort (0-65535)"
+            )
+            is_valid = False
+
+    if not is_valid:
+
+        issue_id = result.register_issue(
+            checker_bundle_name=constants.BUNDLE_NAME,
+            checker_id=basic_constants.CHECKER_ID,
+            description="Issue flagging when revMajor revMinor attribute of header are missing or invalid",
+            level=IssueSeverity.ERROR,
+            rule_uid=rule_uid,
+        )
+
+        result.add_xml_location(
+            checker_bundle_name=constants.BUNDLE_NAME,
+            checker_id=basic_constants.CHECKER_ID,
+            issue_id=issue_id,
+            xpath=tree.getpath(file_header_tag),
+            description=f"header tag has invalid or missing version info",
+        )
+
+        return False
+    else:
+        logging.info(f"- header version correctly defined: {rev_major}.{rev_minor}")
+    return True

qc_opendrive/checks/geometry/geometry_checker.py

@@ -16,6 +16,28 @@
 )
 
 
+def skip_checks(result: Result) -> None:
+
+    if geometry_constants.CHECKER_ID not in result.get_checker_ids(
+        constants.BUNDLE_NAME
+    ):
+        result.register_checker(
+            checker_bundle_name=constants.BUNDLE_NAME,
+            checker_id=geometry_constants.CHECKER_ID,
+            description="Check if xml properties of input file are properly set",
+            summary="",
+        )
+
+    logging.error(
+        f"Invalid xml input file. Checker {geometry_constants.CHECKER_ID} skipped"
+    )
+    result.set_checker_status(
+        checker_bundle_name=constants.BUNDLE_NAME,
+        checker_id=geometry_constants.CHECKER_ID,
+        status=StatusType.SKIPPED,
+    )
+
+
 def run_checks(config: Configuration, result: Result) -> None:
     logging.info("Executing geometry checks")