Add yml validation schema generation and schema checker

doc/usage.adoc

@@ -59,6 +59,16 @@ python rules2yml.py # Generates the rule directory
 osivalidator --data data/20240221T141700Z_sv_300_2112_10_one_moving_object.osi --rules rules -p
 ----
 
+The rules2yml.py generates a yml file for each OSI proto file containing the rules specified in OSI.
+The yml files are located in the specified rules folder given as an input parameter.
+Additionally, the script generates respective yml schema files to validate the rule yml files in <rules_folder>/schema.
+The schema files contain the message names of the original OSI proto file and a list of applicable rules.
+If a rule has an associated value, e.g. a string or a number, the type of the value is also checked.
+When executing osivalidator, all rule files are validated against their respective schema.
+
+If needed, the rules folder can be copied and modified for specific use cases, e.g. by adding or removing certain rules.
+This way, osivalidation can be used with different sets of rules.
+
 After successfully running the validation the following output is
 generated:
 

osivalidator/osi_general_validator.py

@@ -46,7 +46,7 @@ def command_line_arguments():
     parser.add_argument(
         "--rules",
         "-r",
-        help="Directory with text files containig rules. ",
+        help="Directory with yml files containing rules. ",
         default=os.path.join(dir_path, "rules"),
         type=str,
     )
@@ -142,7 +142,12 @@ def main():
 
     # Collect Validation Rules
     print("Collect validation rules ...")
-    VALIDATION_RULES.from_yaml_directory(args.rules)
+    try:
+        VALIDATION_RULES.from_yaml_directory(args.rules)
+    except Exception as e:
+        trace.close()
+        print("Error collecting validation rules:", e)
+        exit(1)
 
     # Pass all timesteps or the number specified
     if args.timesteps != -1:

osivalidator/osi_rules.py

@@ -12,6 +12,7 @@
 
 from ruamel.yaml import YAML
 from pathlib import Path
+import yamale
 
 import osi_rules_implementations
 
@@ -31,6 +32,31 @@ def __init__(self):
             "orientation_acceleration",
         }
 
+    def validate_rules_yml(self, file=None):
+        """Validate rule yml files against schema."""
+
+        # Read schema file
+        directory = os.path.dirname(file)
+        filename, file_extension = os.path.splitext(os.path.basename(file))
+        schema_file = directory + os.sep + "schema" + os.sep + filename + "_schema.yml"
+        if os.path.exists(schema_file):
+            schema = yamale.make_schema(schema_file)
+        else:
+            print(f"WARNING: No schema file found for {file}.\n")
+            return
+
+        # Create a Data object
+        data = yamale.make_data(file)
+
+        # Validate data against the schema. Throws a ValueError if data is invalid.
+        try:
+            yamale.validate(schema, data)
+        except yamale.yamale_error.YamaleError as exc:
+            print(exc.message)
+            return False
+
+        return True
+
     def from_yaml_directory(self, path=None):
         """Collect validation rules found in the directory."""
 
@@ -39,13 +65,18 @@ def from_yaml_directory(self, path=None):
             path = os.path.join(dir_path, "rules")
 
         exts = (".yml", ".yaml")
-        try:
-            for filename in os.listdir(path):
-                if filename.startswith("osi_") and filename.endswith(exts):
+        rule_file_errors = dict()
+        for filename in os.listdir(path):
+            if filename.startswith("osi_") and filename.endswith(exts):
+                if self.validate_rules_yml(os.path.join(path, filename)):
                     self.from_yaml_file(os.path.join(path, filename))
+                else:
+                    print(f"WARNING: Invalid rule file: {filename}.\n")
+                    rule_file_errors[filename] = rule_file_errors.get(filename, 0) + 1
 
-        except FileNotFoundError:
-            print("Error while reading files OSI-rules. Exiting!")
+        if rule_file_errors:
+            print(f"Errors per file: {rule_file_errors}")
+            raise Exception("Errors were found in the OSI rule files.")
 
     def from_yaml_file(self, path):
         """Import from a file"""

requirements.txt

@@ -1,4 +1,5 @@
 tqdm>=4.66.1
+yamale>=5.0.0
 tabulate>=0.9.0
 ruamel.yaml>=0.18.5
 defusedxml>=0.7.1

rules2yml.py

@@ -41,6 +41,115 @@ def gen_yml_rules(dir_name="rules", full_osi=False):
 
     if not os.path.exists(dir_name):
         os.makedirs(dir_name)
+    if not os.path.exists(dir_name + "/schema"):
+        os.makedirs(dir_name + "/schema")
+
+    for file in glob("open-simulation-interface/*.proto*"):
+        filename = file.split("open-simulation-interface/")[1].split(".proto")[0]
+
+        if os.path.exists(f"{dir_name}/{filename}.yml"):
+            continue
+
+        with open(f"{dir_name}/schema/{filename}_schema.yml", "a") as schema_file:
+            with open(file, "rt") as fin:
+                isEnum = False
+                numMessage = 0
+                saveStatement = ""
+                prevMainField = False  # boolean, that the previous field has children
+
+                for line in fin:
+                    if file.find(".proto") != -1:
+                        # Search for comment ("//").
+                        matchComment = re.search("//", line)
+                        if matchComment is not None:
+                            statement = line[: matchComment.start()]
+                        else:
+                            statement = line
+
+                        # Add part of the statement from last line.
+                        statement = saveStatement + " " + statement
+
+                        # New line is not necessary. Remove for a better output.
+                        statement = statement.replace("\n", "")
+
+                        # Is statement complete
+                        matchSep = re.search(r"[{};]", statement)
+                        if matchSep is None:
+                            saveStatement = statement
+                            statement = ""
+                        else:
+                            saveStatement = statement[matchSep.end() :]
+                            statement = statement[: matchSep.end()]
+
+                        # Search for "enum".
+                        matchEnum = re.search(r"\benum\b", statement)
+                        if matchEnum is not None:
+                            isEnum = True
+
+                        # Search for a closing brace.
+                        matchClosingBrace = re.search("}", statement)
+                        if isEnum is True and matchClosingBrace is not None:
+                            isEnum = False
+                            continue
+
+                        # Check if not inside an enum.
+                        if isEnum is False:
+                            # Search for "message".
+                            matchMessage = re.search(r"\bmessage\b", statement)
+                            if matchMessage is not None:
+                                # a new message or a new nested message
+                                numMessage += 1
+                                endOfLine = statement[matchMessage.end() :]
+                                matchName = re.search(r"\b\w[\S]*\b", endOfLine)
+                                if matchName is not None and prevMainField is False:
+                                    # Check previous main field to exclude empty fields from sensor specific file
+                                    matchNameConv = re.search(
+                                        r"\b[A-Z][a-zA-Z0-9]*\b",
+                                        endOfLine[matchName.start() : matchName.end()],
+                                    )
+                                    schema_file.write(
+                                        2 * (numMessage - 1) * " "
+                                        + f"{matchNameConv.group(0)}:\n"
+                                    )
+                                    prevMainField = True
+
+                            elif re.search(r"\bextend\b", statement) is not None:
+                                # treat extend as message
+                                numMessage += 1
+
+                            # Search for a closing brace.
+                            matchClosingBrace = re.search("}", statement)
+                            if numMessage > 0 and matchClosingBrace is not None:
+                                numMessage -= 1
+
+                            if matchComment is None and len(saveStatement) == 0:
+                                if numMessage > 0 or isEnum == True:
+                                    if statement.find(";") != -1:
+                                        field = statement.strip().split()[2]
+                                        schema_file.write(
+                                            (2 * numMessage) * " "
+                                            + f"{field}: any(list(include('rules', required=False)), null(), required=False)\n"
+                                        )
+                                        prevMainField = False
+                schema_file.write(
+                    "---\n"
+                    "rules:\n"
+                    "  is_greater_than: num(required=False)\n"
+                    "  is_greater_than_or_equal_to: num(required=False)\n"
+                    "  is_less_than_or_equal_to: num(required=False)\n"
+                    "  is_less_than: num(required=False)\n"
+                    "  is_equal_to: any(num(), bool(), required=False)\n"
+                    "  is_different_to: num(required=False)\n"
+                    "  is_globally_unique: str(required=False)\n"
+                    "  refers_to: str(required=False)\n"
+                    "  is_iso_country_code: str(required=False)\n"
+                    "  is_set: str(required=False)\n"
+                    "  check_if: list(include('rules', required=False),required=False)\n"
+                    "  do_check: any(required=False)\n"
+                    "  target: any(required=False)\n"
+                    "  first_element: any(required=False)\n"
+                    "  last_element: any(required=False)"
+                )
 
     for file in glob("open-simulation-interface/*.proto*"):
         filename = file.split("open-simulation-interface/")[1].split(".proto")[0]
@@ -54,6 +163,7 @@ def gen_yml_rules(dir_name="rules", full_osi=False):
                 numMessage = 0
                 shiftCounter = False
                 saveStatement = ""
+                prevMainField = False  # boolean, that the previous field has children
                 rules = []
 
                 for line in fin:
@@ -120,7 +230,7 @@ def gen_yml_rules(dir_name="rules", full_osi=False):
                                 numMessage += 1
                                 endOfLine = statement[matchMessage.end() :]
                                 matchName = re.search(r"\b\w[\S]*\b", endOfLine)
-                                if matchName is not None:
+                                if matchName is not None and prevMainField is False:
                                     # Test case 10: Check name - no special char -
                                     # start with a capital letter
                                     matchNameConv = re.search(
@@ -132,6 +242,7 @@ def gen_yml_rules(dir_name="rules", full_osi=False):
                                         2 * (numMessage - 1) * " "
                                         + f"{matchNameConv.group(0)}:\n"
                                     )
+                                    prevMainField = True
 
                             elif re.search(r"\bextend\b", statement) is not None:
                                 # treat extend as message
@@ -168,6 +279,8 @@ def gen_yml_rules(dir_name="rules", full_osi=False):
                                         yml_file.write(
                                             (2 * numMessage) * " " + f"{field}:\n"
                                         )
+                                        prevMainField = False
+
                                         # If option --full-osi is enabled:
                                         # Check if is_set is already a rule for the current field, if not, add it.
                                         if full_osi and not any(
@@ -233,6 +346,7 @@ def gen_yml_rules(dir_name="rules", full_osi=False):
                                                         (2 * numMessage + 8) * " "
                                                         + f"- {rule_list[2]}: {rule_list[3]}\n"
                                                     )
+
                                                 # Standalone rules
                                                 elif any(
                                                     list_item

tests/test_validation_rules.py

@@ -1,13 +1,18 @@
 """Module for test class of OSIValidationRules class"""
 
 import unittest
+from glob import *
+import os
+import shutil
+import yamale
 from osivalidator.osi_rules import (
     Rule,
     TypeRulesContainer,
     ProtoMessagePath,
     OSIRules,
     OSIRuleNode,
 )
+from rules2yml import gen_yml_rules
 
 
 class TestValidationRules(unittest.TestCase):
@@ -58,6 +63,53 @@ def test_parse_yaml(self):
 
         self.assertEqual(field["is_set"], rule_check)
 
+    def test_yaml_generation(self):
+        gen_yml_rules("unit_test_rules/")
+
+        num_proto_files = len(glob("open-simulation-interface/*.proto*"))
+        num_rule_files = len(glob("unit_test_rules/*.yml"))
+        num_rule_schema_files = len(glob("unit_test_rules/schema/*.yml"))
+        self.assertEqual(num_proto_files, num_rule_files)
+        self.assertEqual(num_rule_files, num_rule_schema_files)
+
+        # clean up
+        if os.path.isdir("unit_test_rules"):
+            shutil.rmtree("unit_test_rules")
+
+    def test_yaml_schema_fail(self):
+        gen_yml_rules("unit_test_rules/")
+
+        # alter exemplary rule for fail check
+        raw_sensorspecific = """RadarSpecificObjectData:
+  rcs:
+LidarSpecificObjectData:
+  maximum_measurement_distance_sensor:
+    - is_greater_than_or_equal_to: 0
+  probability:
+    - is_less_than_or_equal_to: x
+    - is_greater_than_or_equal_to: 0
+  trilateration_status:
+  trend:
+  signalway:
+  Signalway:
+    sender_id:
+    receiver_id:
+"""
+
+        os.remove("unit_test_rules/osi_sensorspecific.yml")
+        with open("unit_test_rules/osi_sensorspecific.yml", "w") as rule_file:
+            rule_file.write(raw_sensorspecific)
+
+        validation_rules = OSIRules()
+        validation_output = validation_rules.validate_rules_yml(
+            "unit_test_rules/osi_sensorspecific.yml"
+        )
+        self.assertEqual(validation_output, False)
+
+        # clean up
+        if os.path.isdir("unit_test_rules"):
+            shutil.rmtree("unit_test_rules")
+
 
 if __name__ == "__main__":
     unittest.main()