Upload 5 files

Agent.py

@@ -0,0 +1,154 @@
+import json
+import re
+from huggingface_hub import InferenceClient
+from schemas import TakeoffRequest
+from calculator import calculate_takeoff_roll_data
+import os
+# Initialize the client with your token
+hf_token = os.environ.get("Token_Apertus")
+
+# Fallback for local testing if needed, or raise error if missing
+if not hf_token:
+    # Optional: You can try to look for a local .env file or just warn the user
+    print("Warning: HF_TOKEN not found in environment variables.")
+
+client = InferenceClient(token=hf_token)
+
+def _extract_json_string(text):
+    """Helper to strip markdown code blocks if the LLM adds them."""
+    pattern = r"```json\s*(.*?)\s*```"
+    match = re.search(pattern, text, re.DOTALL)
+    if match:
+        return match.group(1)
+    return text
+
+def call_apertus_llm(system_prompt, user_prompt):
+    """
+    Calls the Apertus Instruct model via Hugging Face API.
+    """
+    
+    # 1. Define the model (Use the Instruct version for chat!)
+    model_id = "swiss-ai/Apertus-8B-Instruct-2509"
+
+    # 2. Format the messages
+    # The InferenceClient automatically applies the correct chat template 
+    # (user/assistant tags) required by Apertus.
+    messages = [
+        {"role": "system", "content": system_prompt},
+        {"role": "user", "content": user_prompt}
+    ]
+
+    try:
+        # 3. Call the API
+        response = client.chat_completion(
+            model=model_id,
+            messages=messages,
+            max_tokens=500,
+            temperature=0.1, # Keep low for JSON data extraction
+            seed=42          # Optional: Helps keep results deterministic
+        )
+
+        # 4. Extract the content
+        # The API returns a structured object; we just need the content string.
+        raw_content = response.choices[0].message.content
+
+        # 5. Clean JSON (strips markdown ```json ... ``` if present)
+        return _extract_json_string(raw_content)
+
+    except Exception as e:
+        # Graceful error handling for API timeouts or loading errors
+        print(f"API Error: {e}")
+        return "{}"
+
+class TakeoffAgent:
+    def __init__(self):
+        self.current_state = TakeoffRequest()
+    
+    def process_message(self, user_message: str):
+        """
+        1. Parse user message with LLM to update state.
+        2. Check correctness/completeness.
+        3. Calculate or Ask for more info.
+        """
+        
+        # 1. System Prompt construction
+        system_instruction = """
+        You are an extraction assistant for a PA-28 Flight Calculator.
+        Extract parameters from the user text into JSON format matching these keys:
+        - altitude_ft (float)
+        - qnh_hpa (float)
+        - temperature_c (float)
+        - weight_kg (float)
+        - wind_type ("Headwind" or "Tailwind")
+        - wind_speed_kt (float)
+        - safety_factor (float)
+        
+        Return ONLY valid JSON.
+        """
+        
+        # 2. Call LLM (In real life, pass the history + new message)
+        llm_response_json = call_apertus_llm(system_instruction, user_message)
+        
+        # 3. Update Pydantic Model (State Management)
+        try:
+            new_data = json.loads(llm_response_json)
+            # Update only fields that are present in the new extraction
+            updated_fields = self.current_state.model_dump(exclude_defaults=True)
+            updated_fields.update(new_data)
+            self.current_state = TakeoffRequest(**updated_fields)
+        except Exception as e:
+            # If LLM returns garbage, just ignore extraction for this turn
+            pass
+
+        # 4. Check Completeness (Guardrails)
+        if not self.current_state.is_complete():
+            missing = self.current_state.get_missing_fields()
+            response_text = f"I updated your flight parameters. I still need: {', '.join(missing)}.\n\n"
+            response_text += f"**Current State:**\n{self._format_state_summary()}"
+            return response_text
+
+        # 5. Check Correctness (Validation Logic)
+        warnings = []
+        if self.current_state.temperature_c > 45:
+            warnings.append("⚠️ Warning: Temperature is extremely high.")
+        if self.current_state.weight_kg > 1160:
+             warnings.append("⚠️ Warning: Weight exceeds typical MTOW.")
+
+        # 6. Run Calculation
+        try:
+            result = calculate_takeoff_roll_data(
+                indicated_altitude_ft=self.current_state.altitude_ft,
+                qnh_hpa=self.current_state.qnh_hpa,
+                temperature_c=self.current_state.temperature_c,
+                weight_kg=self.current_state.weight_kg,
+                wind_type=self.current_state.wind_type,
+                wind_speed=self.current_state.wind_speed_kt,
+                safety_factor=self.current_state.safety_factor
+            )
+            
+            # 7. Formulate Response
+            response = "### ✅ Takeoff Performance Calculated\n\n"
+            if warnings:
+                response += "**Alerts:**\n" + "\n".join(warnings) + "\n\n"
+            
+            response += f"**Environmental:**\n"
+            response += f"- Pressure Alt: {result['pressure_altitude']:.0f} ft\n"
+            response += f"- Density Alt: {result['density_altitude']:.0f} ft\n\n"
+            
+            response += f"**Ground Roll:**\n"
+            response += f"- Base: {result['ground_roll']['base']:.1f} ft\n"
+            response += f"- Corrections: Weight {result['ground_roll']['weight_adj']:.1f}, Wind {result['ground_roll']['wind_adj']:.1f}\n"
+            response += f"- **Final: {result['ground_roll']['final_m']:.0f} meters** ({result['ground_roll']['final_ft']:.0f} ft)\n\n"
+            
+            response += f"**50ft Obstacle:**\n"
+            response += f"- **Final: {result['obstacle_50ft']['final_m']:.0f} meters** ({result['obstacle_50ft']['final_ft']:.0f} ft)\n"
+            
+            return response
+            
+        except Exception as e:
+            return f"Error in calculation: {str(e)}"
+
+    def _format_state_summary(self):
+        s = self.current_state
+        return (f"- Alt: {s.altitude_ft} ft\n- Temp: {s.temperature_c} C\n"
+                f"- Weight: {s.weight_kg} kg\n- Wind: {s.wind_speed_kt} kt ({s.wind_type})")

TakeoffRequest.py

@@ -0,0 +1,35 @@
+from pydantic import BaseModel, Field, field_validator
+from typing import Optional, Literal
+
+class TakeoffRequest(BaseModel):
+    altitude_ft: Optional[float] = Field(None, description="Indicated altitude in feet")
+    qnh_hpa: Optional[float] = Field(1013.25, description="QNH pressure setting in hPa")
+    temperature_c: Optional[float] = Field(None, description="Outside air temperature in Celsius")
+    weight_kg: Optional[float] = Field(None, description="Aircraft mass in kg")
+    wind_type: Literal["Headwind", "Tailwind"] = Field("Headwind", description="Direction of wind component")
+    wind_speed_kt: Optional[float] = Field(None, description="Wind speed in knots")
+    safety_factor: float = Field(1.0, description="Safety factor multiplier (usually 1.0 to 1.5)")
+
+    @field_validator('weight_kg')
+    def check_weight(cls, v):
+        if v is not None and (v < 800 or v > 1500):
+             # Soft warning logic could go here, but for now we just validate bounds roughly
+             pass 
+        return v
+        
+    def is_complete(self) -> bool:
+        """Checks if all mandatory fields are present."""
+        return all(x is not None for x in [
+            self.altitude_ft, 
+            self.temperature_c, 
+            self.weight_kg, 
+            self.wind_speed_kt
+        ])
+    
+    def get_missing_fields(self) -> list[str]:
+        missing = []
+        if self.altitude_ft is None: missing.append("Altitude")
+        if self.temperature_c is None: missing.append("Temperature")
+        if self.weight_kg is None: missing.append("Aircraft Weight")
+        if self.wind_speed_kt is None: missing.append("Wind Speed")
+        return missing

app.py

@@ -0,0 +1,23 @@
+import gradio as gr
+from agent import TakeoffAgent
+
+# Initialize the agent
+agent = TakeoffAgent()
+
+def chat_response(message, history):
+    # The agent handles state, parsing, and calculation logic
+    response = agent.process_message(message)
+    return response
+
+# Clean Chat Interface
+with gr.Blocks(theme=gr.themes.Soft()) as demo:
+    gr.Markdown("#✈️ PA-28-181 AI Performance Engineer")
+    gr.Markdown("Describe your conditions naturally. Example: *'I am at 2000ft, 25C, QNH 1013, 1090kg with 10kt headwind'*")
+    
+    chat_interface = gr.ChatInterface(
+        fn=chat_response,
+        type="messages",
+    )
+
+if __name__ == "__main__":
+    demo.launch()

calculate_takeoff_roll_data.py

@@ -0,0 +1,163 @@
+import math
+
+# --- Data Digitized from the PA-28-181 Chart ---
+TAKEOFF_DATA = {
+    0:    {8.283945422852682: 892.4445513013161, 29.982568972771546: 1283.66893069664},
+    1000: {-3.8119853338943273: 888.693875097673, 30.775981246618983: 1507.1707639598485},
+    2000: {-12.181282683176057: 892.3002945242533, 30.905812345975832: 1674.6528821301918},
+    3000: {-22.421109575043573: 882.9236040151468, 31.045260563803566: 1854.5410831279678},
+    4000: {-29.855142153032396: 893.021578409569, 30.908216625593553: 2077.7544028370503},
+    5000: {-37.61856103864879: 878.2112159644166, 30.775981246618983: 2307.1707639598485},
+}
+
+TAKEOFF_DATA_50Ft = {
+    1000 : {-5.699053711848606: 1500, 4.250299880047962: 1770.491803278689, 12.581634013061446: 2024.5901639344265, 21.543382646941225: 2319.6721311475417, 29.2123150739704: 2598.3606557377057},
+    2000 : {-14.972007464676082:1494.6680885097298, -8.914956011730212:1681.9514796054377, -0.6291655558517704:1942.5486536923481, 5.737136763529719:2154.3588376432945, 12.092775259930704:2382.564649426819, 28.936283657691277:2985.670487869901},
+    3000 : {-23.417755265262592:1480.0053319114895, -18.011197014129564:1667.422020794454, -12.60463876299653:1854.8387096774186, -5.60917088776327:2099.306851506264, -0.21860837110104114:2311.3169821380957, 10.26392961876833:2694.4148227139426, 21.370301252999198:3118.368435083977, 28.978938949613436:3420.087976539589},
+    4000 : {-35.78245801119701:1490.735803785657, -27.155425219941343:1726.6728872300714, -21.73287123433751:1889.496134364169, -12.828579045587837:2199.1468941615562, -3.62036790189282:2541.5222607304713, 2.0954412156758337:2753.4657424686748, 10.64782724606772:3104.1722207411353, 20.122633964276204:3536.6568914956006, 28.63236470274593:3952.9458810983733},
+    5000 : { -37.53132498000533:1679.6187683284456, -28.595041322314046:1940.0826446280985, -18.11783524393494:2331.378299120234, -8.621700879765399:2731.0717142095436, 1.519594774726741:3138.829645427885, 12.1940815782458:3726.806185017328, 20.047987203412433:4151.426286323647, 25.03332444681419:4486.470274593441},
+}
+REFERENCE_WEIGHT_LBS = 2850.0
+MIN_CHART_WEIGHT_LBS = 2050.0
+
+def _interpolate_1d(x, x_points, y_points):
+    if not x_points or not y_points: return 0
+    if x <= x_points[0]: return y_points[0]
+    if x >= x_points[-1]: return y_points[-1]
+    for i in range(len(x_points) - 1):
+        if x_points[i] <= x <= x_points[i+1]:
+            x1, x2 = x_points[i], x_points[i+1]
+            y1, y2 = y_points[i], y_points[i+1]
+            if (x2 - x1) == 0: return y1
+            return y1 + (y2 - y1) * ((x - x1) / (x2 - x1))
+    return y_points[0]
+
+def _get_distance_at_temp_and_alt(temp, alt, data):
+    alt_points = sorted(data.keys())
+    if not alt_points: return 0
+    alt_low, alt_high = -1, -1
+    if alt <= alt_points[0]: alt_low = alt_high = alt_points[0]
+    elif alt >= alt_points[-1]: alt_low = alt_high = alt_points[-1]
+    else:
+        for i in range(len(alt_points) - 1):
+            if alt_points[i] <= alt <= alt_points[i+1]:
+                alt_low, alt_high = alt_points[i], alt_points[i+1]
+                break
+    
+    if alt_low not in data: return 0
+    temp_points_low = sorted(data[alt_low].keys())
+    dist_points_low = [data[alt_low][t] for t in temp_points_low]
+    dist_at_alt_low = _interpolate_1d(temp, temp_points_low, dist_points_low)
+    
+    if alt_low == alt_high: return dist_at_alt_low
+    
+    if alt_high not in data: return dist_at_alt_low
+    temp_points_high = sorted(data[alt_high].keys())
+    dist_points_high = [data[alt_high][t] for t in temp_points_high]
+    dist_at_alt_high = _interpolate_1d(temp, temp_points_high, dist_points_high)
+    
+    return _interpolate_1d(alt, [alt_low, alt_high], [dist_at_alt_low, dist_at_alt_high])
+
+def _calculate_weight_correction(base_distance, weight_lbs):
+    if weight_lbs >= REFERENCE_WEIGHT_LBS:
+        return 0.0
+    x_points = [1999.6411139821994, 2544.621494879893]
+    y_points = [840.3100775193798, 1410.8527131782948]
+    distance_at_actual_weight = _interpolate_1d(weight_lbs, x_points, y_points)
+    distance_at_reference_weight = _interpolate_1d(REFERENCE_WEIGHT_LBS, x_points, y_points)
+    weight_correction_delta = distance_at_actual_weight - distance_at_reference_weight
+    return weight_correction_delta
+
+def _calculate_weight_correction_50ft(base_distance, weight_lbs):
+    if weight_lbs >= REFERENCE_WEIGHT_LBS:
+        return 0.0
+    x_points = [2074.7658688865768, 2545.629552549428]
+    y_points = [1557.622268470344, 2718.652445369407]
+    distance_at_actual_weight = _interpolate_1d(weight_lbs, x_points, y_points)
+    distance_at_reference_weight = _interpolate_1d(REFERENCE_WEIGHT_LBS, x_points, y_points)
+    weight_correction_delta = distance_at_actual_weight - distance_at_reference_weight
+    return weight_correction_delta
+
+def _calculate_headwind_correction(wind_knots):
+    x_points = [0.16104294478526526, 15.04722311914756]
+    y_points = [1139.2960929932183, 847.9173393606702]
+    distance_at_actual_wind = _interpolate_1d(wind_knots, x_points, y_points)
+    distance_at_zero_wind = _interpolate_1d(0, x_points, y_points)
+    return distance_at_actual_wind - distance_at_zero_wind
+
+def _calculate_headwind_correction_50ft(wind_knots):
+    x_points = [-0.4118297401879545, 14.90049751243781]
+    y_points = [2083.6557950985825, 1657.0849456421615]
+    distance_at_actual_wind = _interpolate_1d(wind_knots, x_points, y_points)
+    distance_at_zero_wind = _interpolate_1d(0, x_points, y_points)
+    return distance_at_actual_wind - distance_at_zero_wind
+
+def _calculate_tailwind_correction(wind_knots):
+    x_points = [0.16104294478526526, 5.268404907975452]
+    y_points = [1139.2960929932183, 1414.1427187600893]
+    distance_at_actual_wind = _interpolate_1d(wind_knots, x_points, y_points)
+    distance_at_zero_wind = _interpolate_1d(0, x_points, y_points)
+    return distance_at_actual_wind - distance_at_zero_wind
+
+def _calculate_tailwind_correction_50ft(wind_knots):
+    x_points = [-0.10779436152570554, 4.483139856274192]
+    y_points = [1681.684171733924, 2061.912658927585]
+    distance_at_actual_wind = _interpolate_1d(wind_knots, x_points, y_points)
+    distance_at_zero_wind = _interpolate_1d(0, x_points, y_points)
+    return distance_at_actual_wind - distance_at_zero_wind
+
+def calculate_density_altitude(pressure_altitude_ft, outside_air_temp_c):
+    isa_temp_c = 15 - (2 * (pressure_altitude_ft / 1000))
+    return pressure_altitude_ft + (120 * (outside_air_temp_c - isa_temp_c))
+
+def calculate_takeoff_roll_data(indicated_altitude_ft, qnh_hpa, temperature_c, weight_kg, wind_type, wind_speed, safety_factor):
+    # Common calculations
+    weight_lbs = weight_kg * 2.20462
+    pressure_altitude = indicated_altitude_ft + ((1013.2 - qnh_hpa) * 27)
+    density_altitude = calculate_density_altitude(pressure_altitude, temperature_c)
+
+    # --- 0ft (Ground Roll) Calculation Path ---
+    base_distance_0ft = _get_distance_at_temp_and_alt(temperature_c, pressure_altitude, TAKEOFF_DATA)
+    weight_delta_0ft = _calculate_weight_correction(base_distance_0ft, weight_lbs)
+    distance_after_weight_0ft = base_distance_0ft + weight_delta_0ft
+    wind_delta_0ft = 0.0
+    if wind_type == "Headwind":
+        wind_delta_0ft = _calculate_headwind_correction(wind_speed)
+    elif wind_type == "Tailwind":
+        wind_delta_0ft = _calculate_tailwind_correction(wind_speed)
+    distance_after_wind_0ft = distance_after_weight_0ft + wind_delta_0ft
+    final_distance_ft_0ft = distance_after_wind_0ft * safety_factor
+    final_distance_m_0ft = final_distance_ft_0ft * 0.3048
+
+    # --- 50ft Obstacle Calculation Path ---
+    base_distance_50ft = _get_distance_at_temp_and_alt(temperature_c, pressure_altitude, TAKEOFF_DATA_50Ft)
+    weight_delta_50ft = _calculate_weight_correction_50ft(base_distance_50ft, weight_lbs)
+    distance_after_weight_50ft = base_distance_50ft + weight_delta_50ft
+    wind_delta_50ft = 0.0
+    if wind_type == "Headwind":
+        wind_delta_50ft = _calculate_headwind_correction_50ft(wind_speed)
+    elif wind_type == "Tailwind":
+        wind_delta_50ft = _calculate_tailwind_correction_50ft(wind_speed)
+    distance_after_wind_50ft = distance_after_weight_50ft + wind_delta_50ft
+    final_distance_ft_50ft = distance_after_wind_50ft * safety_factor
+    final_distance_m_50ft = final_distance_ft_50ft * 0.3048
+    
+    # Return dictionary for easy access
+    return {
+        "pressure_altitude": pressure_altitude,
+        "density_altitude": density_altitude,
+        "ground_roll": {
+            "base": base_distance_0ft,
+            "weight_adj": weight_delta_0ft,
+            "wind_adj": wind_delta_0ft,
+            "final_ft": final_distance_ft_0ft,
+            "final_m": final_distance_m_0ft
+        },
+        "obstacle_50ft": {
+            "base": base_distance_50ft,
+            "weight_adj": weight_delta_50ft,
+            "wind_adj": wind_delta_50ft,
+            "final_ft": final_distance_ft_50ft,
+            "final_m": final_distance_m_50ft
+        }
+    }

requirements.txt

@@ -0,0 +1 @@
+gradio