4f4037b4ae
- nats_consumer/grpc_server: annotate nc/sub/server attrs (were inferred as None-only from __init__, breaking on the real assignment later). - active_read: narrow on 'if names:' instead of the derived 'persisted' bool so mypy sees names as non-Optional inside the publish_reading call. - pyproject: exclude ingestor/proto/ (protoc/grpcio-tools output) from mypy's file walk and skip following it as an import - it's generated, not hand-maintained, and mypy can't see its runtime-built message classes.
69 lines
2.6 KiB
Python
69 lines
2.6 KiB
Python
"""gRPC input adapter for on-demand tag reads (contract C4).
|
|
|
|
Thin adapter: unpack the request, call ActiveReadEngine.read_now, map the
|
|
result onto ReadTagResponse. Internal-only for now -- no auth/TLS, per the
|
|
feature owner (no consumer exists yet, likely a frontend later).
|
|
"""
|
|
|
|
import grpc
|
|
from sientia_do.observability.logger import Logger
|
|
|
|
from ingestor.engines.active_read import ActiveReadEngine
|
|
from ingestor.proto import opc_gateway_pb2, opc_gateway_pb2_grpc
|
|
|
|
|
|
class ReadTagServicer(opc_gateway_pb2_grpc.OpcGatewayServicer):
|
|
def __init__(self, active_read: ActiveReadEngine, logger: Logger, slot_lookup=None):
|
|
self.active_read = active_read
|
|
self.logger = logger
|
|
# ponytail: slot_lookup wiring (address -> names from the current slot)
|
|
# lands in Task 18 (app wiring); optional callable keeps this servicer usable now.
|
|
self.slot_lookup = slot_lookup
|
|
|
|
async def ReadTag(self, request, context):
|
|
names = None
|
|
if self.slot_lookup is not None:
|
|
names = self.slot_lookup(request.server_id, request.addr)
|
|
try:
|
|
result = await self.active_read.read_now(request.server_id, request.addr, names=names)
|
|
except Exception as e:
|
|
self.logger.error(f'ReadTag failed for {request.server_id}:{request.addr}: {e}')
|
|
await context.abort(grpc.StatusCode.INTERNAL, f'read failed: {e}')
|
|
return
|
|
return opc_gateway_pb2.ReadTagResponse(
|
|
value=str(result['value']),
|
|
opc_status_code=result['opc_status_code'] or '',
|
|
source_timestamp=result['source_timestamp'] or '',
|
|
server_timestamp=result['server_timestamp'] or '',
|
|
persisted=result['persisted'],
|
|
)
|
|
|
|
|
|
class GrpcServer:
|
|
def __init__(
|
|
self,
|
|
active_read: ActiveReadEngine,
|
|
logger: Logger,
|
|
port: int = 50051,
|
|
slot_lookup=None,
|
|
):
|
|
self.active_read = active_read
|
|
self.logger = logger
|
|
self.port = port
|
|
self.slot_lookup = slot_lookup
|
|
self.server: grpc.aio.Server | None = None
|
|
|
|
async def start(self) -> None:
|
|
self.server = grpc.aio.server()
|
|
opc_gateway_pb2_grpc.add_OpcGatewayServicer_to_server(
|
|
ReadTagServicer(self.active_read, self.logger, self.slot_lookup), self.server
|
|
)
|
|
self.server.add_insecure_port(f'[::]:{self.port}')
|
|
await self.server.start()
|
|
self.logger.info(f'GrpcServer started on port {self.port}')
|
|
|
|
async def stop(self) -> None:
|
|
if self.server is None:
|
|
return
|
|
await self.server.stop(grace=5)
|