How to use the traffic.data.session 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 / traffic / data / basic / airports.py View on Github external
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)
github xoolive / traffic / traffic / data / datasets / __init__.py View on Github external
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']}"
        )
github xoolive / traffic / traffic / data / basic / aircraft.py View on Github external
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,
github xoolive / traffic / traffic / data / basic / runways.py View on Github external
def download_bluesky(self) -> None:  # coverage: ignore
        from .. import session

        self._runways = dict()
        c = session.get(base_url + "/apt.zip")

        with ZipFile(BytesIO(c.content)).open("apt.dat", "r") as fh:
            for line in fh.readlines():
                elems = (
                    line.decode(encoding="ascii", errors="ignore")
                    .strip()
                    .split()
                )
                if len(elems) == 0:
                    continue

                # 1: AIRPORT
                if elems[0] == "1":
                    # Add airport to runway threshold database
                    cur: List[Tuple[Threshold, Threshold]] = list()
                    self._runways[elems[4]] = cur
github xoolive / traffic / traffic / data / basic / navaid.py View on Github external
#  30.580372 -094.384169 FAREL
            fields = line.split()
            navaids.append(
                Navaid(
                    fields[2],
                    "FIX",
                    float(fields[0]),
                    float(fields[1]),
                    float("nan"),
                    None,
                    None,
                    None,
                )
            )

        c = session.get(f"{base_url}/earth_nav.dat")

        for line in c.iter_lines():

            line = line.decode(encoding="ascii", errors="ignore").strip()

            # Skip empty lines or comments
            if len(line) == 0 or line[0] == "#":
                continue

            # Data line => Process fields of this record, separated by a comma
            # Example lines:
            # 2  58.61466599  125.42666626 451   522  30  0.0 A   Aldan NDB
            # 3  31.26894444 -085.72630556 334 11120  40 -3.0 OZR CAIRNS VOR-DME
            # type    lat       lon        elev freq  ?   var id   desc
            #   0      1         2           3    4   5    6   7    8
github xoolive / traffic / traffic / core / flightplan.py View on Github external
def skyvector(self) -> Dict[str, Any]:
        from ..data import session

        c = session.get(
            "https://skyvector.com/api/routes?dep={}&dst={}".format(
                self.origin, self.destination
            )
        )
        c.raise_for_status()

        c = session.get(
            f"https://skyvector.com/api/fpl?cmd=route&route={self.repr}"
        )
        c.raise_for_status()

        return c.json()
github xoolive / traffic / traffic / data / basic / aircraft.py View on Github external
def download_junzis(self) -> None:  # coverage: ignore
        """Downloads the latest version of the aircraft database by @junzis.

        url: https://junzisun.com/adb/download
        """
        from .. import session

        f = session.get("https://junzisun.com/adb/download")
        with zipfile.ZipFile(io.BytesIO(f.content)) as zfile:
            with zfile.open("aircraft_db.csv", "r") as dbfile:
                self._junzis = (
                    pd.read_csv(dbfile, dtype=str)
                    .assign(
                        regid=lambda df: df.regid.str.upper(),
                        mdl=lambda df: df.mdl.str.upper(),
                    )
                    .rename(
                        columns={
                            "icao": "icao24",
                            "regid": "registration",
                            "mdl": "typecode",
                            "type": "model",
                        }
                    )
github xoolive / traffic / traffic / data / faa / __init__.py View on Github external
def download_data(self) -> None:
        from .. import session

        logging.warning(
            f"Downloading data from {self.website}. Please check terms of use."
        )
        c = session.get(self.json_url)
        c.raise_for_status()
        json_contents = c.json()
        with self.cache_file.open("w") as fh:
            json.dump(json_contents, fh)
github xoolive / traffic / traffic / data / basic / navaid.py View on Github external
def download_data(self) -> None:  # coverage: ignore
        """Downloads the latest version of the navaid database from the
        repository.
        """

        from .. import session

        navaids = []
        c = session.get(f"{base_url}/earth_fix.dat")

        for line in c.iter_lines():

            line = line.decode(encoding="ascii", errors="ignore").strip()

            # Skip empty lines or comments
            if len(line) < 3 or line[0] == "#":
                continue

            # Start with valid 2 digit latitude -45. or 52.
            if not ((line[0] == "-" and line[3] == ".") or line[2] == "."):
                continue

            # Data line => Process fields of this record, separated by a comma
            # Example line:
            #  30.580372 -094.384169 FAREL
github xoolive / traffic / traffic / core / flightplan.py View on Github external
def skyvector(self) -> Dict[str, Any]:
        from ..data import session

        c = session.get(
            "https://skyvector.com/api/routes?dep={}&dst={}".format(
                self.origin, self.destination
            )
        )
        c.raise_for_status()

        c = session.get(
            f"https://skyvector.com/api/fpl?cmd=route&route={self.repr}"
        )
        c.raise_for_status()

        return c.json()