Commit 8e513a8b authored by Shayan Hajipour's avatar Shayan Hajipour
Browse files

feat: basic dev of ecoc25 telemetry

parent 785f2570

manifests/nbiservice.yaml

+12 −0
Original line number Diff line number Diff line
@@ -44,6 +44,18 @@ spec: 
              value: "production"  # normal value is "production", change to "development" if developing

            - name: IETF_NETWORK_RENDERER

              value: "LIBYANG"

            - name: NBI_DATABASE

              value: "tfs_nbi"

            - name: CRDB_NAMESPACE

              value: "crdb"

            - name: CRDB_SQL_PORT

              value: "26257"

            - name: CRDB_USERNAME

              value: "tfs"

            - name: CRDB_PASSWORD

              value: "tfs123"

            - name: CRDB_SSLMODE

              value: "require"

          envFrom:

            - secretRef:

                name: kfk-kpi-data

proto/device.proto

+6 −5
Original line number Diff line number Diff line
@@ -24,6 +24,7 @@ service DeviceService { 
  rpc DeleteDevice         (context.DeviceId                          ) returns (context.Empty                                ) {}

  rpc GetInitialConfig     (context.DeviceId                          ) returns (context.DeviceConfig                         ) {}

  rpc MonitorDeviceKpi     (MonitoringSettings                        ) returns (context.Empty                                ) {}

  rpc SSETelemetrySubscribe(monitoring.SSEMonitoringSubscriptionConfig) returns (monitoring.SSEMonitoringSubscriptionResponse ) {}

}



message MonitoringSettings {

proto/monitoring.proto

+18 −0
Original line number Diff line number Diff line
@@ -172,3 +172,21 @@ message AlarmResponse { 
message AlarmList {

  repeated AlarmDescriptor alarm_descriptor = 1;

}



message SSEMonitoringSubscriptionConfig {

    enum ConfigType {

        Subscribe = 0;

        Unsubscribe = 1;

        GetTelemetry = 2;

    }

    context.DeviceId device_id = 1;

    ConfigType config_type = 2;

    string uri = 3;

    string sampling_interval = 4; // in seconds

    string identifier = 5;

}



message SSEMonitoringSubscriptionResponse {

    string identifier = 1;

    string uri = 2;

}

src/device/service/DeviceServiceServicerImpl.py

+29 −1
Original line number Diff line number Diff line
@@ -25,6 +25,7 @@ from common.proto.context_pb2 import ( 
)

from common.proto.device_pb2 import MonitoringSettings

from common.proto.device_pb2_grpc import DeviceServiceServicer

from common.proto.monitoring_pb2 import SSEMonitoringSubscriptionConfig, SSEMonitoringSubscriptionResponse

from common.tools.context_queries.Device import get_device

from common.tools.mutex_queues.MutexQueues import MutexQueues

from context.client.ContextClient import ContextClient
@@ -400,3 +401,30 @@ class DeviceServiceServicerImpl(DeviceServiceServicer): 
            return Empty()

        finally:

            self.mutex_queues.signal_done(device_uuid)



    def SSETelemetrySubscribe(self, request: SSEMonitoringSubscriptionConfig, context : grpc.ServicerContext) -> SSEMonitoringSubscriptionResponse:

        device_id = request.device_id.device_uuid.uuid

        config_type = request.config_type

        context_client = ContextClient()

        device = get_device(

            context_client, device_id, rw_copy=True, include_endpoints=False, include_components=False,

            include_config_rules=True)

        if device is None:

            raise NotFoundException('Device', device_id, extra_details='loading in ConfigureDevice')

        driver : _Driver = get_driver(self.driver_instance_cache, device)

        if config_type == SSEMonitoringSubscriptionConfig.Subscribe:

            r = driver.SubscribeState([(request.uri, 0, float(request.sampling_interval))])

            if len(r) != 1:

                raise OperationFailedException(

                    'SSETelemetrySubscribe', extra_details='Driver returned an unexpected number of responses: {:d}'.format(len(r))

                )

            sub_conf: dict = r[0]

            return SSEMonitoringSubscriptionResponse(identifier=sub_conf['identifier'], uri=sub_conf['uri'])

        if config_type == SSEMonitoringSubscriptionConfig.Unsubscribe:

            r = driver.UnsubscribeState([(request.identifier, 0, 0)])

            if len(r) != 1:

                raise OperationFailedException(

                    'SSETelemetrySubscribe', extra_details='Driver returned an unexpected number of responses: {:d}'.format(len(r))

                )

            return SSEMonitoringSubscriptionResponse()

src/device/service/driver_api/_Driver.py

+1 −1
Original line number Diff line number Diff line
@@ -139,7 +139,7 @@ class _Driver: 
        raise NotImplementedError()



    def SubscribeState(self, subscriptions: List[Tuple[str, float, float]]) -> \

            List[Union[bool, Exception]]:

            List[Union[bool, dict[str, Any], Exception]]:

        """ Subscribe to state information of entire device or

        selected resources. Subscriptions are incremental.

            Driver should keep track of requested resources.