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

Address remaining mypy complaints

Browse source 
Add type hints for the variable that mypy couldn't figure out by context.

Also, one more field should have been marked as Optional[].
This commit is contained in:
sparky8512 2022-09-10 16:19:04 -07:00
parent 8650687f69
commit 6a186b6e3b
2 changed files with 15 additions and 13 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
dish_common.py
View file

@ -16,6 +16,7 @@ from datetime import timezone
import logging import logging
import re import re
import time import time
from typing import List
import grpc import grpc
@ -23,11 +24,11 @@ import starlink_grpc
BRACKETS_RE = re.compile(r"([^[]*)(\[((\d+),|)(\d*)\]|)$") BRACKETS_RE = re.compile(r"([^[]*)(\[((\d+),|)(\d*)\]|)$")
LOOP_TIME_DEFAULT = 0 LOOP_TIME_DEFAULT = 0
STATUS_MODES = ["status", "obstruction_detail", "alert_detail", "location"] STATUS_MODES: List[str] = ["status", "obstruction_detail", "alert_detail", "location"]
HISTORY_STATS_MODES = [ HISTORY_STATS_MODES: List[str] = [
"ping_drop", "ping_run_length", "ping_latency", "ping_loaded_latency", "usage" "ping_drop", "ping_run_length", "ping_latency", "ping_loaded_latency", "usage"
] ]
UNGROUPED_MODES = [] UNGROUPED_MODES: List[str] = []
def create_arg_parser(output_description, bulk_history=True): def create_arg_parser(output_description, bulk_history=True):

21
starlink_grpc.py
View file

@ -362,7 +362,7 @@ period.
from itertools import chain from itertools import chain
import math import math
import statistics import statistics
from typing import Any, Dict, Optional, Sequence, Tuple, get_type_hints from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple, get_type_hints
try: try:
from typing import TypedDict, get_args from typing import TypedDict, get_args
except ImportError: except ImportError:
@ -476,7 +476,7 @@ PingLatencyDict = TypedDict(
"deciles_all_ping_latency[]": Sequence[float], "deciles_all_ping_latency[]": Sequence[float],
"mean_full_ping_latency": float, "mean_full_ping_latency": float,
"deciles_full_ping_latency[]": Sequence[float], "deciles_full_ping_latency[]": Sequence[float],
"stdev_full_ping_latency": float, "stdev_full_ping_latency": Optional[float],
}) })
LoadedLatencyDict = TypedDict( LoadedLatencyDict = TypedDict(
@ -983,6 +983,7 @@ def _compute_sample_range(history, parse_samples: int, start: Optional[int] = No
# Set the range for the requested set of samples. This will iterate # Set the range for the requested set of samples. This will iterate
# sample index in order from oldest to newest. # sample index in order from oldest to newest.
sample_range: Iterable[int]
if start_offset < end_offset: if start_offset < end_offset:
sample_range = range(start_offset, end_offset) sample_range = range(start_offset, end_offset)
else: else:
@ -1030,7 +1031,7 @@ def concatenate_history(history1, history2, samples1: int = -1, start1: Optional
print("WARNING: Appending discontiguous samples. Polling interval probably too short.") print("WARNING: Appending discontiguous samples. Polling interval probably too short.")
new_samples = size2 new_samples = size2
unwrapped = UnwrappedHistory() unwrapped: Any = UnwrappedHistory()
for field in HISTORY_FIELDS: for field in HISTORY_FIELDS:
setattr(unwrapped, field, []) setattr(unwrapped, field, [])
unwrapped.unwrapped = True unwrapped.unwrapped = True
@ -1204,9 +1205,9 @@ def history_stats(
usage_down = 0.0 usage_down = 0.0
usage_up = 0.0 usage_up = 0.0
rtt_full = [] rtt_full: List[float] = []
rtt_all = [] rtt_all: List[Tuple[float, float]] = []
rtt_buckets = [[] for _ in range(15)] rtt_buckets: List[List[float]] = [[] for _ in range(15)]
for i in sample_range: for i in sample_range:
d = history.pop_ping_drop_rate[i] d = history.pop_ping_drop_rate[i]
@ -1274,10 +1275,10 @@ def history_stats(
accum_value += sum(x[0] for x in items) accum_value += sum(x[0] for x in items)
return accum_value / total_weight, result return accum_value / total_weight, result
bucket_samples = [] bucket_samples: List[int] = []
bucket_min = [] bucket_min: List[Optional[float]] = []
bucket_median = [] bucket_median: List[Optional[float]] = []
bucket_max = [] bucket_max: List[Optional[float]] = []
for bucket in rtt_buckets: for bucket in rtt_buckets:
if bucket: if bucket:
bucket_samples.append(len(bucket)) bucket_samples.append(len(bucket))