|
1 | 1 | import os |
2 | 2 | from collections import defaultdict |
3 | 3 | from concurrent.futures import ThreadPoolExecutor |
4 | | -from typing import DefaultDict |
| 4 | +from typing import DefaultDict, Sequence, Mapping |
5 | 5 | from collections.abc import Iterable |
6 | 6 |
|
7 | 7 | from .constants import SECS_IN_DAY |
@@ -63,11 +63,11 @@ def __init__(self, auto_update=True, |
63 | 63 | self.dcbs_fetched_times = TimeRangeHolder() |
64 | 64 |
|
65 | 65 | self.dgps_delays = [] |
66 | | - self.ionex_maps: list[IonexMap] = [] |
67 | | - self.orbits: DefaultDict[str, list[PolyEphemeris]] = defaultdict(list) |
68 | | - self.qcom_polys: DefaultDict[str, list[PolyEphemeris]] = defaultdict(list) |
69 | | - self.navs: DefaultDict[str, list[GPSEphemeris | GLONASSEphemeris]] = defaultdict(list) |
70 | | - self.dcbs: DefaultDict[str, list[DCB]] = defaultdict(list) |
| 66 | + self.ionex_maps: Sequence[IonexMap] = [] |
| 67 | + self.orbits: DefaultDict[str, Sequence[PolyEphemeris]] = defaultdict(list) |
| 68 | + self.qcom_polys: DefaultDict[str, Sequence[PolyEphemeris]] = defaultdict(list) |
| 69 | + self.navs: DefaultDict[str, Sequence[GPSEphemeris | GLONASSEphemeris]] = defaultdict(list) |
| 70 | + self.dcbs: DefaultDict[str, Sequence[DCB]] = defaultdict(list) |
71 | 71 |
|
72 | 72 | self.cached_ionex: IonexMap | None = None |
73 | 73 | self.cached_dgps = None |
@@ -160,16 +160,16 @@ def get_dgps_corrections(self, time, recv_pos): |
160 | 160 | self.cached_dgps = latest_data |
161 | 161 | return latest_data |
162 | 162 |
|
163 | | - def add_qcom_polys(self, new_ephems: dict[str, list[Ephemeris]]): |
| 163 | + def add_qcom_polys(self, new_ephems: Mapping[str, Sequence[Ephemeris]]): |
164 | 164 | self._add_ephems(new_ephems, self.qcom_polys) |
165 | 165 |
|
166 | | - def add_orbits(self, new_ephems: dict[str, list[Ephemeris]]): |
| 166 | + def add_orbits(self, new_ephems: Mapping[str, Sequence[Ephemeris]]): |
167 | 167 | self._add_ephems(new_ephems, self.orbits) |
168 | 168 |
|
169 | | - def add_navs(self, new_ephems: dict[str, list[Ephemeris]]): |
| 169 | + def add_navs(self, new_ephems: Mapping[str, Sequence[Ephemeris]]): |
170 | 170 | self._add_ephems(new_ephems, self.navs) |
171 | 171 |
|
172 | | - def _add_ephems(self, new_ephems: dict[str, list[Ephemeris]], ephems_dict): |
| 172 | + def _add_ephems(self, new_ephems: Mapping[str, Sequence[Ephemeris]], ephems_dict): |
173 | 173 | for k, v in new_ephems.items(): |
174 | 174 | if len(v) > 0: |
175 | 175 | if self.clear_old_ephemeris: |
@@ -208,7 +208,7 @@ def download_and_parse(constellation, parse_rinex_nav_func): |
208 | 208 | end_day = GPSTime(time.week, SECS_IN_DAY * (1 + (time.tow // SECS_IN_DAY))) |
209 | 209 | self.navs_fetched_times.add(begin_day, end_day) |
210 | 210 |
|
211 | | - def download_parse_orbit(self, gps_time: GPSTime, skip_before_epoch=None) -> dict[str, list[Ephemeris]]: |
| 211 | + def download_parse_orbit(self, gps_time: GPSTime, skip_before_epoch=None) -> Mapping[str, Sequence[PolyEphemeris]]: |
212 | 212 | # Download multiple days to be able to polyfit at the start-end of the day |
213 | 213 | time_steps = [gps_time - SECS_IN_DAY, gps_time, gps_time + SECS_IN_DAY] |
214 | 214 | with ThreadPoolExecutor() as executor: |
|
0 commit comments