Skip to content

Commit

Permalink
Fixing py3.7 err
Browse files Browse the repository at this point in the history
  • Loading branch information
ampledata committed Jan 13, 2025
1 parent 93a88d4 commit 9d5c032
Showing 1 changed file with 42 additions and 43 deletions.
85 changes: 42 additions & 43 deletions adsbxcot/classes.py
Original file line number Diff line number Diff line change
Expand Up @@ -37,49 +37,6 @@ def __init__(self, queue: asyncio.Queue, config: SectionProxy) -> None:
self.session: Union[aiohttp.ClientSession, None] = None
self.altitudes: dict = {}

async def handle_data(self, data: list) -> None:
"""Marshal ADS-B data into CoT, and put it onto a TX queue."""
if not isinstance(data, list):
self._logger.warning("Invalid aircraft data, should be a Python list.")
return None

if not data:
self._logger.warning("Empty aircraft list")
return None

lod = len(data)
i = 1
for craft in data:
i += 1
await self.process_craft(craft)
icao: str = craft.get("hex", craft.get("icao", ""))
self._logger.debug("Handling %s/%s ICAO: %s", i, lod, icao)

def calc_altitude(self, craft: dict) -> dict:
"""Calculate altitude based on barometric and geometric altitude."""
alt_baro = craft.get("alt_baro", "")
alt_geom = craft.get("alt_geom", "")

if not alt_baro:
return {}

if alt_baro == "ground":
return {}

alt_baro = float(alt_baro)
if alt_geom:
self.altitudes["alt_geom"] = float(alt_geom)
self.altitudes["alt_baro"] = alt_baro
elif "alt_baro" in self.altitudes and "alt_geom" in self.altitudes:
ref_alt_baro = float(self.altitudes["alt_baro"])
alt_baro_offset = alt_baro - ref_alt_baro
return {
"x_alt_baro_offset": alt_baro_offset,
"x_alt_geom": ref_alt_baro + alt_baro_offset,
}

return {}

async def process_craft(self, craft: dict) -> Optional[str]:
"""Process individual aircraft data."""
if not isinstance(craft, dict):
Expand Down Expand Up @@ -128,6 +85,48 @@ async def process_craft(self, craft: dict) -> Optional[str]:
await self.put_queue(event)
return icao

async def handle_data(self, data: list) -> None:
"""Marshal ADS-B data into CoT, and put it onto a TX queue."""
if not isinstance(data, list):
self._logger.warning("Invalid aircraft data, should be a Python list.")
return None

if not data:
self._logger.warning("Empty aircraft list")
return None

lod = len(data)
i = 1
for craft in data:
i += 1
icao = await self.process_craft(craft)
self._logger.debug("Handling %s/%s ICAO: %s", i, lod, icao)

def calc_altitude(self, craft: dict) -> dict:
"""Calculate altitude based on barometric and geometric altitude."""
alt_baro = craft.get("alt_baro", "")
alt_geom = craft.get("alt_geom", "")

if not alt_baro:
return {}

if alt_baro == "ground":
return {}

alt_baro = float(alt_baro)
if alt_geom:
self.altitudes["alt_geom"] = float(alt_geom)
self.altitudes["alt_baro"] = alt_baro
elif "alt_baro" in self.altitudes and "alt_geom" in self.altitudes:
ref_alt_baro = float(self.altitudes["alt_baro"])
alt_baro_offset = alt_baro - ref_alt_baro
return {
"x_alt_baro_offset": alt_baro_offset,
"x_alt_geom": ref_alt_baro + alt_baro_offset,
}

return {}

async def get_feed(self, url: str) -> None:
"""
ADS-B Aggregator API Client wrapper.
Expand Down

0 comments on commit 9d5c032

Please sign in to comment.