bsherriff/aviation
1
1
Fork 0
Code Issues 10 Pull Requests Actions Packages Projects 1 Releases Wiki Activity

Updated metar querying logic and parsing

Browse Source
This commit is contained in:
Ben Sherriff
2023-12-19 12:35:47 -05:00
parent 97be61e297
commit 6ecd9598e7
9 changed files with 198 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
service/docker-compose.yml
View File

@@ -22,6 +22,8 @@ services:
redis:
image: redis:latest
container_name: weather-redis
volumes:
- redis:/data
ports:
- ${REDIS_PORT:-6379}:6379
networks:
@@ -72,6 +74,7 @@ services:
volumes:
db:
db_logs:
redis:
minio:
networks:

3
service/migrations/000000_airports/up.sql
View File

@@ -7,8 +7,7 @@ CREATE TABLE IF NOT EXISTS airports (
iso_country TEXT NOT NULL,
iso_region TEXT NOT NULL,
municipality TEXT NOT NULL,
iata_code TEXT NOT NULL,
local_code TEXT NOT NULL,
has_metar BOOLEAN NOT NULL DEFAULT FALSE,
point GEOMETRY(POINT,4326) NOT NULL,
data JSONB NOT NULL
);

2
service/migrations/000001_metars/up.sql
View File

@@ -1,6 +1,6 @@
CREATE TABLE IF NOT EXISTS metars (
id INTEGER PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
station_id TEXT NOT NULL,
icao TEXT NOT NULL,
observation_time TIMESTAMP NOT NULL,
raw_text TEXT NOT NULL,
data JSONB NOT NULL

116
service/src/airports/model.rs
View File

@@ -1,3 +1,4 @@
use std::fmt::Display;
use std::str::FromStr;
use crate::db;
@@ -12,16 +13,19 @@ use serde::{Deserialize, Serialize};
#[derive(Serialize, Deserialize)]
pub struct Airport {
pub icao: String,
pub category: String,
pub category: AirportCategory,
pub name: String,
pub elevation_ft: f32,
pub iso_country: String,
pub iso_region: String,
pub municipality: String,
pub iata_code: String,
pub local_code: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub iata_code: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub local_code: Option<String>,
pub latitude: f64,
pub longitude: f64,
#[serde(skip_serializing_if = "Option::is_none")]
pub has_tower: Option<bool>,
}
@@ -29,14 +33,13 @@ impl Into<QueryAirport> for Airport {
fn into(self) -> QueryAirport {
return QueryAirport {
icao: self.icao.clone(),
category: self.category.clone(),
category: self.category.clone().to_string(),
name: self.name.clone(),
elevation_ft: self.elevation_ft,
iso_country: self.iso_country.clone(),
iso_region: self.iso_region.clone(),
municipality: self.municipality.clone(),
iata_code: self.iata_code.clone(),
local_code: self.local_code.clone(),
has_metar: false,
point: Point::new(self.longitude, self.latitude, Some(4326)),
data: match serde_json::to_value(&self) {
Ok(d) => d,
@@ -55,6 +58,57 @@ impl From<QueryAirport> for Airport {
}
}
#[derive(Serialize, Deserialize, Clone, Debug)]
pub enum AirportCategory {
#[serde(rename = "small_airport")]
Small,
#[serde(rename = "medium_airport")]
Medium,
#[serde(rename = "large_airport")]
Large,
#[serde(rename = "heliport")]
Heliport,
#[serde(rename = "closed")]
Closed,
#[serde(rename = "seaplane_base")]
Seaplane,
#[serde(rename = "balloonport")]
Balloonport,
#[serde(rename = "unknown")]
Unknown
}
impl FromStr for AirportCategory {
type Err = ();
fn from_str(s: &str) -> Result<Self, Self::Err> {
match s {
"small_airport" => Ok(AirportCategory::Small),
"medium_airport" => Ok(AirportCategory::Medium),
"large_airport" => Ok(AirportCategory::Large),
"heliport" => Ok(AirportCategory::Heliport),
"closed" => Ok(AirportCategory::Closed),
"seaplane_base" => Ok(AirportCategory::Seaplane),
"balloonport" => Ok(AirportCategory::Balloonport),
_ => Ok(AirportCategory::Unknown)
}
}
}
impl Display for AirportCategory {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
AirportCategory::Small => write!(f, "small_airport"),
AirportCategory::Medium => write!(f, "medium_airport"),
AirportCategory::Large => write!(f, "large_airport"),
AirportCategory::Heliport => write!(f, "heliport"),
AirportCategory::Closed => write!(f, "closed"),
AirportCategory::Seaplane => write!(f, "seaplane_base"),
AirportCategory::Balloonport => write!(f, "balloonport"),
AirportCategory::Unknown => write!(f, "unknown")
}
}
}
#[derive(Serialize, Deserialize, AsChangeset, Insertable, Queryable, QueryableByName)]
#[diesel(table_name = airports)]
pub struct QueryAirport {
@@ -65,8 +119,7 @@ pub struct QueryAirport {
pub iso_country: String,
pub iso_region: String,
pub municipality: String,
pub iata_code: String,
pub local_code: String,
pub has_metar: bool,
pub point: Point,
pub data: serde_json::Value
}
@@ -75,7 +128,7 @@ pub struct QueryAirport {
pub struct QueryFilters {
pub search: Option<String>,
pub bounds: Option<Polygon<Point>>,
pub categories: Option<Vec<String>>,
pub categories: Option<Vec<AirportCategory>>,
pub order_field: Option<QueryOrderField>,
pub order_by: Option<QueryOrderBy>
}
@@ -144,33 +197,34 @@ impl QueryAirport {
let mut query: String = "SELECT * FROM airports".to_string();
query = format!("{} {}", query, QueryAirport::build_filter_query(&filters)?);
query = format!("{} ORDER BY has_metar DESC", query);
if let Some(order_by) = &filters.order_by {
match order_by {
QueryOrderBy::Asc => {
if let Some(order_field) = &filters.order_field {
query = match order_field {
QueryOrderField::Icao => format!("{} ORDER BY icao ASC", query),
QueryOrderField::Name => format!("{} ORDER BY name ASC", query),
QueryOrderField::Category => format!("{} ORDER BY category ASC", query),
QueryOrderField::Country => format!("{} ORDER BY iso_country ASC", query),
QueryOrderField::Region => format!("{} ORDER BY iso_region ASC", query),
QueryOrderField::Municipality => format!("{} ORDER BY municipality ASC", query),
QueryOrderField::Iata => format!("{} ORDER BY iata_code ASC", query),
QueryOrderField::Local => format!("{} ORDER BY local_code ASC", query),
QueryOrderField::Icao => format!("{}, icao ASC", query),
QueryOrderField::Name => format!("{}, name ASC", query),
QueryOrderField::Category => format!("{}, category ASC", query),
QueryOrderField::Country => format!("{}, iso_country ASC", query),
QueryOrderField::Region => format!("{}, iso_region ASC", query),
QueryOrderField::Municipality => format!("{}, municipality ASC", query),
QueryOrderField::Iata => format!("{}, iata_code ASC", query),
QueryOrderField::Local => format!("{}, local_code ASC", query),
};
};
},
QueryOrderBy::Desc => {
if let Some(order_field) = &filters.order_field {
query = match order_field {
QueryOrderField::Icao => format!("{} ORDER BY icao DESC", query),
QueryOrderField::Name => format!("{} ORDER BY name DESC", query),
QueryOrderField::Category => format!("{} ORDER BY category DESC", query),
QueryOrderField::Country => format!("{} ORDER BY iso_country DESC", query),
QueryOrderField::Region => format!("{} ORDER BY iso_region DESC", query),
QueryOrderField::Municipality => format!("{} ORDER BY municipality DESC", query),
QueryOrderField::Iata => format!("{} ORDER BY iata_code DESC", query),
QueryOrderField::Local => format!("{} ORDER BY local_code DESC", query),
QueryOrderField::Icao => format!("{}, icao DESC", query),
QueryOrderField::Name => format!("{}, name DESC", query),
QueryOrderField::Category => format!("{}, category DESC", query),
QueryOrderField::Country => format!("{}, iso_country DESC", query),
QueryOrderField::Region => format!("{}, iso_region DESC", query),
QueryOrderField::Municipality => format!("{}, municipality DESC", query),
QueryOrderField::Iata => format!("{}, iata_code DESC", query),
QueryOrderField::Local => format!("{}, local_code DESC", query),
};
};
}
@@ -219,10 +273,12 @@ impl QueryAirport {
}
}
if let Some(categories) = &filters.categories {
parts.push(format!("({})", categories.iter().map(|category| format!("category = '{}'", category)).collect::<Vec<String>>().join(" OR ")));
parts.push(format!("({})", categories.iter().map(|category| format!("category = '{}'", category.to_string())).collect::<Vec<String>>().join(" OR ")));
}
if let Some(search) = &filters.search {
let search_strs = vec!["icao", "name", "iso_country", "iso_region", "municipality", "iata_code", "local_code"];
// Sanitize search to only allow [a-zA-Z0-9-\\s]
let search = search.chars().filter(|c| c.is_alphanumeric() || *c == '-' || *c == ' ').collect::<String>();
let search_strs = vec!["icao", "name", "iso_country", "iso_region", "municipality"];
parts.push(format!("({})", search_strs.iter().map(|s| format!("{} ILIKE '%{}%'", s, search)).collect::<Vec<String>>().join(" OR ")));
}
@@ -233,7 +289,7 @@ impl QueryAirport {
return Ok(query);
}
pub fn find(icao: String) -> Result<Self, ServiceError> {
pub fn get(icao: &str) -> Result<Self, ServiceError> {
let mut conn = db::connection()?;
let airport = airports::table.filter(airports::icao.eq(icao)).first(&mut conn)?;
Ok(airport)
@@ -261,10 +317,10 @@ impl QueryAirport {
Ok(inserted_airports)
}
pub fn update(icao: String, airport: Self) -> Result<Self, ServiceError> {
pub fn update(airport: Self) -> Result<Self, ServiceError> {
let mut conn = db::connection()?;
let airport = diesel::update(airports::table)
.filter(airports::icao.eq(icao))
.filter(airports::icao.eq(airport.icao.clone()))
.set(airport)
.get_result(&mut conn)?;
Ok(airport)

10
service/src/airports/routes.rs
View File

@@ -1,7 +1,7 @@
use std::str::FromStr;
use futures_util::stream::StreamExt as _;
use crate::{airports::{QueryAirport, QueryFilters, QueryOrderField, QueryOrderBy, Airport}, db::{Response, Metadata}, auth::{JwtAuth, verify_role}};
use crate::{airports::{QueryAirport, QueryFilters, QueryOrderField, QueryOrderBy, Airport, AirportCategory}, db::{Response, Metadata}, auth::{JwtAuth, verify_role}};
use actix_multipart::Multipart;
use actix_web::{delete, get, post, put, web, HttpResponse, HttpRequest, ResponseError};
use log::{error, warn};
@@ -70,7 +70,7 @@ async fn get_all(req: HttpRequest) -> HttpResponse {
let mut filters = QueryFilters::default();
filters.search = params.search.clone();
filters.categories = match &params.categories {
Some(c) => Some(c.split(",").map(|s| s.to_string()).collect()),
Some(c) => Some(c.split(",").map(|s| AirportCategory::from_str(s).unwrap()).collect()),
None => None
};
filters.bounds = match &params.bounds {
@@ -163,7 +163,7 @@ async fn get_all(req: HttpRequest) -> HttpResponse {
#[get("/{icao}")]
async fn get(icao: web::Path<String>) -> HttpResponse {
match QueryAirport::find(icao.into_inner()) {
match QueryAirport::get(&icao.into_inner()) {
Ok(a) => {
let airport: Airport = a.into();
HttpResponse::Ok().json(Response {
@@ -198,13 +198,13 @@ async fn create(airport: web::Json<Airport>, auth: JwtAuth) -> HttpResponse {
}
#[put("/{icao}")]
async fn update(icao: web::Path<String>, airport: web::Json<Airport>, auth: JwtAuth) -> HttpResponse {
async fn update(_icao: web::Path<String>, airport: web::Json<Airport>, auth: JwtAuth) -> HttpResponse {
let _ = match verify_role(&auth, "admin") {
Ok(_) => {},
Err(err) => return ResponseError::error_response(&err)
};
let query_airport: QueryAirport = airport.into_inner().into();
match QueryAirport::update(icao.into_inner(), query_airport) {
match QueryAirport::update(query_airport) {
Ok(a) => {
let airport: Airport = a.into();
HttpResponse::Ok().json(airport)

5
service/src/db/schema.rs
View File

@@ -9,8 +9,7 @@ diesel::table! {
iso_country -> Text,
iso_region -> Text,
municipality -> Text,
iata_code -> Text,
local_code -> Text,
has_metar -> Bool,
point -> Geometry,
data -> Jsonb
}
@@ -19,7 +18,7 @@ diesel::table! {
diesel::table! {
metars (id) {
id -> Integer,
station_id -> Text,
icao -> Text,
observation_time -> Timestamp,
raw_text -> Text,
data -> Jsonb,

105
service/src/metars/model.rs
View File

@@ -1,3 +1,4 @@
use crate::airports::QueryAirport;
use crate::{error_handler::ServiceError, db};
use crate::db::schema::metars::{self};
use chrono::Datelike;
@@ -17,7 +18,9 @@ pub struct QualityControlFlags {
#[serde(skip_serializing_if = "Option::is_none")]
pub maintenance_indicator_on: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub corrected: Option<bool>
pub corrected: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub no_significant_change: Option<bool>,
}
impl Default for QualityControlFlags {
@@ -28,6 +31,7 @@ impl Default for QualityControlFlags {
auto_station_with_precipication: None,
maintenance_indicator_on: None,
corrected: None,
no_significant_change: None,
}
}
}
@@ -36,14 +40,17 @@ impl Default for QualityControlFlags {
pub struct SkyCondition {
pub sky_cover: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub cloud_base_ft_agl: Option<i32>
pub cloud_base_ft_agl: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub significant_convective_clouds: Option<String>
}
impl Default for SkyCondition {
fn default() -> Self {
SkyCondition {
sky_cover: "".to_string(),
cloud_base_ft_agl: None
cloud_base_ft_agl: None,
significant_convective_clouds: None,
}
}
}
@@ -91,9 +98,9 @@ pub struct Metar {
#[serde(skip_serializing_if = "Option::is_none")]
pub wind_dir_degrees: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub wind_speed_kt: Option<i32>,
pub wind_speed_kt: Option<f64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub wind_gust_kt: Option<i32>,
pub wind_gust_kt: Option<f64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub variable_wind_dir_degrees: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
@@ -203,27 +210,39 @@ impl Metar {
metar.quality_control_flags.corrected = Some(true);
metar_parts.remove(0);
}
if !metar_parts.is_empty() && metar_parts[0] == "NOSIG" {
metar.quality_control_flags.no_significant_change = Some(true);
metar_parts.remove(0);
}
// Wind Direction and Speed
let wind_re = regex::Regex::new(r"^(?:[0-9]{3}|VRB)[0-9]{2}KT$").unwrap();
let wind_gust_re = regex::Regex::new(r"^(?:[0-9]{3}|VRB)[0-9]{2}G[0-9]{2}KT$").unwrap();
let wind_re = regex::Regex::new(r"^(?:[0-9]{3}|VRB)[0-9]{2}(?:KT|MPS)$").unwrap();
let wind_gust_re = regex::Regex::new(r"^(?:[0-9]{3}|VRB)[0-9]{2}G[0-9]{2}(?:KT|MPS)$").unwrap();
if !metar_parts.is_empty() && wind_re.is_match(metar_parts[0]) {
let wind = metar_parts[0];
metar_parts.remove(0);
let wind_dir_degrees = &wind[0..3];
let wind_speed_kt = &wind[3..5];
metar.wind_dir_degrees = Some(wind_dir_degrees.to_string());
metar.wind_speed_kt = Some(wind_speed_kt.parse::<i32>().unwrap());
let mut wind_speed_kt = wind[3..5].to_string();
// Convert m/s to kt
if wind.len() == 8 {
wind_speed_kt = (wind_speed_kt.parse::<f64>().unwrap() * 1.94384).to_string();
}
metar.wind_speed_kt = Some(wind_speed_kt.parse::<f64>().unwrap());
} else if !metar_parts.is_empty() && wind_gust_re.is_match(metar_parts[0]) {
let wind = metar_parts[0];
metar_parts.remove(0);
let wind_dir_degrees = &wind[0..3];
let wind_speed_kt = &wind[3..5];
metar.wind_dir_degrees = Some(wind_dir_degrees.to_string());
metar.wind_speed_kt = Some(wind_speed_kt.parse::<i32>().unwrap());
// Gust
let wind_gust_kt = &wind[6..8];
metar.wind_gust_kt = Some(wind_gust_kt.parse::<i32>().unwrap());
let mut wind_speed_kt = wind[3..5].to_string();
let mut wind_gust_kt = wind[6..8].to_string();
// Convert m/s to kt
if wind.len() == 9 {
wind_speed_kt = (wind_speed_kt.parse::<f64>().unwrap() * 1.94384).to_string();
wind_gust_kt = (wind_gust_kt.parse::<f64>().unwrap() * 1.94384).to_string();
}
metar.wind_speed_kt = Some(wind_speed_kt.parse::<f64>().unwrap());
metar.wind_gust_kt = Some(wind_gust_kt.parse::<f64>().unwrap());
}
// Variable Wind Direction
@@ -273,7 +292,7 @@ impl Metar {
// Runway Visual Range
let rvr_re = regex::Regex::new(r"^R[0-9]{1,3}(?:L|R|C)?/[PM]?[0-9]{4}FT$").unwrap();
let variable_rvr_re = regex::Regex::new(r"^R[0-9]{1,3}(?:L|R|C)?/[PM]?[0-9]{4}V[PM]?[0-9]{4}FT$").unwrap();
while !metar_parts.is_empty() && rvr_re.is_match(metar_parts[0]) || variable_rvr_re.is_match(metar_parts[0]) {
while !metar_parts.is_empty() && (rvr_re.is_match(metar_parts[0]) || variable_rvr_re.is_match(metar_parts[0])) {
let rvr_string = metar_parts[0];
metar_parts.remove(0);
let mut rvr = RunwayVisualRange::default();
@@ -303,7 +322,7 @@ impl Metar {
}
// Sky Condition
let sky_condition_re = regex::Regex::new(r"^(?:CLR|SKC|(?:FEW|SCT|BKN|OVC|VV)([0-9]{3})?)$").unwrap();
let sky_condition_re = regex::Regex::new(r"^(?:CLR|SKC|CAVOK|NSC|NCD|(?:FEW|SCT|BKN|OVC|VV)([0-9]{3})?(?:CB|TCU)?)$").unwrap();
while !metar_parts.is_empty() && sky_condition_re.is_match(metar_parts[0]) {
let sky_condition_string = metar_parts[0];
metar_parts.remove(0);
@@ -311,7 +330,20 @@ impl Metar {
let sky_cover = &sky_condition_string[0..3];
sky_condition.sky_cover = sky_cover.to_string();
if sky_condition_string.len() > 3 {
sky_condition.cloud_base_ft_agl = Some(sky_condition_string[3..sky_condition_string.len()].parse::<i32>().unwrap() * 100);
// Parse out the next three digits
let cloud_base_ft_agl = &sky_condition_string[3..6];
sky_condition.cloud_base_ft_agl = match cloud_base_ft_agl.parse::<i32>() {
Ok(c) => Some(c * 100),
Err(err) => {
warn!("Unable to parse cloud base in {}: {}", sky_condition_string, err);
None
}
};
if sky_condition_string.len() > 6 {
// Parse out the next two digits
let scc = &sky_condition_string[6..8];
sky_condition.significant_convective_clouds = Some(scc.to_string());
}
}
metar.sky_condition.push(sky_condition);
}
@@ -525,7 +557,7 @@ impl Metar {
let mut insert_metars: Vec<InsertMetar> = vec![];
for metar in metars {
insert_metars.push(InsertMetar {
station_id: metar.station_id.to_string(),
icao: metar.station_id.to_string(),
observation_time: metar.observation_time,
raw_text: metar.raw_text.to_string(),
data: serde_json::to_value(metar).unwrap()
@@ -552,6 +584,13 @@ impl Metar {
}
trace!("Retrieving missing METAR data for {:?}", missing_icaos);
let missing_icaos_string: Vec<String> = missing_icaos.iter().map(|icao| format!("{}", icao.to_string())).collect();
let mut airports: Vec<QueryAirport> = vec![];
missing_icaos_string.clone().iter().for_each(|icao| {
match QueryAirport::get(icao) {
Ok(a) => airports.push(a),
Err(_) => {}
}
});
let mut missing_metars = Self::get_remote_metars(missing_icaos_string.join(",")).await;
if missing_metars.len() > 0 {
let insert_metars = Self::to_insert(&missing_metars);
@@ -559,6 +598,27 @@ impl Metar {
Ok(rows) => trace!("Inserted {} metar rows", rows),
Err(err) => warn!("Unable to insert metar data; {}", err)
};
// Update airports with the appropriate has_metar flag
airports.iter().for_each(|airport| {
if missing_metars.iter().any(|metar| metar.station_id == airport.icao) {
let updated = QueryAirport {
icao: airport.icao.to_string(),
category: airport.category.to_string(),
name: airport.name.to_string(),
elevation_ft: airport.elevation_ft,
iso_country: airport.iso_country.to_string(),
iso_region: airport.iso_region.to_string(),
municipality: airport.municipality.to_string(),
has_metar: true,
point: airport.point,
data: airport.data.to_owned()
};
match QueryAirport::update(updated) {
Ok(_) => {},
Err(err) => warn!("Unable to update airport with has_metar flag; {}", err)
}
}
});
}
let mut metars: Vec<Metar> = vec![];
metars.append(&mut missing_metars);
@@ -570,7 +630,7 @@ impl Metar {
#[derive(Serialize, Deserialize, AsChangeset, Insertable)]
#[diesel(table_name = metars)]
struct InsertMetar {
station_id: String,
icao: String,
observation_time: chrono::NaiveDateTime,
raw_text: String,
data: serde_json::Value
@@ -590,7 +650,7 @@ impl InsertMetar {
#[diesel(table_name = metars)]
struct QueryMetar {
id: i32,
station_id: String,
icao: String,
observation_time: chrono::NaiveDateTime,
raw_text: String,
data: serde_json::Value
@@ -598,11 +658,12 @@ struct QueryMetar {
impl QueryMetar {
fn get_all(icaos: &Vec<&str>) -> Result<Vec<QueryMetar>, ServiceError> {
// Sanitize search to only allow [a-zA-Z0-9]
let icaos = icaos.iter().map(|icao| icao.chars().filter(|c| c.is_alphanumeric()).collect::<String>()).collect::<Vec<String>>();
let station_query: Vec<String> = icaos.iter().map(|icao| format!("'{}'", icao.to_string())).collect();
let mut conn = db::connection()?;
let db_metars: Vec<Self> = match sql_query(
format!("SELECT DISTINCT ON (station_id) * FROM metars WHERE station_id IN ({}) ORDER BY station_id, observation_time DESC", station_query.join(","))
format!("SELECT DISTINCT ON (icao) * FROM metars WHERE icao IN ({}) ORDER BY icao, observation_time DESC", station_query.join(","))
).load(&mut conn) {
Ok(m) => m,
Err(err) => return Err(ServiceError { status: 500, message: format!("{}", err) })