mirror of
https://github.com/craigerl/aprsd.git
synced 2026-10-09 16:50:31 -04:00
hemna
f30f7640ac
Replace the flat GPS fields on GPSPacket and its subclasses with a
composed Position value object (has-a instead of is-a), collapsing the
Packet -> GPSPacket -> {Beacon, MicE, Telemetry, Object, Weather}
data spine. Position owns the coordinates, symbol, movement/PHG data
and the conversion/formatting logic (zulu time, DDM, human_info).
- GPSPacket holds position: Optional[Position]; subclass defaults
preserved via _position_default() (ObjectPacket 'r', WeatherPacket '_')
- to_dict()/from_dict() flatten/unflatten so the on-disk JSON format
is unchanged; verified via objectstore save/load round-trip
- GPSPacket inherits DataClassJsonMixin directly because the
@dataclass_json decorator overwrites to_dict()/from_dict() and would
clobber the flattening overrides
- constructor is breaking: position=Position(...) instead of
latitude=.../longitude=...; updated tx.py, log.py and all tests
Full test suite (775) passes; ruff format + lint clean.
209 lines
8.8 KiB
Python
209 lines
8.8 KiB
Python
import json
|
|
import unittest
|
|
|
|
import aprslib
|
|
|
|
from aprsd import packets
|
|
from tests import fake
|
|
|
|
|
|
class TestWeatherPacket(unittest.TestCase):
|
|
"""Test WeatherPacket JSON serialization."""
|
|
|
|
def test_weather_packet_to_json(self):
|
|
"""Test WeatherPacket.to_json() method."""
|
|
packet = packets.WeatherPacket(
|
|
from_call=fake.FAKE_FROM_CALLSIGN,
|
|
to_call=fake.FAKE_TO_CALLSIGN,
|
|
position=packets.Position(
|
|
latitude=37.7749,
|
|
longitude=-122.4194,
|
|
symbol='_',
|
|
symbol_table='/',
|
|
comment='Test weather comment',
|
|
),
|
|
wind_speed=10.5,
|
|
wind_direction=180,
|
|
wind_gust=15.0,
|
|
temperature=72.5,
|
|
rain_1h=0.1,
|
|
rain_24h=0.5,
|
|
rain_since_midnight=0.3,
|
|
humidity=65,
|
|
pressure=1013.25,
|
|
)
|
|
|
|
json_str = packet.to_json()
|
|
self.assertIsInstance(json_str, str)
|
|
json_dict = json.loads(json_str)
|
|
self.assertEqual(json_dict['_type'], 'WeatherPacket')
|
|
self.assertEqual(json_dict['from_call'], fake.FAKE_FROM_CALLSIGN)
|
|
self.assertEqual(json_dict['to_call'], fake.FAKE_TO_CALLSIGN)
|
|
self.assertEqual(json_dict['latitude'], 37.7749)
|
|
self.assertEqual(json_dict['longitude'], -122.4194)
|
|
self.assertEqual(json_dict['symbol'], '_')
|
|
self.assertEqual(json_dict['wind_speed'], 10.5)
|
|
self.assertEqual(json_dict['wind_direction'], 180)
|
|
self.assertEqual(json_dict['wind_gust'], 15.0)
|
|
self.assertEqual(json_dict['temperature'], 72.5)
|
|
self.assertEqual(json_dict['rain_1h'], 0.1)
|
|
self.assertEqual(json_dict['rain_24h'], 0.5)
|
|
self.assertEqual(json_dict['rain_since_midnight'], 0.3)
|
|
self.assertEqual(json_dict['humidity'], 65)
|
|
self.assertEqual(json_dict['pressure'], 1013.25)
|
|
self.assertEqual(json_dict['comment'], 'Test weather comment')
|
|
|
|
def test_weather_packet_from_dict(self):
|
|
"""Test WeatherPacket.from_dict() method."""
|
|
packet_dict = {
|
|
'_type': 'WeatherPacket',
|
|
'from_call': fake.FAKE_FROM_CALLSIGN,
|
|
'to_call': fake.FAKE_TO_CALLSIGN,
|
|
'latitude': 37.7749,
|
|
'longitude': -122.4194,
|
|
'symbol': '_',
|
|
'symbol_table': '/',
|
|
'wind_speed': 10.5,
|
|
'wind_direction': 180,
|
|
'wind_gust': 15.0,
|
|
'temperature': 72.5,
|
|
'rain_1h': 0.1,
|
|
'rain_24h': 0.5,
|
|
'rain_since_midnight': 0.3,
|
|
'humidity': 65,
|
|
'pressure': 1013.25,
|
|
'comment': 'Test weather comment',
|
|
}
|
|
packet = packets.WeatherPacket.from_dict(packet_dict)
|
|
self.assertIsInstance(packet, packets.WeatherPacket)
|
|
self.assertEqual(packet.from_call, fake.FAKE_FROM_CALLSIGN)
|
|
self.assertEqual(packet.to_call, fake.FAKE_TO_CALLSIGN)
|
|
self.assertEqual(packet.position.latitude, 37.7749)
|
|
self.assertEqual(packet.position.longitude, -122.4194)
|
|
self.assertEqual(packet.position.symbol, '_')
|
|
self.assertEqual(packet.wind_speed, 10.5)
|
|
self.assertEqual(packet.wind_direction, 180)
|
|
self.assertEqual(packet.wind_gust, 15.0)
|
|
self.assertEqual(packet.temperature, 72.5)
|
|
self.assertEqual(packet.rain_1h, 0.1)
|
|
self.assertEqual(packet.rain_24h, 0.5)
|
|
self.assertEqual(packet.rain_since_midnight, 0.3)
|
|
self.assertEqual(packet.humidity, 65)
|
|
self.assertEqual(packet.pressure, 1013.25)
|
|
self.assertEqual(packet.position.comment, 'Test weather comment')
|
|
|
|
def test_weather_packet_round_trip(self):
|
|
"""Test WeatherPacket round-trip: to_json -> from_dict."""
|
|
original = packets.WeatherPacket(
|
|
from_call=fake.FAKE_FROM_CALLSIGN,
|
|
to_call=fake.FAKE_TO_CALLSIGN,
|
|
position=packets.Position(
|
|
latitude=37.7749,
|
|
longitude=-122.4194,
|
|
symbol='_',
|
|
symbol_table='/',
|
|
comment='Test weather comment',
|
|
),
|
|
wind_speed=10.5,
|
|
wind_direction=180,
|
|
wind_gust=15.0,
|
|
temperature=72.5,
|
|
rain_1h=0.1,
|
|
rain_24h=0.5,
|
|
rain_since_midnight=0.3,
|
|
humidity=65,
|
|
pressure=1013.25,
|
|
)
|
|
|
|
json_str = original.to_json()
|
|
packet_dict = json.loads(json_str)
|
|
restored = packets.WeatherPacket.from_dict(packet_dict)
|
|
self.assertEqual(restored.from_call, original.from_call)
|
|
self.assertEqual(restored.to_call, original.to_call)
|
|
self.assertEqual(restored.position.latitude, original.position.latitude)
|
|
self.assertEqual(restored.position.longitude, original.position.longitude)
|
|
self.assertEqual(restored.position.symbol, original.position.symbol)
|
|
self.assertEqual(restored.wind_speed, original.wind_speed)
|
|
self.assertEqual(restored.wind_direction, original.wind_direction)
|
|
self.assertEqual(restored.wind_gust, original.wind_gust)
|
|
self.assertEqual(restored.temperature, original.temperature)
|
|
self.assertEqual(restored.rain_1h, original.rain_1h)
|
|
self.assertEqual(restored.rain_24h, original.rain_24h)
|
|
self.assertEqual(restored.rain_since_midnight, original.rain_since_midnight)
|
|
self.assertEqual(restored.humidity, original.humidity)
|
|
self.assertEqual(restored.pressure, original.pressure)
|
|
self.assertEqual(restored.position.comment, original.position.comment)
|
|
self.assertEqual(restored._type, original._type)
|
|
|
|
def test_weather_packet_from_raw_string(self):
|
|
"""Test WeatherPacket creation from raw APRS string."""
|
|
packet_raw = 'FW9222>APRS,TCPXX*,qAX,CWOP-6:@122025z2953.94N/08423.77W_232/003g006t084r000p032P000h80b10157L745.DsWLL'
|
|
packet_dict = aprslib.parse(packet_raw)
|
|
packet = packets.factory(packet_dict)
|
|
self.assertIsInstance(packet, packets.WeatherPacket)
|
|
# Test to_json
|
|
json_str = packet.to_json()
|
|
self.assertIsInstance(json_str, str)
|
|
json_dict = json.loads(json_str)
|
|
self.assertEqual(json_dict['_type'], 'WeatherPacket')
|
|
# Test from_dict round trip
|
|
restored = packets.factory(json_dict)
|
|
self.assertIsInstance(restored, packets.WeatherPacket)
|
|
self.assertEqual(restored.from_call, packet.from_call)
|
|
self.assertEqual(restored.temperature, packet.temperature)
|
|
self.assertEqual(restored.humidity, packet.humidity)
|
|
self.assertEqual(restored.pressure, packet.pressure)
|
|
self.assertEqual(restored.wind_speed, packet.wind_speed)
|
|
self.assertEqual(restored.wind_direction, packet.wind_direction)
|
|
|
|
def test_key_with_raw_timestamp(self):
|
|
"""WeatherPacket.key returns from_call:raw_timestamp when raw_timestamp is set."""
|
|
packet = packets.WeatherPacket(
|
|
from_call=fake.FAKE_FROM_CALLSIGN,
|
|
to_call=fake.FAKE_TO_CALLSIGN,
|
|
raw_timestamp='121234z',
|
|
)
|
|
self.assertEqual(packet.key, f'{fake.FAKE_FROM_CALLSIGN}:121234z')
|
|
|
|
def test_key_with_wx_raw_timestamp(self):
|
|
"""WeatherPacket.key returns from_call:wx_raw_timestamp when only wx_raw_timestamp is set."""
|
|
packet = packets.WeatherPacket(
|
|
from_call=fake.FAKE_FROM_CALLSIGN,
|
|
to_call=fake.FAKE_TO_CALLSIGN,
|
|
wx_raw_timestamp='121230z',
|
|
)
|
|
self.assertEqual(packet.key, f'{fake.FAKE_FROM_CALLSIGN}:121230z')
|
|
|
|
def test_key_without_timestamps_is_not_none(self):
|
|
"""WeatherPacket.key must not return None when both timestamps are absent.
|
|
|
|
Previously the property fell off the end without a return statement when
|
|
neither raw_timestamp nor wx_raw_timestamp was set, silently returning None.
|
|
"""
|
|
packet = packets.WeatherPacket(
|
|
from_call=fake.FAKE_FROM_CALLSIGN,
|
|
to_call=fake.FAKE_TO_CALLSIGN,
|
|
)
|
|
key = packet.key
|
|
self.assertIsNotNone(key)
|
|
self.assertIsInstance(key, str)
|
|
# Key must include from_call as prefix
|
|
self.assertTrue(key.startswith(fake.FAKE_FROM_CALLSIGN))
|
|
|
|
def test_key_without_timestamps_no_collision(self):
|
|
"""Two WeatherPackets from the same station with no timestamps must not
|
|
share the same key, otherwise PacketList / dupe-filter will collide.
|
|
|
|
The fallback uses id(self) which is unique per object instance.
|
|
"""
|
|
p1 = packets.WeatherPacket(
|
|
from_call=fake.FAKE_FROM_CALLSIGN,
|
|
to_call=fake.FAKE_TO_CALLSIGN,
|
|
)
|
|
p2 = packets.WeatherPacket(
|
|
from_call=fake.FAKE_FROM_CALLSIGN,
|
|
to_call=fake.FAKE_TO_CALLSIGN,
|
|
)
|
|
# Distinct objects — keys must differ regardless of timing
|
|
self.assertNotEqual(p1.key, p2.key)
|