Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
import pandas as pd
from shapely.geometry import LineString
from ..core.mixins import DataFrameMixin, ShapelyMixin # type: ignore
def split(data: pd.DataFrame, value, unit) -> Iterator[pd.DataFrame]:
diff = data.timestamp.diff().values
if diff.max() > np.timedelta64(value, unit):
yield from split(data.iloc[:diff.argmax()], value, unit)
yield from split(data.iloc[diff.argmax():], value, unit)
else:
yield data
class Flight(ShapelyMixin, DataFrameMixin):
def plot(self, ax, **kwargs):
if 'projection' in ax.__dict__:
from cartopy.crs import PlateCarree
kwargs['transform'] = PlateCarree()
for subflight in self.split(10, 'm'):
ax.plot(subflight.data.longitude,
subflight.data.latitude,
**kwargs)
def split(self, value=10, unit='m'):
for data in split(self.data, value, unit):
yield Flight(data)
@property
def start(self):
@property
def shape(self):
data = self.data[self.data.longitude.notnull()]
return LineString(zip(data.longitude, data.latitude))
def _repr_html_(self):
cumul = ''
for flight in self.split():
title = f'<b>{", ".join(flight.callsign)}</b>'
title += f' ({", ".join(flight.icao24)})'
no_wrap_div = '<div style="white-space: nowrap">{}</div>'
cumul += title + no_wrap_div.format(flight._repr_svg_())
return cumul
class ADSB(DataFrameMixin):
def __getitem__(self, index: str) -> Optional[Flight]:
try:
value16 = int(index, 16)
data = self.data[self.data.icao24 == index]
except ValueError:
data = self.data[self.data.callsign == index]
if data.shape[0] > 0:
return Flight(data)
def _ipython_key_completions_(self):
return {*self.aircraft, *self.callsigns}
def __iter__(self):
for _, df in self.data.groupby('icao24'):
return instance
def __str__(self) -> str:
if self.reply is None:
return "[empty]"
s = ElementTree.tostring(self.reply)
return minidom.parseString(s).toprettyxml(indent=" ")
def __repr__(self) -> str:
res = str(self)
if len(res) > 1000:
return res[:1000] + "..."
return res
class FlightList(DataFrameMixin, B2BReply):
def __init__(self, *args, **kwargs):
if len(args) == 0 and "data" not in kwargs:
super().__init__(data=None, **kwargs)
else:
super().__init__(*args, **kwargs)
@classmethod
def fromB2BReply(cls, r: B2BReply):
assert r.reply is not None
return cls.fromET(r.reply)
@classmethod
def fromET(cls, tree: ElementTree.Element) -> "B2BReply":
instance = cls()
instance.reply = tree
instance.build_df()
pointDistance=lambda df: df.pointDistance.astype(int),
altitude=lambda df: df.flightLevel.astype(int) * 100,
geoPointId=lambda df: df.geoPointId.apply(parse_coordinates),
latitude=lambda df: df.geoPointId.apply(itemgetter(0)),
longitude=lambda df: df.geoPointId.apply(itemgetter(1)),
icao24=self.icao24,
callsign=self.callsign,
flight_id=self.ifpsId,
origin=self.origin,
destination=self.destination,
)
.drop(columns=["geoPointId", "relDist", "isVisible"])
)
class AllFT(DataFrameMixin):
__slots__ = ("data",)
@classmethod
def from_allft(
cls: Type[AllFTTypeVar], filename: Union[str, Path, BytesIO]
) -> AllFTTypeVar:
allft = (
pd.read_csv(
filename,
header=None,
names=allft_fields,
sep=";",
dtype={
"aobt": str,
"iobt": str,
"cobt": str,
elt = self.reply.find("location/flightLevels/max/level")
return (
int(elt.text) if elt is not None and elt.text is not None else 999
)
def __getattr__(self, name) -> str:
cls = type(self)
assert self.reply is not None
elt = self.reply.find(name)
if elt is not None:
return elt.text # type: ignore
msg = "{.__name__!r} object has no attribute {!r}"
raise AttributeError(msg.format(cls, name))
class RegulationList(DataFrameMixin, B2BReply):
def __init__(self, *args, **kwargs):
if len(args) == 0 and "data" not in kwargs:
super().__init__(data=None, **kwargs)
else:
super().__init__(*args, **kwargs)
@classmethod
def fromB2BReply(cls, r: B2BReply):
assert r.reply is not None
return cls.fromET(r.reply)
@classmethod
def fromET(
cls: Type[RegulationListTypeVar], tree: ElementTree.Element
) -> RegulationListTypeVar:
instance = cls()
def geoencode(self) -> alt.Chart: # coverage: ignore
"""Returns an `altair `_ encoding of the
shape to be composed in an interactive visualization.
"""
return (
alt.Chart(
self.data.query(
"latitude == latitude and longitude == longitude"
)[["latitude", "longitude"]]
)
.encode(latitude="latitude", longitude="longitude")
.mark_line()
)
class GeoDBMixin(DataFrameMixin):
def extent(
self: T,
extent: Union[str, ShapelyMixin, Tuple[float, float, float, float]],
buffer: float = 0.5,
) -> Optional[T]:
"""
Selects the subset of data inside the given extent.
The parameter extent may be passed as:
- a string to query OSM Nominatim service;
- the result of an OSM Nominatim query;
- any kind of shape (including airspaces);
- extents (west, east, south, north)
This works with databases like airways, airports or navaids.
f'x <= {flight.max("x")} + {lateral_separation} and '
f'y >= {flight.min("y")} - {lateral_separation} and '
f'y <= {flight.max("y")} + {lateral_separation} and '
f'altitude >= {flight.min("altitude")} - {vertical_separation} and '
f'altitude <= {flight.max("altitude")} + {vertical_separation} and '
f'timestamp <= "{flight.stop}" and '
f'timestamp >= "{flight.start}" '
)
if clipped is None:
continue
for second in clipped:
yield flight, second
class CPA(DataFrameMixin):
def aggregate(
self, lateral_separation: float = 5, vertical_separation: float = 1000
) -> "CPA":
return (
self.assign(
tmp_lateral=lambda df: df.lateral / lateral_separation,
tmp_vertical=lambda df: df.vertical / vertical_separation,
)
.assign(
aggregated=lambda df: df[["tmp_lateral", "tmp_vertical"]].max(
axis=1
)
)
.drop(columns=["tmp_lateral", "tmp_vertical"])
)
else None,
callsign=self.aircraftId,
origin=self.aerodromeOfDeparture,
destination=self.aerodromeOfDestination,
flight_id=self.flight_id,
EOBT=self.estimatedOffBlockTime,
typecode=self.aircraftType,
)
)
# https://github.com/python/mypy/issues/2511
FlightListTypeVar = TypeVar("FlightListTypeVar", bound="FlightList")
class FlightList(DataFrameMixin, B2BReply):
def __init__(self, *args, **kwargs):
if len(args) == 0 and "data" not in kwargs:
super().__init__(data=None, **kwargs)
else:
super().__init__(*args, **kwargs)
@classmethod
def fromB2BReply(
cls: Type[FlightListTypeVar], r: B2BReply
) -> FlightListTypeVar:
assert r.reply is not None
return cls.fromET(r.reply)
@classmethod
def fromET(
cls: Type[FlightListTypeVar], tree: ElementTree.Element
def encode_time_dump1090(times: pd.Series) -> pd.Series:
if isinstance(times.iloc[0], pd.datetime):
times = times.astype(np.int64) * 1e-9
ref_time = times.iloc[0]
rel_times = times - ref_time
rel_times = rel_times * 12e6
rel_times = rel_times.apply(lambda row: hex(int(row))[2:].zfill(12))
return rel_times
encode_time: Dict[str, Callable[[pd.Series], pd.Series]] = {
"dump1090": encode_time_dump1090
}
class RawData(DataFrameMixin):
def __add__(self: T, other: T) -> T:
return self.__class__.from_list([self, other])
@classmethod
def from_list(cls: Type[T], elts: Iterable[Optional[T]]) -> T:
res = cls(
pd.concat(list(x.data for x in elts if x is not None), sort=False)
)
return res.sort_values("mintime")
def decode(
self: T,
reference: Union[None, str, Airport, Tuple[float, float]] = None,
*,
uncertainty: bool = False,
progressbar: Union[bool, Callable[[Iterable], Iterable]] = True,
lat_2=bounds[3],
lat_0=(bounds[1] + bounds[3]) / 2,
lon_0=(bounds[0] + bounds[2]) / 2,
)
transformer = pyproj.Transformer.from_proj(
pyproj.Proj("epsg:4326"), projection, always_xy=True
)
projected_shape = transform(transformer.transform, self.shape,)
if not projected_shape.is_valid:
warnings.warn("The chosen projection is invalid for current shape")
return projected_shape
class GeographyMixin(DataFrameMixin):
"""Adds Euclidean coordinates to a latitude/longitude DataFrame."""
def compute_xy(
self: T, projection: Union[pyproj.Proj, crs.Projection, None] = None
) -> T:
"""Enrich the structure with new x and y columns computed through a
projection of the latitude and longitude columns.
The source projection is WGS84 (EPSG 4326).
The default destination projection is a Lambert Conformal Conical
projection centered on the data inside the dataframe.
Other valid projections are available:
- as ``pyproj.Proj`` objects;
- as ``cartopy.crs.Projection`` objects.