Create flight_status.py

tools/flight_status.py

@@ -0,0 +1,78 @@
+from typing import Any, Optional
+from smolagents.tools import Tool
+import requests
+
+class FlightStatus(Tool):
+    name = "flight_status"
+    description = "Fetches the status of a given flight based on the flight number using AviationStack API."
+    inputs = {
+        'flight': {'type': 'string', 'description': 'Flight Code to fetch the information.'},
+        'api_key': {'type': 'string', 'description': 'The API key for accessing the AviationStack data'}
+    }
+    output_type = {'flight_info': {'type': 'dict', 'description': 'A dictionary containing the flight status information or an error message'}}
+
+    def forward(self) -> dict:
+        """
+        Fetches the status of a flight by querying the AviationStack API.
+        Returns a dictionary containing flight status or error details.
+        """
+        flight_number = self.inputs.get('flight')
+        #api_key = self.inputs.get('api_key')
+        api_key = ""
+
+        # Ensure flight number and API key are provided
+        if not flight_number or not api_key:
+            return {'error': 'Missing flight number or API key'}
+
+        base_url = "http://api.aviationstack.com/v1/flights"
+        
+        # Parameters for the API request
+        params = {
+            'access_key': api_key,
+            'flight_iata': flight_number  # Flight number in IATA format (e.g., "AA100")
+        }
+
+        try:
+            # Send a GET request to the AviationStack API
+            #response = requests.get(base_url, params=params)
+            response = 200
+            
+            # Check if the request was successful
+            # if response.status_code == 200:
+            #     data = response.json()
+
+            #     # If no flight data is found
+            #     if not data.get("data"):
+            #         return {'error': f"No information available for flight {flight_number}."}
+
+            #     flight_data = data['data'][0]  # Get the first flight result
+            if response == 200:
+                flight_data = {'flight_date': '2025-02-16', 'flight_status': 'scheduled', 'departure': {'airport': 'John F Kennedy International', 'timezone': 'America/New_York', 'iata': 'JFK', 'icao': 'KJFK', 'terminal': '8', 'gate': '4', 'delay': None, 'scheduled': '2025-02-16T19:00:00+00:00', 'estimated': '2025-02-16T19:00:00+00:00', 'actual': None, 'estimated_runway': None, 'actual_runway': None}, 'arrival': {'airport': 'Heathrow', 'timezone': 'Europe/London', 'iata': 'LHR', 'icao': 'EGLL', 'terminal': '3', 'gate': None, 'baggage': None, 'delay': None, 'scheduled': '2025-02-17T06:50:00+00:00', 'estimated': '2025-02-17T06:50:00+00:00', 'actual': None, 'estimated_runway': None, 'actual_runway': None}, 'airline': {'name': 'American Airlines', 'iata': 'AA', 'icao': 'AAL'}, 'flight': {'number': '100', 'iata': 'AA100', 'icao': 'AAL100', 'codeshared': None}, 'aircraft': None, 'live': None}
+
+                # Extract relevant flight information
+                flight_info = {
+                    'flight_number': flight_data['flight']['iata'],
+                    'airline': flight_data['airline']['name'],
+                    'status': flight_data['flight_status'],
+                    'departure': {
+                        'airport': flight_data['departure']['airport'],
+                        'estimated': flight_data['departure'].get('estimated', 'N/A'),
+                        'actual': flight_data['departure'].get('estimated_runway', 'N/A'),
+                    },
+                    'arrival': {
+                        'airport': flight_data['arrival']['airport'],
+                        'estimated': flight_data['arrival'].get('estimated', 'N/A'),
+                        'actual': flight_data['arrival'].get('estimated_runway', 'N/A'),
+                    },
+                }
+
+                return flight_info
+            
+            else:
+                return {'error': f"Error fetching data for {flight_number}. Status code: {response.status_code}"}
+        
+        except Exception as e:
+            return {'error': f"An error occurred: {str(e)}"}
+
+    def __init__(self, *args, **kwargs):
+        self.is_initialized = True  # Marking the tool as initialized