How to use traffic - 10 common examples

def download_airports(self) -> None:  # coverage: ignore
        from .. import session

        f = session.get(
            "https://ourairports.com/data/airports.csv", stream=True
        )
        total = int(f.headers["Content-Length"])
        buffer = io.BytesIO()
        for chunk in tqdm(
            f.iter_content(1024),
            total=total // 1024 + 1 if total % 1024 > 0 else 0,
            desc="airports @ourairports.com",
        ):
            buffer.write(chunk)

        buffer.seek(0)
        df = pd.read_csv(buffer)

        f = session.get("https://ourairports.com/data/countries.csv",)
        buffer = io.BytesIO(f.content)

def download_data(dataset: Dict[str, str]) -> io.BytesIO:
    from .. import session

    f = session.get(dataset["url"], stream=True)
    total = int(f.headers["Content-Length"])
    buffer = io.BytesIO()
    for chunk in tqdm(
        f.iter_content(1024),
        total=total // 1024 + 1 if total % 1024 > 0 else 0,
        desc="download",
    ):
        buffer.write(chunk)

    buffer.seek(0)

    compute_md5 = md5(buffer.getbuffer()).hexdigest()
    if compute_md5 != dataset["md5sum"]:
        raise RuntimeError(
            f"Error in MD5 check: {compute_md5} instead of {dataset['md5sum']}"
        )

def download_opensky(self):  # coverage: ignore
        """Downloads the latest version of the OpenSky aircraft database.

        url: https://opensky-network.org/aircraft-database
        """
        from .. import session

        logging.warning("Downloading OpenSky aircraft database")
        file_url = (
            "https://opensky-network.org/datasets/metadata/aircraftDatabase.csv"
        )
        f = session.get(file_url, stream=True)
        total = int(f.headers["Content-Length"])
        buffer = io.BytesIO()
        for chunk in tqdm(
            f.iter_content(1024),
            total=total // 1024 + 1 if total % 1024 > 0 else 0,
            desc="download",
        ):
            buffer.write(chunk)

        buffer.seek(0)
        self._opensky = pd.read_csv(
            buffer,
            dtype={"icao24": str, "operator": str},
            skiprows=[1],
            engine="c",
            keep_default_na=False,

def test_bearing() -> None:
    ajaccio: Flight = get_sample(calibration, "ajaccio")
    ext_navaids = navaids.extent(ajaccio)
    assert ext_navaids is not None
    vor = ext_navaids["AJO"]
    assert vor is not None
    subset = ajaccio.bearing(vor).query("bearing.diff().abs() < .01")
    assert subset is not None
    assert (
        sum(
            1
            for chunk in subset.split("1T")
            if chunk.duration > pd.Timedelta("5 minutes")
        )
        == 7
    )

def test_resample_unwrapped() -> None:
    # https://github.com/xoolive/traffic/issues/41

    df = pd.DataFrame.from_records(
        [
            (pd.Timestamp("2019-01-01 12:00:00Z"), 345),
            (pd.Timestamp("2019-01-01 12:00:30Z"), 355),
            (pd.Timestamp("2019-01-01 12:01:00Z"), 5),
            (pd.Timestamp("2019-01-01 12:01:30Z"), 15),
        ],
        columns=["timestamp", "track"],
    )

    resampled = Flight(df).resample("1s")
    assert resampled.query("50 < track < 300") is None

    resampled_10 = Flight(df).resample(10)
    assert len(resampled_10) == 10

"altitude": 36000,
            "callsign": "WZZ1066",
            "flight_id": "231619151",
            "icao24": "471f52",
        },
        {
            "timestamp": pd.Timestamp("2019-07-02 15:15:52+0000", tz="UTC"),
            "longitude": 0.5097222166666667,
            "latitude": 47.71388888333333,
            "altitude": 36000,
            "callsign": "WZZ1066",
            "flight_id": "231619151",
            "icao24": "471f52",
        },
    ]
    flight = Flight(pd.DataFrame.from_records(records))
    assert flight.clip(eurofirs["LFBB"]) is None

def test_getter() -> None:
    airport = airports["LHR"]
    assert airport is not None
    assert airport.icao == "EGLL"
    assert airport.iata == "LHR"
    assert airport.country == "United Kingdom"
    assert airport.name == "London Heathrow Airport"
    lat, lon = airport.latlon
    assert max(abs(lat - 51.471626), abs(lon + 0.467081)) < 1e-2

def test_closest_point() -> None:
    from traffic.data import navaids, airports

    item = belevingsvlucht.between(
        "2018-05-30 16:00", "2018-05-30 17:00"
    ).closest_point(  # type: ignore
        [
            airports["EHLE"],  # type: ignore
            airports["EHAM"],  # type: ignore
            navaids["NARAK"],  # type: ignore
        ]
    )
    res = f"{item.timestamp:%H:%M:%S}, {item.point}, {item.distance:.2f}m"
    assert res == "16:53:46, Lelystad Airport, 49.11m"

@pytest.mark.skipif(aircraft.opensky_db is not None, reason="OpenSky database")
def test_opensky_dl() -> None:
    aircraft.download_opensky()
    assert aircraft.opensky_db is not None

def test_getter() -> None:
    a = aircraft["PH-BHA"]
    assert a.icao24.iloc[0] == "4851ad"
    assert a.typecode.iloc[0] == "B789"
    a = aircraft["39b415"]
    assert a.registration.iloc[0] == "F-HNAV"
    assert a.typecode.iloc[0] == "BE20"