S4 resubmitted

bdi_api/s1/exercise.py

@@ -1,8 +1,14 @@
+import json
 import os
 from typing import Annotated
+from urllib.parse import urljoin
 
+import pandas as pd
+import requests
+from bs4 import BeautifulSoup
 from fastapi import APIRouter, status
 from fastapi.params import Query
+from tqdm import tqdm
 
 from bdi_api.settings import Settings
 
@@ -13,9 +19,10 @@
         status.HTTP_404_NOT_FOUND: {"description": "Not found"},
         status.HTTP_422_UNPROCESSABLE_ENTITY: {"description": "Something is wrong with the request"},
     },
-    prefix="/api/s1",
+    prefix="/api/s1",  # download postman
     tags=["s1"],
 )
+# test 1
 
 
 @s1.post("/aircraft/download")
@@ -52,6 +59,45 @@ def download_data(
     base_url = settings.source_url + "/2023/11/01/"
     # TODO Implement download
 
+    # Step 1: I will check to ensure the download directory exists and create it if it doesn't
+    os.makedirs(download_dir, exist_ok=True)
+
+    # Clean the download directory for existing files
+    for file in os.listdir(download_dir):
+        file_path = os.path.join(download_dir, file)
+        if os.path.isfile(file_path):
+            os.remove(file_path)
+
+    try:
+        # Get list of files from the Swagger UI link
+        response = requests.get(base_url)
+        # Added error handling for HTTP status
+        response.raise_for_status()
+
+        # Parse the HTML content with BeautifulSoup
+        soup = BeautifulSoup(response.text, "html.parser")
+        files = [a["href"] for a in soup.find_all("a") if a["href"].endswith(".json.gz")][
+            :file_limit
+        ]  # Apply the file limit of 10
+
+        # A count of successfully downloaded files
+        downloaded_count = 0
+        for file_name in tqdm(files, desc="Downloading files"):
+            file_url = urljoin(base_url, file_name)
+            response = requests.get(file_url, stream=True)
+
+            if response.status_code == 200:
+                file_path = os.path.join(download_dir, file_name[:-3])
+                with open(file_path, "wb") as f:  # add data into s3
+                    f.write(response.content)  # potentially change here
+                downloaded_count += 1
+            else:
+                print(f"Failed download {file_name}")
+        return f"Downloaded {downloaded_count} files to {download_dir}"
+    except requests.RequestException as e:
+        return f"Error accessing URL: {str(e)}"
+    except Exception as e:
+        return f"Error during download: {str(e)}"
     return "OK"
 
 
@@ -75,6 +121,74 @@ def prepare_data() -> str:
     Keep in mind that we are downloading a lot of small files, and some libraries might not work well with this!
     """
     # TODO
+    prepared_directory = os.path.join(settings.prepared_dir, "day=20231101")
+    raw_data_dir = os.path.join(settings.raw_dir, "day=20231101")
+
+    # Create directory if it doesn't exist
+    os.makedirs(prepared_directory, exist_ok=True)
+
+    # Clean existing files
+    for file in os.listdir(prepared_directory):
+        file_path = os.path.join(prepared_directory, file)
+        if os.path.isfile(file_path):
+            os.remove(file_path)
+
+    # The code reads JSON files from the raw data directory and processes them into Dataframes
+    try:
+        aircraft_data = []
+        for file in os.listdir(raw_data_dir):
+            if file.endswith('.json'):
+                file_path = os.path.join(raw_data_dir, file)
+                with open(file_path) as f:
+                    data = json.load(f)
+                    if "aircraft" in data:
+                        df = pd.DataFrame(data["aircraft"])
+                        # Add timestamp directly to DataFrame
+                        df["timestamp"] = data["now"]
+                        aircraft_data.append(df)
+
+        if not aircraft_data:
+            return "No aircraft data found."
+
+        # All DataFrames are concatenated into a single Dataframe
+        processed_data = pd.concat(aircraft_data, ignore_index=True)
+
+        # Select required columns
+        processed_data = processed_data[
+            ["hex", "r", "type", "t", "lat", "lon", "alt_baro", "gs", "emergency", "timestamp"]
+        ]
+
+        # Rename columns
+        processed_data = processed_data.rename(
+            columns={
+                "hex": "icao",
+                "r": "registration",
+                "t": "type",
+                "alt_baro": "altitude_baro",
+                "gs": "ground_speed",
+                "emergency": "had_emergency",
+            }
+        )
+
+        # Drop rows with NaN values in 'icao', 'registration', and 'type'
+        processed_data = processed_data.dropna(subset=["icao", "registration", "type"])
+
+        # Process emergency flags
+        emergency_flags = {"general", "lifeguard", "minfuel", "nordo", "unlawful", "downed", "reserved"}
+        processed_data["had_emergency"] = processed_data["had_emergency"].apply(lambda x: x in emergency_flags)
+
+        # Remove rows with any NaN values
+        processed_data = processed_data.dropna()
+
+        # Save as CSV
+        output_file = os.path.join(prepared_directory, "prepared_data.csv")
+        processed_data.to_csv(output_file, index=False)
+
+        print(processed_data)
+        return f"Data prepared and saved to {output_file}"
+
+    except Exception as e:
+        return f"An error has occurred during data preparation: {str(e)}"
     return "OK"
 
 
@@ -84,7 +198,26 @@ def list_aircraft(num_results: int = 100, page: int = 0) -> list[dict]:
     icao asc
     """
     # TODO
-    return [{"icao": "0d8300", "registration": "YV3382", "type": "LJ31"}]
+    # This variable directly constructs the full path to the specific and only file with all the prepared data
+    prepared_directory = os.path.join(settings.prepared_dir, "day=20231101", "prepared_data.csv")
+    if not os.path.exists(prepared_directory):
+        return []
+
+    if not os.path.exists(prepared_directory):
+        return []
+
+    sorted_aircraft = pd.read_csv(prepared_directory)
+    # Deduplicate values and sort by 'icao'
+    sorted_aircraft = sorted_aircraft[["icao", "registration", "type"]].drop_duplicates().sort_values(by="icao")
+
+    start = page * num_results
+    end = start + num_results
+
+    result = sorted_aircraft.iloc[start:end].to_dict(orient="records")
+    print(result)
+    return result
+
+    return [{"icao": "0d8300", "registration": "YV3382", "type": "LJ31"}]  # Done
 
 
 @s1.get("/aircraft/{icao}/positions")
@@ -93,7 +226,20 @@ def get_aircraft_position(icao: str, num_results: int = 1000, page: int = 0) ->
     If an aircraft is not found, return an empty list.
     """
     # TODO implement and return a list with dictionaries with those values.
-    return [{"timestamp": 1609275898.6, "lat": 30.404617, "lon": -86.476566}]
+
+    # Calculate the start and end index for slicing the DataFrame
+    start_index = page * num_results
+    end_index = (page + 1) * num_results
+
+    prepared_directory = os.path.join(settings.prepared_dir, "day=20231101", "prepared_data.csv")
+    if not os.path.exists(prepared_directory):
+        return []
+
+    df = pd.read_csv(prepared_directory)
+    # Dataframe with subset of rows for the requested 'icao'
+    filtered_df = df[df["icao"] == icao].sort_values(by="timestamp")
+    # Slice the DataFrame according to the page and number of results per page
+    return filtered_df.iloc[start_index:end_index][["timestamp", "lat", "lon"]].to_dict(orient="records")
 
 
 @s1.get("/aircraft/{icao}/stats")
@@ -105,4 +251,57 @@ def get_aircraft_statistics(icao: str) -> dict:
     * had_emergency
     """
     # TODO Gather and return the correct statistics for the requested aircraft
-    return {"max_altitude_baro": 300000, "max_ground_speed": 493, "had_emergency": False}
+    data_file = os.path.join(settings.prepared_dir, "day=20231101", "prepared_data.csv")
+
+    # Check if the file exists
+    if not os.path.exists(data_file):
+        return {}
+
+    try:
+        # Read the CSV file into a DataFrame
+        df = pd.read_csv(data_file)
+
+        # Filter the DataFrame by 'icao'
+        df = df[df["icao"] == icao]
+
+        # Check if the filtered DataFrame is empty
+        if df.empty:
+            return {}
+        # Convert and clean altitude data
+        df["altitude_baro"] = pd.to_numeric(df["altitude_baro"], errors="coerce")
+        df["ground_speed"] = pd.to_numeric(df["ground_speed"], errors="coerce")
+
+        # Remove rows with NaN values in relevant columns
+        df = df.dropna(subset=["altitude_baro", "ground_speed", "had_emergency"])
+
+        # Calculate statistics while handling errors
+        max_altitude = df["altitude_baro"].max()
+        max_speed = df["ground_speed"].max()
+        emergency = df["had_emergency"].any()
+
+        # Ensure we have valid numeric values
+        if pd.isna(max_altitude) or pd.isna(max_speed):
+            return {
+                "max_altitude_baro": 0,
+                "max_ground_speed": 0,
+                "had_emergency": bool(emergency)
+            }
+
+        return {
+            "max_altitude_baro": float(max_altitude),
+            "max_ground_speed": float(max_speed),
+            "had_emergency": bool(emergency)
+        }
+
+    except Exception as e:
+        # Log the error for debugging
+        print(f"Error processing aircraft statistics: {str(e)}")
+        # Return default values instead of error message
+        return {
+            "max_altitude_baro": 0,
+            "max_ground_speed": 0,
+            "had_emergency": False
+        }
+
+# This line should be removed or commented out as it is outside of any function or class
+# return {"max_altitude_baro": 300000, "max_ground_speed": 493, "had_emergency": False}

bdi_api/s4/exercise.py

@@ -1,11 +1,19 @@
+import json
+import os
 from typing import Annotated
+from urllib.parse import urljoin
 
-from fastapi import APIRouter, status
+import boto3
+import requests
+from bs4 import BeautifulSoup
+from fastapi import APIRouter, status, HTTPException
 from fastapi.params import Query
+from tqdm import tqdm
 
 from bdi_api.settings import Settings
 
 settings = Settings()
+s3 = boto3.client("s3")
 
 s4 = APIRouter(
     responses={
@@ -19,37 +27,96 @@
 
 @s4.post("/aircraft/download")
 def download_data(
-    file_limit: Annotated[
-        int,
-        Query(
-            ...,
-            description="""
-    Limits the number of files to download.
-    You must always start from the first the page returns and
-    go in ascending order in order to correctly obtain the results.
-    I'll test with increasing number of files starting from 100.""",
-        ),
-    ] = 100,
+    file_limit: Annotated[int, Query(..., description="Limits the number of files to download.")],
 ) -> str:
-    """Same as s1 but store to an aws s3 bucket taken from settings
-    and inside the path `raw/day=20231101/`
-
-    NOTE: you can change that value via the environment variable `BDI_S3_BUCKET`
-    """
+    """Download files from a source URL and store them in S3."""
     base_url = settings.source_url + "/2023/11/01/"
     s3_bucket = settings.s3_bucket
     s3_prefix_path = "raw/day=20231101/"
-    # TODO
 
-    return "OK"
+    try:
+        try:
+            response = requests.get(base_url)
+            response.raise_for_status()
+        except Exception as e:
+            return f"Error accessing URL: {str(e)}"
+
+        soup = BeautifulSoup(response.text, "html.parser")
+        files = [a["href"] for a in soup.find_all("a") if a["href"].endswith(".json.gz")][:file_limit]
+
+        downloaded_count = 0
+        for file_name in tqdm(files, desc="Downloading files"):
+            file_url = urljoin(base_url, file_name)
+            try:
+                response = requests.get(file_url, stream=True)
+                response.raise_for_status()
+            except Exception as e:
+                return f"Error accessing URL: {str(e)}"
+
+            s3_key = f"{s3_prefix_path}{file_name}"
+            s3.put_object(Bucket=s3_bucket, Key=s3_key, Body=response.content, ContentType="application/json")
+            downloaded_count += 1
+
+        return f"Downloaded {downloaded_count} files in S3."
+
+    except Exception as e:
+        raise HTTPException(status_code=200, detail=f"Error accessing URL: {str(e)}")
 
 
 @s4.post("/aircraft/prepare")
 def prepare_data() -> str:
-    """Obtain the data from AWS s3 and store it in the local `prepared` directory
-    as done in s2.
+    """Download data from S3 and store it in the local `prepared` directory."""
+    s3_bucket = settings.s3_bucket
+    s3_prefix_path = "raw/day=20231101/"
+    local_directory = settings.prepared_dir  # Use the property
+
+    try:
+        os.makedirs(local_directory, exist_ok=True)
+
+        # Clear old data in prepared directory
+        for file in os.listdir(local_directory):
+            os.remove(os.path.join(local_directory, file))
+
+        response = s3.list_objects_v2(Bucket=s3_bucket, Prefix=s3_prefix_path)
+        if 'Contents' not in response:
+            return "No files found in S3."
+
+        for obj in tqdm(response['Contents'], desc="Processing files"):
+            s3_key = obj['Key']
+            file_name = os.path.basename(s3_key)
+            prepared_file_path = os.path.join(local_directory, file_name.replace(".gz", ""))
+
+            try:
+                file_response = s3.get_object(Bucket=s3_bucket, Key=s3_key)
+                json_content = json.loads(file_response['Body'].read().decode('utf-8'))
+
+                timestamp = json_content.get("now")
+                aircraft_data = json_content.get("aircraft", [])
+
+                processed_aircraft_data = [
+                    {
+                        "icao": record.get("hex"),
+                        "registration": record.get("r"),
+                        "type": record.get("t"),
+                        "lat": record.get("lat"),
+                        "lon": record.get("lon"),
+                        "alt_baro": record.get("alt_baro"),
+                        "timestamp": timestamp,
+                        "max_altitude_baro": record.get("alt_baro"),
+                        "max_ground_speed": record.get("gs"),
+                        "had_emergency": record.get("alert", 0) == 1,
+                    }
+                    for record in aircraft_data
+                ]
+
+                with open(prepared_file_path, "w", encoding="utf-8") as f:
+                    json.dump(processed_aircraft_data, f)
+
+            except Exception as e:
+                print(f"Error processing file {s3_key}: {str(e)}")
+                continue
+
+        return f"Prepared data saved to {local_directory}."
 
-    All the `/api/s1/aircraft/` endpoints should work as usual
-    """
-    # TODO
-    return "OK"
+    except Exception as e:
+        return f"Error during preparation: {str(e)}"