Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def get_flight(filename: str, directory: Path) -> Union[Flight, Traffic]:
flight: Union[None, Flight, Traffic] = Traffic.from_file(
directory / f"{filename}.json.gz", dtype={"icao24": str}
)
if flight is None:
raise RuntimeError(f"File {filename}.json.gz not found in {directory}")
icao24 = set(flight.data.icao24)
if len(icao24) == 1:
# easier way to cast...
flight = Flight(flight.data)
# -- Dealing with time-like features --
if "hour" in flight.data.columns:
flight = flight.assign(
hour=lambda df: pd.to_datetime(df.hour * 1e9).dt.tz_localize("utc")
)
if "last_position" in flight.data.columns:
flight = flight.assign(
last_position=lambda df: pd.to_datetime(
options = {
"caption": "Open file",
"filter": (
"Pandas DataFrame (*.pkl);;"
"CSV files (*.csv);;"
"Sqlite3 files (*.db)"
),
# "filter": "Data files (*.csv *.pkl)",
"directory": os.path.expanduser("~"),
}
self.filename = QFileDialog.getOpenFileName(self, **options)[0]
if self.filename == "":
return
self.filename = Path(self.filename)
self._traffic = Traffic.from_file(self.filename)
assert self._traffic is not None
self._tview = self._traffic.sort_values("timestamp")
assert self._tview is not None
self.open_dropdown.setItemText(0, self.filename.name)
self.map_plot.default_plot(self._tview)
self.identifier_select.clear()
for callsign in sorted(cast(Set[str], self._tview.callsigns)):
self.identifier_select.addItem(callsign)
self.set_time_range()
self.set_float_columns()
)
squawk7700 = Traffic.from_file(filename)
assert squawk7700 is not None
squawk7700 = squawk7700.merge(
metadata.drop(columns=["callsign", "icao24"]), on="flight_id"
)
squawk7700.metadata = metadata # type: ignore
return squawk7700
datasets = dict(
paris_toulouse_2017=dict(
url="https://ndownloader.figshare.com/files/20291055",
md5sum="e869a60107fdb9f092f5abbb3b43a2c0",
filename="city_pair_dataset.parquet",
reader=Traffic.from_file,
),
airspace_bordeaux_2017=dict(
url="https://ndownloader.figshare.com/files/20291040",
md5sum="f7b057f12cc735a15b93984b9ae7b8fc",
filename="airspace_dataset.parquet",
reader=Traffic.from_file,
),
landing_zurich_2019=dict(
url="https://ndownloader.figshare.com/files/20291079",
md5sum="c5577f450424fa74ca673ed8a168c67f",
filename="landing_dataset.parquet",
reader=Traffic.from_file,
),
squawk7700=dict(
url=f"{_squawk7700_url}/files/squawk7700_trajectories.parquet.gz",
md5sum="8ed5375d92bd7b7b94ea03c0533d959e",
group = parser.add_mutually_exclusive_group()
group.add_argument(
"--head",
dest="head",
action="store_true",
help="print the first lines of the traffic file",
)
parser.add_argument(
"callsigns",
nargs=argparse.REMAINDER,
help="display specific information about specific flights",
)
args = parser.parse_args(args)
t = Traffic.from_file(args.filename)
assert t is not None
print("Traffic with {} identifiers".format(len(t)))
print(t)
print()
print("with the following features:")
print(t.data.columns)
if args.head:
print()
print("Head of the DataFrame:")
print(t.data.head())
if args.callsigns:
for callsign in args.callsigns:
print()
def _squawk7700_reader(filename):
metadata = get_dataset(
dict(
url=f"{_squawk7700_url}/files/squawk7700_metadata.csv",
md5sum="4d33ff84a05742136f3733692bf944e6",
filename="squawk7700_metadata.csv",
reader=pd.read_csv,
)
)
squawk7700 = Traffic.from_file(filename)
assert squawk7700 is not None
squawk7700 = squawk7700.merge(
metadata.drop(columns=["callsign", "icao24"]), on="flight_id"
)
squawk7700.metadata = metadata # type: ignore
return squawk7700