config_tools: add a slicer of XML schema

We plan to add the following attributes to element definitions in the XML
schema:

  - acrn:applicable-vms, which specify if an element applies to a
    pre-launched VM, the service VM or a post-launched VM

  - acrn:views, which specify if an element shall appear in the basic or
    advanced tab in the configurator.

In order to reduce the attributes above to existing XML technologies, we
need to create new complex types that lists all config items that applies
to a pre-launched VM, the service VM or a post-launched VM, or that should
be shown in the basic or advanced view. Such types can then be used to
replace the original, all-in-one type during validation or configurator
rendering.

When unspecified, an element always applies under all possible
circumstances.

To realize this slicing mechanism, this patch adds a generic class
implementing the common part of slicing XML schema types and two
specific-purpose slicers according to the applicable VMs or views
attributes.

v2 -> v3:

  * Update configdoc.xsl to recognize types in xs:alternative nodes.

Tracked-On: #6690
Signed-off-by: Junjie Mao <junjie.mao@intel.com>

doc/scripts/configdoc.xsl

@@ -39,7 +39,16 @@
     <xsl:param name="level"/>
     <xsl:param name="prefix"/>
 
-    <xsl:variable name="ty" select="@type"/>
+    <xsl:variable name="ty">
+      <xsl:choose>
+	<xsl:when test="@type">
+	  <xsl:value-of select="@type" />
+	</xsl:when>
+	<xsl:otherwise>
+	  <xsl:value-of select=".//xs:alternative[1]/@type" />
+	</xsl:otherwise>
+      </xsl:choose>
+    </xsl:variable>
 
     <!-- Only visit elements having complex types. Those having simple types are
          described as an option.. -->

misc/config_tools/scenario_config/default_populator.py

@@ -11,6 +11,7 @@ import argparse
 from scenario_transformer import ScenarioTransformer
 
 from pipeline import PipelineObject, PipelineStage, PipelineEngine
+from schema_slicer import SlicingSchemaByVMTypeStage
 
 class DefaultValuePopulator(ScenarioTransformer):
     def add_missing_nodes(self, xsd_element_node, xml_parent_node, new_node_index):
@@ -51,6 +52,7 @@ def main(args):
     pipeline.add_stages([
         LXMLLoadStage("schema"),
         XMLLoadStage("scenario"),
+        SlicingSchemaByVMTypeStage(),
         DefaultValuePopulatingStage(),
     ])
 

misc/config_tools/scenario_config/schema_slicer.py

@@ -0,0 +1,248 @@
+#!/usr/bin/env python3
+#
+# Copyright (C) 2022 Intel Corporation.
+#
+# SPDX-License-Identifier: BSD-3-Clause
+#
+
+import os
+import argparse
+from copy import deepcopy
+
+from pipeline import PipelineObject, PipelineStage, PipelineEngine
+
+class SchemaTypeSlicer:
+    xpath_ns = {
+        "xs": "http://www.w3.org/2001/XMLSchema",
+        "acrn": "https://projectacrn.org",
+    }
+
+    @classmethod
+    def get_node(cls, element, xpath):
+        return element.find(xpath, namespaces=cls.xpath_ns)
+
+    @classmethod
+    def get_nodes(cls, element, xpath):
+        return element.findall(xpath, namespaces=cls.xpath_ns)
+
+    def __init__(self, etree):
+        self.etree = etree
+
+    def get_type_definition(self, type_name):
+        type_node = self.get_node(self.etree, f"//xs:complexType[@name='{type_name}']")
+        if type_node is None:
+            type_node = self.get_node(self.etree, f"//xs:simpleType[@name='{type_name}']")
+        return type_node
+
+    def slice_element_list(self, element_list_node, new_nodes):
+        sliced = False
+
+        for element_node in self.get_nodes(element_list_node, "xs:element"):
+            if not self.is_element_needed(element_node):
+                element_list_node.remove(element_node)
+                sliced = True
+                continue
+
+            # For embedded complex type definition, also slice in place. If the sliced type contains no sub-element,
+            # remove the element itself, too.
+            element_type_node = self.get_node(element_node, "xs:complexType")
+            if element_type_node is not None:
+                new_sub_nodes = self.slice(element_type_node, in_place=True)
+                if len(self.get_nodes(element_type_node, ".//xs:element")) > 0:
+                    new_nodes.extend(new_sub_nodes)
+                else:
+                    element_list_node.remove(element_node)
+                continue
+
+            # For external type definition, create a copy to slice. If the sliced type contains no sub-element, remove
+            # the element itself.
+            element_type_name = element_node.get("type")
+            if element_type_name:
+                element_type_node = self.get_type_definition(element_type_name)
+                if element_type_node is not None:
+                    sliced_type_name = self.get_name_of_slice(element_type_name)
+
+                    # If a sliced type already exists, do not duplicate the effort
+                    type_node = self.get_type_definition(sliced_type_name)
+                    if type_node is not None:
+                        element_node.set("type", sliced_type_name)
+                        sliced = True
+                    else:
+                        new_sub_nodes = self.slice(element_type_node)
+                        if len(new_sub_nodes) == 0:
+                            continue
+                        elif new_sub_nodes[-1].tag.endswith("simpleType") or len(self.get_nodes(new_sub_nodes[-1], ".//xs:element")) > 0:
+                            new_nodes.extend(new_sub_nodes)
+                            element_node.set("type", sliced_type_name)
+                            sliced = True
+                        else:
+                            element_list_node.remove(element_node)
+
+        return sliced
+
+    def slice_restriction(self, restriction_node):
+        sliced = False
+
+        for restriction in self.get_nodes(restriction_node, "xs:enumeration"):
+            if not self.is_element_needed(restriction):
+                restriction_node.remove(restriction)
+                sliced = True
+
+        return sliced
+
+    def slice(self, type_node, in_place=False, force_copy=False):
+        new_nodes = []
+        sliced = False
+
+        if in_place:
+            new_type_node = type_node
+        else:
+            new_type_node = deepcopy(type_node)
+            type_name = type_node.get("name")
+            if type_name != None:
+                sliced_type_name = self.get_name_of_slice(type_name)
+                new_type_node.set("name", sliced_type_name)
+
+        element_list_node = self.get_node(new_type_node, "xs:all")
+        if element_list_node is not None:
+            sliced = self.slice_element_list(element_list_node, new_nodes)
+
+        restriction_node = self.get_node(new_type_node, "xs:restriction")
+        if restriction_node is not None:
+            sliced = self.slice_restriction(restriction_node)
+
+        if not in_place and (sliced or force_copy):
+            new_nodes.append(new_type_node)
+        return new_nodes
+
+    def is_element_needed(self, element_node):
+        return True
+
+    def get_name_of_slice(self, name):
+        return f"Sliced{name}"
+
+class SlicingSchemaByVMTypeStage(PipelineStage):
+    uses = {"schema_etree"}
+    provides = {"schema_etree"}
+
+    class VMTypeSlicer(SchemaTypeSlicer):
+        def is_element_needed(self, element_node):
+            annot_node = self.get_node(element_node, "xs:annotation")
+            if annot_node is None:
+                return True
+            applicable_vms = annot_node.get("{https://projectacrn.org}applicable-vms")
+            return applicable_vms is None or applicable_vms.find(self.vm_type_indicator) >= 0
+
+        def get_name_of_slice(self, name):
+            return f"{self.type_prefix}{name}"
+
+    class PreLaunchedTypeSlicer(VMTypeSlicer):
+        vm_type_indicator = "pre-launched"
+        type_prefix = "PreLaunched"
+
+    class ServiceVMTypeSlicer(VMTypeSlicer):
+        vm_type_indicator = "service-vm"
+        type_prefix = "Service"
+
+    class PostLaunchedTypeSlicer(VMTypeSlicer):
+        vm_type_indicator = "post-launched"
+        type_prefix = "PostLaunched"
+
+    def run(self, obj):
+        schema_etree = obj.get("schema_etree")
+
+        vm_type_name = "VMConfigType"
+        vm_type_node = SchemaTypeSlicer.get_node(schema_etree, f"//xs:complexType[@name='{vm_type_name}']")
+        slicers = [
+            self.PreLaunchedTypeSlicer(schema_etree),
+            self.ServiceVMTypeSlicer(schema_etree),
+            self.PostLaunchedTypeSlicer(schema_etree)
+        ]
+
+        for slicer in slicers:
+            new_nodes = slicer.slice(vm_type_node, force_copy=True)
+            for n in new_nodes:
+                schema_etree.getroot().append(n)
+
+        for node in SchemaTypeSlicer.get_nodes(schema_etree, "//xs:complexType[@name='ACRNConfigType']//xs:element[@name='vm']//xs:alternative"):
+            test = node.get("test")
+            if test.find("PRE_LAUNCHED_VM") >= 0:
+                node.set("type", slicers[0].get_name_of_slice(vm_type_name))
+            elif test.find("SERVICE_VM") >= 0:
+                node.set("type", slicers[1].get_name_of_slice(vm_type_name))
+            elif test.find("POST_LAUNCHED_VM") >= 0:
+                node.set("type", slicers[2].get_name_of_slice(vm_type_name))
+
+        obj.set("schema_etree", schema_etree)
+
+class SlicingSchemaByViewStage(PipelineStage):
+    uses = {"schema_etree"}
+    provides = {"schema_etree"}
+
+    class ViewSlicer(SchemaTypeSlicer):
+        def is_element_needed(self, element_node):
+            annot_node = self.get_node(element_node, "xs:annotation")
+            if annot_node is None:
+                return True
+            views = annot_node.get("{https://projectacrn.org}views")
+            return views is None or views.find(self.view_indicator) >= 0
+
+        def get_name_of_slice(self, name):
+            if name.find("ConfigType") >= 0:
+                return name.replace("ConfigType", f"{self.type_prefix}ConfigType")
+            else:
+                return f"{self.type_prefix}{name}"
+
+    class BasicViewSlicer(ViewSlicer):
+        view_indicator = "basic"
+        type_prefix = "Basic"
+
+    class AdvancedViewSlicer(ViewSlicer):
+        view_indicator = "advanced"
+        type_prefix = "Advanced"
+
+    def run(self, obj):
+        schema_etree = obj.get("schema_etree")
+
+        type_nodes = list(filter(lambda x: x.get("name") and x.get("name").endswith("VMConfigType"), SchemaTypeSlicer.get_nodes(schema_etree, "//xs:complexType")))
+        type_nodes.append(SchemaTypeSlicer.get_node(schema_etree, "//xs:complexType[@name = 'HVConfigType']"))
+
+        slicers = [
+            self.BasicViewSlicer(schema_etree),
+            self.AdvancedViewSlicer(schema_etree),
+        ]
+
+        for slicer in slicers:
+            for type_node in type_nodes:
+                new_nodes = slicer.slice(type_node, force_copy=True)
+                for n in new_nodes:
+                    schema_etree.getroot().append(n)
+
+        obj.set("schema_etree", schema_etree)
+
+def main(args):
+    from lxml_loader import LXMLLoadStage
+
+    pipeline = PipelineEngine(["schema_path"])
+    pipeline.add_stages([
+        LXMLLoadStage("schema"),
+        SlicingSchemaByVMTypeStage(),
+        SlicingSchemaByViewStage(),
+    ])
+
+    obj = PipelineObject(schema_path = args.schema)
+    pipeline.run(obj)
+    obj.get("schema_etree").write(args.out)
+
+    print(f"Sliced schema written to {args.out}")
+
+if __name__ == "__main__":
+    config_tools_dir = os.path.join(os.path.dirname(__file__), "..")
+    schema_dir = os.path.join(config_tools_dir, "schema")
+
+    parser = argparse.ArgumentParser(description="Slice a given scenario schema by VM types and views")
+    parser.add_argument("out", nargs="?", default=os.path.join(schema_dir, "sliced.xsd"), help="Path where the output is placed")
+    parser.add_argument("--schema", default=os.path.join(schema_dir, "config.xsd"), help="the XML schema that defines the syntax of scenario XMLs")
+    args = parser.parse_args()
+
+    main(args)

misc/config_tools/scenario_config/validator.py

@@ -20,6 +20,7 @@ except ImportError:
     sys.exit(0)
 
 from pipeline import PipelineObject, PipelineStage, PipelineEngine
+from schema_slicer import SlicingSchemaByVMTypeStage
 from default_populator import DefaultValuePopulatingStage
 
 def existing_file_type(parser):
@@ -174,6 +175,7 @@ def main(args):
     validator_construction_pipeline.add_stages([
         LXMLLoadStage("schema"),
         LXMLLoadStage("datachecks"),
+        SlicingSchemaByVMTypeStage(),
         ValidatorConstructionStage(),
     ])
 

misc/config_tools/schema/config.xsd

@@ -450,7 +450,7 @@ board by configuring the hypervisor image features and capabilities such as
 setting up the log and the serial port.</xs:documentation>
       </xs:annotation>
     </xs:element>
-    <xs:element name="vm" type="VMConfigType" maxOccurs="unbounded">
+    <xs:element name="vm" maxOccurs="unbounded">
       <xs:annotation>
         <xs:documentation>VM configuration includes **scenario-based** VM configuration
 information that is used to describe the characteristics and attributes for
@@ -458,6 +458,9 @@ all VMs in a user scenario. It also includes **launch script-based** VM
 configuration information, where parameters are passed to the device model
 to launch post-launched User VMs.</xs:documentation>
       </xs:annotation>
+      <xs:alternative test=".//load_order = 'PRE_LAUNCHED_VM'" type="VMConfigType" />
+      <xs:alternative test=".//load_order = 'SERVICE_VM'" type="VMConfigType" />
+      <xs:alternative test=".//load_order = 'POST_LAUNCHED_VM'" type="VMConfigType" />
     </xs:element>
   </xs:all>
   <xs:attribute name="board" type="xs:string" use="required" />