MCPcopy Create free account
hub / github.com/Tong89/smartNode / fetch_group

Function fetch_group

backend/physics/tle_source.py:218–270  ·  view source on GitHub ↗

按星座分组名称拉取 TLE 条目。 首先尝试从 Celestrak 在线拉取;若拉取失败或超时,则回退到本地缓存。 若既无网络也无缓存,返回空列表并记录警告。 参数 ---- group : str 星座/分组标识,例如 ``"starlink"``、``"gps-ops"``、``"stations"``。 支持别名映射(见 ``_GROUP_ALIASES``)。 force_refresh : bool 为 True 时即使缓存未过期也强制拉取,默认 False。 返回

(
    group: str,
    force_refresh: bool = False,
)

Source from the content-addressed store, hash-verified

216# ---------------------------------------------------------------------------
217
218def fetch_group(
219 group: str,
220 force_refresh: bool = False,
221) -> List[TleEntry]:
222 """按星座分组名称拉取 TLE 条目。
223
224 首先尝试从 Celestrak 在线拉取;若拉取失败或超时,则回退到本地缓存。
225 若既无网络也无缓存,返回空列表并记录警告。
226
227 参数
228 ----
229 group : str
230 星座/分组标识,例如 ``"starlink"``、``"gps-ops"``、``"stations"``。
231 支持别名映射(见 ``_GROUP_ALIASES``)。
232 force_refresh : bool
233 为 True 时即使缓存未过期也强制拉取,默认 False。
234
235 返回
236 ----
237 list of TleEntry
238 每条记录为 (name, line1, line2, epoch_utc) 元组。
239 """
240 canonical = _GROUP_ALIASES.get(group, group)
241
242 # 检查缓存新鲜度
243 if not force_refresh and _cache_is_fresh(canonical):
244 logger.debug("使用新鲜 TLE 缓存: group=%s", canonical)
245 cached = _read_cache(canonical)
246 if cached:
247 return _parse_tle_text(cached)
248
249 # 尝试在线拉取
250 url = _CELESTRAK_TLE_URL.format(group=canonical)
251 try:
252 raw_text = _http_get(url, timeout=_FETCH_TIMEOUT)
253 if raw_text and len(raw_text.strip()) > 0:
254 _write_cache(canonical, raw_text)
255 return _parse_tle_text(raw_text)
256 else:
257 logger.warning("Celestrak 返回空响应: group=%s url=%s", canonical, url)
258 except urllib.error.URLError as exc:
259 logger.warning("Celestrak 拉取失败 (URLError): group=%s, error=%s", canonical, exc)
260 except Exception as exc: # pragma: no cover
261 logger.warning("Celestrak 拉取异常: group=%s, error=%s", canonical, exc)
262
263 # 回退到缓存
264 cached = _read_cache(canonical)
265 if cached:
266 logger.info("网络不可用,使用离线 TLE 缓存: group=%s", canonical)
267 return _parse_tle_text(cached)
268
269 logger.warning("无法获取 TLE:在线拉取失败且无本地缓存: group=%s", canonical)
270 return []
271
272
273def fetch_by_catnr(catnr: int, force_refresh: bool = False) -> List[TleEntry]:

Callers 1

tle_importFunction · 0.90

Calls 6

_cache_is_freshFunction · 0.85
_read_cacheFunction · 0.85
_parse_tle_textFunction · 0.85
_http_getFunction · 0.85
_write_cacheFunction · 0.85
getMethod · 0.45

Tested by

no test coverage detected