Create OpenSky Network Document Loader

[AndrewEffendi]

Checked

• I searched existing ideas and did not find a similar one
• I added a very descriptive title
• I've clearly described the feature request and motivation for it

Feature request

A Document Loader for https://opensky-network.org/

The OpenSky Network API provides endpoints for accessing both real-time and historical flight data:

All State Vectors: https://opensky-network.org/api/states/all
Flight by time interval: https://opensky-network.org/api/flights/all?begin={begin}&end={end}
Flight by aircraft: https://opensky-network.org/api/flights/aircraft?icao24={icao24}&begin={begin}&end={end}
Arrivals by Airport: https://opensky-network.org/api/flights/arrival?airport={airport}&begin={begin}&end={end}
Departures by Airport: https://opensky-network.org/api/flights/departure?airport={airport}&begin={begin}&end={end}

Motivation

The OpenSky Network is an open-source initiative aimed at providing transparent access to air traffic data, which can greatly enhance aviation research and applications. By developing a dedicated document loader, users will be able to seamlessly integrate flight data into their applications or analytical workflows, enabling insights into air traffic patterns, flight tracking, and operational analysis using LLMs.

Proposal (If applicable)

We are a group of four students from the University of Toronto working to develop this feature. If this idea is approved, we hope to submit a PR by mid-November.

Here is an outline of the changes we are anticipating to make:

Modifications to existing files
We plan to modify the following file:

libs/community/langchain_community/document_loaders/__init__.py to register our loader

libs/community/langchain_community/utilities/__init__.py to register our API Wrapper

New files
We plan to create the following files:

libs/community/langchain_community/document_loaders/openSky.py where we will define the class OpenSkyLoader which inherits from BaseLoader and OpenSkyAPIWrapper

libs/community/langchain_community/utilities/openSky.py where we will define the class OpenSkyAPIWrapper which inherits from BaseModel

Some Sample Code

class OpenSkyLoader(BaseLoader):
    """Load flight data from the OpenSky API."""

    def __init__(self, api_wrapper: OpenSkyAPIWrapper, mode: str, params: Optional[Dict[str, Any]] = None) -> None:
        """Initialize the loader with the API wrapper and parameters."""
        super().__init__()
        self.api_wrapper = api_wrapper
        self.mode = mode
        self.params = params or {}

    def lazy_load(self) -> Iterator[Document]:
        """Lazily load flight data based on the specified flight type."""

class OpenSkyAPIWrapper(BaseModel):

    def fetch_all_state_vectors(self) -> List[Dict[str, Any]]:
        response = requests.get("https://opensky-network.org/api/states/all")
        json_data = response.json()
        states = json_data.get("states", [])
        return self.parse_state_vectors(json_data['time'], states)

    def fetch_flights_by_time_interval(self, begin: int, end: int) -> List[Dict[str, Any]]:
        url = f"https://opensky-network.org/api/flights/all?begin={begin}&end={end}"
        response = requests.get(url)
        data = response.json()
        return self.parse_flight_data(data)

    def fetch_flights_by_aircraft(self, icao24: str, begin: int, end: int) -> List[Dict[str, Any]]:
        url = f"https://opensky-network.org/api/flights/aircraft?icao24={icao24}&begin={begin}&end={end}"
        response = requests.get(url)
        data = response.json()
        return self.parse_flight_data(data)

    def fetch_arrivals_by_airport(self, airport: str, begin: int, end: int) -> List[Dict[str, Any]]:
        url = f"https://opensky-network.org/api/flights/arrival?airport={airport}&begin={begin}&end={end}"
        response = requests.get(url)
        data = response.json()
        return self.parse_flight_data(data)

    def fetch_departures_by_airport(self, airport: str, begin: int, end: int) -> List[Dict[str, Any]]:
        url = f"https://opensky-network.org/api/flights/departure?airport={airport}&begin={begin}&end={end}"
        response = requests.get(url)
        data = response.json()
        return self.parse_flight_data(data)

    def parse_flight_data(self, flight_data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """Parse flight data from the API response."""

    def parse_state_vectors(self, timestamp: int, states: List[List[Any]]) -> List[Dict[str, Any]]:
        """Parse state vector data from the API response."""

[AndrewEffendi]

I have made a PR here
#27324

[AndrewEffendi]

Hi @efriis
Could we get an issue open for this idea? I would love to hear any feedback you have on this idea.