proto/
├── onto/
│   ├── common/v1/         # Common type definitions
│   │   ├── types.proto    # RID, Timestamp, Pagination, etc.
│   │   └── errors.proto   # Unified error codes
│   ├── ontology/v1/       # Ontology service
│   ├── action/v1/         # Action service
│   ├── query/v1/          # Query service
│   ├── subscription/v1/   # Subscription service
│   └── custom/v1/         # Custom extension services

// 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;

// Device Monitoring Service
service DeviceMonitorService {
    // Single device query
    rpc GetDevice(GetDeviceRequest) returns (GetDeviceResponse);

    // Batch device query
    rpc ListDevices(ListDevicesRequest) returns (ListDevicesResponse);

    // Report device telemetry data
    rpc ReportTelemetry(ReportTelemetryRequest) returns (ReportTelemetryResponse);

    // Batch report (client streaming)
    rpc BatchReportTelemetry(stream TelemetryDataPoint) returns (BatchReportResponse);

    // Real-time telemetry stream (server streaming)
    rpc StreamTelemetry(StreamTelemetryRequest) returns (stream TelemetryDataPoint);

    // Bidirectional streaming: real-time commands and responses
    rpc DeviceCommandChannel(stream DeviceCommand) returns (stream CommandResponse);

    // Trigger device alert
    rpc TriggerAlert(TriggerAlertRequest) returns (TriggerAlertResponse);

    // Query alert history
    rpc ListAlerts(ListAlertsRequest) returns (ListAlertsResponse);
}

// ---- Message definitions ----

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;
    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;
}

# Install gRPC tools
pip install grpcio grpcio-tools

# Generate Python code
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

# Generated files:
# python-sdk/generated/onto/device/v1/
# ├── device_service_pb2.py       # Message classes
# ├── device_service_pb2.pyi      # Type stubs
# └── device_service_pb2_grpc.py  # Service stubs and 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
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):
    """Device monitoring gRPC service implementation"""

    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):
        """Query a single device"""
        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 ReportTelemetry(self, request, context):
        """Report telemetry data"""
        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):
        """Client streaming: batch telemetry report"""
        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):
        """Server streaming: real-time telemetry push"""
        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()  # Heartbeat
        finally:
            del self._telemetry_subscribers[subscriber_id]
            logger.info(f"Telemetry subscriber disconnected: {subscriber_id}")

    async def DeviceCommandChannel(self, request_iterator, context):
        """Bidirectional streaming: device command channel"""
        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):
        """Trigger alert"""
        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):
        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 = 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):
        await asyncio.sleep(0.1)
        return {"status": "ok"}

    def _to_proto_device(self, device) -> pb2.Device:
        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

logger = logging.getLogger(__name__)


async def serve():
    """Start gRPC server"""

    device_repo = DeviceRepository()
    telemetry_store = TelemetryStore()
    alert_manager = AlertManager()

    server = grpc.aio.server(
        options=[
            ("grpc.max_send_message_length", 50 * 1024 * 1024),
            ("grpc.max_receive_message_length", 50 * 1024 * 1024),
            ("grpc.keepalive_time_ms", 30000),
            ("grpc.keepalive_timeout_ms", 10000),
            ("grpc.keepalive_permit_without_calls", True),
        ],
    )

    servicer = DeviceMonitorServicer(device_repo, telemetry_store, alert_manager)
    pb2_grpc.add_DeviceMonitorServiceServicer_to_server(servicer, server)

    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)

    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())

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:
    """Device monitoring gRPC client"""

    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:
            return self.stub.GetDevice(request, timeout=5.0)
        except grpc.RpcError as e:
            if e.code() == grpc.StatusCode.NOT_FOUND:
                return None
            raise

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

    def __enter__(self):
        return self

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

import grpc
import asyncio
from typing import AsyncIterator


class AsyncDeviceMonitorClient:
    """Device monitoring async gRPC client"""

    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]:
        """Stream telemetry data"""
        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 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} exceeds threshold 80",
                ))

asyncio.run(main())

class AuthInterceptor(grpc.aio.ServerInterceptor):
    """Server-side authentication interceptor"""

    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 = 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:]

        try:
            user_context = await self.token_validator.validate(token)
            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.mark.asyncio
async def test_get_device_found(servicer):
    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)
    context = MagicMock()
    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):
    servicer.device_repo.get.return_value = None
    request = pb2.GetDeviceRequest(device_rid="ri.device.999")
    context = MagicMock()

    await servicer.GetDevice(request, context)
    context.set_code.assert_called_once_with(grpc.StatusCode.NOT_FOUND)

@pytest.fixture(scope="module")
async def grpc_channel():
    channel = grpc.aio.insecure_channel("localhost:50051")
    yield channel
    await channel.close()


@pytest.mark.integration
@pytest.mark.asyncio
async def test_full_workflow(grpc_channel):
    stub = pb2_grpc.DeviceMonitorServiceStub(grpc_channel)

    # 1. Report telemetry
    ts = Timestamp()
    ts.GetCurrentTime()
    report = 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.accepted_count == 1

    # 2. Query device
    device_resp = await stub.GetDevice(
        pb2.GetDeviceRequest(device_rid="ri.device.test-001", include_latest_telemetry=True)
    )
    assert device_resp.device.device_rid == "ri.device.test-001"

    # 3. Trigger alert
    alert_resp = await stub.TriggerAlert(
        pb2.TriggerAlertRequest(
            device_rid="ri.device.test-001",
            alert_type="TEMPERATURE_HIGH",
            severity="CRITICAL",
            message="Temperature exceeds threshold",
        )
    )
    assert alert_resp.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

@ontology_function(
    name="check_device_health",
    description="Check device health status",
    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)
        )

        anomalies = []
        for dp in device.latest_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(),
        }