proto/
├── onto/
│   ├── common/v1/         # 公共类型定义
│   │   ├── types.proto    # RID, Timestamp, Pagination 等
│   │   └── errors.proto   # 统一错误码
│   ├── ontology/v1/       # 本体服务
│   ├── action/v1/         # Action 服务
│   ├── query/v1/          # 查询服务
│   ├── subscription/v1/   # 订阅服务
│   └── custom/v1/         # 自定义扩展服务

// proto/onto/device/v1/device_service.proto
syntax = "proto3";

package onto.device.v1;

import "onto/common/v1/types.proto";
import "google/protobuf/timestamp.proto";
import "google/protobuf/empty.proto";

option java_package = "com.onto.device.v1";
option java_multiple_files = true;
option python_package = "onto.device.v1";

// 设备监控服务
service DeviceMonitorService {
    // 单个设备查询
    rpc GetDevice(GetDeviceRequest) returns (GetDeviceResponse);

    // 批量设备查询
    rpc ListDevices(ListDevicesRequest) returns (ListDevicesResponse);

    // 上报设备遥测数据
    rpc ReportTelemetry(ReportTelemetryRequest) returns (ReportTelemetryResponse);

    // 批量上报（客户端流式）
    rpc BatchReportTelemetry(stream TelemetryDataPoint) returns (BatchReportResponse);

    // 实时遥测数据流（服务端流式）
    rpc StreamTelemetry(StreamTelemetryRequest) returns (stream TelemetryDataPoint);

    // 双向流：实时命令与响应
    rpc DeviceCommandChannel(stream DeviceCommand) returns (stream CommandResponse);

    // 触发设备告警
    rpc TriggerAlert(TriggerAlertRequest) returns (TriggerAlertResponse);

    // 查询告警历史
    rpc ListAlerts(ListAlertsRequest) returns (ListAlertsResponse);
}

// ---- 消息定义 ----

message Device {
    string device_rid = 1;
    string device_name = 2;
    string device_type = 3;          // SENSOR, ACTUATOR, GATEWAY
    string location = 4;
    DeviceStatus status = 5;
    map<string, string> metadata = 6;
    google.protobuf.Timestamp last_seen = 7;
    google.protobuf.Timestamp registered_at = 8;
}

enum DeviceStatus {
    DEVICE_STATUS_UNSPECIFIED = 0;
    DEVICE_STATUS_ONLINE = 1;
    DEVICE_STATUS_OFFLINE = 2;
    DEVICE_STATUS_MAINTENANCE = 3;
    DEVICE_STATUS_ERROR = 4;
}

message TelemetryDataPoint {
    string device_rid = 1;
    string metric_name = 2;
    double value = 3;
    string unit = 4;
    google.protobuf.Timestamp timestamp = 5;
    map<string, string> tags = 6;
}

message GetDeviceRequest {
    string device_rid = 1;
    bool include_latest_telemetry = 2;
}

message GetDeviceResponse {
    Device device = 1;
    repeated TelemetryDataPoint latest_telemetry = 2;
}

message ListDevicesRequest {
    string device_type = 1;         // 可选过滤
    DeviceStatus status = 2;        // 可选过滤
    string location_prefix = 3;     // 位置前缀匹配
    onto.common.v1.Pagination pagination = 4;
}

message ListDevicesResponse {
    repeated Device devices = 1;
    onto.common.v1.PaginationInfo pagination_info = 2;
}

message ReportTelemetryRequest {
    repeated TelemetryDataPoint data_points = 1;
}

message ReportTelemetryResponse {
    int32 accepted_count = 1;
    int32 rejected_count = 2;
    repeated string rejected_reasons = 3;
}

message BatchReportResponse {
    int64 total_received = 1;
    int64 total_accepted = 2;
    int64 total_rejected = 3;
}

message StreamTelemetryRequest {
    string device_rid = 1;          // 可选，空则订阅所有设备
    repeated string metric_names = 2; // 可选，空则订阅所有指标
    int32 throttle_ms = 3;          // 节流间隔
}

message DeviceCommand {
    string command_id = 1;
    string device_rid = 2;
    string command_type = 3;        // REBOOT, CONFIG_UPDATE, FIRMWARE_UPDATE
    map<string, string> parameters = 4;
    int32 timeout_seconds = 5;
}

message CommandResponse {
    string command_id = 1;
    string device_rid = 2;
    CommandStatus status = 3;
    string message = 4;
    map<string, string> result = 5;
}

enum CommandStatus {
    COMMAND_STATUS_UNSPECIFIED = 0;
    COMMAND_STATUS_RECEIVED = 1;
    COMMAND_STATUS_EXECUTING = 2;
    COMMAND_STATUS_SUCCESS = 3;
    COMMAND_STATUS_FAILED = 4;
    COMMAND_STATUS_TIMEOUT = 5;
}

message TriggerAlertRequest {
    string device_rid = 1;
    string alert_type = 2;
    string severity = 3;            // INFO, WARNING, CRITICAL
    string message = 4;
    map<string, string> context = 5;
}

message TriggerAlertResponse {
    string alert_id = 1;
    google.protobuf.Timestamp triggered_at = 2;
}

message ListAlertsRequest {
    string device_rid = 1;
    string severity = 2;
    google.protobuf.Timestamp from_time = 3;
    google.protobuf.Timestamp to_time = 4;
    onto.common.v1.Pagination pagination = 5;
}

message ListAlertsResponse {
    repeated Alert alerts = 1;
    onto.common.v1.PaginationInfo pagination_info = 2;
}

message Alert {
    string alert_id = 1;
    string device_rid = 2;
    string alert_type = 3;
    string severity = 4;
    string message = 5;
    map<string, string> context = 6;
    bool acknowledged = 7;
    google.protobuf.Timestamp triggered_at = 8;
    google.protobuf.Timestamp acknowledged_at = 9;
}

# 安装 gRPC 工具
pip install grpcio grpcio-tools

# 生成 Python 代码
python -m grpc_tools.protoc \
    --proto_path=proto/ \
    --python_out=python-sdk/generated/ \
    --grpc_python_out=python-sdk/generated/ \
    --pyi_out=python-sdk/generated/ \
    proto/onto/device/v1/device_service.proto

# 生成的文件：
# python-sdk/generated/onto/device/v1/
# ├── device_service_pb2.py       # 消息类
# ├── device_service_pb2.pyi      # 类型存根
# └── device_service_pb2_grpc.py  # 服务存根和 Servicer

// build.gradle.kts
plugins {
    id("com.google.protobuf") version "0.9.4"
}

dependencies {
    implementation("io.grpc:grpc-netty-shaded:1.62.2")
    implementation("io.grpc:grpc-protobuf:1.62.2")
    implementation("io.grpc:grpc-stub:1.62.2")
    compileOnly("org.apache.tomcat:annotations-api:6.0.53")
}

protobuf {
    protoc {
        artifact = "com.google.protobuf:protoc:3.25.3"
    }
    plugins {
        id("grpc") {
            artifact = "io.grpc:protoc-gen-grpc-java:1.62.2"
        }
    }
    generateProtoTasks {
        all().forEach { task ->
            task.plugins {
                id("grpc")
            }
        }
    }
}

# intelligence-Layer/device_monitor/grpc_server.py
import grpc
from concurrent import futures
from datetime import datetime, timezone
from typing import AsyncIterator
import asyncio
import logging

from onto.device.v1 import device_service_pb2 as pb2
from onto.device.v1 import device_service_pb2_grpc as pb2_grpc
from google.protobuf.timestamp_pb2 import Timestamp

logger = logging.getLogger(__name__)


class DeviceMonitorServicer(pb2_grpc.DeviceMonitorServiceServicer):
    """设备监控 gRPC 服务实现"""

    def __init__(self, device_repo, telemetry_store, alert_manager):
        self.device_repo = device_repo
        self.telemetry_store = telemetry_store
        self.alert_manager = alert_manager
        self._telemetry_subscribers: dict[str, asyncio.Queue] = {}

    async def GetDevice(self, request, context):
        """查询单个设备"""
        device = await self.device_repo.get(request.device_rid)

        if device is None:
            context.set_code(grpc.StatusCode.NOT_FOUND)
            context.set_details(f"Device {request.device_rid} not found")
            return pb2.GetDeviceResponse()

        response = pb2.GetDeviceResponse(
            device=self._to_proto_device(device),
        )

        if request.include_latest_telemetry:
            telemetry = await self.telemetry_store.get_latest(
                request.device_rid, limit=10
            )
            for t in telemetry:
                response.latest_telemetry.append(self._to_proto_telemetry(t))

        return response

    async def ListDevices(self, request, context):
        """批量查询设备"""
        filters = {}
        if request.device_type:
            filters["device_type"] = request.device_type
        if request.status != pb2.DEVICE_STATUS_UNSPECIFIED:
            filters["status"] = request.status
        if request.location_prefix:
            filters["location_prefix"] = request.location_prefix

        page_size = request.pagination.page_size or 20
        page_token = request.pagination.page_token or ""

        devices, next_token, total = await self.device_repo.list(
            filters=filters,
            page_size=page_size,
            page_token=page_token,
        )

        return pb2.ListDevicesResponse(
            devices=[self._to_proto_device(d) for d in devices],
            pagination_info=pb2.onto.common.v1.PaginationInfo(
                next_page_token=next_token,
                total_count=total,
            ),
        )

    async def ReportTelemetry(self, request, context):
        """上报遥测数据"""
        accepted = 0
        rejected = 0
        rejected_reasons = []

        for dp in request.data_points:
            try:
                await self.telemetry_store.write(
                    device_rid=dp.device_rid,
                    metric_name=dp.metric_name,
                    value=dp.value,
                    unit=dp.unit,
                    timestamp=dp.timestamp.ToDatetime(),
                    tags=dict(dp.tags),
                )
                accepted += 1

                # 通知流式订阅者
                await self._notify_subscribers(dp)

            except ValueError as e:
                rejected += 1
                rejected_reasons.append(f"{dp.device_rid}/{dp.metric_name}: {e}")

        return pb2.ReportTelemetryResponse(
            accepted_count=accepted,
            rejected_count=rejected,
            rejected_reasons=rejected_reasons,
        )

    async def BatchReportTelemetry(self, request_iterator, context):
        """客户端流式：批量上报遥测数据"""
        total_received = 0
        total_accepted = 0
        total_rejected = 0
        batch = []
        batch_size = 100

        async for data_point in request_iterator:
            total_received += 1
            batch.append(data_point)

            if len(batch) >= batch_size:
                accepted, rejected = await self._flush_batch(batch)
                total_accepted += accepted
                total_rejected += rejected
                batch.clear()

        # 刷新最后一批
        if batch:
            accepted, rejected = await self._flush_batch(batch)
            total_accepted += accepted
            total_rejected += rejected

        return pb2.BatchReportResponse(
            total_received=total_received,
            total_accepted=total_accepted,
            total_rejected=total_rejected,
        )

    async def StreamTelemetry(self, request, context):
        """服务端流式：实时推送遥测数据"""
        subscriber_id = f"stream-{id(context)}"
        queue = asyncio.Queue(maxsize=1000)
        self._telemetry_subscribers[subscriber_id] = queue

        logger.info(f"New telemetry subscriber: {subscriber_id}")

        try:
            while not context.cancelled():
                try:
                    data_point = await asyncio.wait_for(queue.get(), timeout=30.0)

                    # 应用过滤
                    if request.device_rid and data_point.device_rid != request.device_rid:
                        continue
                    if request.metric_names and data_point.metric_name not in request.metric_names:
                        continue

                    yield data_point

                    # 节流
                    if request.throttle_ms > 0:
                        await asyncio.sleep(request.throttle_ms / 1000.0)

                except asyncio.TimeoutError:
                    # 发送心跳（空数据点）
                    yield pb2.TelemetryDataPoint()

        finally:
            del self._telemetry_subscribers[subscriber_id]
            logger.info(f"Telemetry subscriber disconnected: {subscriber_id}")

    async def DeviceCommandChannel(self, request_iterator, context):
        """双向流式：设备命令通道"""
        async for command in request_iterator:
            logger.info(f"Command received: {command.command_type} -> {command.device_rid}")

            # 确认收到
            yield pb2.CommandResponse(
                command_id=command.command_id,
                device_rid=command.device_rid,
                status=pb2.COMMAND_STATUS_RECEIVED,
                message="Command received, executing...",
            )

            # 异步执行命令
            try:
                result = await self._execute_command(command)
                yield pb2.CommandResponse(
                    command_id=command.command_id,
                    device_rid=command.device_rid,
                    status=pb2.COMMAND_STATUS_SUCCESS,
                    message="Command executed successfully",
                    result=result,
                )
            except TimeoutError:
                yield pb2.CommandResponse(
                    command_id=command.command_id,
                    device_rid=command.device_rid,
                    status=pb2.COMMAND_STATUS_TIMEOUT,
                    message=f"Command timed out after {command.timeout_seconds}s",
                )
            except Exception as e:
                yield pb2.CommandResponse(
                    command_id=command.command_id,
                    device_rid=command.device_rid,
                    status=pb2.COMMAND_STATUS_FAILED,
                    message=str(e),
                )

    async def TriggerAlert(self, request, context):
        """触发告警"""
        alert_id = await self.alert_manager.trigger(
            device_rid=request.device_rid,
            alert_type=request.alert_type,
            severity=request.severity,
            message=request.message,
            context=dict(request.context),
        )

        now = Timestamp()
        now.GetCurrentTime()

        return pb2.TriggerAlertResponse(
            alert_id=alert_id,
            triggered_at=now,
        )

    # ---- 内部方法 ----

    async def _notify_subscribers(self, data_point):
        """通知所有流式订阅者"""
        dead_subscribers = []
        for sub_id, queue in self._telemetry_subscribers.items():
            try:
                queue.put_nowait(data_point)
            except asyncio.QueueFull:
                logger.warning(f"Subscriber {sub_id} queue full, dropping data point")

    async def _flush_batch(self, batch):
        """批量写入遥测数据"""
        accepted = 0
        rejected = 0
        for dp in batch:
            try:
                await self.telemetry_store.write(
                    device_rid=dp.device_rid,
                    metric_name=dp.metric_name,
                    value=dp.value,
                    unit=dp.unit,
                    timestamp=dp.timestamp.ToDatetime(),
                    tags=dict(dp.tags),
                )
                accepted += 1
            except Exception:
                rejected += 1
        return accepted, rejected

    async def _execute_command(self, command):
        """执行设备命令"""
        # 实际实现中会通过 MQTT/CoAP 下发到设备
        await asyncio.sleep(0.1)  # 模拟
        return {"status": "ok"}

    def _to_proto_device(self, device) -> pb2.Device:
        """将领域模型转为 Protobuf 消息"""
        ts = Timestamp()
        ts.FromDatetime(device.last_seen)

        reg_ts = Timestamp()
        reg_ts.FromDatetime(device.registered_at)

        return pb2.Device(
            device_rid=device.rid,
            device_name=device.name,
            device_type=device.device_type,
            location=device.location,
            status=device.status,
            metadata=device.metadata,
            last_seen=ts,
            registered_at=reg_ts,
        )

    def _to_proto_telemetry(self, t) -> pb2.TelemetryDataPoint:
        ts = Timestamp()
        ts.FromDatetime(t.timestamp)
        return pb2.TelemetryDataPoint(
            device_rid=t.device_rid,
            metric_name=t.metric_name,
            value=t.value,
            unit=t.unit,
            timestamp=ts,
            tags=t.tags,
        )

# intelligence-Layer/device_monitor/server.py
import grpc
from grpc_reflection.v1alpha import reflection
import asyncio
import signal
import logging

from onto.device.v1 import device_service_pb2 as pb2
from onto.device.v1 import device_service_pb2_grpc as pb2_grpc
from device_monitor.grpc_server import DeviceMonitorServicer
from device_monitor.repositories import DeviceRepository
from device_monitor.stores import TelemetryStore
from device_monitor.alerts import AlertManager

logger = logging.getLogger(__name__)


async def serve():
    """启动 gRPC 服务器"""

    # 初始化依赖
    device_repo = DeviceRepository()
    telemetry_store = TelemetryStore()
    alert_manager = AlertManager()

    # 创建 gRPC 服务器
    server = grpc.aio.server(
        futures.ThreadPoolExecutor(max_workers=10),
        options=[
            ("grpc.max_send_message_length", 50 * 1024 * 1024),    # 50MB
            ("grpc.max_receive_message_length", 50 * 1024 * 1024),  # 50MB
            ("grpc.keepalive_time_ms", 30000),                      # 30s
            ("grpc.keepalive_timeout_ms", 10000),                   # 10s
            ("grpc.keepalive_permit_without_calls", True),
        ],
    )

    # 注册服务
    servicer = DeviceMonitorServicer(device_repo, telemetry_store, alert_manager)
    pb2_grpc.add_DeviceMonitorServiceServicer_to_server(servicer, server)

    # 启用 gRPC 反射（方便调试）
    service_names = (
        pb2.DESCRIPTOR.services_by_name["DeviceMonitorService"].full_name,
        reflection.SERVICE_NAME,
    )
    reflection.enable_server_reflection(service_names, server)

    # 绑定端口
    listen_addr = "[::]:50051"
    server.add_insecure_port(listen_addr)

    logger.info(f"Starting gRPC server on {listen_addr}")
    await server.start()

    # 优雅关闭
    async def shutdown(sig):
        logger.info(f"Received {sig.name}, shutting down...")
        await server.stop(grace=5)  # 5 秒优雅关闭

    loop = asyncio.get_event_loop()
    for sig in (signal.SIGTERM, signal.SIGINT):
        loop.add_signal_handler(sig, lambda s=sig: asyncio.create_task(shutdown(s)))

    await server.wait_for_termination()


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(serve())

# python-sdk/ontology_sdk/grpc_client/device_client.py
import grpc
from onto.device.v1 import device_service_pb2 as pb2
from onto.device.v1 import device_service_pb2_grpc as pb2_grpc


class DeviceMonitorClient:
    """设备监控 gRPC 客户端"""

    def __init__(self, endpoint: str = "localhost:50051"):
        self.channel = grpc.insecure_channel(endpoint)
        self.stub = pb2_grpc.DeviceMonitorServiceStub(self.channel)

    def get_device(self, device_rid: str, include_telemetry: bool = False):
        """查询单个设备"""
        request = pb2.GetDeviceRequest(
            device_rid=device_rid,
            include_latest_telemetry=include_telemetry,
        )
        try:
            response = self.stub.GetDevice(request, timeout=5.0)
            return response
        except grpc.RpcError as e:
            if e.code() == grpc.StatusCode.NOT_FOUND:
                return None
            raise

    def list_devices(self, device_type: str = "", status=None, page_size: int = 20):
        """批量查询设备"""
        request = pb2.ListDevicesRequest(
            device_type=device_type,
            pagination=pb2.onto.common.v1.Pagination(page_size=page_size),
        )
        if status is not None:
            request.status = status

        return self.stub.ListDevices(request, timeout=10.0)

    def report_telemetry(self, data_points: list[dict]):
        """上报遥测数据"""
        proto_points = []
        for dp in data_points:
            ts = Timestamp()
            ts.FromDatetime(dp["timestamp"])
            proto_points.append(pb2.TelemetryDataPoint(
                device_rid=dp["device_rid"],
                metric_name=dp["metric_name"],
                value=dp["value"],
                unit=dp.get("unit", ""),
                timestamp=ts,
                tags=dp.get("tags", {}),
            ))

        request = pb2.ReportTelemetryRequest(data_points=proto_points)
        return self.stub.ReportTelemetry(request, timeout=10.0)

    def close(self):
        self.channel.close()

    def __enter__(self):
        return self

    def __exit__(self, *args):
        self.close()

# python-sdk/ontology_sdk/grpc_client/async_device_client.py
import grpc
import asyncio
from typing import AsyncIterator

from onto.device.v1 import device_service_pb2 as pb2
from onto.device.v1 import device_service_pb2_grpc as pb2_grpc


class AsyncDeviceMonitorClient:
    """设备监控异步 gRPC 客户端"""

    def __init__(self, endpoint: str = "localhost:50051"):
        self.channel = grpc.aio.insecure_channel(endpoint)
        self.stub = pb2_grpc.DeviceMonitorServiceStub(self.channel)

    async def stream_telemetry(
        self,
        device_rid: str = "",
        metric_names: list[str] | None = None,
    ) -> AsyncIterator[pb2.TelemetryDataPoint]:
        """流式接收遥测数据"""
        request = pb2.StreamTelemetryRequest(
            device_rid=device_rid,
            metric_names=metric_names or [],
            throttle_ms=100,
        )

        async for data_point in self.stub.StreamTelemetry(request):
            if data_point.device_rid:  # 跳过心跳
                yield data_point

    async def batch_report(
        self,
        data_points: AsyncIterator[pb2.TelemetryDataPoint],
    ) -> pb2.BatchReportResponse:
        """客户端流式批量上报"""
        return await self.stub.BatchReportTelemetry(data_points)

    async def command_channel(self, commands: AsyncIterator[pb2.DeviceCommand]):
        """双向流式命令通道"""
        async for response in self.stub.DeviceCommandChannel(commands):
            yield response

    async def close(self):
        await self.channel.close()

    async def __aenter__(self):
        return self

    async def __aexit__(self, *args):
        await self.close()


# 使用示例
async def main():
    async with AsyncDeviceMonitorClient("localhost:50051") as client:
        # 流式接收温度传感器数据
        async for dp in client.stream_telemetry(
            device_rid="ri.device.sensor-001",
            metric_names=["temperature"],
        ):
            print(f"[{dp.timestamp}] {dp.metric_name}: {dp.value}{dp.unit}")

            # 温度异常时触发告警
            if dp.value > 80.0:
                await client.stub.TriggerAlert(pb2.TriggerAlertRequest(
                    device_rid=dp.device_rid,
                    alert_type="TEMPERATURE_HIGH",
                    severity="CRITICAL",
                    message=f"Temperature {dp.value}°C exceeds threshold 80°C",
                    context={"current_value": str(dp.value)},
                ))

asyncio.run(main())

class AuthInterceptor(grpc.aio.ServerInterceptor):
    """服务端认证拦截器"""

    def __init__(self, token_validator):
        self.token_validator = token_validator
        self._public_methods = {"/grpc.reflection.v1alpha.ServerReflection/"}

    async def intercept_service(self, continuation, handler_call_details):
        method = handler_call_details.method

        # 公开方法不需要认证
        for public in self._public_methods:
            if method.startswith(public):
                return await continuation(handler_call_details)

        # 从 metadata 提取 token
        metadata = dict(handler_call_details.invocation_metadata or [])
        token = metadata.get("authorization", "")

        if not token.startswith("Bearer "):
            return self._unauthenticated("Missing or invalid Authorization header")

        token = token[7:]  # 去掉 "Bearer " 前缀

        try:
            user_context = await self.token_validator.validate(token)
            # 将用户信息注入 context（通过 metadata 传递）
            handler_call_details.invocation_metadata.append(
                ("x-user-id", user_context.user_id)
            )
        except InvalidTokenError as e:
            return self._unauthenticated(str(e))

        return await continuation(handler_call_details)

    def _unauthenticated(self, details: str):
        async def abort(ignored_request, context):
            await context.abort(grpc.StatusCode.UNAUTHENTICATED, details)
        return grpc.unary_unary_rpc_method_handler(abort)

import time
from prometheus_client import Counter, Histogram

grpc_requests_total = Counter(
    "grpc_requests_total",
    "Total gRPC requests",
    ["method", "status"],
)

grpc_request_duration = Histogram(
    "grpc_request_duration_seconds",
    "gRPC request duration",
    ["method"],
)


class MetricsInterceptor(grpc.aio.ServerInterceptor):
    """指标收集拦截器"""

    async def intercept_service(self, continuation, handler_call_details):
        method = handler_call_details.method
        start = time.perf_counter()

        try:
            response = await continuation(handler_call_details)
            grpc_requests_total.labels(method=method, status="OK").inc()
            return response
        except grpc.aio.AioRpcError as e:
            grpc_requests_total.labels(method=method, status=e.code().name).inc()
            raise
        finally:
            duration = time.perf_counter() - start
            grpc_request_duration.labels(method=method).observe(duration)

import pytest
import grpc
from unittest.mock import AsyncMock, MagicMock

from device_monitor.grpc_server import DeviceMonitorServicer
from onto.device.v1 import device_service_pb2 as pb2


@pytest.fixture
def servicer():
    return DeviceMonitorServicer(
        device_repo=AsyncMock(),
        telemetry_store=AsyncMock(),
        alert_manager=AsyncMock(),
    )


@pytest.fixture
def context():
    ctx = MagicMock()
    ctx.cancelled.return_value = False
    return ctx


@pytest.mark.asyncio
async def test_get_device_found(servicer, context):
    """测试设备查询——存在的设备"""
    servicer.device_repo.get.return_value = MagicMock(
        rid="ri.device.001",
        name="Temperature Sensor",
        device_type="SENSOR",
        location="Building A / Floor 3",
        status=pb2.DEVICE_STATUS_ONLINE,
        metadata={"firmware": "2.1.0"},
        last_seen=datetime.now(timezone.utc),
        registered_at=datetime.now(timezone.utc),
    )

    request = pb2.GetDeviceRequest(
        device_rid="ri.device.001",
        include_latest_telemetry=False,
    )

    response = await servicer.GetDevice(request, context)
    assert response.device.device_rid == "ri.device.001"
    assert response.device.device_name == "Temperature Sensor"


@pytest.mark.asyncio
async def test_get_device_not_found(servicer, context):
    """测试设备查询——不存在的设备"""
    servicer.device_repo.get.return_value = None

    request = pb2.GetDeviceRequest(device_rid="ri.device.999")
    response = await servicer.GetDevice(request, context)

    context.set_code.assert_called_once_with(grpc.StatusCode.NOT_FOUND)


@pytest.mark.asyncio
async def test_report_telemetry(servicer, context):
    """测试遥测数据上报"""
    ts = Timestamp()
    ts.GetCurrentTime()

    request = pb2.ReportTelemetryRequest(
        data_points=[
            pb2.TelemetryDataPoint(
                device_rid="ri.device.001",
                metric_name="temperature",
                value=25.5,
                unit="°C",
                timestamp=ts,
            ),
        ]
    )

    response = await servicer.ReportTelemetry(request, context)
    assert response.accepted_count == 1
    assert response.rejected_count == 0

import pytest
import grpc
import asyncio

from onto.device.v1 import device_service_pb2 as pb2
from onto.device.v1 import device_service_pb2_grpc as pb2_grpc


@pytest.fixture(scope="module")
async def grpc_channel():
    """创建测试用 gRPC channel"""
    channel = grpc.aio.insecure_channel("localhost:50051")
    yield channel
    await channel.close()


@pytest.fixture
def stub(grpc_channel):
    return pb2_grpc.DeviceMonitorServiceStub(grpc_channel)


@pytest.mark.integration
@pytest.mark.asyncio
async def test_full_workflow(stub):
    """端到端测试：上报数据 -> 查询设备 -> 触发告警"""

    # 1. 上报遥测数据
    ts = Timestamp()
    ts.GetCurrentTime()

    report_response = await stub.ReportTelemetry(
        pb2.ReportTelemetryRequest(
            data_points=[
                pb2.TelemetryDataPoint(
                    device_rid="ri.device.test-001",
                    metric_name="temperature",
                    value=85.0,
                    unit="°C",
                    timestamp=ts,
                ),
            ]
        )
    )
    assert report_response.accepted_count == 1

    # 2. 查询设备
    device_response = await stub.GetDevice(
        pb2.GetDeviceRequest(
            device_rid="ri.device.test-001",
            include_latest_telemetry=True,
        )
    )
    assert device_response.device.device_rid == "ri.device.test-001"

    # 3. 触发告警
    alert_response = await stub.TriggerAlert(
        pb2.TriggerAlertRequest(
            device_rid="ri.device.test-001",
            alert_type="TEMPERATURE_HIGH",
            severity="CRITICAL",
            message="Temperature exceeds threshold",
        )
    )
    assert alert_response.alert_id

from ontology_sdk import OntoPlatform

platform = OntoPlatform(base_url="http://localhost:8080", token="admin-token")

# 注册自定义服务
platform.services.register(
    name="device-monitor",
    endpoint="device-monitor:50051",
    proto_descriptor="onto.device.v1.DeviceMonitorService",
    health_check_path="/grpc.health.v1.Health/Check",
    metadata={
        "version": "1.0.0",
        "owner": "iot-team",
        "description": "IoT Device Monitoring Service",
    },
)

from ontology_sdk.functions import ontology_function
from ontology_sdk.grpc_client import DeviceMonitorClient

@ontology_function(
    name="check_device_health",
    description="检查设备健康状态",
    parameters={"device_rid": "string"},
    returns="DeviceHealthReport",
)
async def check_device_health(device_rid: str) -> dict:
    async with AsyncDeviceMonitorClient("device-monitor:50051") as client:
        device = await client.stub.GetDevice(
            pb2.GetDeviceRequest(
                device_rid=device_rid,
                include_latest_telemetry=True,
            )
        )

        # 分析遥测数据，生成健康报告
        telemetry = device.latest_telemetry
        anomalies = []
        for dp in telemetry:
            if dp.metric_name == "temperature" and dp.value > 80:
                anomalies.append(f"High temperature: {dp.value}°C")
            if dp.metric_name == "cpu_usage" and dp.value > 95:
                anomalies.append(f"High CPU: {dp.value}%")

        return {
            "device_rid": device_rid,
            "status": "HEALTHY" if not anomalies else "UNHEALTHY",
            "anomalies": anomalies,
            "last_seen": device.device.last_seen.ToDatetime().isoformat(),
        }