danvics/immich-automated-selfie-timelapse
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

small improvements to the rust backend

Browse source
This commit is contained in:
Arnaud_Cayrol 2026-02-12 20:20:12 +01:00
parent 8d6446bdd6
commit 2e649dd540
24 changed files with 426 additions and 283 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
Cargo.toml
View file

@ -51,6 +51,7 @@ futures-util = "0.3"
async-trait = "0.1"
dotenvy = "0.15"
unicode-normalization = "0.1"
urlencoding = "2.1.3"
[dev-dependencies]
tokio-test = "0.4"

39
src/config.rs
View file

@ -56,7 +56,7 @@ impl Default for Config {
}
/// Immich API connection settings.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[derive(Clone, Serialize, Deserialize)]
pub struct ApiConfig {
/// API key for authentication.
pub api_key: String,
@ -69,6 +69,16 @@ pub struct ApiConfig {
pub timeout_secs: u64,
}
impl std::fmt::Debug for ApiConfig {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ApiConfig")
.field("api_key", &"[REDACTED]")
.field("base_url", &self.base_url)
.field("timeout_secs", &self.timeout_secs)
.finish()
}
}
fn default_timeout() -> u64 {
30
}
@ -371,11 +381,6 @@ impl AlignmentConfig {
"Alignment eye_distance must be between 0.0 and 1.0 (exclusive)".to_string(),
));
}
if self.eye_distance < 0.05 || self.eye_distance > 0.4 {
return Err(Error::Config(
"Alignment eye_distance should be between 0.05 and 0.4 for best results".to_string(),
));
}
Ok(())
}
@ -569,6 +574,11 @@ impl Default for ProcessingConfig {
impl ProcessingConfig {
/// Validate all step configurations.
pub fn validate(&self) -> Result<()> {
if self.max_workers == 0 {
return Err(Error::Config(
"max_workers must be greater than 0".to_string(),
));
}
self.face_resolution.validate()?;
self.blur.validate()?;
self.brightness.validate()?;
@ -620,6 +630,9 @@ impl Default for VideoConfig {
}
}
/// Valid video codecs that can be used with FFmpeg.
const VALID_CODECS: &[&str] = &["libx264", "libx265", "libvpx", "libvpx-vp9", "libaom-av1"];
impl VideoConfig {
/// Validate the configuration values.
pub fn validate(&self) -> Result<()> {
@ -638,6 +651,13 @@ impl VideoConfig {
"Video CRF must be between 0 and 51".to_string(),
));
}
if !VALID_CODECS.contains(&self.codec.as_str()) {
return Err(Error::Config(format!(
"Video codec must be one of: {}, got '{}'",
VALID_CODECS.join(", "),
self.codec
)));
}
Ok(())
}
}
@ -769,10 +789,15 @@ impl Config {
/// Only saves `processing` and `video` sections - API credentials are
/// intentionally excluded as they should come from environment variables.
pub fn save_to_file(&self, path: impl AsRef<std::path::Path>) -> Result<()> {
let path = path.as_ref();
let persistable = PersistableConfig::from(self);
let content = toml::to_string_pretty(&persistable)
.map_err(|e| Error::Config(format!("Failed to serialize config: {}", e)))?;
std::fs::write(path, content)?;
// Write to a temp file then atomically rename to avoid partial writes
let tmp_path = path.with_extension("toml.tmp");
std::fs::write(&tmp_path, content)?;
std::fs::rename(&tmp_path, path)?;
Ok(())
}
}

26
src/immich_api/mod.rs
View file

@ -5,6 +5,7 @@ use crate::error::{Error, Result};
use reqwest::Client;
use serde::{Deserialize, Serialize};
use std::time::Duration;
use urlencoding::encode as urlencode;
/// Client for interacting with the Immich API.
#[derive(Clone)]
@ -190,15 +191,23 @@ impl ImmichClient {
break;
}
page += 1;
if page > 1000 {
tracing::warn!("Pagination limit reached (1000 pages), stopping asset fetch");
break;
}
}
tracing::info!("Found {} assets for person {}", all_assets.len(), person_id);
Ok(all_assets)
}
/// Maximum download size for a single asset (100 MB).
const MAX_DOWNLOAD_SIZE: u64 = 100 * 1024 * 1024;
/// Download an asset's original image.
pub async fn download_asset(&self, asset_id: &str) -> Result<bytes::Bytes> {
let url = format!("{}/assets/{}/original", self.base_url, asset_id);
let encoded_id = urlencode(asset_id);
let url = format!("{}/assets/{}/original", self.base_url, encoded_id);
let response = self
.client
@ -215,6 +224,18 @@ impl ImmichClient {
)));
}
// Check Content-Length before downloading to reject unexpectedly large files
if let Some(content_length) = response.content_length() {
if content_length > Self::MAX_DOWNLOAD_SIZE {
return Err(Error::ImmichApi(format!(
"Asset {} is too large ({} bytes, max {} bytes)",
asset_id,
content_length,
Self::MAX_DOWNLOAD_SIZE
)));
}
}
let bytes = response.bytes().await?;
Ok(bytes)
}
@ -249,7 +270,8 @@ impl ImmichClient {
/// Get a person's thumbnail image.
/// Returns the image bytes and content-type.
pub async fn get_person_thumbnail(&self, person_id: &str) -> Result<(bytes::Bytes, String)> {
let url = format!("{}/people/{}/thumbnail", self.base_url, person_id);
let encoded_id = urlencode(person_id);
let url = format!("{}/people/{}/thumbnail", self.base_url, encoded_id);
// tracing::debug!("Fetching thumbnail from: {}", url);
let response = self

12
src/job/mod.rs
View file

@ -57,12 +57,12 @@ impl TimeIntervalTracker {
// Get or create the atomic counter for this bucket
let counter = {
// Try read lock first
let buckets = self.buckets.read().unwrap();
let buckets = self.buckets.read().expect("bucket lock poisoned");
if let Some(counter) = buckets.get(&key) {
counter.clone()
} else {
drop(buckets);
let mut buckets = self.buckets.write().unwrap();
let mut buckets = self.buckets.write().expect("bucket lock poisoned");
buckets
.entry(key)
.or_insert_with(|| Arc::new(AtomicU32::new(0)))
@ -84,7 +84,7 @@ impl TimeIntervalTracker {
/// This is a non-mutating check used to skip processing early.
pub fn is_full(&self, timestamp: &str) -> bool {
let key = self.bucket_key(timestamp);
let buckets = self.buckets.read().unwrap();
let buckets = self.buckets.read().expect("bucket lock poisoned");
if let Some(counter) = buckets.get(&key) {
counter.load(Ordering::SeqCst) >= self.max_photos
} else {
@ -95,7 +95,7 @@ impl TimeIntervalTracker {
/// Compute the bucket key from a timestamp string.
fn bucket_key(&self, timestamp: &str) -> String {
// Parse date from ISO 8601 timestamp (e.g. "2024-01-15" or "2024-01-15T12:34:56Z")
let date_str = &timestamp[..timestamp.len().min(10)];
let date_str = timestamp.get(..10).unwrap_or(timestamp);
let date = NaiveDate::parse_from_str(date_str, "%Y-%m-%d").unwrap_or_default();
match self.time_range {
@ -289,7 +289,8 @@ async fn run_job_inner(
tracing::debug!("Pipeline steps: {:?}", pipeline.step_ids());
// Create photo limit tracker if enabled
let time_interval_tracker = TimeIntervalTracker::new(&config.processing.time_interval).map(Arc::new);
let time_interval_tracker =
TimeIntervalTracker::new(&config.processing.time_interval).map(Arc::new);
// Process images in parallel with concurrency limit
let completed = Arc::new(AtomicU32::new(0));
@ -460,6 +461,7 @@ async fn run_job_inner(
&output_dirs.images,
&output_dirs.video,
&config.video,
Some(&cancel_token),
|frame, total| {
// Sync callback - just log progress. State updates would require
// async context or channels, which adds complexity for minimal benefit

46
src/job/processing.rs
View file

@ -88,7 +88,11 @@ pub async fn process_single_asset(
// Early check: skip if the time slot is already full (avoids wasting processing time)
if let Some(tracker) = time_interval {
if tracker.is_full(&timestamp) {
tracing::debug!("Asset {} skipped early: time slot already full (timestamp: {})", asset_id, timestamp);
tracing::debug!(
"Asset {} skipped early: time slot already full (timestamp: {})",
asset_id,
timestamp
);
skip_stats.increment("time_interval");
return AssetProcessResult::Skipped {
asset_id: asset_id.clone(),
@ -137,7 +141,11 @@ pub async fn process_single_asset(
// Check photo limit before saving
if let Some(tracker) = time_interval {
if !tracker.try_claim(&timestamp) {
tracing::debug!("Asset {} skipped: time interval too short (timestamp: {})", asset_id, timestamp);
tracing::debug!(
"Asset {} skipped: time interval too short (timestamp: {})",
asset_id,
timestamp
);
skip_stats.increment("time_interval");
return AssetProcessResult::Skipped {
asset_id,
@ -161,16 +169,32 @@ pub async fn process_single_asset(
let filename = format!("{}_{}.jpg", safe_timestamp, asset_id);
let output_path = output_dirs.images.join(&filename);
// Encode and save final image
let mut buffer = Cursor::new(Vec::new());
if let Err(e) = image.write_to(&mut buffer, ImageFormat::Jpeg) {
return AssetProcessResult::Error {
asset_id,
error: format!("Failed to encode image: {}", e),
};
}
// Encode image in a blocking task (CPU-bound JPEG compression)
let encoded = tokio::task::spawn_blocking(move || {
let mut buffer = Cursor::new(Vec::new());
image
.write_to(&mut buffer, ImageFormat::Jpeg)
.map(|_| buffer.into_inner())
})
.await;
if let Err(e) = tokio::fs::write(&output_path, buffer.into_inner()).await {
let jpeg_bytes = match encoded {
Ok(Ok(bytes)) => bytes,
Ok(Err(e)) => {
return AssetProcessResult::Error {
asset_id,
error: format!("Failed to encode image: {}", e),
};
}
Err(e) => {
return AssetProcessResult::Error {
asset_id,
error: format!("Image encoding task panicked: {}", e),
};
}
};
if let Err(e) = tokio::fs::write(&output_path, jpeg_bytes).await {
return AssetProcessResult::Error {
asset_id,
error: format!("Failed to save image: {}", e),

19
src/models/dlib_landmarks.rs
View file

@ -24,7 +24,9 @@ pub struct DlibLandmarks {
predictor: Mutex<LandmarkPredictor>,
}
// Safety: The Mutex ensures thread-safe access to the inner types
// Safety: The Mutex ensures exclusive access to the inner dlib types (FaceDetector and
// LandmarkPredictor), which are not Send/Sync themselves. All access goes through
// Mutex::lock(), guaranteeing only one thread uses them at a time.
unsafe impl Send for DlibLandmarks {}
unsafe impl Sync for DlibLandmarks {}
@ -81,7 +83,19 @@ impl DlibLandmarks {
pixels: &[u8],
face_rect: Option<(i64, i64, i64, i64)>,
) -> Result<Landmarks> {
// Verify buffer is large enough for the given dimensions
assert!(
pixels.len() >= width * height * 3,
"pixel buffer too small: need {} bytes for {}x{} RGB, got {}",
width * height * 3,
width,
height,
pixels.len()
);
// Create image matrix for dlib
// Safety: We verified above that `pixels` has at least width*height*3 bytes,
// matching the RGB layout that ImageMatrix::new expects.
let matrix = unsafe { ImageMatrix::new(width, height, pixels.as_ptr()) };
// Lock detector and predictor
@ -129,8 +143,7 @@ impl DlibLandmarks {
.map(|p| Point::new(p.x() as f32, p.y() as f32))
.collect();
Landmarks::new(points)
.ok_or_else(|| Error::Model("Could not detect 68 facial landmarks".to_string()))
Landmarks::new(points).map_err(Error::Model)
}
}

7
src/pipeline/crop_utils.rs
View file

@ -7,6 +7,7 @@ use crate::immich_api::FaceData;
use crate::pipeline::BoundingBox;
use image::imageops::FilterType;
use image::{DynamicImage, GenericImageView, RgbImage};
use std::borrow::Cow;
/// Result of cropping a face from an image.
pub struct CropResult {
@ -101,7 +102,11 @@ fn crop_with_replicate_fill(
y_offset: i32,
size: u32,
) -> DynamicImage {
let rgb = img.to_rgb8();
// Avoid allocation if the image is already RGB8
let rgb: Cow<'_, RgbImage> = match img.as_rgb8() {
Some(r) => Cow::Borrowed(r),
None => Cow::Owned(img.to_rgb8()),
};
let (w, h) = (rgb.width() as i32, rgb.height() as i32);
let out = RgbImage::from_fn(size, size, |px, py| {

26
src/pipeline/debug_utils.rs
View file

@ -2,8 +2,34 @@
//!
//! Shared functions for generating debug visualizations in pipeline steps.
use super::traits::{computed_keys, PipelineContext};
use image::{Rgb, RgbImage};
/// Extract the face region pixel coordinates from the pipeline context.
///
/// Uses the FACE_RECT computed value if available (face region within the cropped image),
/// otherwise falls back to the full image dimensions.
///
/// Returns `(x1, y1, x2, y2)` clamped to valid image bounds.
pub fn face_rect_pixels(
ctx: &PipelineContext,
img_width: u32,
img_height: u32,
) -> (u32, u32, u32, u32) {
if let Some(face_rect) = ctx
.get_computed(computed_keys::FACE_RECT)
.and_then(|v| v.as_face_rect())
{
let x1 = (face_rect.x1.max(0.0) as u32).min(img_width.saturating_sub(1));
let y1 = (face_rect.y1.max(0.0) as u32).min(img_height.saturating_sub(1));
let x2 = (face_rect.x2.max(0.0) as u32).min(img_width);
let y2 = (face_rect.y2.max(0.0) as u32).min(img_height);
(x1, y1, x2, y2)
} else {
(0, 0, img_width, img_height)
}
}
/// Draw simple text using a basic 5x7 pixel font.
///
/// # Arguments

2
src/pipeline/mod.rs
View file

@ -21,7 +21,7 @@ mod traits;
mod types;
pub use crop_utils::{crop_face_with_intermediate, CropResult};
pub use debug_utils::draw_simple_text;
pub use debug_utils::{draw_simple_text, face_rect_pixels};
pub use orientation::load_image_with_orientation;
pub use traits::*;
pub use types::*;

23
src/pipeline/steps/alignment.rs
View file

@ -60,6 +60,17 @@ impl ProcessingStep for AlignmentStep {
let left_eye = landmarks.left_eye_center();
let right_eye = landmarks.right_eye_center();
// Guard against coincident eyes (would cause division by zero in transform)
let eye_distance =
((right_eye.x - left_eye.x).powi(2) + (right_eye.y - left_eye.y).powi(2)).sqrt();
if eye_distance < 1e-6 {
return StepOutcome::Skip {
ctx,
reason: "alignment".to_string(),
detail: Some("Eye positions are too close together for alignment".to_string()),
};
}
// Calculate the transformation matrix
let transform = calculate_eye_alignment_transform(
left_eye,
@ -269,15 +280,21 @@ fn apply_affine_transform(
let r = (p00[0] as f32 * (1.0 - dx) * (1.0 - dy)
+ p10[0] as f32 * dx * (1.0 - dy)
+ p01[0] as f32 * (1.0 - dx) * dy
+ p11[0] as f32 * dx * dy) as u8;
+ p11[0] as f32 * dx * dy)
.round()
.clamp(0.0, 255.0) as u8;
let g = (p00[1] as f32 * (1.0 - dx) * (1.0 - dy)
+ p10[1] as f32 * dx * (1.0 - dy)
+ p01[1] as f32 * (1.0 - dx) * dy
+ p11[1] as f32 * dx * dy) as u8;
+ p11[1] as f32 * dx * dy)
.round()
.clamp(0.0, 255.0) as u8;
let b = (p00[2] as f32 * (1.0 - dx) * (1.0 - dy)
+ p10[2] as f32 * dx * (1.0 - dy)
+ p01[2] as f32 * (1.0 - dx) * dy
+ p11[2] as f32 * dx * dy) as u8;
+ p11[2] as f32 * dx * dy)
.round()
.clamp(0.0, 255.0) as u8;
Rgb([r, g, b])
});

39
src/pipeline/steps/blur.rs
View file

@ -9,7 +9,8 @@
use crate::config::Config;
use crate::pipeline::{
computed_keys, draw_simple_text, ComputedValue, PipelineContext, ProcessingStep, StepOutcome,
computed_keys, draw_simple_text, face_rect_pixels, ComputedValue, PipelineContext,
ProcessingStep, StepOutcome,
};
use async_trait::async_trait;
use image::{DynamicImage, Rgb};
@ -115,25 +116,8 @@ impl ProcessingStep for BlurStep {
Err(e) => return StepOutcome::Error { ctx, error: e },
};
let img_width = image.width();
let img_height = image.height();
// Use FACE_RECT if available (face region within the cropped image),
// otherwise analyze the entire cropped image
let (x1, y1, x2, y2) = if let Some(face_rect) = ctx
.get_computed(computed_keys::FACE_RECT)
.and_then(|v| v.as_face_rect())
{
// Use the face rectangle within the cropped image
let x1 = (face_rect.x1.max(0.0) as u32).min(img_width.saturating_sub(1));
let y1 = (face_rect.y1.max(0.0) as u32).min(img_height.saturating_sub(1));
let x2 = (face_rect.x2.max(0.0) as u32).min(img_width);
let y2 = (face_rect.y2.max(0.0) as u32).min(img_height);
(x1, y1, x2, y2)
} else {
// No face rect available, use the entire cropped image
(0, 0, img_width, img_height)
};
let (img_width, img_height) = (image.width(), image.height());
let (x1, y1, x2, y2) = face_rect_pixels(&ctx, img_width, img_height);
let gradient_magnitude =
Self::calculate_gradient_magnitude_in_region(image, x1, y1, x2, y2);
@ -172,20 +156,7 @@ impl ProcessingStep for BlurStep {
// Create a copy for visualization
let mut debug_img = rgb.clone();
// Draw the face bounding box (FACE_RECT if available, otherwise full image)
let (x1, y1, x2, y2) = if let Some(face_rect) = ctx
.get_computed(computed_keys::FACE_RECT)
.and_then(|v| v.as_face_rect())
{
let x1 = (face_rect.x1.max(0.0) as u32).min(width.saturating_sub(1));
let y1 = (face_rect.y1.max(0.0) as u32).min(height.saturating_sub(1));
let x2 = (face_rect.x2.max(0.0) as u32).min(width);
let y2 = (face_rect.y2.max(0.0) as u32).min(height);
(x1, y1, x2, y2)
} else {
// No face rect, show that we analyzed the full cropped image
(0, 0, width, height)
};
let (x1, y1, x2, y2) = face_rect_pixels(ctx, width, height);
// Draw rectangle outline (cyan color for visibility)
let rect_color = Rgb([0, 255, 255]);

51
src/pipeline/steps/brightness.rs
View file

@ -6,7 +6,8 @@
use crate::config::Config;
use crate::pipeline::{
computed_keys, draw_simple_text, ComputedValue, PipelineContext, ProcessingStep, StepOutcome,
computed_keys, draw_simple_text, face_rect_pixels, ComputedValue, PipelineContext,
ProcessingStep, StepOutcome,
};
use async_trait::async_trait;
use image::{DynamicImage, Rgb};
@ -34,8 +35,8 @@ impl BrightnessStep {
x2: u32,
y2: u32,
) -> f32 {
let rgb = image.to_rgb8();
let (img_width, img_height) = (rgb.width(), rgb.height());
let luma = image.to_luma8();
let (img_width, img_height) = luma.dimensions();
// Clamp coordinates to image bounds
let x1 = x1.min(img_width.saturating_sub(1));
@ -56,14 +57,12 @@ impl BrightnessStep {
return 0.0;
}
// Calculate average luminance using standard RGB to luminance conversion
// Y = 0.299*R + 0.587*G + 0.114*B
// Sum luminance values from the pre-computed luma image
// (to_luma8 uses the standard Y = 0.299*R + 0.587*G + 0.114*B conversion)
let mut total_luminance: u64 = 0;
for y in y1..y2 {
for x in x1..x2 {
let pixel = rgb.get_pixel(x, y);
let [r, g, b] = pixel.0;
total_luminance += (299 * r as u64 + 587 * g as u64 + 114 * b as u64) / 1000;
total_luminance += luma.get_pixel(x, y)[0] as u64;
}
}
@ -93,25 +92,8 @@ impl ProcessingStep for BrightnessStep {
Err(e) => return StepOutcome::Error { ctx, error: e },
};
let img_width = image.width();
let img_height = image.height();
// Use FACE_RECT if available (face region within the cropped image),
// otherwise analyze the entire cropped image
let (x1, y1, x2, y2) = if let Some(face_rect) = ctx
.get_computed(computed_keys::FACE_RECT)
.and_then(|v| v.as_face_rect())
{
// Use the face rectangle within the cropped image
let x1 = (face_rect.x1.max(0.0) as u32).min(img_width.saturating_sub(1));
let y1 = (face_rect.y1.max(0.0) as u32).min(img_height.saturating_sub(1));
let x2 = (face_rect.x2.max(0.0) as u32).min(img_width);
let y2 = (face_rect.y2.max(0.0) as u32).min(img_height);
(x1, y1, x2, y2)
} else {
// No face rect available, use the entire cropped image
(0, 0, img_width, img_height)
};
let (img_width, img_height) = (image.width(), image.height());
let (x1, y1, x2, y2) = face_rect_pixels(&ctx, img_width, img_height);
let brightness = Self::calculate_brightness_in_region(image, x1, y1, x2, y2);
@ -157,20 +139,7 @@ impl ProcessingStep for BrightnessStep {
// Create a copy for visualization
let mut debug_img = rgb.clone();
// Draw the face bounding box (FACE_RECT if available, otherwise full image)
let (x1, y1, x2, y2) = if let Some(face_rect) = ctx
.get_computed(computed_keys::FACE_RECT)
.and_then(|v| v.as_face_rect())
{
let x1 = (face_rect.x1.max(0.0) as u32).min(width.saturating_sub(1));
let y1 = (face_rect.y1.max(0.0) as u32).min(height.saturating_sub(1));
let x2 = (face_rect.x2.max(0.0) as u32).min(width);
let y2 = (face_rect.y2.max(0.0) as u32).min(height);
(x1, y1, x2, y2)
} else {
// No face rect, show that we analyzed the full cropped image
(0, 0, width, height)
};
let (x1, y1, x2, y2) = face_rect_pixels(ctx, width, height);
// Draw rectangle outline (cyan color for visibility)
let rect_color = Rgb([0, 255, 255]);

1
src/pipeline/steps/decode.rs
View file

@ -38,6 +38,7 @@ impl ProcessingStep for DecodeImageStep {
match load_image_with_orientation(raw_bytes) {
Ok(image) => {
ctx.image = Some(image);
ctx.raw_bytes = None; // Free raw bytes after successful decode
StepOutcome::Continue(ctx)
}
Err(e) => StepOutcome::Skip {

7
src/pipeline/steps/eye_filter.rs
View file

@ -65,7 +65,7 @@ impl ProcessingStep for EyeFilterStep {
StepOutcome::Continue(ctx)
}
fn debug_visualize(&self, ctx: &PipelineContext, _config: &Config) -> Option<DynamicImage> {
fn debug_visualize(&self, ctx: &PipelineContext, config: &Config) -> Option<DynamicImage> {
// Get landmarks for eye visualization
let landmarks: &Landmarks = ctx
.get_computed(computed_keys::LANDMARKS)
@ -114,12 +114,13 @@ impl ProcessingStep for EyeFilterStep {
// Draw landmarks on the zoomed image (coordinates transformed to zoomed space)
let points = landmarks.points();
let left_color = if ear.left >= 0.2 {
let min_ear = config.processing.eye_filter.min_ear;
let left_color = if ear.left >= min_ear {
Rgb([0, 255, 0])
} else {
Rgb([255, 0, 0])
};
let right_color = if ear.right >= 0.2 {
let right_color = if ear.right >= min_ear {
Rgb([0, 255, 0])
} else {
Rgb([255, 0, 0])

14
src/pipeline/steps/head_pose.rs
View file

@ -235,15 +235,21 @@ impl ProcessingStep for HeadPoseStep {
image.clone()
};
// Run inference on the face crop
let pose = match model.estimate(&face_image) {
Ok(p) => p,
Err(e) => {
// Run inference on the face crop (blocking CPU work offloaded from async runtime)
let pose = match tokio::task::spawn_blocking(move || model.estimate(&face_image)).await {
Ok(Ok(p)) => p,
Ok(Err(e)) => {
return StepOutcome::Error {
ctx,
error: format!("Head pose estimation failed: {}", e),
};
}
Err(e) => {
return StepOutcome::Error {
ctx,
error: format!("Head pose inference task panicked: {}", e),
};
}
};
// Store pose in computed values

4
src/pipeline/traits.rs
View file

@ -259,7 +259,7 @@ pub trait ProcessingStep: Send + Sync {
/// Computed value keys that this step depends on (must be present in `ctx.computed`).
///
/// Return a list of keys from `computed_keys` that must be available for this step.
/// The pipeline will verify these dependencies are satisfied before execution.
/// These are informational only; the pipeline does not currently enforce them.
fn dependencies(&self) -> Vec<&'static str> {
vec![]
}
@ -267,7 +267,7 @@ pub trait ProcessingStep: Send + Sync {
/// Computed value keys that this step provides (sets in `ctx.computed`).
///
/// Return a list of keys from `computed_keys` that this step will set.
/// This is used for documentation and dependency verification.
/// These are informational only; the pipeline does not currently enforce them.
fn provides(&self) -> Vec<&'static str> {
vec![]
}

82
src/pipeline/types.rs
View file

@ -25,6 +25,16 @@ pub struct BoundingBox {
}
impl BoundingBox {
/// Create a new bounding box, ensuring x1 <= x2 and y1 <= y2.
pub fn new(x1: f32, y1: f32, x2: f32, y2: f32) -> Self {
Self {
x1: x1.min(x2),
y1: y1.min(y2),
x2: x1.max(x2),
y2: y1.max(y2),
}
}
pub fn width(&self) -> f32 {
self.x2 - self.x1
}
@ -55,10 +65,16 @@ pub const LANDMARK_COUNT: usize = 68;
impl Landmarks {
/// Create a new Landmarks from a vector of points.
///
/// Returns `None` if the vector doesn't contain exactly 68 points.
pub fn new(points: Vec<Point>) -> Option<Self> {
let points: [Point; LANDMARK_COUNT] = points.try_into().ok()?;
Some(Self { points })
/// Returns an error if the vector doesn't contain exactly 68 points.
pub fn new(points: Vec<Point>) -> Result<Self, String> {
let len = points.len();
let points: [Point; LANDMARK_COUNT] = points.try_into().map_err(|_| {
format!(
"Expected exactly {} landmark points, got {}",
LANDMARK_COUNT, len
)
})?;
Ok(Self { points })
}
/// Get all landmark points.
@ -192,61 +208,3 @@ impl EyeAspectRatio {
self.left >= threshold && self.right >= threshold
}
}
/// Result of processing a single face.
#[derive(Debug)]
pub struct ProcessedFace {
/// The aligned and cropped face image data.
pub image_data: Vec<u8>,
/// Original asset ID.
pub asset_id: String,
/// Timestamp for sorting/naming.
pub timestamp: String,
}
/// Reason why an image was skipped during processing.
#[derive(Debug, Clone, PartialEq)]
pub enum SkipReason {
FaceTooSmall,
EyesClosed,
HeadTurned,
TooDark,
TooBright,
NoFaceDetected,
DownloadFailed,
DecodeFailed,
CropFailed,
Cancelled,
}
impl std::fmt::Display for SkipReason {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
SkipReason::FaceTooSmall => write!(f, "Face too small"),
SkipReason::EyesClosed => write!(f, "Eyes closed"),
SkipReason::HeadTurned => write!(f, "Head turned"),
SkipReason::TooDark => write!(f, "Too dark"),
SkipReason::TooBright => write!(f, "Too bright"),
SkipReason::NoFaceDetected => write!(f, "No face detected"),
SkipReason::DownloadFailed => write!(f, "Download failed"),
SkipReason::DecodeFailed => write!(f, "Decode failed"),
SkipReason::CropFailed => write!(f, "Crop failed"),
SkipReason::Cancelled => write!(f, "Cancelled"),
}
}
}
/// Processing result for a single asset.
#[derive(Debug)]
pub enum AssetResult {
/// Successfully processed.
Success(ProcessedFace),
/// Skipped (face too small, bad pose, etc.).
Skipped {
asset_id: String,
reason: SkipReason,
detail: Option<String>,
},
/// Error during processing.
Error { asset_id: String, error: String },
}

18
src/utils.rs
View file

@ -34,13 +34,9 @@ pub fn sanitize_folder_name(name: Option<&str>, id: &str) -> String {
})
.collect();
// Trim whitespace/underscores and limit length
// Trim whitespace/underscores and limit length (char-based to avoid UTF-8 boundary panics)
let trimmed = sanitized.trim_matches(|c: char| c.is_whitespace() || c == '_');
if trimmed.len() > 50 {
trimmed[..50].to_string()
} else {
trimmed.to_string()
}
trimmed.chars().take(50).collect()
}
#[cfg(test)]
@ -75,7 +71,15 @@ mod tests {
fn test_sanitize_folder_name_limits_length() {
let long_name = "A".repeat(100);
let result = sanitize_folder_name(Some(&long_name), "id");
assert_eq!(result.len(), 50);
assert_eq!(result.chars().count(), 50);
}
#[test]
fn test_sanitize_folder_name_limits_length_multibyte() {
// CJK characters are multi-byte in UTF-8 but pass is_alphanumeric()
let long_name = "\u{4e2d}".repeat(100); // 100 CJK characters
let result = sanitize_folder_name(Some(&long_name), "id");
assert_eq!(result.chars().count(), 50);
}
#[test]

71
src/video/ffmpeg.rs
View file

@ -6,6 +6,7 @@ use std::path::Path;
use std::process::Stdio;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::process::Command;
use tokio_util::sync::CancellationToken;
/// Compile images into a timelapse video.
///
@ -13,11 +14,13 @@ use tokio::process::Command;
/// * `input_dir` - Directory containing JPEG images (sorted alphabetically for order)
/// * `output_path` - Path for the output video file
/// * `config` - Video configuration
/// * `cancel_token` - Optional cancellation token to kill the ffmpeg process
/// * `progress_callback` - Called with (current_frame, total_frames)
pub async fn compile_timelapse<F>(
input_dir: &Path,
output_path: &Path,
config: &VideoConfig,
cancel_token: Option<&CancellationToken>,
mut progress_callback: F,
) -> Result<()>
where
@ -73,6 +76,35 @@ where
file.write_all(concat_content.as_bytes()).await?;
file.flush().await?;
// Use a closure to ensure concat file is cleaned up on all paths
let result = run_ffmpeg(
&concat_path,
output_path,
config,
cancel_token,
image_count,
&mut progress_callback,
)
.await;
// Always clean up concat file, even on error
let _ = tokio::fs::remove_file(&concat_path).await;
result
}
/// Run the ffmpeg process and handle progress/cancellation.
async fn run_ffmpeg<F>(
concat_path: &Path,
output_path: &Path,
config: &VideoConfig,
cancel_token: Option<&CancellationToken>,
image_count: u32,
progress_callback: &mut F,
) -> Result<()>
where
F: FnMut(u32, u32),
{
// Build ffmpeg command using concat demuxer
let mut cmd = Command::new("ffmpeg");
cmd.arg("-y") // Overwrite output
@ -81,7 +113,7 @@ where
.arg("-safe")
.arg("0") // Allow absolute paths
.arg("-i")
.arg(&concat_path)
.arg(concat_path)
.arg("-c:v")
.arg(&config.codec)
.arg("-crf")
@ -119,15 +151,33 @@ where
stderr_output
});
// Parse progress from stdout
// Parse progress from stdout, with cancellation support
let mut reader = BufReader::new(stdout).lines();
while let Some(line) = reader.next_line().await? {
if line.starts_with("frame=") {
if let Some(frame_str) = line.strip_prefix("frame=") {
if let Ok(frame) = frame_str.trim().parse::<u32>() {
// Clamp to image_count - ffmpeg can report higher values internally
let clamped_frame = frame.min(image_count);
progress_callback(clamped_frame, image_count);
// Create a dummy token for when no cancel_token is provided
let owned_token = CancellationToken::new();
let token = cancel_token.unwrap_or(&owned_token);
loop {
tokio::select! {
_ = token.cancelled() => {
// Kill the ffmpeg process on cancellation
let _ = child.kill().await;
return Err(Error::Cancelled);
}
line = reader.next_line() => {
match line? {
Some(line) => {
if line.starts_with("frame=") {
if let Some(frame_str) = line.strip_prefix("frame=") {
if let Ok(frame) = frame_str.trim().parse::<u32>() {
let clamped_frame = frame.min(image_count);
progress_callback(clamped_frame, image_count);
}
}
}
}
None => break, // EOF
}
}
}
@ -136,9 +186,6 @@ where
let status = child.wait().await?;
let stderr_output = stderr_handle.await.unwrap_or_default();
// Clean up concat file
let _ = tokio::fs::remove_file(&concat_path).await;
if !status.success() {
// Get last few lines of stderr for the error message
let error_lines: Vec<&str> = stderr_output.lines().rev().take(10).collect();

24
src/web/handlers/config.rs
View file

@ -2,7 +2,8 @@
use crate::config::{
AlignmentConfig, BlurConfig, BrightnessConfig, EyeFilterConfig, FaceResolutionConfig,
HeadPoseConfig, OutputConfig, TimeIntervalConfig, ProcessingConfig, TimestampConfig, VideoConfig,
HeadPoseConfig, OutputConfig, ProcessingConfig, TimeIntervalConfig, TimestampConfig,
VideoConfig,
};
use crate::web::state::AppState;
use axum::{extract::State, http::StatusCode, response::Json};
@ -254,8 +255,8 @@ pub async fn update_config(
})?;
}
// Update config
{
// Update config in-memory, then save outside the lock
let config_snapshot = {
let mut config = state.config.write().await;
if let Some(proc) = update.processing {
@ -306,12 +307,17 @@ pub async fn update_config(
}
}
// Persist to file (non-blocking, best-effort)
if let Err(e) = config.save_to_file("config.toml") {
tracing::warn!("Failed to persist config to file: {}", e);
} else {
tracing::info!("Configuration updated and persisted to config.toml");
}
config.clone()
}; // Write lock dropped here
// Persist to file outside the lock using spawn_blocking for the fs write
let save_result =
tokio::task::spawn_blocking(move || config_snapshot.save_to_file("config.toml")).await;
match save_result {
Ok(Ok(())) => tracing::info!("Configuration updated and persisted to config.toml"),
Ok(Err(e)) => tracing::warn!("Failed to persist config to file: {}", e),
Err(e) => tracing::warn!("Config save task panicked: {}", e),
}
// Return updated config

83
src/web/handlers/output.rs
View file

@ -50,6 +50,19 @@ pub struct BulkDeleteResponse {
pub remaining_images: u32,
}
/// Check if a path exists asynchronously.
async fn path_exists(path: &std::path::Path) -> bool {
tokio::fs::metadata(path).await.is_ok()
}
/// Check if a path is a directory asynchronously.
async fn path_is_dir(path: &std::path::Path) -> bool {
tokio::fs::metadata(path)
.await
.map(|m| m.is_dir())
.unwrap_or(false)
}
/// List all output folders with their stats.
pub async fn list_output_folders(
State(state): State<AppState>,
@ -59,7 +72,7 @@ pub async fn list_output_folders(
let mut folders = Vec::new();
if output_dir.exists() {
if path_exists(output_dir).await {
let mut entries = tokio::fs::read_dir(output_dir).await.map_err(|e| {
(
StatusCode::INTERNAL_SERVER_ERROR,
@ -74,7 +87,7 @@ pub async fn list_output_folders(
)
})? {
let path = entry.path();
if path.is_dir() {
if path_is_dir(&path).await {
let name = entry.file_name().to_string_lossy().to_string();
// Count images in the images subfolder
@ -82,7 +95,7 @@ pub async fn list_output_folders(
let mut image_count = 0u32;
let mut size_bytes = 0u64;
if images_dir.exists() {
if path_exists(&images_dir).await {
if let Ok(mut img_entries) = tokio::fs::read_dir(&images_dir).await {
while let Ok(Some(img_entry)) = img_entries.next_entry().await {
let img_path = img_entry.path();
@ -98,7 +111,7 @@ pub async fn list_output_folders(
// Check for video file (named after the folder/person)
let video_filename = format!("{}.mp4", name);
let has_video = path.join(&video_filename).exists();
let has_video = path_exists(&path.join(&video_filename)).await;
folders.push(OutputFolderInfo {
name,
@ -129,7 +142,7 @@ pub async fn cleanup_all_output(
let output_dir = &config.output_dir;
// Remove all contents of output directory
if output_dir.exists() {
if path_exists(output_dir).await {
let mut entries = tokio::fs::read_dir(output_dir).await.map_err(|e| {
(
StatusCode::INTERNAL_SERVER_ERROR,
@ -145,7 +158,7 @@ pub async fn cleanup_all_output(
)
})? {
let path = entry.path();
if path.is_dir() {
if path_is_dir(&path).await {
tokio::fs::remove_dir_all(&path).await.map_err(|e| {
(
StatusCode::INTERNAL_SERVER_ERROR,
@ -187,14 +200,14 @@ pub async fn cleanup_output_folder(
let folder_path = config.output_dir.join(&folder_name);
if !folder_path.exists() {
return Err((
let folder_meta = tokio::fs::metadata(&folder_path).await.map_err(|_| {
(
StatusCode::NOT_FOUND,
format!("Folder '{}' not found", folder_name),
));
}
)
})?;
if !folder_path.is_dir() {
if !folder_meta.is_dir() {
return Err((
StatusCode::BAD_REQUEST,
format!("'{}' is not a directory", folder_name),
@ -228,7 +241,7 @@ pub async fn list_folder_images(
let images_dir = config.output_dir.join(&folder_name).join("images");
if !images_dir.exists() {
if !path_exists(&images_dir).await {
return Err((
StatusCode::NOT_FOUND,
format!("Folder '{}' not found or has no images", folder_name),
@ -275,11 +288,8 @@ pub async fn list_folder_images(
let total_count = images.len() as u32;
let video_filename = format!("{}.mp4", folder_name);
let video_exists = config
.output_dir
.join(&folder_name)
.join(&video_filename)
.exists();
let video_exists =
path_exists(&config.output_dir.join(&folder_name).join(&video_filename)).await;
Ok(Json(FolderImagesResponse {
folder_name,
@ -319,7 +329,7 @@ pub async fn delete_single_image(
.join("images")
.join(&filename);
if !file_path.exists() {
if !path_exists(&file_path).await {
return Err((
StatusCode::NOT_FOUND,
format!("Image '{}' not found in folder '{}'", filename, folder_name),
@ -359,7 +369,7 @@ pub async fn delete_images_bulk(
let images_dir = config.output_dir.join(&folder_name).join("images");
if !images_dir.exists() {
if !path_exists(&images_dir).await {
return Err((
StatusCode::NOT_FOUND,
format!("Folder '{}' not found or has no images", folder_name),
@ -381,13 +391,9 @@ pub async fn delete_images_bulk(
}
let file_path = images_dir.join(filename);
if file_path.exists() {
match tokio::fs::remove_file(&file_path).await {
Ok(_) => deleted_count += 1,
Err(_) => failed_count += 1,
}
} else {
failed_count += 1;
match tokio::fs::remove_file(&file_path).await {
Ok(_) => deleted_count += 1,
Err(_) => failed_count += 1,
}
}
@ -420,11 +426,6 @@ pub async fn compile_folder_video(
State(state): State<AppState>,
Path(folder_name): Path<String>,
) -> Result<Json<StartResponse>, (StatusCode, String)> {
state
.ensure_no_job_running()
.await
.map_err(|e| (StatusCode::CONFLICT, e.to_string()))?;
validate_path_component(&folder_name, "folder name")
.map_err(|e| (StatusCode::BAD_REQUEST, e.to_string()))?;
@ -433,7 +434,7 @@ pub async fn compile_folder_video(
let folder_path = config.output_dir.join(&folder_name);
let images_dir = folder_path.join("images");
if !images_dir.exists() {
if !path_exists(&images_dir).await {
return Err((
StatusCode::NOT_FOUND,
format!("Folder '{}' not found or has no images", folder_name),
@ -457,9 +458,14 @@ pub async fn compile_folder_video(
));
}
// Set status to compiling video (bypasses terminal state check)
state
.reset_progress(Progress {
// Atomically check-and-set: hold write lock across both check and status update
// to prevent TOCTOU race conditions between concurrent requests.
{
let mut progress = state.progress.write().await;
if progress.status == JobStatus::Running || progress.status == JobStatus::CompilingVideo {
return Err((StatusCode::CONFLICT, "A job is already running".to_string()));
}
let new_progress = Progress {
status: JobStatus::CompilingVideo,
completed: 0,
total: image_count,
@ -467,8 +473,10 @@ pub async fn compile_folder_video(
skip_stats: SkipStats::default(),
person_id: None,
person_name: Some(folder_name.clone()),
})
.await;
};
*progress = new_progress.clone();
let _ = state.progress_tx.send(new_progress);
}
// Create cancellation token
let cancel_token = state.create_cancel_token().await;
@ -486,6 +494,7 @@ pub async fn compile_folder_video(
&images_dir,
&output_path,
&video_config,
Some(&cancel_token),
|current, total| {
// Check for cancellation
if cancel_token.is_cancelled() {

58
src/web/handlers/processing.rs
View file

@ -45,18 +45,8 @@ pub struct ProgressResponse {
pub async fn get_progress(State(state): State<AppState>) -> Json<ProgressResponse> {
let progress = state.progress.read().await;
let status_str = match &progress.status {
JobStatus::Idle => "idle",
JobStatus::Running => "running",
JobStatus::Cancelling => "cancelling",
JobStatus::CompilingVideo => "compiling_video",
JobStatus::Completed => "completed",
JobStatus::Cancelled => "cancelled",
JobStatus::Error(_) => "error",
};
Json(ProgressResponse {
status: status_str.to_string(),
status: progress.status.as_str().to_string(),
completed: progress.completed,
total: progress.total,
message: progress.message.clone(),
@ -75,22 +65,48 @@ pub struct StartRequest {
pub date_to: Option<String>,
}
/// Validate that a person_id looks reasonable (non-empty, reasonable length, safe characters).
fn validate_person_id(person_id: &str) -> Result<(), (StatusCode, String)> {
if person_id.is_empty() {
return Err((
StatusCode::BAD_REQUEST,
"person_id must not be empty".to_string(),
));
}
if person_id.len() > 128 {
return Err((
StatusCode::BAD_REQUEST,
"person_id is too long (max 128 characters)".to_string(),
));
}
if !person_id
.chars()
.all(|c| c.is_alphanumeric() || c == '-' || c == '_')
{
return Err((
StatusCode::BAD_REQUEST,
"person_id contains invalid characters (only alphanumeric, hyphens, underscores allowed)"
.to_string(),
));
}
Ok(())
}
/// Start processing for a person.
pub async fn start_processing(
State(state): State<AppState>,
Json(request): Json<StartRequest>,
) -> Result<Json<StartResponse>, (StatusCode, String)> {
// Check if already running
validate_person_id(&request.person_id)?;
// Atomically check-and-set: hold write lock across both check and status update
// to prevent TOCTOU race conditions between concurrent requests.
{
let progress = state.progress.read().await;
let mut progress = state.progress.write().await;
if progress.status == JobStatus::Running || progress.status == JobStatus::CompilingVideo {
return Err((StatusCode::CONFLICT, "A job is already running".to_string()));
}
}
// Reset progress with person info (bypasses terminal state check)
state
.reset_progress(Progress {
let new_progress = Progress {
status: JobStatus::Running,
completed: 0,
total: 0,
@ -98,8 +114,10 @@ pub async fn start_processing(
skip_stats: SkipStats::default(),
person_id: Some(request.person_id.clone()),
person_name: request.person_name.clone(),
})
.await;
};
*progress = new_progress.clone();
let _ = state.progress_tx.send(new_progress);
}
// Create cancellation token
let cancel_token = state.create_cancel_token().await;

37
src/web/handlers/ws.rs
View file

@ -1,7 +1,7 @@
//! WebSocket handler for real-time progress updates.
use crate::web::handlers::ProgressResponse;
use crate::web::state::{AppState, JobStatus};
use crate::web::state::AppState;
use axum::{
extract::{
ws::{Message, WebSocket},
@ -47,15 +47,24 @@ async fn handle_socket(socket: WebSocket, state: AppState) {
// Send progress updates to the client
let mut send_task = tokio::spawn(async move {
while let Ok(progress) = progress_rx.recv().await {
let response = progress_to_response(&progress);
match serde_json::to_string(&response) {
Ok(json) => {
if sender.send(Message::Text(json.into())).await.is_err() {
break;
loop {
match progress_rx.recv().await {
Ok(progress) => {
let response = progress_to_response(&progress);
match serde_json::to_string(&response) {
Ok(json) => {
if sender.send(Message::Text(json.into())).await.is_err() {
break;
}
}
Err(_) => break,
}
}
Err(_) => break,
Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
tracing::debug!("WebSocket client lagged, skipped {} messages", n);
continue;
}
Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
}
}
});
@ -75,18 +84,8 @@ async fn handle_socket(socket: WebSocket, state: AppState) {
fn progress_to_response(progress: &crate::web::state::Progress) -> ProgressResponse {
use crate::web::handlers::SkipStatsResponse;
let status_str = match &progress.status {
JobStatus::Idle => "idle",
JobStatus::Running => "running",
JobStatus::Cancelling => "cancelling",
JobStatus::CompilingVideo => "compiling_video",
JobStatus::Completed => "completed",
JobStatus::Cancelled => "cancelled",
JobStatus::Error(_) => "error",
};
ProgressResponse {
status: status_str.to_string(),
status: progress.status.as_str().to_string(),
completed: progress.completed,
total: progress.total,
message: progress.message.clone(),

19
src/web/state.rs
View file

@ -28,6 +28,25 @@ impl JobStatus {
JobStatus::Completed | JobStatus::Cancelled | JobStatus::Error(_)
)
}
/// Get the status as a string identifier for API responses.
pub fn as_str(&self) -> &str {
match self {
JobStatus::Idle => "idle",
JobStatus::Running => "running",
JobStatus::Cancelling => "cancelling",
JobStatus::CompilingVideo => "compiling_video",
JobStatus::Completed => "completed",
JobStatus::Cancelled => "cancelled",
JobStatus::Error(_) => "error",
}
}
}
impl std::fmt::Display for JobStatus {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}", self.as_str())
}
}
/// Detailed statistics for skipped images.