Verification of Reactive Systems — Safety, Liveness and Real-Time Properties

class ReactiveVerifier:
    def __init__(self, states, transitions):
        self.states = states
        self.transitions = transitions

    def check_safety(self, bad_states):
        for s in self.states:
            if s in bad_states:
                return False, f"Initial state {s} is bad"
        visited = set()
        stack = list(self.states.keys())
        while stack:
            s = stack.pop()
            if s in visited:
                continue
            visited.add(s)
            if s in bad_states:
                return False, f"Bad state {s} reachable"
            stack.extend(self.transitions.get(s, []))
        return True, "Safety holds"

states = {"s1": {"x": 0}, "s2": {"x": 1}, "s3": {"x": 2}}
transitions = {"s1": ["s2"], "s2": ["s1", "s3"], "s3": ["s3"]}
bad = {"s3"}  # x should never reach 2
rv = ReactiveVerifier(states, transitions)
safe, msg = rv.check_safety(bad)
print(f"Safety check: {msg}")

def check_liveness(system, request_states, grant_states):
    for s in request_states:
        path = [s]
        current = s
        visited_at_depth = {s: 0}
        for step in range(100):
            next_states = system.transitions.get(current, [])
            if not next_states:
                break
            current = next_states[0]
            path.append(current)
            if current in grant_states:
                return True, f"Grant reached in {step+1} steps"
        if current in visited_at_depth:
            return False, "Cycle without grant detected"
    return True, "All requests eventually granted"

req_states = ["s1"]
grant_states = ["s3"]
live, msg = check_liveness(ReactiveVerifier(states, transitions), req_states, grant_states)
print(f"Liveness check: {msg}")

class TimedReactiveVerifier:
    def __init__(self):
        self.deadlines = {}  # state -> max time

    def add_deadline(self, state, max_time):
        self.deadlines[state] = max_time

    def check_response_time(self, path, trigger_state, response_state):
        for i, s in enumerate(path):
            if s == trigger_state:
                for j in range(i+1, min(i+10, len(path))):
                    if path[j] == response_state:
                        time_elapsed = j - i
                        max_time = self.deadlines.get(trigger_state, float('inf'))
                        if time_elapsed <= max_time:
                            return True, f"Response in {time_elapsed} units"
                        return False, f"Response took {time_elapsed}, max {max_time}"
                return False, "No response found"
        return True, "No trigger encountered"

trv = TimedReactiveVerifier()
trv.add_deadline("request", 5)
path = ["idle", "request", "busy", "processing", "response", "idle"]
ok, msg = trv.check_response_time(path, "request", "response")
print(f"Timing check: {msg}")

class LustreNode:
    def __init__(self):
        self.pre_count = 0
        self.initialized = False

    def step(self, inc):
        if not self.initialized:
            self.pre_count = 0
            self.initialized = True
            count = 0
        else:
            count = self.pre_count + (1 if inc else 0)
        self.pre_count = count
        return count

node = LustreNode()
inputs = [True, False, True, True, False]
outputs = [node.step(inc) for inc in inputs]
print(f"Outputs: {outputs}")

def verify_lustre_property(node, inputs, property_fn):
    node.initialized = False
    for i, inc in enumerate(inputs):
        out = node.step(inc)
        if not property_fn(i, out):
            return False, f"Property failed at step {i}, output {out}"
    return True, "Property holds for all steps"

node = LustreNode()
prop = lambda step, val: val >= 0 and val <= len(inputs)
ok, msg = verify_lustre_property(node, [True, True, True, True, True], prop)
print(f"Lustre property: {msg}")

class TimedAutomaton:
    def __init__(self, locations, edges, clocks, invariants):
        self.locations = locations
        self.edges = edges  # (from, guard, reset, to)
        self.clocks = clocks
        self.invariants = invariants

    def can_fire(self, edge, clock_vals):
        loc, guard, reset, to = edge
        return eval(guard, dict(zip(self.clocks, clock_vals)))

    def fire(self, edge, clock_vals):
        loc, guard, reset, to = edge
        new_clocks = list(clock_vals)
        if reset:
            new_clocks[self.clocks.index(reset)] = 0
        return to, new_clocks

ta = TimedAutomaton(
    ["idle", "active", "error"],
    [("idle", "x >= 0", None, "active"),
     ("active", "x >= 5", "x", "error"),
     ("active", "x < 5", None, "idle")],
    ["x"],
    {}
)
clocks = [0.0]
print(f"Can fire edge 0: {ta.can_fire(ta.edges[0], clocks)}")