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

Add data group for GPS location data

Browse source 
Add a new mode group, location, for physical location data. This requires access to the location data being enabled on the dish in order to return data; see README file for more details. At least for now, this functionality should be considered experimental.

This is being lumped into the "status" category for the purposes of database schema, but it's a separate grpc request from the other status data, and can fail at times when the other status request does not, so it has a separate function in the starlink_grpc module.
This commit is contained in:
sparky8512 2022-09-10 14:43:50 -07:00
parent 29c813b51a
commit 29699f0f59
3 changed files with 132 additions and 27 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
README.md
View file

@ -24,6 +24,14 @@ If you really care about the details here or wish to minimize your package requi
This step is no longer required, nor is it particularly recommended, so the details have been moved to [this Wiki article](https://github.com/sparky8512/starlink-grpc-tools/wiki/gRPC-Protocol-Modules).
### Enabling access to location data
This step is only required if you want to use the `location` data group with the grpc scripts. Note that it will allow any device on your local (home) network to access the physical location (GPS) data for your dish. If you are not comfortable with that, then do not enable it.
Access to location data must be enabled per dish and currently (2022-Sep), this can only be done using the Starlink mobile app, version 2022.09.0 or later. It cannot be done using the browser app. To enable access, you must be logged in to your Starlink account. You can log in by pressing the user icon in the upper left corner of the main screen of the app. Once logged in, from the main screen, select SETTINGS, then select ADVANCED, then select DEBUG DATA. Scroll down and you should see a toggle switch for "allow access on local network" in a section labelled STARLINK LOCATION, which should be off by default. Turn that switch on to enable access or off to disable it. This may move in the future, and there is no guarantee the ability to enable this feature will remain in the app.
Note that the Starlink mobile app can be pretty finicky and painfully slow. It's best to wait for the screens to load completely before going on to the next one.
## Usage
Of the 3 groups below, the grpc scripts are really the only ones being actively developed. The others are mostly by way of example of what could be done with the underlying data.

66
dish_common.py
View file

@ -23,7 +23,7 @@ import starlink_grpc
BRACKETS_RE = re.compile(r"([^[]*)(\[((\d+),|)(\d*)\]|)$")
LOOP_TIME_DEFAULT = 0
STATUS_MODES = ["status", "obstruction_detail", "alert_detail"]
STATUS_MODES = ["status", "obstruction_detail", "alert_detail", "location"]
HISTORY_STATS_MODES = [
"ping_drop", "ping_run_length", "ping_latency", "ping_loaded_latency", "usage"
]
@ -124,7 +124,11 @@ def run_arg_parser(parser, need_id=False, no_stdout_errors=False):
parser.error("Poll loops arg must be 2 or greater to be meaningful")
# for convenience, set flags for whether any mode in a group is selected
opts.status_mode = bool(set(STATUS_MODES).intersection(opts.mode))
status_set = set(STATUS_MODES)
opts.status_mode = bool(status_set.intersection(opts.mode))
status_set.remove("location")
# special group for any status mode other than location
opts.pure_status_mode = bool(status_set.intersection(opts.mode))
opts.history_stats_mode = bool(set(HISTORY_STATS_MODES).intersection(opts.mode))
opts.bulk_mode = "bulk_history" in opts.mode
@ -167,6 +171,7 @@ class GlobalState:
self.poll_count = 0
self.accum_history = None
self.first_poll = True
self.warn_once_location = True
def shutdown(self):
self.context.close()
@ -252,30 +257,41 @@ def add_data_numeric(data, category, add_item, add_sequence):
def get_status_data(opts, gstate, add_item, add_sequence):
if opts.status_mode:
timestamp = int(time.time())
try:
groups = starlink_grpc.status_data(context=gstate.context)
status_data, obstruct_detail, alert_detail = groups[0:3]
except starlink_grpc.GrpcError as e:
if "status" in opts.mode:
if opts.need_id and gstate.dish_id is None:
conn_error(opts, "Dish unreachable and ID unknown, so not recording state")
return 1, None
if opts.verbose:
print("Dish unreachable")
add_item("state", "DISH_UNREACHABLE", "status")
return 0, timestamp
conn_error(opts, "Failure getting status: %s", str(e))
return 1, None
if opts.need_id:
gstate.dish_id = status_data["id"]
del status_data["id"]
add_data = add_data_numeric if opts.numeric else add_data_normal
if "status" in opts.mode:
add_data(status_data, "status", add_item, add_sequence)
if "obstruction_detail" in opts.mode:
add_data(obstruct_detail, "status", add_item, add_sequence)
if "alert_detail" in opts.mode:
add_data(alert_detail, "status", add_item, add_sequence)
if opts.pure_status_mode or opts.need_id and gstate.dish_id is None:
try:
groups = starlink_grpc.status_data(context=gstate.context)
status_data, obstruct_detail, alert_detail = groups[0:3]
except starlink_grpc.GrpcError as e:
if "status" in opts.mode:
if opts.need_id and gstate.dish_id is None:
conn_error(opts, "Dish unreachable and ID unknown, so not recording state")
return 1, None
if opts.verbose:
print("Dish unreachable")
add_item("state", "DISH_UNREACHABLE", "status")
return 0, timestamp
conn_error(opts, "Failure getting status: %s", str(e))
return 1, None
if opts.need_id:
gstate.dish_id = status_data["id"]
del status_data["id"]
if "status" in opts.mode:
add_data(status_data, "status", add_item, add_sequence)
if "obstruction_detail" in opts.mode:
add_data(obstruct_detail, "status", add_item, add_sequence)
if "alert_detail" in opts.mode:
add_data(alert_detail, "status", add_item, add_sequence)
if "location" in opts.mode:
try:
location = starlink_grpc.location_data(context=gstate.context)
except starlink_grpc.GrpcError as e:
conn_error(opts, "Failure getting location: %s", str(e))
return 1, None
if location["latitude"] is None and gstate.warn_once_location:
logging.warning("Location data not enabled. See README for more details.")
gstate.warn_once_location = False
add_data(location, "status", add_item, add_sequence)
return 0, timestamp
elif opts.need_id and gstate.dish_id is None:
try:

85
starlink_grpc.py
View file

@ -143,6 +143,20 @@ their nature, but the field names are pretty self-explanatory.
: **alert_power_supply_thermal_throttle** : Alert corresponding with bit 9 (bit
mask 512) in *alerts*.
Location data
-------------
This group holds information about the physical location of the user terminal.
This group of fields should be considered EXPERIMENTAL, due to the requirement
to authorize access to location data on the user terminal.
: **latitude** : Latitude part of the current location, in degrees, or None if
location data is not available.
: **longitude** : Longitude part of the current location, in degrees, or None if
location data is not available.
: **altitude** : Altitude part of the current location, (probably) in meters, or
None if location data is not available.
General history data
--------------------
This set of fields contains data relevant to all the other history groups.
@ -414,7 +428,16 @@ ObstructionDict = TypedDict(
AlertDict = Dict[str, bool]
HistGeneralDict = TypedDict("HistGeneralDict", {"samples": int, "end_counter": int})
LocationDict = TypedDict("LocationDict", {
"latitude": Optional[float],
"longitude": Optional[float],
"altitude": Optional[float],
})
HistGeneralDict = TypedDict("HistGeneralDict", {
"samples": int,
"end_counter": int,
})
HistBulkDict = TypedDict(
"HistBulkDict", {
@ -464,7 +487,10 @@ LoadedLatencyDict = TypedDict(
"load_bucket_max_latency[]": Sequence[Optional[float]],
})
UsageDict = TypedDict("UsageDict", {"download_usage": int, "upload_usage": int})
UsageDict = TypedDict("UsageDict", {
"download_usage": int,
"upload_usage": int,
})
# For legacy reasons, there is a slight difference between the field names
# returned in the actual data vs the *_field_names functions. This is a map of
@ -750,6 +776,61 @@ def status_data(
}, alerts
def get_location(context: Optional[ChannelContext] = None):
"""Fetch location data and return it in grpc structure format.
Args:
context (ChannelContext): Optionally provide a channel for reuse
across repeated calls. If an existing channel is reused, the RPC
call will be retried at most once, since connectivity may have
been lost and restored in the time since it was last used.
Raises:
grpc.RpcError: Communication or service error.
"""
def grpc_call(channel):
if imports_pending:
resolve_imports(channel)
stub = device_pb2_grpc.DeviceStub(channel)
response = stub.Handle(device_pb2.Request(get_location={}), timeout=REQUEST_TIMEOUT)
return response.get_location
return call_with_channel(grpc_call, context=context)
def location_data(context: Optional[ChannelContext] = None) -> LocationDict:
"""Fetch current location data.
Args:
context (ChannelContext): Optionally provide a channel for reuse
across repeated calls.
Returns:
A dict mapping location data field names to their values. Values will
be set to None in the case that location request is not enabled (ie:
not authorized).
Raises:
GrpcError: Failed getting location info from the Starlink user terminal.
"""
try:
location = get_location(context)
except grpc.RpcError as e:
if isinstance(e, grpc.Call) and e.code() is grpc.StatusCode.PERMISSION_DENIED:
return {
"latitude": None,
"longitude": None,
"altitude": None,
}
raise GrpcError(e)
return {
"latitude": location.lla.lat,
"longitude": location.lla.lon,
"altitude": location.lla.alt,
}
def history_bulk_field_names():
"""Return the field names of the bulk history data.