How to use the traffic.data.adsb.opensky_impala.ImpalaError 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 / adsb / opensky_impala.py View on Github external
Impala._parseErrorMsg.format(path=new_path)
                        + (error + "\n" + content)
                    )

                if df.shape[0] > 0:
                    return df.drop_duplicates()

        error_msg: Optional[str] = None
        with cachename.open("r") as fh:
            output = fh.readlines()
            if any(elt.startswith("ERROR:") for elt in output):
                error_msg = "".join(output[:-1])

        if error_msg is not None:
            cachename.unlink()
            raise ImpalaError(error_msg)

        return None
github xoolive / traffic / traffic / data / adsb / opensky_impala.py View on Github external
return "".join(fh.readlines())

            if count > 0:
                s.seek(0)
                try:
                    # otherwise pandas would parse 1234e5 as 123400000.0
                    df = pd.read_csv(s, dtype={"icao24": str, "callsign": str})
                except ParserError as error:
                    for x in re.finditer(r"line (\d)+,", error.args[0]):
                        line_nb = int(x.group(1))
                        with cachename.open("r") as fh:
                            content = fh.readlines()[line_nb - 1]

                    new_path = Path(gettempdir()) / cachename.name
                    cachename.rename(new_path)
                    raise ImpalaError(
                        Impala._parseErrorMsg.format(path=new_path)
                        + (error + "\n" + content)
                    )

                if df.shape[0] > 0:
                    return df.drop_duplicates()

        error_msg: Optional[str] = None
        with cachename.open("r") as fh:
            output = fh.readlines()
            if any(elt.startswith("ERROR:") for elt in output):
                error_msg = "".join(output[:-1])

        if error_msg is not None:
            cachename.unlink()
            raise ImpalaError(error_msg)