GitHub Viewer
import networkx as nx
import heapq
import datetime
import logging
logging.basicConfig(level=logging.INFO)
node_types = {
"sensor": {
"default_state": {"value": 0},
"on_threshold_event": lambda node, event: logging.info(f"Sensor {node} event: {event}")
},
"controller": {
"default_state": {"status": "idle"},
"on_state_change": lambda node, event: logging.info(f"Controller {node} event: {event}")
}
}
event_queue = []
def add_node_with_type(graph, node_id, node_type, **kwargs):
state = node_types[node_type]["default_state"].copy()
state.update(kwargs)
graph.add_node(node_id, node_type=node_type, **state)
def add_event(event_type, node, priority, **kwargs):
event = {"event_type": event_type, "node": node, "timestamp": datetime.datetime.now(), "priority": priority, **kwargs}
heapq.heappush(event_queue, (priority, event))
logging.info(f"Event added: {event}")
def process_events(graph):
while event_queue:
priority, event = heapq.heappop(event_queue)
logging.info(f"Processing: {event}")
node_type = graph.nodes[event["node"]]["node_type"]
if event["event_type"] == "threshold_event" and "on_threshold_event" in node_types[node_type]:
node_types[node_type]["on_threshold_event"](event["node"], event)
elif event["event_type"] == "state_change" and "on_state_change" in node_types[node_type]:
node_types[node_type]["on_state_change"](event["node"], event)
graph = nx.Graph()
add_node_with_type(graph, "sensor1", "sensor", value=5)
add_node_with_type(graph, "controller1", "controller", status="active")
add_event("threshold_event", "sensor1", 1, value=10)
add_event("state_change", "controller1", 2, status="inactive")
process_events(graph)