# 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()