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