From db83a7f0428c9c8b8759beedc2e327b1cdd04766 Mon Sep 17 00:00:00 2001 From: sparky8512 <76499194+sparky8512@users.noreply.github.com> Date: Thu, 4 Feb 2021 20:02:45 -0800 Subject: [PATCH] Switch generated protobuf module imports around The way I had them, it was hiding the fact that there was no explicit import for the spacex.api.device.dish_pb2 module. --- dump_dish_status.py | 13 +++++++------ starlink_grpc.py | 27 ++++++++++++++------------- 2 files changed, 21 insertions(+), 19 deletions(-) diff --git a/dump_dish_status.py b/dump_dish_status.py index fd393e2..4992d05 100644 --- a/dump_dish_status.py +++ b/dump_dish_status.py @@ -1,16 +1,17 @@ #!/usr/bin/python3 -"""Simple example of get_status request use grpc call directly.""" +"""Simple example of get_status request using grpc call directly.""" import grpc -import spacex.api.device.device_pb2 -import spacex.api.device.device_pb2_grpc +from spacex.api.device import device_pb2 +from spacex.api.device import device_pb2_grpc +from spacex.api.device import dish_pb2 # Note that if you remove the 'with' clause here, you need to separately # call channel.close() when you're done with the gRPC connection. with grpc.insecure_channel("192.168.100.1:9200") as channel: - stub = spacex.api.device.device_pb2_grpc.DeviceStub(channel) - response = stub.Handle(spacex.api.device.device_pb2.Request(get_status={})) + stub = device_pb2_grpc.DeviceStub(channel) + response = stub.Handle(device_pb2.Request(get_status={})) # Dump everything print(response) @@ -20,4 +21,4 @@ print("Software version:", response.dish_get_status.device_info.software_version # Check if connected print("Connected" if response.dish_get_status.state == - spacex.api.device.dish_pb2.DishState.CONNECTED else "Not connected") + dish_pb2.DishState.CONNECTED else "Not connected") diff --git a/starlink_grpc.py b/starlink_grpc.py index 5054edc..3c72765 100644 --- a/starlink_grpc.py +++ b/starlink_grpc.py @@ -300,8 +300,9 @@ import statistics import grpc -import spacex.api.device.device_pb2 -import spacex.api.device.device_pb2_grpc +from spacex.api.device import device_pb2 +from spacex.api.device import device_pb2_grpc +from spacex.api.device import dish_pb2 class GrpcError(Exception): @@ -352,7 +353,7 @@ def status_field_names(): names, and obstruction detail field names, in that order. """ alert_names = [] - for field in spacex.api.device.dish_pb2.DishAlerts.DESCRIPTOR.fields: + for field in dish_pb2.DishAlerts.DESCRIPTOR.fields: alert_names.append("alert_" + field.name) return [ @@ -406,7 +407,7 @@ def status_field_types(): ], [ float, # wedges_fraction_obstructed[] float, # valid_s - ], [bool] * len(spacex.api.device.dish_pb2.DishAlerts.DESCRIPTOR.fields) + ], [bool] * len(dish_pb2.DishAlerts.DESCRIPTOR.fields) def get_status(context=None): @@ -423,15 +424,15 @@ def get_status(context=None): """ if context is None: with grpc.insecure_channel("192.168.100.1:9200") as channel: - stub = spacex.api.device.device_pb2_grpc.DeviceStub(channel) - response = stub.Handle(spacex.api.device.device_pb2.Request(get_status={})) + stub = device_pb2_grpc.DeviceStub(channel) + response = stub.Handle(device_pb2.Request(get_status={})) return response.dish_get_status while True: channel, reused = context.get_channel() try: - stub = spacex.api.device.device_pb2_grpc.DeviceStub(channel) - response = stub.Handle(spacex.api.device.device_pb2.Request(get_status={})) + stub = device_pb2_grpc.DeviceStub(channel) + response = stub.Handle(device_pb2.Request(get_status={})) return response.dish_get_status except grpc.RpcError: context.close() @@ -495,7 +496,7 @@ def status_data(context=None): "id": status.device_info.id, "hardware_version": status.device_info.hardware_version, "software_version": status.device_info.software_version, - "state": spacex.api.device.dish_pb2.DishState.Name(status.state), + "state": dish_pb2.DishState.Name(status.state), "uptime": status.device_state.uptime_s, "snr": status.snr, "seconds_to_first_nonempty_slot": status.seconds_to_first_nonempty_slot, @@ -680,15 +681,15 @@ def get_history(context=None): """ if context is None: with grpc.insecure_channel("192.168.100.1:9200") as channel: - stub = spacex.api.device.device_pb2_grpc.DeviceStub(channel) - response = stub.Handle(spacex.api.device.device_pb2.Request(get_history={})) + stub = device_pb2_grpc.DeviceStub(channel) + response = stub.Handle(device_pb2.Request(get_history={})) return response.dish_get_history while True: channel, reused = context.get_channel() try: - stub = spacex.api.device.device_pb2_grpc.DeviceStub(channel) - response = stub.Handle(spacex.api.device.device_pb2.Request(get_history={})) + stub = device_pb2_grpc.DeviceStub(channel) + response = stub.Handle(device_pb2.Request(get_history={})) return response.dish_get_history except grpc.RpcError: context.close()