Pyk: KCFG stores aliases for nodes

pyk/src/pyk/kcfg.py

@@ -31,7 +31,7 @@
     substToMlPred,
 )
 from .ktool import KPrint
-from .utils import add_indent, compare_short_hashes, shorten_hashes
+from .utils import add_indent, compare_short_hashes, shorten_hash
 
 
 class KCFG(Container[Union['KCFG.Node', 'KCFG.Edge', 'KCFG.Cover']]):
@@ -117,6 +117,7 @@ def pretty_print(self, kprint: KPrint) -> List[str]:
     _target: Set[str]
     _expanded: Set[str]
     _verified: Set[Tuple[str, str]]
+    _aliases: Dict[str, str]
     _lock: RLock
 
     def __init__(self):
@@ -127,6 +128,7 @@ def __init__(self):
         self._target = set()
         self._expanded = set()
         self._verified = set()
+        self._aliases = dict()
         self._lock = RLock()
 
     def __contains__(self, item: object) -> bool:
@@ -197,6 +199,7 @@ def to_dict(self) -> Dict[str, Any]:
         target = sorted(self._target)
         expanded = sorted(self._expanded)
         verified = [{"source": source_id, "target": target_id} for source_id, target_id in sorted(self._verified)]
+        aliases = self._aliases
 
         res = {
             'nodes': nodes,
@@ -206,6 +209,7 @@ def to_dict(self) -> Dict[str, Any]:
             'target': target,
             'expanded': expanded,
             'verified': verified,
+            'aliases': aliases
         }
         return {k: v for k, v in res.items() if v}
 
@@ -253,8 +257,15 @@ def resolve(node_id: str) -> str:
         for verified_ids in dct.get('verified') or []:
             cfg.add_verified(resolve(verified_ids['source']), resolve(verified_ids['target']))
 
+        for alias, id in dct.get('aliases', {}).items():
+            cfg.add_alias(alias=alias, node_id=resolve(id))
+
         return cfg
 
+    def aliases(self, node_id: str) -> List[str]:
+        node_id = self._resolve(node_id)
+        return [alias for alias, value in self._aliases.items() if node_id == value]
+
     def to_json(self) -> str:
         return json.dumps(self.to_dict(), sort_keys=True)
 
@@ -263,9 +274,9 @@ def from_json(s: str) -> 'KCFG':
         return KCFG.from_dict(json.loads(s))
 
     def node_short_info(self, node: Node) -> str:
-        attrs = self.node_attrs(node.id)
+        attrs = self.node_attrs(node.id) + ['@' + alias for alias in self.aliases(node.id)]
         attr_string = ' (' + ', '.join(attrs) + ')' if attrs else ''
-        return shorten_hashes(node.id) + attr_string
+        return shorten_hash(node.id) + attr_string
 
     def pretty_print(self, kprint: KPrint) -> List[str]:
 
@@ -311,7 +322,13 @@ def _print_subgraph(indent: str, curr_node: KCFG.Node, prior_on_trace: List[KCFG
                     ret.append(new_indent.rstrip())
             return ret
 
-        return [self.node_short_info(self.init[0])] + _print_subgraph('', self.init[0], [self.init[0]])
+        ret = []
+        init = self.init
+        while init:
+            ret.append(self.node_short_info(init[0]))
+            ret.extend(_print_subgraph('', init[0], [init[0]]))
+            init = [node for node in self.nodes if node not in processed_nodes]
+        return ret
 
     def to_dot(self, kprint: KPrint) -> str:
         def _short_label(label):
@@ -320,9 +337,8 @@ def _short_label(label):
         graph = Digraph()
 
         for node in self.nodes:
-            nodeAttrs = self.node_attrs(node.id)
-            classAttrs = ' '.join(nodeAttrs)
-            label = shorten_hashes(node.id) + (classAttrs and ' ' + classAttrs)
+            label = self.node_short_info(node)
+            classAttrs = ' '.join(self.node_attrs(node.id))
             attrs = {'class': classAttrs} if classAttrs else {}
             graph.node(name=node.id, label=label, **attrs)
 
@@ -351,21 +367,25 @@ def _short_label(label):
 
         return graph.source
 
-    def _resolve_all(self, short_id: str) -> List[str]:
-        return [node_id for node_id in self._nodes if compare_short_hashes(short_id, node_id)]
+    def _resolve_all(self, id_like: str) -> List[str]:
+        if id_like.startswith('@'):
+            if id_like[1:] in self._aliases:
+                return [self._aliases[id_like[1:]]]
+            raise ValueError(f'Unknown alias: {id_like}')
+        return [node_id for node_id in self._nodes if compare_short_hashes(id_like, node_id)]
 
-    def _resolve_or_none(self, short_id: str) -> Optional[str]:
-        matches = self._resolve_all(short_id)
+    def _resolve_or_none(self, id_like: str) -> Optional[str]:
+        matches = self._resolve_all(id_like)
         if not matches:
             return None
         if len(matches) > 1:
-            raise ValueError(f'Multiple nodes for pattern: {short_id} (matches e.g. {matches[0]} and {matches[1]})')
+            raise ValueError(f'Multiple nodes for pattern: {id_like} (matches e.g. {matches[0]} and {matches[1]})')
         return matches[0]
 
-    def _resolve(self, short_id: str) -> str:
-        match = self._resolve_or_none(short_id)
+    def _resolve(self, id_like: str) -> str:
+        match = self._resolve_or_none(id_like)
         if not match:
-            raise ValueError(f'Unknown node: {short_id}')
+            raise ValueError(f'Unknown node: {id_like}')
         return match
 
     def node(self, node_id: str) -> Node:
@@ -419,6 +439,9 @@ def remove_node(self, node_id: str) -> None:
         self._expanded.discard(node_id)
         self._verified = set((source_id, target_id) for source_id, target_id in self._verified if source_id != node_id and target_id != node_id)
 
+        for alias in [alias for alias, id in self._aliases.items() if id == node_id]:
+            del self._aliases[alias]
+
     def edge(self, source_id: str, target_id: str) -> Optional[Edge]:
         source_id = self._resolve(source_id)
         target_id = self._resolve(target_id)
@@ -537,6 +560,14 @@ def add_verified(self, source_id: str, target_id: str) -> None:
         target_id = self._resolve(target_id)
         self._verified.add((source_id, target_id))
 
+    def add_alias(self, alias: str, node_id: str) -> None:
+        if '@' in alias:
+            raise ValueError('Alias may not contain "@"')
+        if alias in self._aliases:
+            raise ValueError(f'Duplicate alias "{alias}"')
+        node_id = self._resolve(node_id)
+        self._aliases[alias] = node_id
+
     def remove_init(self, node_id: str) -> None:
         node_id = self._resolve(node_id)
         if node_id not in self._init:
@@ -562,6 +593,11 @@ def remove_verified(self, source_id: str, target_id: str) -> None:
             raise ValueError(f'Edge is not verified: {(source_id, target_id)}')
         self._verified.remove((source_id, target_id))
 
+    def remove_alias(self, alias: str) -> None:
+        if alias not in self._aliases:
+            raise ValueError(f'Alias does not exist: {alias}')
+        self._aliases.pop(alias)
+
     def discard_init(self, node_id: str) -> None:
         node_id = self._resolve(node_id)
         self._init.discard(node_id)

pyk/src/pyk/tests/test_kcfg.py

@@ -6,17 +6,13 @@
 from ..kcfg import KCFG
 from ..ktool import KPrint
 from ..prelude import token
-from ..utils import shorten_hashes
+from ..utils import shorten_hash
 
 
 def nid(i: int) -> str:
     return node(i).id
 
 
-def short_id(i: int) -> str:
-    return shorten_hashes(nid(i))
-
-
 # over 10 is variables
 def term(i: int) -> CTerm:
     inside: KInner = token(i)
@@ -263,11 +259,64 @@ def test_paths_between(self):
             },
         )
 
+    def test_resolve(self):
+        # Given
+        d = {
+            'nodes': node_dicts(4),
+            'edges': edge_dicts((0, 1), (0, 2), (1, 2), (1, 3), (2, 3), (3, 0)),
+        }
+        cfg = KCFG.from_dict(d)
+
+        self.assertEqual(node(1), cfg.node('d33..d8'))
+        self.assertEqual(node(1), cfg.node(node(1).id))
+
+        # Matches no nodes
+        with self.assertRaises(ValueError, msg='Unknown node: deadbeef..d8'):
+            self.assertEqual(node(1), cfg.node('deadbeef..d8'))
+
+        # Matches all nodes
+        with self.assertRaisesRegex(ValueError, 'Multiple nodes for pattern: ...'):
+            cfg.node('..')
+
+        # Matches node(0) and node(2)
+        with self.assertRaisesRegex(ValueError, 'Multiple nodes for pattern: ...'):
+            cfg.node('3..')
+
+    def test_aliases(self):
+        # Given
+        d = {
+            'nodes': node_dicts(2),
+            'edges': edge_dicts((0, 1)),
+            'aliases': {'foo': nid(1)}
+        }
+
+        cfg = KCFG.from_dict(d)
+        self.assertEqual(cfg.node('@foo'), node(1))
+
+        cfg.add_alias('bar', node(0).id)
+        cfg.add_alias('bar2', node(0).id)
+        self.assertEqual(cfg.node('@bar'), node(0))
+        self.assertEqual(cfg.node('@bar2'), node(0))
+        cfg.remove_alias('bar')
+        with self.assertRaises(ValueError, msg='Unknown alias: @bar'):
+            cfg.node('@bar')
+
+        with self.assertRaises(ValueError, msg='Duplicate alias "bar2"'):
+            cfg.add_alias('bar2', node(1).id)
+        with self.assertRaises(ValueError, msg='Alias may not contain "@"'):
+            cfg.add_alias('@buzz', node(1).id)
+        with self.assertRaises(ValueError, msg=f'Unknown node: {nid(3)}'):
+            cfg.add_alias('buzz', node(3).id)
+
+        cfg.remove_node(nid(1))
+        cfg.create_node(term(1))
+
     def test_pretty_print(self):
         d = {
             'init': [nid(0)],
             'target': [nid(6)],
             'nodes': node_dicts(12),
+            'aliases': {'foo': nid(3), 'bar': nid(3)},
                                                              # Each of the branching edges have given depth=0 # noqa: E131
             'edges': edge_dicts((0, 1), (1, 2, 5), (2, 3),   # Initial Linear segment
                                 (3, 4, 0), (4, 5), (5, 2),   # Loops back
@@ -276,46 +325,49 @@ def test_pretty_print(self):
                                 (3, 7, 0), (7, 6),           # Go to previous terminal node not as loop
                                 (3, 11, 0), (11, 8)          # Covered
                                 ),
-            'covers': cover_dicts((8, 11))                   # Loops back
+            'covers': cover_dicts((8, 11)),                  # Loops back
+            'expanded': [nid(i) for i in [0, 1, 2, 3, 4, 5, 7, 11]],
         }
         cfg = KCFG.from_dict(d)
 
-        print(set(map(lambda node: node.id, cfg.reachable_nodes(nid(5), reverse=True, traverse_covers=True))))
+        def _short_hash(i) -> str:
+            return shorten_hash(nid(i))
 
-        # TODO: Why are all nodes (besides the target) frontiers?
-        # TODO: Add a cover
         self.maxDiff = None
         actual = '\n'.join(cfg.pretty_print(mock_kprint())) + '\n'
         self.assertMultiLineEqual(actual,
-                                  f"{short_id(0)} (init, frontier)\n"
+                                  f"{_short_hash(0)} (init, expanded)\n"
                                   f"│  (1 step)\n"
-                                  f"├  {short_id(1)} (frontier)\n"
+                                  f"├  {_short_hash(1)} (expanded)\n"
                                   f"│  (5 steps)\n"
-                                  f"├  {short_id(2)} (frontier)\n"
+                                  f"├  {_short_hash(2)} (expanded)\n"
                                   f"│  (1 step)\n"
-                                  f"├  {short_id(3)} (frontier)\n"
-                                  f"┢━ {short_id(4)} (frontier)\n"
+                                  f"├  {_short_hash(3)} (expanded, @foo, @bar)\n"
+                                  f"┢━ {_short_hash(4)} (expanded)\n"
                                   f"┃   │  (1 step)\n"
-                                  f"┃   ├  {short_id(5)} (frontier)\n"
+                                  f"┃   ├  {_short_hash(5)} (expanded)\n"
                                   f"┃   │  (1 step)\n"
-                                  f"┃   ├  {short_id(2)} (frontier)\n"
+                                  f"┃   ├  {_short_hash(2)} (expanded)\n"
                                   f"┃   ┊ (looped back)\n"
                                   f"┃\n"
-                                  f"┣━ {short_id(5)} (frontier)\n"
+                                  f"┣━ {_short_hash(5)} (expanded)\n"
                                   f"┃   ┊ (continues as previously)\n"
                                   f"┃\n"
-                                  f"┣━ {short_id(6)} (target, leaf)\n"
+                                  f"┣━ {_short_hash(6)} (target, leaf)\n"
                                   f"┃\n"
-                                  f"┣━ {short_id(7)} (frontier)\n"
+                                  f"┣━ {_short_hash(7)} (expanded)\n"
                                   f"┃   │  (1 step)\n"
-                                  f"┃   └  {short_id(6)} (target, leaf)\n"
+                                  f"┃   └  {_short_hash(6)} (target, leaf)\n"
                                   f"┃\n"
-                                  f"┗━ {short_id(11)} (frontier)\n"
+                                  f"┗━ {_short_hash(11)} (expanded)\n"
                                   f"    │  (1 step)\n"
-                                  f"    ├  {short_id(8)} (leaf)\n"
+                                  f"    ├  {_short_hash(8)} (leaf)\n"
                                   f"    │  constraint: KApply(label=KLabel(name='#Top', params=(KSort(name='GeneratedTopCell'),)), args=())\n"
                                   f"    │  subst:\n"
                                   f"    │    KApply(label=KLabel(name='#Equals', params=(KSort(name='K'), KSort(name='K'))), args=(KVariable(name='V11'), KToken(token='8', sort=KSort(name='Int'))))\n"
-                                  f"    ├  {short_id(11)} (frontier)\n"
+                                  f"    ├  {_short_hash(11)} (expanded)\n"
                                   f"    ┊ (looped back)\n\n"
+                                  f""
+                                  f"{_short_hash(9)} (frontier, leaf)\n"
+                                  f"{_short_hash(10)} (frontier, leaf)\n"
                                   )

pyk/src/pyk/tests/test_utils.py

@@ -0,0 +1,26 @@
+from unittest import TestCase
+
+from ..utils import deconstruct_short_hash
+
+
+class ShortHashTest(TestCase):
+    def test_parse_short_id(self):
+        # prefix with / without the dots, suffix, both prefix and suffix, full hash, etc).
+        self.assertEqual(deconstruct_short_hash('..abcdef'), ('', 'abcdef'))
+        self.assertEqual(deconstruct_short_hash('abcdef..'), ('abcdef', ''))
+        self.assertEqual(deconstruct_short_hash('abcdef..12345'), ('abcdef', '12345'))
+
+        full_hash = '0001000200030004000500060007000800010002000300040005000600070008'
+        self.assertEqual(deconstruct_short_hash(full_hash), (full_hash, full_hash))
+
+        # Bad short hash: Has digits between dots
+        with self.assertRaises(ValueError, msg='Bad short hash: 3.c62e73544...'):
+            deconstruct_short_hash('3.c62e73544...')
+
+        # Bad short hash: Has non hex digits
+        with self.assertRaises(ValueError, msg='Bad short hash: 3...XXX'):
+            deconstruct_short_hash('3...XXX')
+
+        # Bad short hash: Has more than two dots
+        with self.assertRaises(ValueError, msg='Bad short hash: 3.....adf'):
+            deconstruct_short_hash('3...adf')

pyk/src/pyk/utils.py

@@ -117,6 +117,10 @@ def add_indent(indent: str, lines: List[str]) -> List[str]:
     return list(map(lambda line: indent + line, lines))
 
 
+def is_hexstring(x: str) -> bool:
+    return all(c in string.hexdigits for c in x)
+
+
 # Hashes
 
 def hash_str(x: Any) -> str:
@@ -129,7 +133,7 @@ def is_hash(x: Any) -> bool:
     # NB! currently only sha256 in hexdec form is detected
     # 2b9e b7c5 441e 9f7e 97f9 a4e5 fc04 a0f7 9f62 c8e9 605a ad1e 02db e8de 3c21 0422
     # 1    2    3    4    5    6    7    8    9    10   11   12   13   14   15   16
-    return type(x) is str and len(x) == 64 and all(c in string.hexdigits for c in x)
+    return type(x) is str and len(x) == 64 and is_hexstring(x)
 
 
 def shorten_hash(h: str, leftChars=6, rightChars=6) -> str:
@@ -159,7 +163,17 @@ def shorten_hashes(value: Any, leftChars=6, rightChars=6) -> Any:
     return result
 
 
+def deconstruct_short_hash(h: str) -> Tuple[str, str]:
+    x = h.lower()
+    if is_hash(x):
+        return (x, x)
+    (l, sep, r) = x.partition('..')
+    if sep == '..' and is_hexstring(l) and is_hexstring(r):
+        return (l, r)
+    raise ValueError(f'Bad short hash: {h}')
+
+
 def compare_short_hashes(lhs: str, rhs: str):
-    left, right = lhs.split('.'), rhs.split('.')
-    (l0, l1, r0, r1) = (left[0].upper(), left[-1].upper(), right[0].upper(), right[-1].upper())
+    (l0, l1) = deconstruct_short_hash(lhs)
+    (r0, r1) = deconstruct_short_hash(rhs)
     return (l0.startswith(r0) or r0.startswith(l0)) and (l1.endswith(r1) or r1.endswith(l1))