Shipping a New Tidbyt App: LAX Departures

pixlet render lax_departures.star --gif --magnify 10

import json
import os
import boto3
import http.client
from datetime import datetime, timedelta

dynamodb = boto3.client('dynamodb')
CACHE_KEY = "lax_departures"
TABLE_NAME = os.environ.get("DDB_TABLE", "FlightCache")
CACHE_TTL_HOURS = 3

from datetime import datetime, timedelta

from datetime import datetime, timedelta

def extract_departures(data, now_utc, limit=10):
    departures = []

    for flight in data.get("departures", []):
        try:
            sched_utc = flight["movement"]["scheduledTime"]["utc"]
            sched_local = flight["movement"]["scheduledTime"]["local"]

            sched_dt_utc = datetime.strptime(sched_utc, "%Y-%m-%d %H:%MZ")
            sched_dt_local = datetime.strptime(sched_local[:16], "%Y-%m-%d %H:%M")  # drop tz

            if now_utc - timedelta(minutes=3) <= sched_dt_utc <= now_utc + timedelta(hours=3):
                departures.append({
                    "scheduled_time": sched_dt_local.strftime("%H:%M"),
                    "destination": flight["movement"]["airport"]["iata"],
                    "number": flight["number"],
                    "callsign": flight.get("callSign", "Unknown"),
                    "status": flight.get("status"),
                    "airline": flight["airline"]["iata"],
                    "aircraft": flight.get("aircraft", {}).get("model", ""),
                    "codeshare_status": flight.get("codeshareStatus", "Unknown"),
                    "is_past": sched_dt_utc < now_utc,
                    "_sort_key": sched_dt_utc  # store UTC datetime for sorting
                })
        except Exception:
            continue

    # Sort by UTC time
    departures.sort(key=lambda x: x["_sort_key"])

    # Remove _sort_key from each item before returning
    for dep in departures:
        dep.pop("_sort_key", None)

    return departures[:limit]




def lambda_handler(event, context):
    now = datetime.utcnow()

    # 1. Check DynamoDB cache
    try:
        response = dynamodb.get_item(
            TableName=TABLE_NAME,
            Key={"pk": {"S": CACHE_KEY}}
        )
        if "Item" in response:
            timestamp_str = response["Item"]["timestamp"]["S"]
            raw_data_str = response["Item"]["data"]["S"]
            cached_time = datetime.fromisoformat(timestamp_str)

            if now - cached_time < timedelta(hours=CACHE_TTL_HOURS):
                print("Using cached data.")
                data = json.loads(raw_data_str)
                cleaned = {
                    "updated": now.isoformat(),
                    "departures": extract_departures(data, now)
                }
                return {
                    "statusCode": 200,
                    "headers": {"Content-Type": "application/json"},
                    "body": json.dumps(cleaned)
                }

    except Exception as e:
        print(f"Error checking cache: {e}")

    # 2. Fetch from External API
    print("Fetching fresh data from External API")
    try:
        conn = http.client.HTTPSConnection("URL")

        headers = {
            'x-rapidapi-key': os.environ["API_KEY"],
            'x-rapidapi-host': "HOST"
        }

        path = (
            "/flights/airports/iata/LAX"
            "?offsetMinutes=-5"
            "&durationMinutes=720"
            "&withLeg=false"
            "&direction=Departure"
            "&withCancelled=true"
            "&withCodeshared=false"
            "&withCargo=false"
            "&withPrivate=false"
            "&withLocation=false"
        )

        conn.request("GET", path, headers=headers)
        res = conn.getresponse()
        raw_data = res.read()

        data = json.loads(raw_data.decode("utf-8"))

        # 3. Store full raw data in DynamoDB
        dynamodb.put_item(
            TableName=TABLE_NAME,
            Item={
                "pk": {"S": CACHE_KEY},
                "timestamp": {"S": now.isoformat()},
                "data": {"S": json.dumps(data)}
            }
        )

        # 4. Prepare cleaned response for Tidbyt
        cleaned = {
            "updated": now.isoformat(),
            "departures": extract_departures(data, now)
        }

        return {
            "statusCode": 200,
            "headers": {"Content-Type": "application/json"},
            "body": json.dumps(cleaned)
        }

    except Exception as e:
        print(f"Error fetching from API: {e}")
        return {
            "statusCode": 500,
            "body": json.dumps({"error": "Failed to fetch fresh data"})
        }

load("encoding/json.star", "json")
load("http.star", "http")
load("render.star", "render")

# Fetch data from your deployed Lambda API
def fetch_data():
    res = http.get("https://ENDPOINT.execute-api.us-east-1.amazonaws.com/lax-departures")
    if res.status_code != 200:
        return None

    data = json.decode(res.body())
    if type(data.get("departures", None)) != "list":
        return None

    return data["departures"][:6]  # First 6 flights max

# Build one page of output
def make_page(flights, page_index, total_pages):
    rows = []

    # Header row with left and right components
    rows.append(render.Row(children = [
        render.Text("Next From LAX", font = "tb-8", color = "#fcf7c5"),
        render.Text(" {}/{}".format(page_index + 1, total_pages), font = "5x8", color = "#666666"),
    ]))

    # Add flight rows
    for flight in flights:
        status = flight.get("status", "").lower()

        if "canceled" in status:
            time_color = "#ff5555"  # red
        elif flight.get("is_past"):
            time_color = "#888888"  # gray
        else:
            time_color = "#a8ffb0"  # light green

        rows.append(
            render.Row(children = [
                render.Text(flight["scheduled_time"], font = "5x8", color = time_color),
                render.Text(" {} {}".format(flight["airline"], flight["destination"]), font = "5x8"),
            ]),
        )

    return render.Column(children = rows)

# Main entrypoint
def main():
    flights = fetch_data()
    if flights == None:
        return render.Root(child = render.Text("Error", font = "5x8", color = "#ff0000"))

    pages = []
    total_pages = (min(len(flights), 6) + 2) // 3  # ceil(len / 3)

    for page_index, i in enumerate(range(0, min(len(flights), 6), 3)):
        page_flights = flights[i:i + 3]
        page = make_page(page_flights, page_index, total_pages)

        # Repeat page N times to control timing
        for _ in range(200):
            pages.append(page)

    return render.Root(
        child = render.Sequence(children = pages),
    )