事件总线 API

Event Bus Client

class neoruntime_ipc_sdk.events.Event(topic: 'str', payload: 'dict[str, Any]', source: 'str' = '', event_id: 'str' = '', timestamp_ns: 'int' = 0, metadata: 'dict[str, str]'=<factory>)

Bases: object

topic: str

payload: dict[str, Any]

source: str = ''

event_id: str = ''

timestamp_ns: int = 0

metadata: dict[str, str]

to_json()

classmethod from_proto(msg)

__init__(topic, payload, source='', event_id='', timestamp_ns=0, metadata=<factory>)

class neoruntime_ipc_sdk.events.TopicInfo(topic: 'str', subscriber_count: 'int', total_messages: 'int', last_message_ts: 'int')

Bases: object

topic: str

subscriber_count: int

total_messages: int

last_message_ts: int

__init__(topic, subscriber_count, total_messages, last_message_ts)

class neoruntime_ipc_sdk.events.EventClient(endpoint=None)

Bases: GrpcClient

Event Bus Client

Usage:

events = EventClient()

events.publish("app/alert", {"type": "person_detected"})

for event in events.subscribe("model/*/detections"):
    print(f"Received: {event.topic}")

__init__(endpoint=None)

close()

publish(topic, payload, persistent=False, ttl_ms=None, metadata=None, compact=False)

publish_batch(events, persistent=False)

subscribe(topic, filters=None, queue_size=100, drop_old=True)

on_event(topic, callback, filters=None)

unsubscribe(topic)

list_topics()

get_topic_info(topic)

get_stats()

get_topic_stats(topic)

EventClient

class neoruntime_ipc_sdk.EventClient(endpoint=None)

Bases: GrpcClient

Event Bus Client

Usage:

events = EventClient()

events.publish("app/alert", {"type": "person_detected"})

for event in events.subscribe("model/*/detections"):
    print(f"Received: {event.topic}")

__init__(endpoint=None)

close()

publish(topic, payload, persistent=False, ttl_ms=None, metadata=None, compact=False)

publish_batch(events, persistent=False)

subscribe(topic, filters=None, queue_size=100, drop_old=True)

on_event(topic, callback, filters=None)

unsubscribe(topic)

list_topics()

get_topic_info(topic)

get_stats()

get_topic_stats(topic)

数据类型

Event

class neoruntime_ipc_sdk.Event(topic: 'str', payload: 'dict[str, Any]', source: 'str' = '', event_id: 'str' = '', timestamp_ns: 'int' = 0, metadata: 'dict[str, str]'=<factory>)

topic: str

payload: dict[str, Any]

source: str = ''

event_id: str = ''

timestamp_ns: int = 0

metadata: dict[str, str]

to_json()

classmethod from_proto(msg)

__init__(topic, payload, source='', event_id='', timestamp_ns=0, metadata=<factory>)

TopicInfo

class neoruntime_ipc_sdk.TopicInfo(topic: 'str', subscriber_count: 'int', total_messages: 'int', last_message_ts: 'int')

topic: str

subscriber_count: int

total_messages: int

last_message_ts: int

__init__(topic, subscriber_count, total_messages, last_message_ts)

使用示例

发布事件

from neoruntime_ipc_sdk import EventClient

events = EventClient()

# 发布简单事件
events.publish("app/status", {"status": "running"})

# 发布带元数据的事件
events.publish("app/detection", {
    "timestamp": 1234567890,
    "objects": [
        {"label": "person", "score": 0.95, "bbox": [100, 200, 50, 100]},
        {"label": "car", "score": 0.88, "bbox": [300, 400, 80, 120]}
    ]
}, metadata={"camera": "cam0", "model": "person_vehicle_v1"})

# 发布持久化事件
events.publish("app/alert", {"type": "person_detected"}, persistent=True)

# 批量发布
events.publish_batch([
    {"topic": "app/event1", "payload": {"data": "value1"}},
    {"topic": "app/event2", "payload": {"data": "value2"}}
])

订阅事件

# 订阅单个主题
for event in events.subscribe("system/temperature"):
    temp = event.payload.get("value")
    print(f"当前温度: {temp}°C")

# 订阅多个主题（通配符）
for event in events.subscribe("model/*/detections"):
    model_name = event.topic.split('/')[1]
    count = len(event.payload.get("objects", []))
    print(f"模型 {model_name} 检测到 {count} 个对象")

# 带过滤订阅
for event in events.subscribe("app/alerts", filters={"severity": "high"}):
    print(f"高严重性告警: {event.payload}")

使用回调函数

def on_alert(event):
    alert_type = event.payload.get("type")
    print(f"收到告警: {alert_type}")

def on_detection(event):
    objects = event.payload.get("objects", [])
    print(f"检测到 {len(objects)} 个对象")

# 注册回调
events.on_event("app/alert", on_alert)
events.on_event("model/*/detections", on_detection)

# 保持运行
import time
while True:
    time.sleep(1)

主题通配符

事件总线支持通配符：

# 匹配 model/person_v1/detections, model/car_v1/detections
events.subscribe("model/*/detections")

# 匹配 app/my_app/alert, app/my_app/status/running
events.subscribe("app/my_app/**")

列出主题

# 列出所有活跃主题
topics = events.list_topics()
for topic in topics:
    print(f"{topic.topic}: {topic.subscriber_count} 订阅者, {topic.total_messages} 条消息")

# 获取特定主题信息
info = events.get_topic_info("app/alerts")
if info:
    print(f"订阅者: {info.subscriber_count}")

获取统计信息

# 获取全局统计
stats = events.get_stats()
print(f"总订阅者数: {stats['total_subscribers']}")
print(f"总主题数: {stats['total_topics']}")
print(f"运行时间: {stats['uptime_ms']}ms")

# 获取特定主题统计
topic_stats = events.get_topic_stats("app/alerts")
print(f"发布数: {topic_stats['published_count']}")
print(f"投递数: {topic_stats['delivered_count']}")
print(f"丢弃数: {topic_stats['dropped_count']}")
print(f"平均延迟: {topic_stats['avg_latency_us']}us")

取消订阅

# 取消订阅
events.unsubscribe("app/alert")

事件过滤

# 订阅并过滤
for event in events.subscribe("model/*/detections"):
    # 只处理高置信度检测
    objects = event.payload.get("objects", [])
    high_conf = [obj for obj in objects if obj.get("score", 0) > 0.9]

    if high_conf:
        print(f"高置信度检测: {len(high_conf)} 个对象")

上下文管理器

# 使用上下文管理器自动管理连接
with EventClient() as events:
    events.publish("app/status", {"status": "running"})
    for event in events.subscribe("app/*"):
        print(f"收到: {event.topic}")

错误处理

from grpc import RpcError

try:
    events.publish("app/status", {"status": "running"})
except RpcError as e:
    print(f"发布失败: {e.details()}")

try:
    for event in events.subscribe("invalid/topic"):
        process_event(event)
except RpcError as e:
    print(f"订阅失败: {e.details()}")