Merge remote-tracking branch 'bjorn/main' into dev

setup.cfg

@@ -94,7 +94,7 @@ use_parentheses = True
 
 [tool:pytest]
 log_cli = False
-log_cli_level = INFO
+log_cli_level = DEBUG
 junit_family = xunit2
 addopts =
     -vv

src/pymbe/graph/calc_lpg.py

@@ -35,8 +35,8 @@ def solve_graph(self, lpg: SysML2LabeledPropertyGraph):
         elements = lpg.model.elements
         for step in self.calculation_list:
             src, tgt, type_ = step
-            src_data = lpg.nodes[src]
-            src_metatype = src_data["@type"]
+            src_data = lpg.model.elements[src]
+            src_metatype = src_data.get("@type")
             source_instances = self.instance_dict.get(src)
             target_instances = self.instance_dict.get(tgt)
             if type_ in ("Assignment", "ValueBinding"):
@@ -73,7 +73,7 @@ def solve_graph(self, lpg: SysML2LabeledPropertyGraph):
                             self.calculation_log.append(f"[FRE]... result includes {target_inst}")
 
                 elif src_metatype == "OperatorExpression":
-                    if src_data["operator"] == "collect":
+                    if src_data.operator == "collect":
                         collect_sub_expressions = []
                         collect_sub_expression_results = []
                         collect_sub_inputs = []
@@ -102,7 +102,7 @@ def solve_graph(self, lpg: SysML2LabeledPropertyGraph):
                                 f"input {input_point}, and path input {path_point}"
                             )
 
-                            if not path_point or path_point.value is None:
+                            if path_point is None or path_point.value is None:
                                 print("Path point value is empty! {path_point}")
                             else:
                                 evaluate_and_apply_collect(
@@ -142,16 +142,16 @@ def solve_graph(self, lpg: SysML2LabeledPropertyGraph):
                             )
 
                 elif src_metatype == "InvocationExpression":
-                    invoke_type = lpg.nodes[src_data["type"][0]["@id"]]
-                    if invoke_type["name"] == "sum":
+                    invoke_type = src_data.type[0]
+                    if invoke_type.name == "sum":
                         sum_inputs = []
 
-                        for member in src_data["input"]:
-                            sum_inputs.append(lpg.nodes[member["@id"]])
+                        for member in src_data.input:
+                            sum_inputs.append(member)
 
                         for index, m0_operator_seq in enumerate(source_instances):
                             input_point = None
-                            input_instances = self.instance_dict[sum_inputs[0]["@id"]]
+                            input_instances = self.instance_dict[sum_inputs[0].get("@id")]
                             for input_inst in input_instances:
                                 if input_inst[0] == m0_operator_seq[0]:
                                     input_point = input_inst[-1]
@@ -165,13 +165,13 @@ def solve_graph(self, lpg: SysML2LabeledPropertyGraph):
                     collect_sub_expressions = []
                     collect_sub_expression_results = []
                     collect_sub_inputs = []
-                    for member in src_data["member"]:
-                        if lpg.nodes[member["@id"]]["@type"] in COLLECTABLE_EXPRESSIONS:
-                            collect_sub_expressions.append(lpg.nodes[member["@id"]])
-                            collect_sub_expression_results.append(lpg.nodes[lpg.nodes[member["@id"]]["result"]["@id"]])
+                    for member in src_data.member:
+                        if member.get("@type") in COLLECTABLE_EXPRESSIONS:
+                            collect_sub_expressions.append(member)
+                            collect_sub_expression_results.append(member.result)
 
-                    for member in src_data["input"]:
-                        collect_sub_inputs.append(lpg.nodes[member["@id"]])
+                    for input in src_data.input:
+                        collect_sub_inputs.append(input)
 
                     # Base sequence is there to filter as appropriate to the expression scope
 
@@ -183,20 +183,20 @@ def solve_graph(self, lpg: SysML2LabeledPropertyGraph):
 
                     for index, m0_operator_seq in enumerate(source_instances):
                         input_point = None
-                        input_instances = self.instance_dict[collect_sub_inputs[0]["@id"]]
+                        input_instances = self.instance_dict[collect_sub_inputs[0]._id]
                         for input_inst in input_instances:
                             if input_inst[0] == m0_operator_seq[0]:
                                 input_point = input_inst[-1]
                         path_point = None
-                        input_instances = self.instance_dict[collect_sub_expression_results[1]["@id"]]
+                        input_instances = self.instance_dict[collect_sub_expression_results[1]._id]
                         for input_inst in input_instances:
                             if input_inst[0] == m0_operator_seq[0]:
                                 path_point = input_inst[-1]
 
-                        self.calculation_log.append(f"[PSE] Calling collect with base = {m0_operator_seq[0]}" +
-                            f", collection input {input_point}, and path input {path_point}")
+                        self.calculation_log.append(f"[PSE] Calling collect with base = {m0_operator_seq[0]}\n" +
+                            f", collection input {input_point}\n, and path input {path_point}")
 
-                        if path_point.value is None:
+                        if path_point is None or path_point.value is None:
                             print("Path point value is empty! " + str(path_point))
                         else:
                             evaluate_and_apply_dot(

src/pymbe/interpretation/calc_dependencies.py

@@ -14,7 +14,7 @@ def generate_execution_order(
     :return:
     """
 
-    all_elements = lpg.nodes
+    all_elements = lpg.model.elements
     eig = lpg.get_projection("Expression Inferred")
 
     execution_pairs = []
@@ -25,7 +25,7 @@ def generate_execution_order(
     roots = [node for node in eig.nodes if eig.in_degree(node) == 0]
 
     for root in roots:
-        context = all_elements[root]['featuringType'][0]['@id']
+        context = all_elements[root].featuringType[0].get('@id')
         execution_contexts[context] = []
 
         calc_order = list(nx.edge_bfs(eig, root))
@@ -37,15 +37,17 @@ def generate_execution_order(
             node = edg[0]
             kind = ''
 
-            if all_elements[node_child]['@type'] == 'Feature' and all_elements[node]['@type'] == 'Feature':
+            if all_elements[node_child].get("@type") == 'Feature' and all_elements[node].get("@type") == 'Feature':
                 kind = 'Assignment'
-            elif all_elements[node_child]['@type'] == 'AttributeUsage' and all_elements[node]['@type'] == 'AttributeUsage':
+            elif all_elements[node_child].get("@type") == 'AttributeUsage' and all_elements[node].get("@type") == 'AttributeUsage':
                 relevant_edge_types = [edg[2] for edg in eig.edges if edg[0] == node and edg[1] == node_child]
                 if "Redefinition^-1" in relevant_edge_types:
                     kind = 'Redefinition'
                 else:
                     kind = 'Assignment'
-            elif all_elements[node_child]['@type'] == 'Feature' and all_elements[node]['@type'] == 'AttributeUsage':
+            elif all_elements[node].get("@type") == 'FeatureReferenceExpression':
+                kind = 'SelectionQuery'
+            elif all_elements[node_child].get("@type") == 'Feature' and all_elements[node].get("@type") == 'AttributeUsage':
                 kind = 'ValueBinding'
             elif (node_child, node, 'ReturnParameterMembership') in lpg.edges_by_type['ReturnParameterMembership']:
                 kind = 'Output'

src/pymbe/interpretation/interpretation.py

@@ -1,3 +1,89 @@
+from dataclasses import dataclass
+from ..model import Element, Model
+from ..label import get_label
+
+class InterpretationSequence(tuple):
+    """
+    A class to represent a single sequence within a model interpretation. Objects of this class should support
+    the drawing of interpretation diagrams and be the eventual target of validity checkers.
+    """
+
+    def __init__(self, elements: list):
+        self.sequence = tuple(elements)
+        self.owning_entry = None
+    def get_line_ends(self):
+        # placeholder for using M1 reference to figure out what the right ends for the connector line are
+        if self.owning_entry.draw_kind == "Line":
+            line_ends = self.owning_entry.base.connectorEnd
+            line_source = None
+            line_target = None
+            for index, line_end in enumerate(line_ends):
+                for entry in self.owning_entry.master_list:
+                    if entry.key == line_end._id:
+                        end_interpretation = entry.value
+                        for seq in end_interpretation:
+                            for item in seq:
+                                if item == self[-1]:
+                                    if index == 0:
+                                        line_source = seq
+                                    if index == 1:
+                                        line_target = seq
+
+            return [line_source, line_target]
+        else:
+            return []
+    def get_nesting_list(self):
+        # placeholder for using M1 reference to figure out what the path of parent shapes are
+        pass
+
+DEF_BOX_KINDS = ("PartDefinition", "PortDefinition")
+USE_BOX_KINDS = ("PartUsage")
+LINE_KINDS = ("ConnectionUsage")
+PORT_BOX_KINDS = ("PortUsage")
+
+
+class InterpretationSet(set):
+    """
+    A class to represent the set of sequences of an interpretation of a single M1 element
+    """
+    def __repr__(self):
+        if len(self) > 10:
+            excerpt = list(self)[0:10]
+            excerpt.append('...')
+            return str(excerpt)
+        else:
+            return super(InterpretationSet, self).__repr__()
+
+
+class InterpretationDictionaryEntry:
+    """
+    A class to represent a key value pair for a master interpretation dictionary, which points from
+    M1 user model elements to a set of sequences of atoms that are the interpretation
+    """
+
+    def __init__(self, m1_base: Element, interprets: set, owner: list = []):
+        self.key = m1_base._id
+        self.value = InterpretationSet()
+        self.base = m1_base
+        self.master_list = owner
+        for item in interprets:
+            self.value.add(item)
+            # link the sequence owning entry back here to leave a breadcrumb for plotting, checking, etc.
+            item.owning_entry = self
+        # build hinting for diagram
+        if m1_base.get("@type") in DEF_BOX_KINDS:
+            self.draw_kind = "Box"
+        elif m1_base.get("@type") in USE_BOX_KINDS:
+            self.draw_kind = "Nested Box"
+        elif m1_base.get("@type") in LINE_KINDS:
+            self.draw_kind = "Line"
+        elif m1_base.get("@type") in PORT_BOX_KINDS:
+            self.draw_kind = "Port"
+
+
+    def __repr__(self):
+        return f'Entry: <{get_label(self.base)}, {self.value}>'
+
 class Instance:
     """
     A class to represent instances of real things in the M0 universe interpreted from the model.
@@ -74,3 +160,22 @@ def shorten_name(name: str, shorten_pre_bake: dict = None) -> str:
                 next_space = name.find(" ", next_space + 1)
             return short_name
     return name
+
+def repack_instance_dictionaries(instance_dict: dict, mdl: Model):
+    """
+    Temporary method to repack the instance dictionaries into objects to be sure this is how we want things to work
+    :param instance_dict: Completed instance dictionary
+    :return: An object version of the instance dictionary
+    """
+
+    instance_list = []
+
+    for key, sequence_set in instance_dict.items():
+        new_set = set()
+        for seq in sequence_set:
+            new_seq = InterpretationSequence(seq)
+            new_set.add(new_seq)
+        entry = InterpretationDictionaryEntry(mdl.elements[key], new_set, instance_list)
+        instance_list.append(entry)
+
+    return instance_list

src/pymbe/interpretation/set_builders.py

@@ -161,7 +161,9 @@ def extend_sequences_with_new_expr(
             expr_string,
             expr,
         )
-        new_sequences.append([seq, new_holder])
+        new_seq = [step for step in seq]
+        new_seq.append(new_holder)
+        new_sequences.append(new_seq)
 
     return new_sequences
 

tests/graph/test_graph_load.py

@@ -8,8 +8,8 @@
 
 
 def test_graph_load(kerbal_lpg):
-    assert len(kerbal_lpg.nodes) == 156
-    assert len(kerbal_lpg.edges) == 389 - 156 # nodes + edges should be all elements from the client
+    assert len(kerbal_lpg.nodes) == 152
+    assert len(kerbal_lpg.edges) == 380 - 152 # nodes + edges should be all elements from the client
 
 
 def test_graph_projection_part_def_node_filter(kerbal_lpg, kerbal_ids_by_type):

tests/interpretation/test_calculation_ordering.py

@@ -20,11 +20,12 @@ def test_kerbal_calc_order1(kerbal_lpg, kerbal_random_stage_5_complete, kerbal_s
 
     # the execution order will be ordered for examination
 
-    sum_1_result = qualified_name_to_id[f'{ROCKET_BUILDING}Liquid Stage::Full Mass: Real::+ (sum (collect (FRE.engines)),'
-                                     f' sum (collect (FRE.tanks))) => $result::sum (collect (FRE.tanks)) => $result::'
-                                     f'sum (collect (FRE.tanks)) <<Feature>>']  # Result of the sum Expression on Full Mass
-    top_plus = qualified_name_to_id[f'{ROCKET_BUILDING}Liquid Stage::Full Mass: Real::+ (sum (collect (FRE.engines)), ' +
-                                 'sum (collect (FRE.tanks))) => $result <<OperatorExpression>>']
+    sum_1_result = qualified_name_to_id[f'{ROCKET_BUILDING}Liquid Stage::Full Mass: Real::+ (sum (engines.Mass' +
+                                 f' (FRE.engines)), sum (tanks.Full Mass (FRE.tanks))) => $result::sum' +
+                                 f' (tanks.Full Mass (FRE.tanks)) => $result::tanks.Full Mass (FRE.tanks) <<Feature>>']
+    top_plus = qualified_name_to_id[f'{ROCKET_BUILDING}Liquid Stage::Full Mass: Real::+ (sum (engines.Mass '
+                                            f'(FRE.engines)), sum (tanks.Full Mass (FRE.tanks))) => '
+                                            f'$result <<OperatorExpression>>']
 
     sum_1_in_dcg = None
     plus_in_dcg = None

tests/interpretation/test_graph_visit_orderings.py

@@ -112,13 +112,14 @@ def test_feature_sequence_templates4(simple_parts_lpg, simple_parts_stable_names
 def test_expression_sequence_templates(kerbal_lpg, kerbal_stable_names):
     *_, qualified_name_to_id = kerbal_stable_names
 
-    top_plus = qualified_name_to_id[f'{ROCKET_BUILDING}Liquid Stage::Full Mass: Real::+ (sum (collect (FRE.engines)), '
-                                 f'sum (collect (FRE.tanks))) => $result <<OperatorExpression>>']
+    top_plus = qualified_name_to_id[f'{ROCKET_BUILDING}Liquid Stage::Full Mass: Real::+ (sum (engines.Mass'
+                                    f' (FRE.engines)), sum (tanks.Full Mass (FRE.tanks))) => '
+                                    f'$result <<OperatorExpression>>']
 
-    fre_1_result = qualified_name_to_id[f'{ROCKET_BUILDING}Liquid Stage::Full Mass: Real::+ (sum (collect'
-                                        f' (FRE.engines)), sum (collect (FRE.tanks))) => $result::sum (collect '
-                                        f'(FRE.tanks)) => $result::collect (FRE.tanks) => $result::FRE.Full Mass (p) =>'
-                                        f' $result::FRE.Full Mass::FRE.Full Mass <<Feature>>']
+    fre_1_result = qualified_name_to_id[f'{ROCKET_BUILDING}Liquid Stage::Full Mass: Real::+ (sum (engines.Mass ' +
+                                 f'(FRE.engines)), sum (tanks.Full Mass (FRE.tanks))) => $result::sum ' +
+                                 f'(tanks.Full Mass (FRE.tanks)) => $result::tanks.Full Mass (FRE.tanks) => ' +
+                                 f'$result::FRE.Full Mass::FRE.Full Mass <<Feature>>']
 
     expr_sequences = build_expression_sequence_templates(lpg=kerbal_lpg)
     top_plus_paths = 0
@@ -141,8 +142,8 @@ def test_expression_sequence_templates(kerbal_lpg, kerbal_stable_names):
         # each sequence should end with the result
         assert kerbal_lpg.nodes[seq[len(seq) - 1]]["@type"] == "Feature"
 
-    assert top_plus_paths == 17
+    assert top_plus_paths == 15
     assert direct_literals == 26
-    assert len(expr_sequences) == 43
+    assert len(expr_sequences) == 41
 
     assert fre_result_found