How to use the traffic.core.Traffic function in traffic

To help you get started, we’ve selected a few traffic examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github xoolive / traffic / scripts / airport_plot.py View on Github external
try:
        from cartotools.osm import request, tags
        request(f'{airport.icao} airport', **tags.airport).plot(ax)
    except Exception:
        pass

    fd = flights.data

    fd = fd[fd.longitude > airport.lon - 1]
    fd = fd[fd.longitude < airport.lon + 1]
    fd = fd[fd.latitude > airport.lat - 1]
    fd = fd[fd.latitude < airport.lat + 1]
    fd = fd.groupby('callsign').filter(
        lambda f: arrival * f.vertical_rate[:-100].mean() < -10)

    for flight in Traffic(fd):
        flight.plot(ax)

    fig.suptitle(f"{airport.name}")
    fig.set_tight_layout(True)

    logging.info(f"Image saved as {output}")
    fig.savefig(output.as_posix())
github xoolive / traffic / traffic / plugins / cesiumjs.py View on Github external
def _onload():
    setattr(Traffic, "to_czml", to_czml)
    setattr(SO6, "to_czml", to_czml)
github xoolive / traffic / traffic / qtgui / layout.py View on Github external
def on_filter(self, max_alt: int, max_time: datetime) -> None:

        assert self._traffic is not None

        west, east, south, north = self.map_plot.ax.get_extent(
            crs=PlateCarree()
        )

        self._tview = self._traffic.before(max_time).sort_values("timestamp")

        if self._tview is None:
            return

        filtered = Traffic.from_flights(
            Flight(f.data.ffill().bfill()) for f in self._tview
        )
        if "altitude" in filtered.data.columns:
            filtered = filtered.query(
                f"altitude != altitude or altitude <= {max_alt}"
            )
        if "latitude" in self._tview.data.columns:
            filtered = filtered.query(
                "latitude != latitude or "
                f"({west} <= longitude <= {east} and "
                f"{south} <= latitude <= {north})"
            )

        self.identifier_select.clear()
        text = self.identifier_input.text()
        # cast is necessary because of the @lru_cache on callsigns which hides
github xoolive / traffic / traffic / data / samples / __init__.py View on Github external
@lru_cache()
def __getattr__(name: str) -> Union[Flight, Traffic]:
    filelist = list(_current_dir.glob(f"**/{name}.json.gz"))
    if len(filelist) == 0:
        msg = f"File {name}.json.gz not found in available samples"
        raise AttributeError(msg)
    return get_flight(name, filelist[0].parent)


airbus_tree: Flight = cast(Flight, __getattr__("airbus_tree"))
belevingsvlucht: Flight = cast(Flight, __getattr__("belevingsvlucht"))
texas_longhorn: Flight = cast(Flight, __getattr__("texas_longhorn"))
quickstart: Traffic = cast(Traffic, __getattr__("quickstart"))
switzerland: Traffic = cast(Traffic, __getattr__("switzerland"))

lfbo_tma = Airspace.from_file(
    Path(__file__).parent / "airspaces" / "LFBOTMA.json"
)
github xoolive / traffic / traffic / data / adsb / decode.py View on Github external
def traffic(self) -> Optional[Traffic]:
        try:
            return Traffic.from_flights(
                self[elt["icao24"]] for elt in self.aircraft
            )
        except ValueError:
            return None
github xoolive / traffic / traffic / plugins / kepler.py View on Github external
def map_add_data(_map, data, *args, **kwargs):
    if any(isinstance(data, c) for c in (Airspace, Flight, Traffic)):
        layer = data.kepler()
        return _old_add_data(_map, layer, *args, **kwargs)

    if any(isinstance(data, c) for c in (list, Generator)):
        layer = dict(
            type="FeatureCollection", features=[elt.kepler() for elt in data]
        )
        return _old_add_data(_map, layer, *args, **kwargs)

    # convenient for airports, navaids, etc.
    if hasattr(data, "data"):
        data = data.data

    return _old_add_data(_map, data, *args, **kwargs)
github xoolive / traffic / traffic / qtgui / layout.py View on Github external
def __init__(self) -> None:

        logging.info("Initialize MainScreen")
        super().__init__(parent=None)

        self._traffic: Optional[Traffic] = None
        self._tview: Optional[Traffic] = None
        self.decoder: Optional[ModeS_Decoder] = None
        self.updateTraffic: Optional[UpdateTraffic] = None
        self.airspace_ready: bool = False
        self.last_interact: datetime = datetime.now()

        airspace_init = AirspaceInitCache(self)
        airspace_init.start()

        self.setWindowTitle("traffic")
        self.setGeometry(10, 10, 920, 720)

        self.set_icons()
        self.set_layout()
        self.set_design()
        self.set_callbacks()
github xoolive / traffic / traffic / plugins / bluesky.py View on Github external
traffic = traffic.after(minimum_time)

    if isinstance(filename, str):
        filename = Path(filename)

    if not filename.parent.exists():
        filename.parent.mkdir(parents=True)

    altitude = (
        "baro_altitude"
        if "baro_altitude" in traffic.data.columns
        else "altitude"
    )

    if "typecode" not in traffic.data.columns:
        traffic = Traffic(
            traffic.data.merge(
                aircraft.data[["icao24", "typecode"]].drop_duplicates("icao24"),
                on="icao24",
                how="inner",
            )
        )

    if "cas" not in traffic.data.columns:
        traffic = Traffic(
            traffic.data.assign(
                cas=vtas2cas(traffic.data.ground_speed, traffic.data[altitude])
            )
        )

    with filename.open("w") as fh:
        t_delta = traffic.data.timestamp - traffic.start_time