Pipeline for policy generation from DDS discovery data

sros2/package.xml

@@ -15,6 +15,7 @@
 
   <exec_depend>python3-lxml</exec_depend>
   <exec_depend>python3-cryptography</exec_depend>
+  <exec_depend>python3-pandas</exec_depend>
 
   <test_depend>ament_copyright</test_depend>
   <test_depend>ament_flake8</test_depend>

sros2/scripts/dds_sql_to_sros2_policy.py

@@ -0,0 +1,151 @@
+#! /usr/bin/env python3
+
+import argparse
+import os
+import pandas as pd
+import sqlite3
+import sys
+
+from lxml import etree
+
+from sros2.policy import (
+    get_policy_schema,
+    get_transport_schema,
+    get_transport_template,
+    POLICY_VERSION,
+)
+
+node_query = """
+SELECT DISTINCT
+    DCPSParticipant.ParticipantData_key,
+    DCPSParticipant.ParticipantData_user_data
+FROM DCPSParticipant
+WHERE DCPSParticipant.ParticipantData_user_data IS NOT NULL;
+"""
+
+_query = """
+SELECT DISTINCT
+    DCPSParticipant.ParticipantData_key,
+    DCPSParticipant.ParticipantData_user_data,
+    DCPS{mode}.{mode}Data_topic_name
+FROM DCPSParticipant 
+INNER JOIN DCPS{mode} ON
+    DCPSParticipant.ParticipantData_key = DCPS{mode}.{mode}Data_participant_key
+AND DCPSParticipant.ParticipantData_user_data IS NOT NULL;
+"""
+
+node_pub_query = _query.format(mode="Publication")
+node_sub_query = _query.format(mode="Subscription")
+
+
+def user_bytes_to_dict(user_bytes):
+    try:
+        user_string = user_bytes.decode('utf8')
+        key_value_list = user_string[:-2].split(';')
+        key_value_dict = dict()
+        for key_values in key_value_list:
+            key, value = key_values.split('=', 1)
+            key_value_dict[key] = value
+        return key_value_dict
+    except:
+        return None
+
+
+def translate_df(df):
+    _df = df['ParticipantData_user_data'] \
+        .apply(user_bytes_to_dict) \
+        .apply(pd.Series)
+    df = pd.concat([df, _df], axis=1)
+    return df
+
+
+def db_to_df(db):
+    node_df = translate_df(pd.read_sql_query(node_query, db))
+    pub_df = translate_df(pd.read_sql_query(node_pub_query, db))
+    sub_df = translate_df(pd.read_sql_query(node_sub_query, db))
+
+    pub_df = pub_df.assign(mode='publish')
+    pub_df = pub_df.rename(columns={"PublicationData_topic_name": "dds_topic"})
+    sub_df = sub_df.assign(mode='subscribe')
+    sub_df = sub_df.rename(columns={"SubscriptionData_topic_name": "dds_topic"})
+
+    df = pd.concat([node_df, pub_df, sub_df])
+    df = df[df['namespace'].notnull()]
+    df.set_index(['namespace', 'name', 'mode'], inplace=True)
+    return df
+
+
+def df_to_dds_policy(df):
+    dds_policy = etree.Element('policy')
+    dds_policy.set("version", POLICY_VERSION,)
+    profiles = etree.SubElement(dds_policy, 'profiles')
+
+    for namespace in df.index.get_level_values('namespace').unique():
+        _df = df.loc[namespace, :, :]
+        for name in _df.index.get_level_values('name').unique():
+            profile = etree.SubElement(profiles, 'profile')
+            profile.set("ns", namespace)
+            profile.set("node", name)
+            __df = df.loc[namespace, name, :]
+            for mode in __df.index.get_level_values('mode').unique():
+                if not pd.isna(mode):
+                    topics = etree.SubElement(profile, 'raws')
+                    topics.set(mode, "ALLOW")
+                    for dds_topic in df['dds_topic'].loc[namespace, name, mode]:
+                        topic = etree.SubElement(topics, 'raw')
+                        topic.text = dds_topic
+    return dds_policy
+
+
+def dds_policy_to_sros2_policy(dds_policy):
+
+    # Parse files
+    sros2_policy_xsd = etree.XMLSchema(
+        etree.parse(
+            get_policy_schema('policy.xsd')))
+    dds_policy_xsd = etree.XMLSchema(
+        etree.parse(
+            get_transport_schema('dds', 'policy.xsd')))
+    dds_demangle_xsl = etree.XSLT(
+        etree.parse(
+            get_transport_template('dds', 'demangle.xsl')))
+
+    # Validate input schema
+    dds_policy_xsd.assertValid(dds_policy)
+
+    # Transform policy
+    sros2_policy = dds_demangle_xsl(dds_policy)
+
+    # Validate output schema
+    sros2_policy_xsd.assertValid(sros2_policy)
+
+    return sros2_policy
+
+
+def main(argv=sys.argv[1:]):
+    parser = argparse.ArgumentParser(
+        formatter_class=argparse.ArgumentDefaultsHelpFormatter)
+    parser.add_argument(
+        '-i', '--input-db', required=True,
+        help='path to SQLite3 database with discovery data')
+    parser.add_argument(
+        '-o', '--output-policy',
+        help='path to XML policy file with generated output')
+    args = parser.parse_args(argv)
+
+    db_uri = 'file:{}?mode=ro'.format(args.input_db)
+    with sqlite3.connect(db_uri, uri=True) as db:
+        df = db_to_df(db)
+
+    dds_policy = df_to_dds_policy(df)
+    sros2_policy = dds_policy_to_sros2_policy(dds_policy)
+
+    if args.output_policy is not None:
+        with open(args.output_policy, 'wb') as f:
+            f.write(etree.tostring(sros2_policy, pretty_print=True))
+    else:
+        print(etree.tostring(sros2_policy, pretty_print=True).decode())
+
+
+if __name__ == '__main__':
+    main()

sros2/sros2/policy/schemas/dds/policy.xsd

@@ -0,0 +1,56 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<xs:schema
+    xmlns:xs="http://www.w3.org/2001/XMLSchema"
+    xmlns:xml="http://www.w3.org/XML/1998/namespace"
+    elementFormDefault="qualified" attributeFormDefault="unqualified">
+    <xs:import namespace="http://www.w3.org/XML/1998/namespace"
+               schemaLocation="http://www.w3.org/2001/03/xml.xsd" />
+    <!-- TODO: Use namespaces for xs:import so schemas can be DRY -->
+
+    <xs:element name="policy" type="Policy" />
+    <xs:complexType name="Policy">
+        <xs:sequence minOccurs="1" maxOccurs="1">
+            <xs:element name="profiles" type="Profiles" />
+        </xs:sequence>
+        <xs:attribute name="version" type="xs:string" use="required" />
+    </xs:complexType>
+
+    <xs:complexType name="Profiles">
+        <xs:sequence minOccurs="1" maxOccurs="unbounded">
+            <xs:element name="profile" type="Profile" />
+        </xs:sequence>
+    </xs:complexType>
+
+    <xs:complexType name="Profile">
+        <xs:sequence minOccurs="0" maxOccurs="unbounded">
+            <xs:choice minOccurs="1" maxOccurs="1">
+                <xs:element name="raws" minOccurs="1" type="RawExpressionList" />
+            </xs:choice>
+        </xs:sequence>
+        <xs:attribute name="ns" type="xs:string" use="required" />
+        <xs:attribute name="node" type="xs:string" use="required" />
+        <xs:attribute ref="xml:base" />
+    </xs:complexType>
+
+    <xs:complexType name="RawExpressionList">
+        <xs:sequence minOccurs="1" maxOccurs="unbounded">
+            <xs:element name="raw" type="Expression" />
+        </xs:sequence>
+        <xs:attribute name="publish" type="RuleQualifier" use="optional" />
+        <xs:attribute name="relay" type="RuleQualifier" use="optional" />
+        <xs:attribute name="subscribe" type="RuleQualifier" use="optional" />
+        <xs:attribute ref="xml:base" />
+    </xs:complexType>
+
+    <xs:simpleType name="Expression">
+        <xs:restriction base="xs:string" />
+    </xs:simpleType>
+
+    <xs:simpleType name="RuleQualifier">
+        <xs:restriction base="xs:string">
+            <xs:enumeration value="ALLOW" />
+            <xs:enumeration value="DENY" />
+        </xs:restriction>
+    </xs:simpleType>
+
+</xs:schema>