1#!/usr/bin/env python3 2# SPDX-License-Identifier: GPL-2.0-only 3# 4# Copyright (C) 2019-2022 Red Hat, Inc. Daniel Bristot de Oliveira <bristot@kernel.org> 5# 6# Automata class: parse an automaton in dot file digraph format into a python object 7# 8# For further information, see: 9# Documentation/trace/rv/deterministic_automata.rst 10 11import ntpath 12 13import lark 14 15class ParseTree: 16 # based on https://graphviz.org/doc/info/lang.html 17 # with the irrelevant stuffs (port and compass) removed 18 grammar = r''' 19 start: "strict"? ("graph" | "digraph") ID? "{" stmt_list "}" 20 21 stmt_list: (stmt ";"? stmt_list)? 22 23 stmt: node_stmt 24 | edge_stmt 25 | attr_stmt 26 | ID "=" ID 27 | subgraph 28 29 attr_stmt: attr_type attr_list 30 31 attr_type: "graph" -> graph 32 | "node" -> node 33 | "edge" -> edge 34 35 attr_list: "[" a_list? "]" attr_list? 36 37 a_list: ID "=" ID (";" | ",")? a_list? 38 39 edge_stmt: (node_id | subgraph) edgerhs attr_list? 40 41 edgerhs: edgeop (node_id | subgraph) edgerhs? 42 43 edgeop: "->" | "--" 44 45 node_stmt: node_id attr_list? 46 47 node_id: ID 48 49 subgraph: ("subgraph" ID?)? "{" stmt_list "}" 50 51 ID: CNAME 52 | /-?(\.[0-9]+|[0-9]+(\.[0-9]*))/ 53 | ESCAPED_STRING 54 55 %import common.CNAME 56 %import common.ESCAPED_STRING 57 %import common.WS 58 %ignore WS 59 ''' 60 61 @staticmethod 62 def parse_edge(tree: lark.Tree) -> tuple[str, str]: 63 # only support a simple node-to-node edge 64 nodes = [] 65 for node in tree.iter_subtrees_topdown(): 66 if node.data == "node_id": 67 nodes.append(node.children[0].strip('"')) 68 69 if len(nodes) != 2: 70 raise AutomataError("Only state-to-state transition is supported") 71 72 return tuple(nodes) 73 74 class ParseNodes(lark.visitors.Visitor): 75 def __init__(self, *args, **kwargs): 76 self.nodes = set() 77 super().__init__(*args, **kwargs) 78 79 def node_stmt(self, tree): 80 node_id = tree.children[0] 81 node = node_id.children[0].strip('"') 82 self.nodes.add(node) 83 84 class ParseEdges(lark.visitors.Visitor): 85 def __init__(self, *args, **kwargs): 86 self.edges = set() 87 super().__init__(*args, **kwargs) 88 89 def edge_stmt(self, tree): 90 edge = ParseTree.parse_edge(tree) 91 self.edges.add(edge) 92 93 class ParseAttributes(lark.visitors.Interpreter): 94 def __init__(self, *args, **kwargs): 95 ''' 96 Stacks of default attributes. [0] is the default 97 attributes for the outermost scope, while [-1] is the 98 default attributes for the current scope. 99 ''' 100 self.default_node_attrs = [{}] 101 self.default_edge_attrs = [{}] 102 103 self.node_attrs = {} 104 self.edge_attrs = {} 105 106 super().__init__(*args, **kwargs) 107 108 @staticmethod 109 def __get_attrs(stmt: lark.Tree) -> dict[str, str]: 110 attrs = {} 111 112 for node in stmt.iter_subtrees(): 113 if node.data == "a_list": 114 attrs[node.children[0]] = node.children[1].strip('"') 115 116 return attrs 117 118 119 def subgraph(self, tree): 120 # We are entering a new scope, inherit the default 121 # attributes of the outer scope 122 self.default_node_attrs.append(self.default_node_attrs[-1].copy()) 123 self.default_edge_attrs.append(self.default_edge_attrs[-1].copy()) 124 125 children = self.visit_children(tree) 126 127 # Exiting the scope 128 del self.default_node_attrs[-1] 129 del self.default_edge_attrs[-1] 130 131 return children 132 133 def node_stmt(self, tree): 134 node_id = tree.children[0] 135 node = node_id.children[0].strip('"') 136 137 attrs = self.default_node_attrs[-1].copy() 138 attrs |= self.__get_attrs(tree) 139 140 if attrs: 141 if node in self.node_attrs: 142 self.node_attrs[node] = attrs | self.node_attrs[node] 143 else: 144 self.node_attrs[node] = attrs 145 146 return self.visit_children(tree) 147 148 def edge_stmt(self, tree): 149 edge = ParseTree.parse_edge(tree) 150 151 attrs = self.default_edge_attrs[-1].copy() 152 attrs |= self.__get_attrs(tree) 153 154 if attrs: 155 if edge in self.edge_attrs: 156 self.edge_attrs[edge] = attrs | self.edge_attrs[edge] 157 else: 158 self.edge_attrs[edge] = attrs 159 160 return self.visit_children(tree) 161 162 def attr_stmt(self, tree): 163 attr_type = tree.children[0].data 164 attrs = self.__get_attrs(tree) 165 166 if attr_type == "node": 167 self.default_node_attrs[-1] |= attrs 168 elif attr_type == "edge": 169 self.default_edge_attrs[-1] |= attrs 170 else: 171 # graph attributes are irrelevant 172 pass 173 174 self.visit_children(tree) 175 176 def __init__(self, dot_file): 177 parser = lark.Lark(self.grammar, parser='lalr') 178 node_parser = self.ParseNodes() 179 edge_parser = self.ParseEdges() 180 attributes_parser = self.ParseAttributes() 181 182 try: 183 with open(dot_file, "r") as f: 184 tree = parser.parse(f.read()) 185 attributes_parser.visit(tree) 186 node_parser.visit(tree) 187 edge_parser.visit(tree) 188 except OSError as exc: 189 raise AutomataError(exc.strerror) from exc 190 except lark.exceptions.UnexpectedInput as exc: 191 raise AutomataError(str(exc)) 192 193 self.nodes = node_parser.nodes 194 self.edges = edge_parser.edges 195 self.node_attrs = attributes_parser.node_attrs 196 self.edge_attrs = attributes_parser.edge_attrs 197 198class ConstraintCondition: 199 def __init__(self, env: str, op: str, val: str, unit=None): 200 self.env = env 201 self.op = op 202 self.val = val 203 self.unit = unit 204 if unit is None: 205 # try to infer unit from constants or parameters 206 val_for_unit = val.lower().replace("()", "") 207 if val_for_unit.endswith("_ns"): 208 self.unit = "ns" 209 if val_for_unit.endswith("_jiffies"): 210 self.unit = "j" 211 212class ConstraintRule: 213 grammar = r''' 214 rule: condition (OP condition)* 215 216 OP: "&&" | "||" 217 218 condition: ENV CMP_OP VAL UNIT? 219 220 ENV: CNAME 221 222 CMP_OP: "==" | "!=" | "<=" | "<" | ">=" | ">" 223 224 VAL: /[0-9]+/ 225 | /[A-Z_]+\(\)/ 226 | /[A-Z_]+/ 227 | /[a-z_]+\(\)/ 228 | /[a-z_]+/ 229 230 UNIT: "ns" | "us" | "ms" | "s" | "j" 231 ''' 232 233 def __init__(self, c: ConstraintCondition): 234 ''' 235 A list of pairs of 236 - the condition (e.g. is_constr_dl == 1) 237 - the logical operator ("||" or "&&") combining this 238 condition with the next one if it exists, otherwise None 239 240 TODO: Perhaps use an abstract syntax tree instead, because 241 this representation cannot capture precedence 242 ''' 243 self.rules = [[c, None]] 244 245 def chain(self, op: str, c: ConstraintCondition): 246 self.rules[-1][1] = op 247 self.rules.append([c, None]) 248 249class ConstraintReset: 250 def __init__(self, env): 251 self.env = env 252 253class StateLabelParser: 254 grammar = r''' 255 label: CNAME ("\\n" condition)? 256 257 %import common.CNAME 258 %import common.WS 259 %ignore WS 260 ''' + ConstraintRule.grammar 261 262 parser = lark.Lark(grammar, parser='lalr', start="label") 263 264 def __init__(self, label: str): 265 try: 266 tree = self.parser.parse(label) 267 except lark.exceptions.UnexpectedInput as exc: 268 raise(AutomataError(f"Unrecognised state \"{label}\"\n{exc}")) 269 270 self.state = tree.children[0] 271 self.constraint = None 272 273 if len(tree.children) == 2: 274 self.constraint = ConstraintCondition(*tree.children[1].children) 275 if self.constraint.op not in ("<", "<="): 276 raise AutomataError("State constraints must be clock expirations like" 277 f" clk<N ({label})") 278 279class EventLabelParser: 280 grammar = r''' 281 events: event ("\\n" event)* 282 283 event: name (";" guard)? 284 285 guard: reset 286 | rule 287 | rule ";" reset 288 | reset ";" rule 289 290 name: CNAME 291 292 reset: "reset" "(" ENV ")" 293 294 %import common.CNAME 295 %import common.WS 296 %ignore WS 297 ''' + ConstraintRule.grammar 298 299 parser = lark.Lark(grammar, parser='lalr', start="events") 300 301 class GetEvents(lark.visitors.Transformer): 302 def guard(self, args): 303 reset = None 304 rule = None 305 for arg in args: 306 if arg.data == "reset": 307 reset = ConstraintReset(arg.children[0]) 308 elif arg.data == "rule": 309 conditions = arg.children 310 rule = ConstraintRule(conditions[0]) 311 for i in range(1, len(conditions), 2): 312 rule.chain(conditions[i], conditions[i + 1]) 313 return reset, rule 314 315 def OP(self, args): 316 return args 317 318 def condition(self, args): 319 return ConstraintCondition(*args) 320 321 def event(self, args): 322 assert(len(args) <= 2) 323 name = args[0] 324 rule, reset = None, None 325 if len(args) == 2: 326 reset, rule = args[1] 327 return name, reset, rule 328 329 def events(self, args): 330 return args 331 332 def name(self, args): 333 return args[0] 334 335 def __init__(self, label: str): 336 try: 337 tree = self.parser.parse(label) 338 self.events = self.GetEvents().transform(tree) 339 except lark.exceptions.UnexpectedInput as exc: 340 raise(AutomataError(f"Unrecognised event \"{label}\"\n{exc}")) 341 342class Transition: 343 def __init__(self, src: str, dst: str, event: str, 344 reset: ConstraintReset, rule: ConstraintRule): 345 self.src = src 346 self.dst = dst 347 self.event = event 348 self.rule = rule 349 self.reset = reset 350 351class State: 352 def __init__(self, name: str, inv: ConstraintCondition): 353 self.name = name 354 self.inv = inv 355 356class AutomataError(Exception): 357 """Exception raised for errors in automata parsing and validation. 358 359 Raised when DOT file processing fails due to invalid format, I/O errors, 360 or malformed automaton definitions. 361 """ 362 363class Automata: 364 """Automata class: Reads a dot file and parses it as an automaton. 365 366 It supports both deterministic and hybrid automata. 367 368 Attributes: 369 dot_file: A dot file with an state_automaton definition. 370 """ 371 372 invalid_state_str = "INVALID_STATE" 373 init_marker = "__init_" 374 375 def __init__(self, file_path, model_name=None): 376 self.__dot_path = file_path 377 self.name = model_name or self.__get_model_name() 378 self.__parse_tree = ParseTree(file_path) 379 self.transitions = self.__parse_transitions() 380 self.states, self.initial_state, self.final_states = self.__parse_states() 381 self.env_types = {} 382 self.env_stored = set() 383 self.constraint_vars = set() 384 self.self_loop_reset_events = set() 385 self.events, self.envs = self.__get_event_variables() 386 self.function = self.__create_matrix() 387 self.events_start, self.events_start_run = self.__store_init_events() 388 self.env_stored = sorted(self.env_stored) 389 self.constraint_vars = sorted(self.constraint_vars) 390 self.self_loop_reset_events = sorted(self.self_loop_reset_events) 391 392 def __get_model_name(self) -> str: 393 basename = ntpath.basename(self.__dot_path) 394 if not basename.endswith(".dot") and not basename.endswith(".gv"): 395 print("not a dot file") 396 raise AutomataError(f"not a dot file: {self.__dot_path}") 397 398 model_name = ntpath.splitext(basename)[0] 399 if not model_name: 400 raise AutomataError(f"not a dot file: {self.__dot_path}") 401 402 return model_name 403 404 def __parse_transitions(self): 405 transitions = [] 406 407 for edge in self.__parse_tree.edges: 408 attr = self.__parse_tree.edge_attrs.get(edge) 409 if not attr: 410 continue 411 412 label = attr.get("label") 413 414 src, dst = edge 415 416 parser = EventLabelParser(label) 417 for event, reset, rule in parser.events: 418 transitions.append(Transition(src, dst, event, reset, rule)) 419 420 transitions.sort(key=lambda t : (t.src, t.event)) 421 return transitions 422 423 def __parse_states(self): 424 initial_state = "" 425 states = [] 426 final_states = [] 427 428 for node in self.__parse_tree.nodes: 429 attr = self.__parse_tree.node_attrs[node] 430 label = attr.get("label") 431 432 if node.startswith(Automata.init_marker): 433 initial_state = node[len(Automata.init_marker):] 434 435 if not label: 436 continue 437 438 parser = StateLabelParser(label) 439 state = State(parser.state, parser.constraint) 440 441 states.append(state) 442 443 shape = attr.get("shape") 444 if shape in ("doublecircle", "ellipse"): 445 final_states.append(state) 446 447 448 initial_state = next((s for s in states if s.name == initial_state), None) 449 if not initial_state: 450 raise AutomataError("The automaton doesn't have an initial state") 451 452 if not final_states: 453 final_states.append(initial_state) 454 455 states.remove(initial_state) 456 states.sort(key=lambda s : s.name) 457 states.insert(0, initial_state) 458 return states, initial_state, final_states 459 460 def __get_event_variables(self) -> tuple[list[str], list[str]]: 461 events: list[str] = [] 462 envs: list[str] = [] 463 464 for transition in self.transitions: 465 events.append(transition.event) 466 467 if transition.reset: 468 envs.append(transition.reset.env) 469 self.env_stored.add(transition.reset.env) 470 if transition.rule: 471 for c, _ in transition.rule.rules: 472 envs.append(c.env) 473 self.__extract_env_var(c) 474 475 for state in self.states: 476 if state.inv: 477 envs.append(state.inv.env) 478 self.__extract_env_var(state.inv) 479 480 return sorted(set(events)), sorted(set(envs)) 481 482 def __extract_env_var(self, constraint: ConstraintCondition): 483 if constraint.unit: 484 self.env_types[constraint.env] = constraint.unit 485 if constraint.val[0].isalpha(): 486 self.constraint_vars.add(constraint.val) 487 488 def __create_matrix(self) -> list[list[str]]: 489 # transform the array into a dictionary 490 events = self.events 491 states = [s.name for s in self.states] 492 events_dict = {} 493 states_dict = {} 494 nr_event = 0 495 for event in events: 496 events_dict[event] = nr_event 497 nr_event += 1 498 499 nr_state = 0 500 for state in states: 501 states_dict[state] = nr_state 502 nr_state += 1 503 504 # declare the matrix.... 505 matrix = [[self.invalid_state_str for _ in range(nr_event)] for _ in range(nr_state)] 506 507 for transition in self.transitions: 508 src, dst = transition.src, transition.dst 509 event = transition.event 510 if src == dst and transition.reset: 511 # those events reset also on self loops 512 self.self_loop_reset_events.add(event) 513 matrix[states_dict[src]][events_dict[event]] = dst 514 515 return matrix 516 517 def __store_init_events(self) -> tuple[list[bool], list[bool]]: 518 events_start = [False] * len(self.events) 519 events_start_run = [False] * len(self.events) 520 for i in range(len(self.events)): 521 curr_event_will_init = 0 522 curr_event_from_init = False 523 curr_event_used = 0 524 for j in range(len(self.states)): 525 if self.function[j][i] != self.invalid_state_str: 526 curr_event_used += 1 527 if self.function[j][i] == self.initial_state.name: 528 curr_event_will_init += 1 529 if self.function[0][i] != self.invalid_state_str: 530 curr_event_from_init = True 531 # this event always leads to init 532 if curr_event_will_init and curr_event_used == curr_event_will_init: 533 events_start[i] = True 534 # this event is only called from init 535 if curr_event_from_init and curr_event_used == 1: 536 events_start_run[i] = True 537 return events_start, events_start_run 538 539 def is_start_event(self, event: str) -> bool: 540 return self.events_start[self.events.index(event)] 541 542 def is_start_run_event(self, event: str) -> bool: 543 # prefer handle_start_event if there 544 if any(self.events_start): 545 return False 546 return self.events_start_run[self.events.index(event)] 547 548 def is_hybrid_automata(self) -> bool: 549 return bool(self.envs) 550