Mirror/intercept
1
0
Fork 0
mirror of https://github.com/smittix/intercept.git synced 2026-04-26 15:50:00 -07:00
Code Issues Packages Projects Releases Wiki Activity

Fix ruff lint errors to unblock CI (import sorting, unused imports, style)

Browse Source
This commit is contained in:
James Smith
2026-03-20 13:51:30 +00:00
parent 90d39f12c1
commit 34e1d25069
14 changed files with 35 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
utils/ground_station/scheduler.py
View File

@@ -22,7 +22,6 @@ from __future__ import annotations
import json
import queue
import threading
import time
import uuid
from datetime import datetime, timedelta, timezone
from pathlib import Path
@@ -283,6 +282,7 @@ class GroundStationScheduler:
) -> list[ScheduledObservation]:
"""Predict passes for each profile and return ScheduledObservation list."""
from skyfield.api import load, wgs84
from utils.satellite_predict import predict_passes as _predict_passes
try:
@@ -525,8 +525,8 @@ class GroundStationScheduler:
def _attach_sigmf_consumer(self, bus, profile, obs_db_id: int | None) -> None:
"""Attach a SigMFConsumer for raw IQ recording."""
from utils.sigmf import SigMFMetadata
from utils.ground_station.consumers.sigmf_writer import SigMFConsumer
from utils.sigmf import SigMFMetadata
meta = SigMFMetadata(
sample_rate=profile.iq_sample_rate,
@@ -681,8 +681,9 @@ class GroundStationScheduler:
def _insert_observation_record(obs: ScheduledObservation, profile) -> int | None:
try:
from utils.database import get_db
from datetime import datetime, timezone
from utils.database import get_db
with get_db() as conn:
cur = conn.execute('''
INSERT INTO ground_station_observations
@@ -719,8 +720,9 @@ def _insert_event_record(obs_db_id: int | None, event_type: str, payload: str) -
if obs_db_id is None:
return
try:
from utils.database import get_db
from datetime import datetime, timezone
from utils.database import get_db
with get_db() as conn:
conn.execute('''
INSERT INTO ground_station_events (observation_id, event_type, payload_json, timestamp)
@@ -768,9 +770,10 @@ def _build_packet_event(payload, source: str) -> dict[str, Any]:
if parsed is None:
try:
from utils.satellite_telemetry import auto_parse
import base64
from utils.satellite_telemetry import auto_parse
for token in text.replace(',', ' ').split():
cleaned = token.strip()
if not cleaned or len(cleaned) < 8:
@@ -794,8 +797,9 @@ def _build_packet_event(payload, source: str) -> dict[str, Any]:
def _insert_recording_record(obs_db_id: int | None, meta_path: Path, data_path: Path, profile) -> None:
try:
from utils.database import get_db
from datetime import datetime, timezone
from utils.database import get_db
size = data_path.stat().st_size if data_path.exists() else 0
with get_db() as conn:
conn.execute('''
@@ -827,9 +831,10 @@ def _insert_output_record(
metadata: dict[str, Any] | None = None,
) -> int | None:
try:
from utils.database import get_db
from datetime import datetime, timezone
from utils.database import get_db
with get_db() as conn:
cur = conn.execute(
'''