aerocontext_awc/

client.rs

//! HTTP client and [`WeatherBriefingProvider`] implementation for the AWC
//! Data API.

use std::sync::Arc;

use aerocontext_core::{
    Area, AreaBriefingRequest, Briefing, ContextProvider, GeoPoint, NavDataSnapshot, Product,
    ProductKind, ProviderError, RouteBriefingRequest, WeatherBriefingProvider,
};
use async_trait::async_trait;
use chrono::Utc;
use reqwest::StatusCode;
use tracing::{debug, warn};
use url::Url;

use crate::decode;

/// Production base URL of the Data API.
///
/// Note the path inversion: the documentation lives at `/data/api/`, the
/// endpoints live under `/api/data/`.
pub const DEFAULT_BASE_URL: &str = "https://aviationweather.gov/api/data/";

/// Configuration for [`AwcClient`].
#[derive(Debug, Clone)]
pub struct AwcConfig {
    /// Total per-request deadline. A hung upstream must never stall a
    /// gather indefinitely.
    pub timeout: std::time::Duration,
    /// Base URL the per-product paths are joined onto. Point this at a
    /// mock server in tests.
    pub base_url: String,
    /// `User-Agent` to send. AWC filters traffic with default agents, so
    /// identify your application here.
    pub user_agent: String,
}

impl Default for AwcConfig {
    fn default() -> Self {
        Self {
            timeout: std::time::Duration::from_secs(30),
            base_url: DEFAULT_BASE_URL.to_owned(),
            user_agent: concat!("aerocontext-awc/", env!("CARGO_PKG_VERSION")).to_owned(),
        }
    }
}

/// Failure constructing an [`AwcClient`].
#[derive(Debug, thiserror::Error)]
pub enum ConfigError {
    /// The configured base URL did not parse or cannot carry paths.
    #[error("invalid AWC base URL {url:?}")]
    InvalidBaseUrl {
        /// The offending URL string.
        url: String,
        /// Parse failure, when the URL did not parse at all.
        #[source]
        cause: Option<url::ParseError>,
    },
    /// The HTTP client could not be constructed.
    #[error("could not build HTTP client")]
    HttpClient(#[source] reqwest::Error),
}

/// aviationweather.gov Data API client.
#[derive(Debug, Clone)]
pub struct AwcClient {
    http: reqwest::Client,
    base_url: Url,
    nav_data: Option<Arc<NavDataSnapshot>>,
}

impl AwcClient {
    /// Build a client from `config`.
    pub fn new(config: AwcConfig) -> Result<Self, ConfigError> {
        let base_url =
            Url::parse(&config.base_url).map_err(|cause| ConfigError::InvalidBaseUrl {
                url: config.base_url.clone(),
                cause: Some(cause),
            })?;
        if base_url.cannot_be_a_base() {
            return Err(ConfigError::InvalidBaseUrl {
                url: config.base_url,
                cause: None,
            });
        }
        let http = reqwest::Client::builder()
            .user_agent(config.user_agent)
            .timeout(config.timeout)
            .connect_timeout(std::time::Duration::from_secs(10))
            .build()
            .map_err(ConfigError::HttpClient)?;
        Ok(Self {
            http,
            base_url,
            nav_data: None,
        })
    }

    /// Attach a cycle-tagged navigation-data snapshot for resolving
    /// identifier-based areas before calling AWC.
    #[must_use]
    pub fn with_nav_data(mut self, nav_data: NavDataSnapshot) -> Self {
        self.nav_data = Some(Arc::new(nav_data));
        self
    }

    fn endpoint(&self, path: &str) -> Result<Url, ProviderError> {
        let mut url = self.base_url.clone();
        url.path_segments_mut()
            .map_err(|()| ProviderError::Unsupported {
                reason: format!("base URL {} cannot carry endpoint paths", self.base_url),
            })?
            .pop_if_empty()
            .push(path);
        Ok(url)
    }

    /// GET one product endpoint and return the body, or `None` when the
    /// API answers `204 No Content` (a valid query that matched no data —
    /// AWC signals "empty" with 204, not with an empty JSON array).
    async fn fetch(
        &self,
        path: &str,
        query: &[(&str, String)],
    ) -> Result<Option<String>, ProviderError> {
        let url = self.endpoint(path)?;
        let context = || format!("fetching {path}");
        debug!(%url, "requesting AWC product");
        let response = self
            .http
            .get(url)
            .query(query)
            .send()
            .await
            .map_err(|cause| ProviderError::Transport {
                context: context(),
                cause: Box::new(cause),
            })?;
        let status = response.status();
        if status == StatusCode::NO_CONTENT {
            return Ok(None);
        }
        let body = response
            .text()
            .await
            .map_err(|cause| ProviderError::Transport {
                context: context(),
                cause: Box::new(cause),
            })?;
        if !status.is_success() {
            return Err(ProviderError::Provider {
                context: context(),
                message: body.chars().take(200).collect(),
                status: Some(status.as_u16()),
            });
        }
        Ok(Some(body))
    }

    async fn fetch_products(
        &self,
        kind: &ProductKind,
        bbox: &str,
        clip: (GeoPoint, GeoPoint),
        lookback_hours: Option<u32>,
    ) -> Result<Vec<Product>, ProviderError> {
        let format = ("format", "json".to_owned());
        let area = ("bbox", bbox.to_owned());
        match kind {
            ProductKind::Metar => {
                let mut query = vec![area, format];
                if let Some(hours) = lookback_hours {
                    query.push(("hours", hours.to_string()));
                }
                self.decode(self.fetch("metar", &query).await?, "metar", decode::metars)
            }
            ProductKind::Taf => self.decode(
                self.fetch("taf", &[area, format]).await?,
                "taf",
                decode::tafs,
            ),
            ProductKind::Pirep => {
                // PIREP names its lookback parameter `age`, not `hours`.
                let mut query = vec![area, format];
                if let Some(hours) = lookback_hours {
                    query.push(("age", hours.to_string()));
                }
                self.decode(self.fetch("pirep", &query).await?, "pirep", decode::pireps)
            }
            ProductKind::Sigmet => {
                // `airsigmet` has no bbox parameter: it returns the whole
                // CONUS set, which must be clipped client-side.
                let body = self.fetch("airsigmet", &[format]).await?;
                let all = self.decode(body, "airsigmet", decode::air_sigmets)?;
                Ok(decode::clip_to_bbox(all, clip))
            }
            other => Err(ProviderError::Unsupported {
                reason: format!("AWC adapter does not serve {other:?} products"),
            }),
        }
    }

    fn decode<T>(
        &self,
        body: Option<String>,
        what: &str,
        decoder: impl Fn(&str) -> Result<T, serde_json::Error>,
    ) -> Result<T, ProviderError>
    where
        T: Default,
    {
        match body {
            None => Ok(T::default()),
            Some(body) => decoder(&body).map_err(|cause| ProviderError::Decode {
                context: format!("decoding {what} JSON"),
                cause: Box::new(cause),
            }),
        }
    }

    fn enclosing_bbox(&self, area: &Area) -> Result<(GeoPoint, GeoPoint), ProviderError> {
        if let Some(bbox) = area.enclosing_bbox() {
            return Ok(bbox);
        }
        let Area::LocationRadius { ident, radius_nm } = area else {
            return Err(ProviderError::Unsupported {
                reason: format!("AWC needs a coordinate area; {area:?} has no equivalent"),
            });
        };
        let nav_data = self
            .nav_data
            .as_ref()
            .ok_or_else(|| ProviderError::Unsupported {
                reason: "AWC needs coordinates for identifier areas; attach a \
                         cycle-tagged navigation-data snapshot"
                    .to_owned(),
            })?;
        let point = nav_data
            .resolve(ident)
            .ok_or_else(|| ProviderError::Unsupported {
                reason: format!(
                    "AWC needs coordinates for {ident}; navigation-data snapshot effective {} \
                     does not contain it",
                    nav_data.cycle.effective_on
                ),
            })?;
        Area::PointRadius {
            center: point.position,
            radius_nm: *radius_nm,
        }
        .enclosing_bbox()
        .ok_or_else(|| ProviderError::Unsupported {
            reason: format!("AWC could not convert {ident} radius {radius_nm} NM to a bbox"),
        })
    }
}

/// Corridor segment length per AWC bbox request: large enough to keep the
/// request count low, small enough that each box stays a sane query.
const ROUTE_SEGMENT_NM: f64 = 250.0;

/// Products fetched when [`AreaBriefingRequest::products`] is empty.
const DEFAULT_PRODUCTS: [ProductKind; 4] = [
    ProductKind::Metar,
    ProductKind::Taf,
    ProductKind::Pirep,
    ProductKind::Sigmet,
];

impl ContextProvider for AwcClient {
    fn name(&self) -> &str {
        "awc"
    }
}

#[async_trait(?Send)]
impl WeatherBriefingProvider for AwcClient {
    async fn area_briefing(
        &self,
        request: &AreaBriefingRequest,
    ) -> Result<Briefing, ProviderError> {
        if request.departure_at.is_some() {
            // AWC serves current observations and active forecasts; the
            // ETD anchors navdata selection upstream but cannot shift
            // these products in time.
            warn!("departure_at has no AWC equivalent; products reflect current conditions");
        }
        let (south_west, north_east) = self.enclosing_bbox(&request.area)?;
        // AWC's bbox parameter is latitude-first: lat0,lon0,lat1,lon1.
        let bbox = format!(
            "{},{},{},{}",
            south_west.lat, south_west.lon, north_east.lat, north_east.lon
        );
        let kinds: &[ProductKind] = if request.products.is_empty() {
            &DEFAULT_PRODUCTS
        } else {
            &request.products
        };
        let mut products = Vec::new();
        // Sequential on purpose: AWC asks for at most one in-flight
        // request per thread.
        for kind in kinds {
            let batch = self
                .fetch_products(
                    kind,
                    &bbox,
                    (south_west, north_east),
                    request.lookback_hours,
                )
                .await?;
            debug!(kind = ?kind, count = batch.len(), "decoded AWC products");
            products.extend(batch);
        }
        Ok(Briefing::new(self.name())
            .with_generated_at(Some(Utc::now()))
            .with_products(products))
    }

    async fn route_briefing(
        &self,
        request: &RouteBriefingRequest,
    ) -> Result<Briefing, ProviderError> {
        // A corridor needs a leg; refuse a degenerate route loudly rather
        // than return an empty brief — parity with the Leidos provider.
        if request.waypoints.len() < 2 {
            return Err(ProviderError::Unsupported {
                reason: "a route briefing needs at least two waypoints".to_owned(),
            });
        }
        // AWC has no route product; the proper composition is one bbox
        // request per corridor segment, merged. Boxes are capped in span
        // so a long route does not become one continent-sized query.
        let kinds: &[ProductKind] = if request.products.is_empty() {
            &DEFAULT_PRODUCTS
        } else {
            &request.products
        };
        let mut products: Vec<Product> = Vec::new();
        for area in request.segment_bboxes(ROUTE_SEGMENT_NM) {
            let (south_west, north_east) = self.enclosing_bbox(&area)?;
            let bbox = format!(
                "{},{},{},{}",
                south_west.lat, south_west.lon, north_east.lat, north_east.lon
            );
            for kind in kinds {
                let batch = self
                    .fetch_products(kind, &bbox, (south_west, north_east), None)
                    .await?;
                for product in batch {
                    // Adjacent boxes overlap; the same station must not
                    // appear twice.
                    if !products.iter().any(|existing| {
                        existing.kind == product.kind && existing.raw_text == product.raw_text
                    }) {
                        products.push(product);
                    }
                }
            }
        }
        Ok(Briefing::new(self.name())
            .with_generated_at(Some(Utc::now()))
            .with_products(products))
    }
}