1#!/usr/bin/env python3 2# SPDX-License-Identifier: GPL-2.0-only 3# 4# Implementation based on 5# Gerth, R., Peled, D., Vardi, M.Y., Wolper, P. (1996). 6# Simple On-the-fly Automatic Verification of Linear Temporal Logic. 7# https://doi.org/10.1007/978-0-387-34892-6_1 8# With extra optimizations 9 10from ply.lex import lex 11from ply.yacc import yacc 12from .automata import AutomataError 13 14# Grammar: 15# ltl ::= opd | ( ltl ) | ltl binop ltl | unop ltl 16# 17# Operands (opd): 18# true, false, user-defined names 19# 20# Unary Operators (unop): 21# always 22# eventually 23# next 24# not 25# 26# Binary Operators (binop): 27# until 28# and 29# or 30# imply 31# equivalent 32 33tokens = ( 34 'AND', 35 'OR', 36 'IMPLY', 37 'UNTIL', 38 'ALWAYS', 39 'EVENTUALLY', 40 'NEXT', 41 'VARIABLE', 42 'LITERAL', 43 'NOT', 44 'LPAREN', 45 'RPAREN', 46 'ASSIGN', 47) 48 49t_AND = r'and' 50t_OR = r'or' 51t_IMPLY = r'imply' 52t_UNTIL = r'until' 53t_ALWAYS = r'always' 54t_NEXT = r'next' 55t_EVENTUALLY = r'eventually' 56t_VARIABLE = r'[A-Z_0-9]+' 57t_LITERAL = r'true|false' 58t_NOT = r'not' 59t_LPAREN = r'\(' 60t_RPAREN = r'\)' 61t_ASSIGN = r'=' 62t_ignore_COMMENT = r'\#.*' 63t_ignore = ' \t\n' 64 65def t_error(t): 66 raise AutomataError(f"Illegal character '{t.value[0]}'") 67 68lexer = lex() 69 70class GraphNode: 71 uid = 0 72 73 def __init__(self, incoming: set['GraphNode'], new, old, _next): 74 self.init = False 75 self.outgoing = set() 76 self.labels = set() 77 self.incoming = incoming.copy() 78 self.new = new.copy() 79 self.old = old.copy() 80 self.next = _next.copy() 81 self.id = GraphNode.uid 82 GraphNode.uid += 1 83 84 def expand(self, node_set): 85 if not self.new: 86 for nd in node_set: 87 if nd.old == self.old and nd.next == self.next: 88 nd.incoming |= self.incoming 89 return node_set 90 91 new_current_node = GraphNode({self}, self.next, set(), set()) 92 return new_current_node.expand({self} | node_set) 93 n = self.new.pop() 94 return n.expand(self, node_set) 95 96 def __lt__(self, other): 97 return self.id < other.id 98 99class ASTNode: 100 uid = 1 101 102 def __init__(self, op): 103 self.op = op 104 self.id = ASTNode.uid 105 ASTNode.uid += 1 106 107 def __hash__(self): 108 return hash(self.op) 109 110 def __eq__(self, other): 111 return self is other 112 113 def __iter__(self): 114 yield self 115 yield from self.op 116 117 def negate(self): 118 self.op = self.op.negate() 119 return self 120 121 def expand(self, node, node_set): 122 return self.op.expand(self, node, node_set) 123 124 def __str__(self): 125 if isinstance(self.op, (Literal, Variable)): 126 return str(self.op) 127 return "val" + str(self.id) 128 129 def normalize(self): 130 # Get rid of: 131 # - ALWAYS 132 # - EVENTUALLY 133 # - IMPLY 134 # And move all the NOT to be inside 135 self.op = self.op.normalize() 136 return self 137 138class BinaryOp: 139 op_str = "not_supported" 140 141 def __init__(self, left: ASTNode, right: ASTNode): 142 self.left = left 143 self.right = right 144 145 def __hash__(self): 146 return hash((self.left, self.right)) 147 148 def __iter__(self): 149 yield from self.left 150 yield from self.right 151 152 def normalize(self): 153 raise NotImplementedError 154 155 def negate(self): 156 raise NotImplementedError 157 158 def _is_temporal(self): 159 raise NotImplementedError 160 161 def is_temporal(self): 162 if self.left.op.is_temporal(): 163 return True 164 if self.right.op.is_temporal(): 165 return True 166 return self._is_temporal() 167 168 @staticmethod 169 def expand(n: ASTNode, node: GraphNode, node_set) -> set[GraphNode]: 170 raise NotImplementedError 171 172class AndOp(BinaryOp): 173 op_str = '&&' 174 175 def normalize(self): 176 return self 177 178 def negate(self): 179 return OrOp(self.left.negate(), self.right.negate()) 180 181 def _is_temporal(self): 182 return False 183 184 @staticmethod 185 def expand(n: ASTNode, node: GraphNode, node_set) -> set[GraphNode]: 186 if not n.op.is_temporal(): 187 node.old.add(n) 188 return node.expand(node_set) 189 190 tmp = GraphNode(node.incoming, 191 node.new | ({n.op.left, n.op.right} - node.old), 192 node.old | {n}, 193 node.next) 194 return tmp.expand(node_set) 195 196class OrOp(BinaryOp): 197 op_str = '||' 198 199 def normalize(self): 200 return self 201 202 def negate(self): 203 return AndOp(self.left.negate(), self.right.negate()) 204 205 def _is_temporal(self): 206 return False 207 208 @staticmethod 209 def expand(n: ASTNode, node: GraphNode, node_set) -> set[GraphNode]: 210 if not n.op.is_temporal(): 211 node.old |= {n} 212 return node.expand(node_set) 213 214 node1 = GraphNode(node.incoming, 215 node.new | ({n.op.left} - node.old), 216 node.old | {n}, 217 node.next) 218 node2 = GraphNode(node.incoming, 219 node.new | ({n.op.right} - node.old), 220 node.old | {n}, 221 node.next) 222 return node2.expand(node1.expand(node_set)) 223 224class UntilOp(BinaryOp): 225 def normalize(self): 226 return self 227 228 def negate(self): 229 return VOp(self.left.negate(), self.right.negate()) 230 231 def _is_temporal(self): 232 return True 233 234 @staticmethod 235 def expand(n: ASTNode, node: GraphNode, node_set) -> set[GraphNode]: 236 node1 = GraphNode(node.incoming, 237 node.new | ({n.op.left} - node.old), 238 node.old | {n}, 239 node.next | {n}) 240 node2 = GraphNode(node.incoming, 241 node.new | ({n.op.right} - node.old), 242 node.old | {n}, 243 node.next) 244 return node2.expand(node1.expand(node_set)) 245 246class VOp(BinaryOp): 247 def normalize(self): 248 return self 249 250 def negate(self): 251 return UntilOp(self.left.negate(), self.right.negate()) 252 253 def _is_temporal(self): 254 return True 255 256 @staticmethod 257 def expand(n: ASTNode, node: GraphNode, node_set) -> set[GraphNode]: 258 node1 = GraphNode(node.incoming, 259 node.new | ({n.op.right} - node.old), 260 node.old | {n}, 261 node.next | {n}) 262 node2 = GraphNode(node.incoming, 263 node.new | ({n.op.left, n.op.right} - node.old), 264 node.old | {n}, 265 node.next) 266 return node2.expand(node1.expand(node_set)) 267 268class ImplyOp(BinaryOp): 269 def normalize(self): 270 # P -> Q === !P | Q 271 return OrOp(self.left.negate(), self.right) 272 273 def _is_temporal(self): 274 return False 275 276 def negate(self): 277 # !(P -> Q) === !(!P | Q) === P & !Q 278 return AndOp(self.left, self.right.negate()) 279 280class UnaryOp: 281 def __init__(self, child: ASTNode): 282 self.child = child 283 284 def __iter__(self): 285 yield from self.child 286 287 def __hash__(self): 288 return hash(self.child) 289 290 def normalize(self): 291 raise NotImplementedError 292 293 def _is_temporal(self): 294 raise NotImplementedError 295 296 def is_temporal(self): 297 if self.child.op.is_temporal(): 298 return True 299 return self._is_temporal() 300 301 def negate(self): 302 raise NotImplementedError 303 304class EventuallyOp(UnaryOp): 305 def __str__(self): 306 return "eventually " + str(self.child) 307 308 def normalize(self): 309 # <>F == true U F 310 return UntilOp(ASTNode(Literal(True)), self.child) 311 312 def _is_temporal(self): 313 return True 314 315 def negate(self): 316 # !<>F == [](!F) 317 return AlwaysOp(self.child.negate()).normalize() 318 319class AlwaysOp(UnaryOp): 320 def normalize(self): 321 # []F === !(true U !F) == false V F 322 new = ASTNode(Literal(False)) 323 return VOp(new, self.child) 324 325 def _is_temporal(self): 326 return True 327 328 def negate(self): 329 # ![]F == <>(!F) 330 return EventuallyOp(self.child.negate()).normalize() 331 332class NextOp(UnaryOp): 333 def normalize(self): 334 return self 335 336 def _is_temporal(self): 337 return True 338 339 def negate(self): 340 # not (next A) == next (not A) 341 self.child = self.child.negate() 342 return self 343 344 @staticmethod 345 def expand(n: ASTNode, node: GraphNode, node_set) -> set[GraphNode]: 346 tmp = GraphNode(node.incoming, 347 node.new, 348 node.old | {n}, 349 node.next | {n.op.child}) 350 return tmp.expand(node_set) 351 352class NotOp(UnaryOp): 353 def __str__(self): 354 return "!" + str(self.child) 355 356 def normalize(self): 357 return self.child.op.negate() 358 359 def negate(self): 360 return self.child.op 361 362 def _is_temporal(self): 363 return False 364 365 @staticmethod 366 def expand(n: ASTNode, node: GraphNode, node_set) -> set[GraphNode]: 367 for f in node.old: 368 if n.op.child is f: 369 return node_set 370 node.old |= {n} 371 return node.expand(node_set) 372 373class Variable: 374 def __init__(self, name: str): 375 self.name = name 376 377 def __hash__(self): 378 return hash(self.name) 379 380 def __iter__(self): 381 yield from () 382 383 def __str__(self): 384 return self.name.lower() 385 386 def negate(self): 387 new = ASTNode(self) 388 return NotOp(new) 389 390 def normalize(self): 391 return self 392 393 def is_temporal(self): 394 return False 395 396 @staticmethod 397 def expand(n: ASTNode, node: GraphNode, node_set) -> set[GraphNode]: 398 for f in node.old: 399 if isinstance(f.op, NotOp) and f.op.child is n: 400 return node_set 401 node.old |= {n} 402 return node.expand(node_set) 403 404class Literal: 405 def __init__(self, value: bool): 406 self.value = value 407 408 def __iter__(self): 409 yield from () 410 411 def __hash__(self): 412 return hash(self.value) 413 414 def __str__(self): 415 if self.value: 416 return "true" 417 return "false" 418 419 def negate(self): 420 self.value = not self.value 421 return self 422 423 def normalize(self): 424 return self 425 426 def is_temporal(self): 427 return False 428 429 @staticmethod 430 def expand(n: ASTNode, node: GraphNode, node_set) -> set[GraphNode]: 431 if not n.op.value: 432 return node_set 433 node.old |= {n} 434 return node.expand(node_set) 435 436def p_spec(p): 437 ''' 438 spec : assign 439 | assign spec 440 ''' 441 if len(p) == 3: 442 p[2].append(p[1]) 443 p[0] = p[2] 444 else: 445 p[0] = [p[1]] 446 447def p_assign(p): 448 ''' 449 assign : VARIABLE ASSIGN ltl 450 ''' 451 p[0] = (p[1], p[3]) 452 453def p_ltl(p): 454 ''' 455 ltl : opd 456 | binop 457 | unop 458 ''' 459 p[0] = p[1] 460 461def p_opd(p): 462 ''' 463 opd : VARIABLE 464 | LITERAL 465 | LPAREN ltl RPAREN 466 ''' 467 if p[1] == "true": 468 p[0] = ASTNode(Literal(True)) 469 elif p[1] == "false": 470 p[0] = ASTNode(Literal(False)) 471 elif p[1] == '(': 472 p[0] = p[2] 473 else: 474 p[0] = ASTNode(Variable(p[1])) 475 476def p_unop(p): 477 ''' 478 unop : ALWAYS ltl 479 | EVENTUALLY ltl 480 | NEXT ltl 481 | NOT ltl 482 ''' 483 if p[1] == "always": 484 op = AlwaysOp(p[2]) 485 elif p[1] == "eventually": 486 op = EventuallyOp(p[2]) 487 elif p[1] == "next": 488 op = NextOp(p[2]) 489 elif p[1] == "not": 490 op = NotOp(p[2]) 491 else: 492 raise AutomataError(f"Invalid unary operator {p[1]}") 493 494 p[0] = ASTNode(op) 495 496def p_binop(p): 497 ''' 498 binop : opd UNTIL ltl 499 | opd AND ltl 500 | opd OR ltl 501 | opd IMPLY ltl 502 ''' 503 if p[2] == "and": 504 op = AndOp(p[1], p[3]) 505 elif p[2] == "until": 506 op = UntilOp(p[1], p[3]) 507 elif p[2] == "or": 508 op = OrOp(p[1], p[3]) 509 elif p[2] == "imply": 510 op = ImplyOp(p[1], p[3]) 511 else: 512 raise AutomataError(f"Invalid binary operator {p[2]}") 513 514 p[0] = ASTNode(op) 515 516parser = yacc() 517 518def parse_ltl(s: str) -> ASTNode: 519 spec = parser.parse(s) 520 521 rule = None 522 subexpr = {} 523 524 for assign in spec: 525 if assign[0] == "RULE": 526 rule = assign[1] 527 else: 528 subexpr[assign[0]] = assign[1] 529 530 if rule is None: 531 raise AutomataError("Please define your specification in the \"RULE = <LTL spec>\" format") 532 533 for node in rule: 534 if not isinstance(node.op, Variable): 535 continue 536 replace = subexpr.get(node.op.name) 537 if replace is not None: 538 node.op = replace.op 539 540 return rule 541 542def create_graph(s: str): 543 atoms = set() 544 545 ltl = parse_ltl(s) 546 for c in ltl: 547 c.normalize() 548 if isinstance(c.op, Variable): 549 atoms.add(c.op.name) 550 551 init = GraphNode(set(), set(), set(), set()) 552 head = GraphNode({init}, {ltl}, set(), set()) 553 graph = sorted(head.expand(set())) 554 555 for i, node in enumerate(graph): 556 # The id assignment during graph generation has gaps. Reassign them 557 node.id = i 558 559 for incoming in node.incoming: 560 if incoming is init: 561 node.init = True 562 else: 563 incoming.outgoing.add(node) 564 for o in node.old: 565 if not o.op.is_temporal(): 566 node.labels.add(str(o)) 567 568 return sorted(atoms), graph, ltl 569