189 lines
8.9 KiB
Python
189 lines
8.9 KiB
Python
# FlightMonitor/data/common_models.py
|
|
"""
|
|
Defines the canonical data models used throughout the FlightMonitor application.
|
|
These models provide a standardized representation of flight data, independent
|
|
of the data provider (e.g., OpenSky, ADS-B Exchange, etc.).
|
|
"""
|
|
|
|
from typing import Optional, Any # For type hints
|
|
import time # Used as a fallback for timestamps if absolutely necessary
|
|
|
|
class CanonicalFlightState:
|
|
"""
|
|
Represents the standardized state of an aircraft at a specific point in time.
|
|
All timestamps are expected to be Unix epoch (seconds, UTC).
|
|
All measurements (altitude, velocity, rate) are in metric units (meters, m/s).
|
|
"""
|
|
def __init__(self,
|
|
icao24: str,
|
|
timestamp: float, # Primary timestamp of the position/state data (UTC epoch)
|
|
last_contact_timestamp: float, # Timestamp of the last message from transponder (UTC epoch)
|
|
latitude: Optional[float],
|
|
longitude: Optional[float],
|
|
on_ground: bool,
|
|
callsign: Optional[str] = None,
|
|
origin_country: Optional[str] = None,
|
|
baro_altitude_m: Optional[float] = None, # Barometric altitude in meters
|
|
geo_altitude_m: Optional[float] = None, # Geometric altitude in meters
|
|
velocity_mps: Optional[float] = None, # Ground speed in meters/second
|
|
true_track_deg: Optional[float] = None, # True track in degrees (0-360, North is 0)
|
|
vertical_rate_mps: Optional[float] = None, # Vertical rate in meters/second (positive=climbing)
|
|
squawk: Optional[str] = None,
|
|
spi: Optional[bool] = None, # Special Purpose Indicator
|
|
position_source: Optional[Any] = None, # Source of position data (e.g., 0 for ADS-B, string name)
|
|
raw_data_provider: Optional[str] = None, # Identifier for the data provider (e.g., "OpenSky")
|
|
# provider_specific_data: Optional[dict] = None # Optional: for any extra data not in canonical model
|
|
):
|
|
"""
|
|
Initializes a CanonicalFlightState object.
|
|
|
|
Args:
|
|
icao24 (str): Unique ICAO 24-bit address of the aircraft (hex string).
|
|
timestamp (float): The primary timestamp (UTC epoch seconds) for this state record,
|
|
ideally from the data provider indicating when the state was valid.
|
|
last_contact_timestamp (float): The timestamp (UTC epoch seconds) of the last message
|
|
received from the aircraft's transponder.
|
|
latitude (Optional[float]): Latitude in decimal degrees (WGS-84).
|
|
longitude (Optional[float]): Longitude in decimal degrees (WGS-84).
|
|
on_ground (bool): True if the aircraft is on the ground.
|
|
callsign (Optional[str]): Callsign of the aircraft (e.g., "DAL123").
|
|
origin_country (Optional[str]): Origin country of the aircraft, often inferred from ICAO24.
|
|
baro_altitude_m (Optional[float]): Barometric altitude in meters.
|
|
geo_altitude_m (Optional[float]): Geometric (GNSS/WGS-84) altitude in meters.
|
|
velocity_mps (Optional[float]): Ground speed in meters per second.
|
|
true_track_deg (Optional[float]): True track/heading in degrees clockwise from North.
|
|
vertical_rate_mps (Optional[float]): Vertical rate in meters per second (positive for climb).
|
|
squawk (Optional[str]): Transponder squawk code.
|
|
spi (Optional[bool]): Special Purpose Indicator flag.
|
|
position_source (Optional[Any]): Identifier for the source of the position data.
|
|
raw_data_provider (Optional[str]): Name of the provider that supplied this data.
|
|
"""
|
|
if not icao24:
|
|
raise ValueError("icao24 cannot be None or empty.")
|
|
|
|
self.icao24: str = icao24.lower().strip() # Standardize to lowercase and strip whitespace
|
|
self.callsign: Optional[str] = callsign.strip() if callsign and callsign.strip() else None
|
|
self.origin_country: Optional[str] = origin_country
|
|
|
|
self.timestamp: float = timestamp
|
|
self.last_contact_timestamp: float = last_contact_timestamp
|
|
|
|
self.latitude: Optional[float] = latitude
|
|
self.longitude: Optional[float] = longitude
|
|
|
|
self.baro_altitude_m: Optional[float] = baro_altitude_m
|
|
self.geo_altitude_m: Optional[float] = geo_altitude_m
|
|
|
|
self.on_ground: bool = on_ground
|
|
self.velocity_mps: Optional[float] = velocity_mps
|
|
self.true_track_deg: Optional[float] = true_track_deg
|
|
self.vertical_rate_mps: Optional[float] = vertical_rate_mps
|
|
|
|
self.squawk: Optional[str] = squawk
|
|
self.spi: Optional[bool] = spi
|
|
self.position_source: Optional[Any] = position_source # Could be int or str based on provider
|
|
|
|
self.raw_data_provider: Optional[str] = raw_data_provider
|
|
# self.provider_specific_data = provider_specific_data or {}
|
|
|
|
def __repr__(self) -> str:
|
|
return (
|
|
f"{self.__class__.__name__}("
|
|
f"icao24='{self.icao24}', callsign='{self.callsign}', "
|
|
f"timestamp={self.timestamp:.0f}, lat={self.latitude}, lon={self.longitude}, "
|
|
f"alt_m={self.baro_altitude_m if self.baro_altitude_m is not None else self.geo_altitude_m}, "
|
|
f"on_ground={self.on_ground}"
|
|
")"
|
|
)
|
|
|
|
def __str__(self) -> str:
|
|
# Provides a more detailed, readable string representation
|
|
details = [f"{attr}={getattr(self, attr)}" for attr in vars(self) if getattr(self, attr) is not None]
|
|
return f"{self.__class__.__name__}({', '.join(details)})"
|
|
|
|
# It might be useful to have a method to convert to a dictionary,
|
|
# e.g., for serialization or if some parts of the code still expect dicts.
|
|
def to_dict(self) -> dict:
|
|
"""Converts the CanonicalFlightState object to a dictionary."""
|
|
return {attr: getattr(self, attr) for attr in vars(self) if not attr.startswith('_')}
|
|
|
|
# Potremmo aggiungere altre classi qui in futuro, come:
|
|
# class CanonicalFlightTrack: ...
|
|
# class CanonicalAirportInfo: ...
|
|
# class CanonicalFlightRoute: ...
|
|
|
|
if __name__ == '__main__':
|
|
# Example usage and test
|
|
print("--- Testing CanonicalFlightState ---")
|
|
try:
|
|
# Test case 1: Minimal data
|
|
ts = time.time()
|
|
state1_data = {
|
|
"icao24": "AABBCC",
|
|
"timestamp": ts,
|
|
"last_contact_timestamp": ts -1, # Last contact slightly before position
|
|
"latitude": 50.123,
|
|
"longitude": 10.456,
|
|
"on_ground": False,
|
|
"raw_data_provider": "TestProvider"
|
|
}
|
|
state1 = CanonicalFlightState(**state1_data)
|
|
print(f"\nState 1 (repr): {repr(state1)}")
|
|
print(f"State 1 (str): {str(state1)}")
|
|
print(f"State 1 (dict): {state1.to_dict()}")
|
|
assert state1.icao24 == "aabbcc"
|
|
assert state1.timestamp == ts
|
|
|
|
# Test case 2: More complete data
|
|
state2_data = {
|
|
"icao24": "123456",
|
|
"callsign": "FLIGHT01 ", # With trailing spaces
|
|
"origin_country": "Testland",
|
|
"timestamp": ts + 10,
|
|
"last_contact_timestamp": ts + 9,
|
|
"latitude": -33.8688,
|
|
"longitude": 151.2093,
|
|
"baro_altitude_m": 10000.0,
|
|
"geo_altitude_m": 10200.5,
|
|
"on_ground": True,
|
|
"velocity_mps": 150.75,
|
|
"true_track_deg": 180.5,
|
|
"vertical_rate_mps": -5.2,
|
|
"squawk": "7000",
|
|
"spi": True,
|
|
"position_source": 0, # ADS-B
|
|
"raw_data_provider": "OpenSky-Test"
|
|
}
|
|
state2 = CanonicalFlightState(**state2_data)
|
|
print(f"\nState 2 (repr): {repr(state2)}")
|
|
assert state2.callsign == "FLIGHT01" # Check stripping
|
|
assert state2.on_ground is True
|
|
assert state2.velocity_mps == 150.75
|
|
|
|
# Test case 3: Missing optional data
|
|
state3 = CanonicalFlightState(
|
|
icao24="abcdef",
|
|
timestamp=ts+20,
|
|
last_contact_timestamp=ts+19,
|
|
latitude=None, # Missing lat
|
|
longitude=None, # Missing lon
|
|
on_ground=False
|
|
)
|
|
print(f"\nState 3 (repr): {repr(state3)}")
|
|
assert state3.latitude is None
|
|
assert state3.callsign is None
|
|
|
|
# Test case 4: Invalid ICAO24 (should raise ValueError)
|
|
print("\nTesting invalid ICAO24 (expect ValueError):")
|
|
try:
|
|
CanonicalFlightState(icao24="", timestamp=ts, last_contact_timestamp=ts, latitude=0, longitude=0, on_ground=False)
|
|
except ValueError as e:
|
|
print(f"Caught expected error: {e}")
|
|
assert "icao24 cannot be None or empty" in str(e)
|
|
|
|
print("\n--- CanonicalFlightState tests passed ---")
|
|
|
|
except Exception as e:
|
|
print(f"Error during CanonicalFlightState tests: {e}")
|
|
import traceback
|
|
traceback.print_exc() |