jeff/starlink-grpc-tools
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Provide type hints for places where ChannelContext is used

Browse source
This commit is contained in:
Jack Boswell (boswelja) 2022-08-20 12:29:48 +12:00
parent 75a8b875c4
commit 4e91755200
1 changed files with 17 additions and 16 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

33
starlink_grpc.py
View file

@ -343,6 +343,7 @@ period.
from itertools import chain from itertools import chain
import math import math
import statistics import statistics
from typing import Tuple
import grpc import grpc
@ -395,24 +396,24 @@ class ChannelContext:
`close()` should be called on the object when it is no longer `close()` should be called on the object when it is no longer
in use. in use.
""" """
def __init__(self, target=None): def __init__(self, target: str | None = None) -> None:
self.channel = None self.channel = None
self.target = "192.168.100.1:9200" if target is None else target self.target = "192.168.100.1:9200" if target is None else target
def get_channel(self): def get_channel(self) -> Tuple[grpc.Channel, bool]:
reused = True reused = True
if self.channel is None: if self.channel is None:
self.channel = grpc.insecure_channel(self.target) self.channel = grpc.insecure_channel(self.target)
reused = False reused = False
return self.channel, reused return self.channel, reused
def close(self): def close(self) -> None:
if self.channel is not None: if self.channel is not None:
self.channel.close() self.channel.close()
self.channel = None self.channel = None
def call_with_channel(function, *args, context=None, **kwargs): def call_with_channel(function, *args, context: ChannelContext = None, **kwargs):
"""Call a function with a channel object. """Call a function with a channel object.
Args: Args:
@ -436,7 +437,7 @@ def call_with_channel(function, *args, context=None, **kwargs):
raise raise
def status_field_names(context=None): def status_field_names(context: ChannelContext | None = None):
"""Return the field names of the status data. """Return the field names of the status data.
Note: Note:
@ -490,7 +491,7 @@ def status_field_names(context=None):
], alert_names ], alert_names
def status_field_types(context=None): def status_field_types(context: ChannelContext | None = None):
"""Return the field types of the status data. """Return the field types of the status data.
Return the type classes for each field. For sequence types, the type of Return the type classes for each field. For sequence types, the type of
@ -540,7 +541,7 @@ def status_field_types(context=None):
], [bool] * len(dish_pb2.DishAlerts.DESCRIPTOR.fields) ], [bool] * len(dish_pb2.DishAlerts.DESCRIPTOR.fields)
def get_status(context=None): def get_status(context: ChannelContext | None = None):
"""Fetch status data and return it in grpc structure format. """Fetch status data and return it in grpc structure format.
Args: Args:
@ -562,7 +563,7 @@ def get_status(context=None):
return call_with_channel(grpc_call, context=context) return call_with_channel(grpc_call, context=context)
def get_id(context=None): def get_id(context: ChannelContext | None = None):
"""Return the ID from the dish status information. """Return the ID from the dish status information.
Args: Args:
@ -583,7 +584,7 @@ def get_id(context=None):
raise GrpcError(e) raise GrpcError(e)
def status_data(context=None): def status_data(context: ChannelContext | None = None):
"""Fetch current status data. """Fetch current status data.
Args: Args:
@ -811,7 +812,7 @@ def history_stats_field_types():
] ]
def get_history(context=None): def get_history(context: ChannelContext | None = None):
"""Fetch history data and return it in grpc structure format. """Fetch history data and return it in grpc structure format.
Args: Args:
@ -939,7 +940,7 @@ def concatenate_history(history1, history2, samples1=-1, start1=None, verbose=Fa
return unwrapped return unwrapped
def history_bulk_data(parse_samples, start=None, verbose=False, context=None, history=None): def history_bulk_data(parse_samples, start=None, verbose=False, context: ChannelContext | None = None, history=None):
"""Fetch history data for a range of samples. """Fetch history data for a range of samples.
Args: Args:
@ -1018,7 +1019,7 @@ def history_ping_stats(parse_samples, verbose=False, context=None):
return history_stats(parse_samples, verbose=verbose, context=context)[0:3] return history_stats(parse_samples, verbose=verbose, context=context)[0:3]
def history_stats(parse_samples, start=None, verbose=False, context=None, history=None): def history_stats(parse_samples, start=None, verbose=False, context: ChannelContext | None = None, history=None):
"""Fetch, parse, and compute ping and usage stats. """Fetch, parse, and compute ping and usage stats.
Note: Note:
@ -1204,7 +1205,7 @@ def history_stats(parse_samples, start=None, verbose=False, context=None, histor
} }
def get_obstruction_map(context=None): def get_obstruction_map(context: ChannelContext | None = None):
"""Fetch obstruction map data and return it in grpc structure format. """Fetch obstruction map data and return it in grpc structure format.
Args: Args:
@ -1227,7 +1228,7 @@ def get_obstruction_map(context=None):
return call_with_channel(grpc_call, context=context) return call_with_channel(grpc_call, context=context)
def obstruction_map(context=None): def obstruction_map(context: ChannelContext | None = None):
"""Fetch current obstruction map data. """Fetch current obstruction map data.
Args: Args:
@ -1252,7 +1253,7 @@ def obstruction_map(context=None):
return tuple((map_data.snr[i:i + cols]) for i in range(0, cols * map_data.num_rows, cols)) return tuple((map_data.snr[i:i + cols]) for i in range(0, cols * map_data.num_rows, cols))
def reboot(context=None): def reboot(context: ChannelContext | None = None):
"""Request dish reboot operation. """Request dish reboot operation.
Args: Args:
@ -1278,7 +1279,7 @@ def reboot(context=None):
raise GrpcError(e) raise GrpcError(e)
def set_stow_state(unstow=False, context=None): def set_stow_state(unstow=False, context: ChannelContext | None = None):
"""Request dish stow or unstow operation. """Request dish stow or unstow operation.
Args: Args: