nonapode/county-sprints
Watch
1
0
Fork
You've already forked county-sprints
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

state

Browse source
This commit is contained in:
Jonas Rabenstein 2026-08-12 17:49:27 +02:00
commit 45bcc13788
3 changed files with 678 additions and 174 deletions
Download patch file Download diff file Expand all files Collapse all files

97
scripts/import-vg250.sh
View file

@ -16,24 +16,55 @@ fi
: "${DATABASE_URL:?DATABASE_URL must be set}"
echo "Available layers:"
ogrinfo "$INPUT" | sed -n '1,80p'
ogrinfo "$INPUT" | sed -n '1,120p'
echo
echo "Importing vg250_krs..."
echo "Inspecting vg250_krs..."
ogrinfo "$INPUT" vg250_krs -so
echo
echo "Importing only vg250_krs..."
# Remove an old failed import.
psql "$DATABASE_URL" <<'SQL'
DROP TABLE IF EXISTS vg250_krs_import CASCADE;
SQL
# IMPORTANT:
# The layer name is explicitly specified here.
ogr2ogr \
-f PostgreSQL \
"$DATABASE_URL" \
"$INPUT" \
vg250_krs \
-nln vg250_krs_import \
-nlt PROMOTE_TO_MULTI \
-t_srs EPSG:4326 \
-lco GEOMETRY_NAME=wkb_geometry \
-overwrite
echo "Import completed."
echo
echo "Source import completed."
echo
echo "Imported source columns:"
psql "$DATABASE_URL" <<'SQL'
SELECT
column_name,
data_type
FROM information_schema.columns
WHERE table_name = 'vg250_krs_import'
ORDER BY ordinal_position;
SQL
echo
echo "Creating county boundaries..."
psql "$DATABASE_URL" <<'SQL'
TRUNCATE county_boundaries RESTART IDENTITY;
BEGIN;
TRUNCATE county_crossings;
TRUNCATE county_boundaries RESTART IDENTITY CASCADE;
INSERT INTO county_boundaries (
name,
@ -41,9 +72,9 @@ INSERT INTO county_boundaries (
)
SELECT
COALESCE(
NULLIF("GEN", ''),
NULLIF("BEZ", ''),
NULLIF("ARS", ''),
NULLIF(TRIM(gen), ''),
NULLIF(TRIM(bez), ''),
NULLIF(TRIM(ars), ''),
'unknown'
) AS name,
@ -52,13 +83,29 @@ SELECT
ST_MakeValid(wkb_geometry),
3
)
)::geometry(MultiPolygon,4326)
)::geometry(MultiPolygon, 4326) AS geometry
FROM vg250_krs_import
WHERE wkb_geometry IS NOT NULL;
WHERE
wkb_geometry IS NOT NULL
AND NOT ST_IsEmpty(wkb_geometry);
COMMIT;
SQL
echo
echo "Removing temporary import table..."
psql "$DATABASE_URL" <<'SQL'
DROP TABLE vg250_krs_import;
SQL
echo
echo "Recreating spatial index..."
psql "$DATABASE_URL" <<'SQL'
DROP INDEX IF EXISTS county_boundaries_geometry_idx;
CREATE INDEX county_boundaries_geometry_idx
ON county_boundaries
@ -68,6 +115,32 @@ ANALYZE county_boundaries;
SQL
echo
echo "Imported counties:"
psql "$DATABASE_URL" \
-c "SELECT id, name FROM county_boundaries ORDER BY name;"
echo "=========================================="
echo " Imported counties"
echo "=========================================="
psql "$DATABASE_URL" <<'SQL'
SELECT
COUNT(*) AS total,
COUNT(*) FILTER (WHERE geometry IS NOT NULL) AS with_geometry,
COUNT(*) FILTER (WHERE ST_IsValid(geometry)) AS valid_geometry
FROM county_boundaries;
SQL
echo
echo "First 20 counties:"
echo
psql "$DATABASE_URL" <<'SQL'
SELECT
id,
name,
ST_GeometryType(geometry) AS geometry_type,
ST_SRID(geometry) AS srid
FROM county_boundaries
ORDER BY name
LIMIT 20;
SQL
echo
echo "Import completed successfully."

429
src/intervals.rs
View file

@ -1,4 +1,4 @@
use anyhow::{Context, Result, anyhow};
use anyhow::{anyhow, Context, Result};
use chrono::{DateTime, NaiveDateTime, Utc};
use reqwest::Client;
use serde_json::Value;
@ -26,7 +26,10 @@ impl IntervalsClient {
/// Create a client using an explicitly supplied personal API key.
///
/// Primarily used by /dev/sync/{api_key}/{athlete_id}.
pub fn with_api_key(config: Config, api_key: impl Into<String>) -> Self {
pub fn with_api_key(
config: Config,
api_key: impl Into<String>,
) -> Self {
Self {
client: Client::new(),
config,
@ -35,13 +38,17 @@ impl IntervalsClient {
}
fn base_url(&self) -> &str {
self.config.intervals_base_url.trim_end_matches('/')
self.config
.intervals_base_url
.trim_end_matches('/')
}
fn api_key(&self) -> Result<&str> {
self.api_key
.as_deref()
.context("Intervals.icu API key is not configured")
.context(
"Intervals.icu API key is not configured",
)
}
fn api_url(&self, path: &str) -> String {
@ -52,30 +59,54 @@ impl IntervalsClient {
)
}
fn authenticated(&self, request: reqwest::RequestBuilder) -> Result<reqwest::RequestBuilder> {
Ok(request.basic_auth("API_KEY", Some(self.api_key()?)))
fn authenticated(
&self,
request: reqwest::RequestBuilder,
) -> Result<reqwest::RequestBuilder> {
Ok(
request.basic_auth(
"API_KEY",
Some(self.api_key()?),
)
)
}
async fn get_json(&self, url: String) -> Result<Value> {
async fn get_json(
&self,
url: String,
) -> Result<Value> {
let response = self
.authenticated(self.client.get(&url))?
.send()
.await
.context("Intervals.icu request failed")?;
.context(
"Intervals.icu request failed",
)?;
let status = response.status();
let body = response
.text()
.await
.context("cannot read Intervals.icu response")?;
.context(
"cannot read Intervals.icu response",
)?;
if !status.is_success() {
return Err(anyhow!("Intervals.icu returned HTTP {}: {}", status, body));
return Err(anyhow!(
"Intervals.icu returned HTTP {}: {}",
status,
body
));
}
serde_json::from_str(&body)
.with_context(|| format!("invalid JSON returned by Intervals.icu: {}", body))
.with_context(|| {
format!(
"invalid JSON returned by Intervals.icu: {}",
body
)
})
}
/// Get the athlete belonging to the configured API key.
@ -83,12 +114,18 @@ impl IntervalsClient {
/// GET /api/v1/athlete/0
pub async fn owner(&self) -> Result<Value> {
let url = self.api_url("athlete/0");
self.get_json(url).await
}
/// Get a specific athlete.
pub async fn athlete(&self, athlete_id: &str) -> Result<Value> {
let url = self.api_url(&format!("athlete/{athlete_id}"));
pub async fn athlete(
&self,
athlete_id: &str,
) -> Result<Value> {
let url = self.api_url(
&format!("athlete/{athlete_id}"),
);
self.get_json(url).await
}
@ -96,8 +133,20 @@ impl IntervalsClient {
/// Get a single activity.
///
/// GET /api/v1/activity/{id}
pub async fn activity(&self, activity_id: &str) -> Result<Value> {
let url = self.api_url(&format!("activity/{activity_id}?intervals=true"));
pub async fn activity(
&self,
activity_id: &str,
) -> Result<Value> {
/*
* Ask Intervals.icu to include interval information,
* which is later used to associate county crossings
* with intervals.
*/
let url = self.api_url(
&format!(
"activity/{activity_id}?intervals=true"
),
);
self.get_json(url).await
}
@ -106,78 +155,180 @@ impl IntervalsClient {
///
/// GET /api/v1/activity/{id}/streams.json
///
/// We explicitly request the streams needed for GPS processing.
pub async fn streams(&self, activity_id: &str) -> Result<Value> {
let url = self.api_url(&format!(
/// We explicitly request the streams needed for GPS
/// processing.
pub async fn streams(
&self,
activity_id: &str,
) -> Result<Value> {
let url = self.api_url(
&format!(
"activity/{activity_id}/streams.json?types=time,latlng"
));
),
);
self.get_json(url).await
tracing::info!(
activity_id = %activity_id,
"requesting Intervals.icu streams"
);
let value =
self.get_json(url).await?;
let stream_count = value
.as_array()
.map(|streams| streams.len())
.unwrap_or(0);
let time_points =
Self::stream_values(
&value,
"time",
)
.map(|values| values.len())
.unwrap_or(0);
let (latitude_points, longitude_points) =
Self::latlng_values(&value)
.map(|(lat, lon)| {
(lat.len(), lon.len())
})
.unwrap_or((0, 0));
tracing::info!(
activity_id = %activity_id,
stream_count,
time_points,
latitude_points,
longitude_points,
"received Intervals.icu streams"
);
Ok(value)
}
/// Find a stream object by type.
pub fn find_stream<'a>(streams: &'a Value, stream_type: &str) -> Option<&'a Value> {
// Normal API response:
//
// [
// {
// "type": "time",
// "data": [...]
// },
// {
// "type": "latlng",
// "data": [...],
// "data2": [...]
// }
// ]
if let Some(array) = streams.as_array() {
return array
.iter()
.find(|stream| stream.get("type").and_then(Value::as_str) == Some(stream_type));
///
/// Normal Intervals.icu response:
///
/// [
/// {
/// "type": "time",
/// "data": [...]
/// },
/// {
/// "type": "latlng",
/// "data": [...],
/// "data2": [...]
/// }
/// ]
///
/// Also tolerates a dictionary-like response.
pub fn find_stream<'a>(
streams: &'a Value,
stream_type: &str,
) -> Option<&'a Value> {
if let Some(array) =
streams.as_array()
{
return array.iter().find(
|stream| {
stream
.get("type")
.and_then(Value::as_str)
== Some(stream_type)
},
);
}
// Also tolerate:
//
// {
// "time": {...},
// "latlng": {...}
// }
streams.get(stream_type)
}
/// Get the `data` array of a normal stream.
pub fn stream_values<'a>(streams: &'a Value, stream_type: &str) -> Option<&'a Vec<Value>> {
let stream = Self::find_stream(streams, stream_type)?;
///
/// For example:
///
/// {
/// "type": "time",
/// "data": [0, 1, 2, 3]
/// }
pub fn stream_values<'a>(
streams: &'a Value,
stream_type: &str,
) -> Option<&'a Vec<Value>> {
let stream =
Self::find_stream(
streams,
stream_type,
)?;
stream.get("data").and_then(Value::as_array)
stream
.get("data")
.and_then(Value::as_array)
}
/// Get the latitude and longitude arrays from the
/// Get latitude and longitude arrays from the
/// Intervals.icu `latlng` stream.
///
/// Intervals.icu represents latlng as:
///
/// {
/// "type": "latlng",
/// "data": [latitude, latitude, ...],
/// "data2": [longitude, longitude, ...]
/// }
///
/// Therefore:
///
/// data = latitude
/// data2 = longitude
///
pub fn latlng_values<'a>(streams: &'a Value) -> Option<(&'a Vec<Value>, &'a Vec<Value>)> {
let stream = Self::find_stream(streams, "latlng")?;
/// It is NOT:
///
/// data = [[lat, lon], [lat, lon], ...]
pub fn latlng_values<'a>(
streams: &'a Value,
) -> Option<(
&'a Vec<Value>,
&'a Vec<Value>,
)> {
let stream =
Self::find_stream(
streams,
"latlng",
)?;
let lat = stream.get("data")?.as_array()?;
let latitudes = stream
.get("data")
.and_then(Value::as_array)?;
let lon = stream.get("data2")?.as_array()?;
let longitudes = stream
.get("data2")
.and_then(Value::as_array)?;
Some((lat, lon))
Some((
latitudes,
longitudes,
))
}
/// Fetch an athlete's activities.
///
/// `athlete_id = "0"` means the athlete belonging to
/// the API key.
pub async fn activities(&self, athlete_id: &str, oldest: &str, newest: &str) -> Result<Value> {
pub async fn activities(
&self,
athlete_id: &str,
oldest: &str,
newest: &str,
) -> Result<Value> {
let url = format!(
"{}?oldest={}&newest={}",
self.api_url(&format!("athlete/{athlete_id}/activities")),
self.api_url(
&format!(
"athlete/{athlete_id}/activities"
),
),
urlencoding::encode(oldest),
urlencoding::encode(newest),
);
@ -186,39 +337,68 @@ impl IntervalsClient {
}
/// Extract the activity start timestamp.
pub fn activity_start(activity: &Value) -> Result<DateTime<Utc>> {
pub fn activity_start(
activity: &Value,
) -> Result<DateTime<Utc>> {
let value = activity
.get("start_date")
.or_else(|| activity.get("start_date_local"))
.or_else(|| activity.get("start_time"))
.ok_or_else(|| anyhow!("activity has no start timestamp"))?;
.or_else(|| {
activity.get("start_date_local")
})
.or_else(|| {
activity.get("start_time")
})
.ok_or_else(|| {
anyhow!(
"activity has no start timestamp"
)
})?;
let string = value
.as_str()
.ok_or_else(|| anyhow!("activity start timestamp is not a string"))?;
.ok_or_else(|| {
anyhow!(
"activity start timestamp is not a string"
)
})?;
parse_datetime(string)
}
/// Try to find an Intervals.icu interval corresponding
/// to a GPS stream index.
pub fn matching_interval(activity: &Value, track_index: usize) -> Option<Value> {
let intervals = activity.get("icu_intervals")?;
pub fn matching_interval(
activity: &Value,
track_index: usize,
) -> Option<Value> {
let intervals =
activity.get("icu_intervals")?;
let array = intervals.as_array()?;
let array =
intervals.as_array()?;
for interval in array {
let start = interval
.get("start_index")
.or_else(|| interval.get("start"));
.or_else(|| {
interval.get("start")
});
let end = interval.get("end_index").or_else(|| interval.get("end"));
let end = interval
.get("end_index")
.or_else(|| {
interval.get("end")
});
let start = start.and_then(Value::as_u64)?;
let start =
start.and_then(Value::as_u64)?;
let end = end.and_then(Value::as_u64)?;
let end =
end.and_then(Value::as_u64)?;
if (start as usize) <= track_index && track_index <= end as usize {
if (start as usize) <= track_index
&& track_index <= end as usize
{
return Some(interval.clone());
}
}
@ -227,18 +407,25 @@ impl IntervalsClient {
}
/// OAuth methods retained for the normal multi-user flow.
pub fn oauth_authorize_url(&self, state: &str) -> Result<String> {
pub fn oauth_authorize_url(
&self,
state: &str,
) -> Result<String> {
let client_id = self
.config
.intervals_client_id
.as_deref()
.context("INTERVALS_CLIENT_ID is not configured")?;
.context(
"INTERVALS_CLIENT_ID is not configured",
)?;
let redirect_uri = self
.config
.intervals_redirect_uri
.as_deref()
.context("INTERVALS_REDIRECT_URI is not configured")?;
.context(
"INTERVALS_REDIRECT_URI is not configured",
)?;
Ok(format!(
"{}/oauth/authorize?client_id={}&redirect_uri={}&scope={}&state={}",
@ -250,22 +437,32 @@ impl IntervalsClient {
))
}
pub async fn exchange_code(&self, code: &str) -> Result<Value> {
pub async fn exchange_code(
&self,
code: &str,
) -> Result<Value> {
let client_id = self
.config
.intervals_client_id
.as_deref()
.context("INTERVALS_CLIENT_ID is not configured")?;
.context(
"INTERVALS_CLIENT_ID is not configured",
)?;
let client_secret = self
.config
.intervals_client_secret
.as_deref()
.context("INTERVALS_CLIENT_SECRET is not configured")?;
.context(
"INTERVALS_CLIENT_SECRET is not configured",
)?;
let response = self
.client
.post(format!("{}/api/oauth/token", self.base_url()))
.post(format!(
"{}/api/oauth/token",
self.base_url()
))
.form(&[
("client_id", client_id),
("client_secret", client_secret),
@ -273,11 +470,15 @@ impl IntervalsClient {
])
.send()
.await
.context("OAuth token request failed")?;
.context(
"OAuth token request failed",
)?;
let status = response.status();
let body = response.text().await?;
let body = response
.text()
.await?;
if !status.is_success() {
return Err(anyhow!(
@ -290,11 +491,20 @@ impl IntervalsClient {
Ok(serde_json::from_str(&body)?)
}
pub fn token_data(response: &Value) -> Result<(String, String, String, String)> {
pub fn token_data(
response: &Value,
) -> Result<(
String,
String,
String,
String,
)> {
let access_token = response
.get("access_token")
.and_then(Value::as_str)
.context("OAuth response has no access_token")?
.context(
"OAuth response has no access_token",
)?
.to_string();
let scope = response
@ -305,7 +515,9 @@ impl IntervalsClient {
let athlete = response
.get("athlete")
.context("OAuth response has no athlete")?;
.context(
"OAuth response has no athlete",
)?;
let athlete_id = athlete
.get("id")
@ -321,27 +533,62 @@ impl IntervalsClient {
let display_name = athlete
.get("name")
.or_else(|| athlete.get("display_name"))
.or_else(|| {
athlete.get("display_name")
})
.and_then(Value::as_str)
.unwrap_or("Intervals.icu athlete")
.unwrap_or(
"Intervals.icu athlete",
)
.to_string();
Ok((access_token, scope, athlete_id, display_name))
Ok((
access_token,
scope,
athlete_id,
display_name,
))
}
}
fn parse_datetime(value: &str) -> Result<DateTime<Utc>> {
if let Ok(value) = DateTime::parse_from_rfc3339(value) {
return Ok(value.with_timezone(&Utc));
fn parse_datetime(
value: &str,
) -> Result<DateTime<Utc>> {
if let Ok(value) =
DateTime::parse_from_rfc3339(value)
{
return Ok(
value.with_timezone(&Utc)
);
}
if let Ok(value) = DateTime::parse_from_str(value, "%Y-%m-%dT%H:%M:%S%z") {
return Ok(value.with_timezone(&Utc));
if let Ok(value) =
DateTime::parse_from_str(
value,
"%Y-%m-%dT%H:%M:%S%z",
)
{
return Ok(
value.with_timezone(&Utc)
);
}
if let Ok(value) = NaiveDateTime::parse_from_str(value, "%Y-%m-%dT%H:%M:%S") {
return Ok(DateTime::<Utc>::from_naive_utc_and_offset(value, Utc));
if let Ok(value) =
NaiveDateTime::parse_from_str(
value,
"%Y-%m-%dT%H:%M:%S",
)
{
return Ok(
DateTime::<Utc>::from_naive_utc_and_offset(
value,
Utc,
)
);
}
Err(anyhow!("cannot parse activity timestamp: {}", value))
Err(anyhow!(
"cannot parse activity timestamp: {}",
value
))
}

306
src/main.rs
View file

@ -8,20 +8,24 @@ mod model;
mod web;
mod webhook;
use anyhow::{Context, Result, anyhow};
use anyhow::{anyhow, Context, Result};
use axum::{
Router,
extract::{Path, State},
response::Json,
routing::{get, post},
Router,
};
use chrono::{DateTime, Duration, Utc};
use serde_json::{Value, json};
use serde_json::{json, Value};
use sqlx::PgPool;
use tower_http::trace::TraceLayer;
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
use crate::{config::Config, intervals::IntervalsClient, model::TrackPoint};
use crate::{
config::Config,
intervals::IntervalsClient,
model::TrackPoint,
};
#[derive(Clone)]
pub struct AppState {
@ -115,7 +119,6 @@ async fn dev_sync(
) -> Result<Json<Value>, axum::http::StatusCode> {
match dev_sync_inner(state, api_key, athlete_id).await {
Ok(result) => Ok(Json(result)),
Err(error) => {
tracing::error!(
error = %error,
@ -129,18 +132,17 @@ async fn dev_sync(
/// One-segment variant:
///
/// /dev/sync/{api_key}
/// GET /dev/sync/{api_key}
///
/// Equivalent to:
///
/// /dev/sync/{api_key}/0
/// GET /dev/sync/{api_key}/0
async fn dev_sync_one(
State(state): State<AppState>,
Path(api_key): Path<String>,
) -> Result<Json<Value>, axum::http::StatusCode> {
match dev_sync_inner(state, api_key, "0".to_string()).await {
Ok(result) => Ok(Json(result)),
Err(error) => {
tracing::error!(
error = %error,
@ -152,7 +154,11 @@ async fn dev_sync_one(
}
}
async fn dev_sync_inner(state: AppState, api_key: String, athlete_id: String) -> Result<Value> {
async fn dev_sync_inner(
state: AppState,
api_key: String,
athlete_id: String,
) -> Result<Value> {
if api_key.trim().is_empty() {
return Err(anyhow!("API key is empty"));
}
@ -167,7 +173,8 @@ async fn dev_sync_inner(state: AppState, api_key: String, athlete_id: String) ->
* This client is explicitly authenticated using the API key
* supplied to the development endpoint.
*/
let client = IntervalsClient::with_api_key(state.config.clone(), api_key.clone());
let client =
IntervalsClient::with_api_key(state.config.clone(), api_key.clone());
/*
* Resolve athlete 0 through Intervals.icu.
@ -197,11 +204,9 @@ async fn dev_sync_inner(state: AppState, api_key: String, athlete_id: String) ->
};
let newest = Utc::now().date_naive();
let oldest = newest - Duration::days(state.config.dev_sync_days);
let oldest_string = oldest.format("%Y-%m-%d").to_string();
let newest_string = newest.format("%Y-%m-%d").to_string();
tracing::info!(
@ -212,13 +217,14 @@ async fn dev_sync_inner(state: AppState, api_key: String, athlete_id: String) ->
);
/*
* IMPORTANT:
*
* Use the resolved athlete ID here, not the original
* "0" value.
* Use the resolved athlete ID here, not the original "0".
*/
let activities = client
.activities(&resolved_athlete_id, &oldest_string, &newest_string)
.activities(
&resolved_athlete_id,
&oldest_string,
&newest_string,
)
.await
.context("cannot fetch development activities")?;
@ -235,14 +241,21 @@ async fn dev_sync_inner(state: AppState, api_key: String, athlete_id: String) ->
/*
* Make sure there is a corresponding local athlete row.
*
* We store the development API key as access_token because
* process_activity() uses the local athlete record and the
* existing activity-processing pipeline expects that field.
* The development API key is stored as access_token so that
* process_activity() can use exactly the same credential.
*/
ensure_dev_athlete(&state, &resolved_athlete_id, &client, &api_key).await?;
ensure_dev_athlete(
&state,
&resolved_athlete_id,
&client,
&api_key,
)
.await?;
for activity in activities_array {
let Some(activity_id) = activity.get("id").and_then(Value::as_str) else {
let Some(activity_id) =
activity.get("id").and_then(Value::as_str)
else {
skipped += 1;
results.push(json!({
@ -339,9 +352,6 @@ async fn ensure_dev_athlete(
/*
* If this athlete already exists, update the development
* credential.
*
* This is particularly useful if the row was originally
* created by an earlier development sync with an empty token.
*/
let existing = sqlx::query(
r#"
@ -401,11 +411,18 @@ async fn ensure_dev_athlete(
Ok(())
}
/// Process one activity.
///
/// The API credential is retrieved from the local athlete record.
/// This is important for development sync because the API key
/// supplied to /dev/sync/{api_key} is stored in access_token.
pub async fn process_activity(
state: AppState,
intervals_athlete_id: String,
activity_id: String,
) -> Result<()> {
use sqlx::Row;
let athlete = sqlx::query(
r#"
SELECT
@ -419,8 +436,6 @@ pub async fn process_activity(
.fetch_optional(&state.db)
.await?;
use sqlx::Row;
let Some(athlete) = athlete else {
return Err(anyhow!(
"no local athlete for Intervals.icu athlete {}",
@ -430,14 +445,21 @@ pub async fn process_activity(
let athlete_db_id: i64 = athlete.try_get("id")?;
let access_token: String = athlete
.try_get("access_token")
.context("local athlete has no access token")?;
/*
* The normal activity-processing path uses the configured
* Intervals client.
* IMPORTANT:
*
* The development endpoint has already ensured that the
* athlete exists locally and that its credential is stored.
* Use the athlete's stored credential rather than creating
* IntervalsClient::new(), which would use only the globally
* configured INTERVALS_API_KEY.
*/
let client = IntervalsClient::new(state.config.clone());
let client = IntervalsClient::with_api_key(
state.config.clone(),
access_token,
);
/*
* Fetch activity metadata.
@ -445,14 +467,22 @@ pub async fn process_activity(
let activity = client
.activity(&activity_id)
.await
.with_context(|| format!("cannot fetch Intervals.icu activity {}", activity_id))?;
.with_context(|| {
format!(
"cannot fetch Intervals.icu activity {}",
activity_id
)
})?;
let start_time = IntervalsClient::activity_start(&activity)?;
/*
* Fetch GPS/time streams.
*/
let streams = client.streams(&activity_id).await.with_context(|| {
let streams = client
.streams(&activity_id)
.await
.with_context(|| {
format!(
"cannot fetch streams for Intervals.icu activity {}",
activity_id
@ -485,19 +515,46 @@ pub async fn process_activity(
let mut tx = state.db.begin().await?;
let activity_db_id =
db::ensure_activity(&mut tx, athlete_db_id, &activity_id, start_time, &activity).await?;
db::ensure_activity(
&mut tx,
athlete_db_id,
&activity_id,
start_time,
&activity,
)
.await?;
db::delete_activity_crossings(&mut tx, activity_db_id).await?;
db::delete_activity_crossings(
&mut tx,
activity_db_id,
)
.await?;
let mut crossing_count = 0usize;
for pair in points.windows(2) {
let crossings = geo::crossings_between(&state.db, &pair[0], &pair[1]).await?;
let crossings =
geo::crossings_between(
&state.db,
&pair[0],
&pair[1],
)
.await?;
for crossing in crossings {
let interval = IntervalsClient::matching_interval(&activity, crossing.track_index);
let interval =
IntervalsClient::matching_interval(
&activity,
crossing.track_index,
);
db::save_crossing(&mut tx, activity_db_id, &crossing, interval.as_ref()).await?;
db::save_crossing(
&mut tx,
activity_db_id,
&crossing,
interval.as_ref(),
)
.await?;
crossing_count += 1;
@ -512,12 +569,17 @@ pub async fn process_activity(
}
}
db::mark_activity_processed(&mut tx, activity_db_id).await?;
db::mark_activity_processed(
&mut tx,
activity_db_id,
)
.await?;
tx.commit().await?;
tracing::info!(
activity_id = %activity_id,
points = points.len(),
crossings = crossing_count,
"activity processed"
);
@ -525,14 +587,65 @@ pub async fn process_activity(
Ok(())
}
fn parse_track(streams: &Value, start_time: DateTime<Utc>) -> Result<Vec<TrackPoint>> {
let times = IntervalsClient::stream_values(streams, "time")
.ok_or_else(|| anyhow!("Intervals.icu response has no time stream"))?;
/// Parse Intervals.icu GPS streams.
///
/// Intervals.icu returns the streams as an array similar to:
///
/// [
/// {
/// "type": "time",
/// "data": [0, 1, 2, 3]
/// },
/// {
/// "type": "latlng",
/// "data": [51.5, 51.5001, 51.5002, 51.5003],
/// "data2": [-0.1, -0.1001, -0.1002, -0.1003]
/// }
/// ]
///
/// IMPORTANT:
///
/// `latlng.data` is latitude.
///
/// `latlng.data2` is longitude.
///
/// It is NOT:
///
/// `latlng.data = [[lat, lon], [lat, lon], ...]`
fn parse_track(
streams: &Value,
start_time: DateTime<Utc>,
) -> Result<Vec<TrackPoint>> {
let times = IntervalsClient::stream_values(
streams,
"time",
)
.ok_or_else(|| {
anyhow!(
"Intervals.icu response has no time stream"
)
})?;
let latlng = IntervalsClient::stream_values(streams, "latlng")
.ok_or_else(|| anyhow!("Intervals.icu response has no latlng stream"))?;
let (latitudes, longitudes) =
IntervalsClient::latlng_values(streams)
.ok_or_else(|| {
anyhow!(
"Intervals.icu response has no usable latlng stream"
)
})?;
let length = times.len().min(latlng.len());
let length = times
.len()
.min(latitudes.len())
.min(longitudes.len());
tracing::info!(
time_points = times.len(),
latitude_points = latitudes.len(),
longitude_points = longitudes.len(),
usable_points = length,
"parsing Intervals.icu GPS streams"
);
let mut result = Vec::with_capacity(length);
@ -541,26 +654,25 @@ fn parse_track(streams: &Value, start_time: DateTime<Utc>) -> Result<Vec<TrackPo
continue;
};
let Some(coordinate) = latlng[index].as_array() else {
let Some(lat) = latitudes[index].as_f64() else {
continue;
};
if coordinate.len() != 2 {
let Some(lon) = longitudes[index].as_f64() else {
continue;
}
let Some(lat) = coordinate[0].as_f64() else {
continue;
};
let Some(lon) = coordinate[1].as_f64() else {
continue;
};
if !lat.is_finite() || !lon.is_finite() || !seconds.is_finite() {
if !seconds.is_finite()
|| !lat.is_finite()
|| !lon.is_finite()
{
continue;
}
/*
* Intervals.icu time values are seconds from the
* beginning of the activity.
*/
let millis = (seconds * 1000.0).round() as i64;
result.push(TrackPoint {
@ -571,6 +683,11 @@ fn parse_track(streams: &Value, start_time: DateTime<Utc>) -> Result<Vec<TrackPo
});
}
tracing::info!(
points = result.len(),
"finished parsing Intervals.icu GPS stream"
);
Ok(result)
}
@ -583,22 +700,89 @@ mod tests {
#[test]
fn interpolation_is_millisecond_precise() {
let a = Utc.timestamp_millis_opt(1000).unwrap();
let b = Utc.timestamp_millis_opt(2000).unwrap();
let result = interpolate_time(a, b, 0.376);
assert_eq!(result.timestamp_millis(), 1376);
assert_eq!(
result.timestamp_millis(),
1376
);
}
#[test]
fn interpolation_rounds_to_nearest_ms() {
let a = Utc.timestamp_millis_opt(1000).unwrap();
let b = Utc.timestamp_millis_opt(2000).unwrap();
let result = interpolate_time(a, b, 0.3764);
assert_eq!(result.timestamp_millis(), 1376);
assert_eq!(
result.timestamp_millis(),
1376
);
}
#[test]
fn parses_intervals_latlng_data_and_data2() {
let streams = serde_json::json!([
{
"type": "time",
"data": [0, 1, 2, 3]
},
{
"type": "latlng",
"data": [
51.5000,
51.5001,
51.5002,
51.5003
],
"data2": [
-0.1000,
-0.1001,
-0.1002,
-0.1003
]
}
]);
let start_time = Utc
.timestamp_opt(1_000_000, 0)
.unwrap();
let points =
super::parse_track(
&streams,
start_time,
)
.unwrap();
assert_eq!(points.len(), 4);
assert_eq!(points[0].index, 0);
assert!(
(points[0].lat - 51.5000).abs() < 0.000001
);
assert!(
(points[0].lon - (-0.1000)).abs() < 0.000001
);
assert_eq!(points[1].index, 1);
assert!(
(points[1].lat - 51.5001).abs() < 0.000001
);
assert!(
(points[1].lon - (-0.1001)).abs() < 0.000001
);
assert_eq!(
points[1].time.timestamp_millis(),
start_time.timestamp_millis() + 1000
);
}
}