mirror of https://github.com/DJ2LS/FreeDATA.git
108 lines
4.0 KiB
Python
108 lines
4.0 KiB
Python
"""
|
|
Created on 05.11.23
|
|
|
|
@author: DJ2LS
|
|
"""
|
|
|
|
import requests
|
|
import json
|
|
import structlog
|
|
|
|
log = structlog.get_logger("explorer")
|
|
|
|
|
|
class Explorer:
|
|
"""Pushes station and last heard data to the FreeDATA explorer.
|
|
|
|
This class collects station information, including callsign, gridsquare,
|
|
frequency, signal strength, version, bandwidth, beacon status, and
|
|
last heard stations, and pushes this data to the FreeDATA explorer API.
|
|
"""
|
|
|
|
def __init__(self, ctx):
|
|
"""Initializes the Explorer.
|
|
|
|
This method initializes the Explorer with the modem version,
|
|
configuration manager, state manager, and the URL of the FreeDATA
|
|
explorer API.
|
|
|
|
Args:
|
|
modem_version (str): The version of the FreeDATA modem.
|
|
config_manager (ConfigManager): The configuration manager object.
|
|
states (StateManager): The state manager object.
|
|
"""
|
|
self.ctx = ctx
|
|
self.modem_version = self.ctx.constants.MODEM_VERSION
|
|
self.explorer_url = self.ctx.constants.EXPLORER_API_URL
|
|
|
|
def push(self):
|
|
"""Pushes station and last heard data to the explorer.
|
|
|
|
This method collects station information from the configuration and
|
|
state manager, formats it as JSON, and sends it to the FreeDATA
|
|
explorer API. It includes error handling and logging for successful
|
|
pushes, failed pushes, and connection issues.
|
|
"""
|
|
|
|
frequency = 0 if self.ctx.state_manager.radio_frequency is None else self.ctx.state_manager.radio_frequency
|
|
band = "USB"
|
|
callsign = f"{self.ctx.config_manager.config['STATION']['mycall']}-{self.ctx.config_manager.config['STATION']['myssid']}"
|
|
gridsquare = str(self.ctx.config_manager.config["STATION"]["mygrid"])
|
|
version = str(self.modem_version)
|
|
bandwidth = str(self.ctx.config_manager.config["MODEM"]["maximum_bandwidth"])
|
|
beacon = str(self.ctx.state_manager.is_beacon_running)
|
|
strength = str(self.ctx.state_manager.s_meter_strength)
|
|
away_from_key = str(self.ctx.state_manager.is_away_from_key)
|
|
|
|
# Stop pushing if default callsign
|
|
if callsign in ["AA1AAA-1", "XX1XXX-6"]:
|
|
return
|
|
|
|
headers = {"Content-Type": "application/json"}
|
|
station_data = {
|
|
"callsign": callsign,
|
|
"gridsquare": gridsquare,
|
|
"frequency": frequency,
|
|
"strength": strength,
|
|
"band": band,
|
|
"version": version,
|
|
"bandwidth": bandwidth,
|
|
"beacon": beacon,
|
|
"lastheard": [],
|
|
"away_from_key": away_from_key,
|
|
}
|
|
|
|
for i in self.ctx.state_manager.heard_stations:
|
|
try:
|
|
callsign = i[0]
|
|
grid = i[1]
|
|
timestamp = i[2]
|
|
frequency = i[6]
|
|
try:
|
|
snr = i[4].split("/")[1] if isinstance(i[4], str) and "/" in i[4] else str(i[4])
|
|
except Exception as e:
|
|
snr = "N/A"
|
|
log.warning("[EXPLORER] SNR parsing failed", e=e)
|
|
station_data["lastheard"].append({
|
|
"callsign": callsign,
|
|
"grid": grid,
|
|
"snr": snr,
|
|
"timestamp": timestamp,
|
|
"frequency": frequency,
|
|
})
|
|
except Exception as e:
|
|
log.debug("[EXPLORER] not publishing station", e=e)
|
|
station_data = json.dumps(station_data)
|
|
try:
|
|
response = requests.post(self.explorer_url, json=station_data, headers=headers)
|
|
if response.status_code == 200:
|
|
log.info("[EXPLORER] Data pushed successfully")
|
|
else:
|
|
log.error(
|
|
"[EXPLORER] Failed to push data",
|
|
status_code=response.status_code,
|
|
response_text=response.text,
|
|
)
|
|
except requests.exceptions.RequestException as e:
|
|
log.warning("[EXPLORER] Connection lost", e=e)
|