JezzWTF/phokus
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Surface failed embeddings and add filter for affected files

Browse Source 
- Fix root cause: embedding_source_path() returns Result<PathBuf>, returning
  Err for videos without a thumbnail instead of silently falling back to the
  raw .mp4 path that CLIP cannot decode
- indexer: split embedding batch into pre-failed (no source) and embeddable
  jobs; pre-failed are marked immediately without hitting the CLIP model
- db: retry_failed_embedding_jobs skips videos still without a thumbnail so
  they no longer re-fail immediately on retry
- Add get_failed_embedding_images command listing failed files + error per folder
- Gallery: amber warning badge on tiles with embedding_status = 'failed'
- BackgroundTasks: fetch and show failed filenames/errors in expanded panel
- Toolbar: conditional 'Failed Embeddings' amber filter pill shown when any
  folder has embedding_failed > 0; filters at DB level via new
  embedding_failed_only param on get_images / count_images
- TagCloud: replaced vocabulary/dictionary label system with representative
  image thumbnails per cluster; results cached in SQLite by image-id hash
This commit is contained in:
LyAhn
2026-04-06 16:54:03 +01:00
parent 6c3fd449ce
commit d0b41119c6
11 changed files with 560 additions and 343 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src-tauri/src/commands.rs
+113 -54
View File
@@ -4,7 +4,7 @@ use crate::indexer;
use crate::vector;
use serde::{Deserialize, Serialize};
use std::path::PathBuf;
use tauri::{AppHandle, Manager, State};
use tauri::{AppHandle, State};
pub type DbState = DbPool;
@@ -22,6 +22,7 @@ pub struct GetImagesParams {
pub search: Option<String>,
pub media_kind: Option<String>,
pub favorites_only: Option<bool>,
pub embedding_failed_only: Option<bool>,
pub sort: Option<String>,
pub offset: Option<i64>,
pub limit: Option<i64>,
@@ -124,8 +125,9 @@ pub async fn get_images(
let search = params.search.as_deref();
let media_kind = params.media_kind.as_deref();
let favorites_only = params.favorites_only.unwrap_or(false);
let embedding_failed_only = params.embedding_failed_only.unwrap_or(false);
let total = db::count_images(&conn, params.folder_id, search, media_kind, favorites_only)
let total = db::count_images(&conn, params.folder_id, search, media_kind, favorites_only, embedding_failed_only)
.map_err(|e| e.to_string())?;
let images = db::get_images(
@@ -134,6 +136,7 @@ pub async fn get_images(
search,
media_kind,
favorites_only,
embedding_failed_only,
sort,
offset,
limit,
@@ -224,77 +227,109 @@ pub async fn semantic_search_images(
Ok(images)
}
#[derive(Serialize)]
#[derive(Serialize, Deserialize)]
pub struct TagCloudEntry {
pub label: String,
pub count: usize,
pub representative_image_id: i64,
pub thumbnail_path: Option<String>,
}
/// Clusters the library's image embeddings with k-means, then labels each cluster by
/// finding the closest word in the vocabulary. The vocabulary is loaded from
/// `{app_data_dir}/vocabulary.txt` if present, otherwise from the bundled default.
/// Vocabulary embeddings are cached to disk — only recomputed when the vocabulary changes.
fn fnv_hash_ids(ids: &[i64]) -> u64 {
let mut h: u64 = 0xcbf29ce484222325;
for &id in ids {
for b in id.to_le_bytes() {
h = h.wrapping_mul(0x100000001b3) ^ (b as u64);
}
}
h
}
/// Clusters the library's image embeddings with k-means and returns one representative
/// image per cluster — the member whose embedding is closest to its cluster centroid.
/// Results are cached in SQLite keyed by a hash of the embedded image IDs, so repeated
/// calls (including across app restarts) return instantly when the library hasn't changed.
#[tauri::command]
pub async fn get_tag_cloud(
app: AppHandle,
db: State<'_, DbState>,
folder_id: Option<i64>,
) -> Result<Vec<TagCloudEntry>, String> {
let app_data_dir = app.path().app_data_dir().map_err(|e| e.to_string())?;
let image_embeddings = {
let embeddings_with_ids = {
let conn = db.get().map_err(|e| e.to_string())?;
vector::get_all_image_embeddings(&conn, folder_id).map_err(|e| e.to_string())?
vector::get_all_image_embeddings_with_ids(&conn, folder_id).map_err(|e| e.to_string())?
};
let n = image_embeddings.len();
let n = embeddings_with_ids.len();
if n < 5 {
return Ok(vec![]);
}
// Load vocabulary (custom file > bundled default)
let vocab = embedder::load_vocabulary(&app_data_dir);
// Compute a hash of the current embedded image IDs (sorted for stability)
let mut sorted_ids: Vec<i64> = embeddings_with_ids.iter().map(|(id, _)| *id).collect();
sorted_ids.sort_unstable();
let current_hash = fnv_hash_ids(&sorted_ids);
// Embed vocabulary — disk-cached, only recomputes when vocabulary changes
let vocab_refs: Vec<&str> = vocab.iter().map(|s| s.as_str()).collect();
let vocab_embeddings = embedder::embed_vocab_cached(&vocab, &app_data_dir)
.map_err(|e| e.to_string())?;
let folder_scope = match folder_id {
Some(id) => format!("folder_{}", id),
None => "all".to_string(),
};
// Choose k proportional to library size, capped at 30
let k = (n / 20).clamp(5, 30);
// Cluster image embeddings
let (centroids, cluster_counts) = kmeans_cosine(&image_embeddings, k, 40);
// Label each cluster with the nearest vocabulary word
let mut entries: Vec<TagCloudEntry> = Vec::new();
let mut used_labels = std::collections::HashSet::new();
let mut order: Vec<usize> = (0..k).collect();
order.sort_unstable_by(|&a, &b| cluster_counts[b].cmp(&cluster_counts[a]));
for ci in order {
let count = cluster_counts[ci];
if count == 0 { continue; }
let centroid = &centroids[ci];
let label = vocab_refs
.iter()
.zip(vocab_embeddings.iter())
.map(|(&word, emb)| {
let sim: f32 = centroid.iter().zip(emb.iter()).map(|(a, b)| a * b).sum();
(word, sim)
})
.max_by(|a, b| a.1.partial_cmp(&b.1).unwrap_or(std::cmp::Ordering::Equal))
.map(|(word, _)| word.to_string())
.unwrap_or_else(|| "other".to_string());
if used_labels.insert(label.clone()) {
entries.push(TagCloudEntry { label, count });
// Try to return a valid SQLite cache
{
let conn = db.get().map_err(|e| e.to_string())?;
if let Some(json) = db::get_tag_cloud_cache(&conn, &folder_scope, current_hash)
.map_err(|e| e.to_string())?
{
if let Ok(entries) = serde_json::from_str::<Vec<TagCloudEntry>>(&json) {
return Ok(entries);
}
}
}
entries.sort_unstable_by(|a, b| b.count.cmp(&a.count));
// Cache miss — run k-means
let ids: Vec<i64> = embeddings_with_ids.iter().map(|(id, _)| *id).collect();
let points: Vec<Vec<f32>> = embeddings_with_ids.into_iter().map(|(_, emb)| emb).collect();
let k = (n / 20).clamp(5, 30);
let (centroids, cluster_counts, assignments) = kmeans_cosine(&points, k, 40);
let mut entries: Vec<TagCloudEntry> = Vec::new();
let mut order: Vec<usize> = (0..k).collect();
order.sort_unstable_by(|&a, &b| cluster_counts[b].cmp(&cluster_counts[a]));
let conn = db.get().map_err(|e| e.to_string())?;
for ci in order {
let count = cluster_counts[ci];
if count == 0 {
continue;
}
let centroid = &centroids[ci];
let best_id = points
.iter()
.enumerate()
.filter(|(i, _)| assignments[*i] == ci)
.map(|(i, p)| (ids[i], dot(centroid, p)))
.max_by(|a, b| a.1.partial_cmp(&b.1).unwrap_or(std::cmp::Ordering::Equal))
.map(|(id, _)| id)
.unwrap_or(0);
let thumbnail_path = db::get_image_by_id(&conn, best_id)
.ok()
.and_then(|img| img.thumbnail_path);
entries.push(TagCloudEntry {
count,
representative_image_id: best_id,
thumbnail_path,
});
}
// Persist to SQLite — ignore write errors (cache is best-effort)
if let Ok(json) = serde_json::to_string(&entries) {
let _ = db::set_tag_cloud_cache(&conn, &folder_scope, current_hash, &json);
}
Ok(entries)
}
@@ -315,7 +350,7 @@ fn kmeans_cosine(
points: &[Vec<f32>],
k: usize,
max_iter: usize,
) -> (Vec<Vec<f32>>, Vec<usize>) {
) -> (Vec<Vec<f32>>, Vec<usize>, Vec<usize>) {
let n = points.len();
let dim = points[0].len();
@@ -375,7 +410,31 @@ fn kmeans_cosine(
let mut counts = vec![0usize; k];
for &a in &assignments { counts[a] += 1; }
(centroids, counts)
(centroids, counts, assignments)
}
#[derive(Serialize)]
pub struct FailedEmbeddingItem {
pub image_id: i64,
pub filename: String,
pub error: Option<String>,
}
#[tauri::command]
pub async fn get_failed_embedding_images(
db: State<'_, DbState>,
folder_id: i64,
) -> Result<Vec<FailedEmbeddingItem>, String> {
let conn = db.get().map_err(|e| e.to_string())?;
let rows = db::get_failed_embedding_images(&conn, folder_id).map_err(|e| e.to_string())?;
Ok(rows
.into_iter()
.map(|(image_id, filename, error)| FailedEmbeddingItem {
image_id,
filename,
error,
})
.collect())
}
#[derive(Serialize)]
src-tauri/src/db.rs
+90 -5
View File
@@ -172,6 +172,13 @@ pub fn migrate(conn: &Connection) -> Result<()> {
updated_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS tag_cloud_cache (
folder_scope TEXT PRIMARY KEY,
image_ids_hash INTEGER NOT NULL,
entries_json TEXT NOT NULL,
created_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE INDEX IF NOT EXISTS idx_images_folder_id ON images(folder_id);
CREATE INDEX IF NOT EXISTS idx_images_modified_at ON images(modified_at);
CREATE INDEX IF NOT EXISTS idx_embedding_jobs_status ON embedding_jobs(status);
@@ -301,11 +308,16 @@ pub fn backfill_embedding_jobs(conn: &Connection) -> Result<usize> {
}
pub fn retry_failed_embedding_jobs(conn: &Connection, folder_id: i64) -> Result<usize> {
// Only re-queue images that are actually embeddable right now.
// Videos without a thumbnail would just fail again immediately, so skip them —
// they will be re-queued automatically once their thumbnail is generated.
let updated = conn.execute(
"INSERT INTO embedding_jobs (image_id, status, attempts, last_error, created_at, updated_at)
SELECT id, 'pending', 0, NULL, datetime('now'), datetime('now')
FROM images
WHERE folder_id = ?1 AND embedding_status = 'failed'
WHERE folder_id = ?1
AND embedding_status = 'failed'
AND NOT (media_kind = 'video' AND thumbnail_path IS NULL)
ON CONFLICT(image_id) DO UPDATE SET
status = 'pending',
last_error = NULL,
@@ -315,7 +327,9 @@ pub fn retry_failed_embedding_jobs(conn: &Connection, folder_id: i64) -> Result<
conn.execute(
"UPDATE images
SET embedding_status = 'pending', embedding_error = NULL
WHERE folder_id = ?1 AND embedding_status = 'failed'",
WHERE folder_id = ?1
AND embedding_status = 'failed'
AND NOT (media_kind = 'video' AND thumbnail_path IS NULL)",
[folder_id],
)?;
Ok(updated)
@@ -802,6 +816,7 @@ pub fn get_images(
search: Option<&str>,
media_kind: Option<&str>,
favorites_only: bool,
embedding_failed_only: bool,
sort: &str,
offset: i64,
limit: i64,
@@ -820,6 +835,7 @@ pub fn get_images(
let search_pattern = search.map(|value| format!("%{}%", value));
let favorites_flag = i64::from(favorites_only);
let embedding_failed_flag = i64::from(embedding_failed_only);
let sql = format!(
"SELECT id, folder_id, path, filename, thumbnail_path, width, height, file_size, created_at, modified_at, mime_type,
media_kind, duration_ms, video_codec, audio_codec, metadata_updated_at, metadata_error,
@@ -829,8 +845,9 @@ pub fn get_images(
AND (?2 IS NULL OR filename LIKE ?2)
AND (?3 IS NULL OR media_kind = ?3)
AND (?4 = 0 OR favorite = 1)
AND (?5 = 0 OR embedding_status = 'failed')
ORDER BY {}
LIMIT ?5 OFFSET ?6",
LIMIT ?6 OFFSET ?7",
order
);
let mut stmt = conn.prepare(&sql)?;
@@ -840,6 +857,7 @@ pub fn get_images(
search_pattern,
media_kind,
favorites_flag,
embedding_failed_flag,
limit,
offset
],
@@ -854,23 +872,52 @@ pub fn count_images(
search: Option<&str>,
media_kind: Option<&str>,
favorites_only: bool,
embedding_failed_only: bool,
) -> Result<i64> {
let search_pattern = search.map(|value| format!("%{}%", value));
let favorites_flag = i64::from(favorites_only);
let embedding_failed_flag = i64::from(embedding_failed_only);
let count = conn.query_row(
"SELECT COUNT(*) FROM images
WHERE (?1 IS NULL OR folder_id = ?1)
AND (?2 IS NULL OR filename LIKE ?2)
AND (?3 IS NULL OR media_kind = ?3)
AND (?4 = 0 OR favorite = 1)",
params![folder_id, search_pattern, media_kind, favorites_flag],
AND (?4 = 0 OR favorite = 1)
AND (?5 = 0 OR embedding_status = 'failed')",
params![
folder_id,
search_pattern,
media_kind,
favorites_flag,
embedding_failed_flag
],
|row| row.get(0),
)?;
Ok(count)
}
pub fn get_failed_embedding_images(
conn: &Connection,
folder_id: i64,
) -> Result<Vec<(i64, String, Option<String>)>> {
let mut stmt = conn.prepare(
"SELECT id, filename, embedding_error
FROM images
WHERE folder_id = ?1 AND embedding_status = 'failed'
ORDER BY filename",
)?;
let rows = stmt.query_map([folder_id], |row| {
Ok((
row.get::<_, i64>(0)?,
row.get::<_, String>(1)?,
row.get::<_, Option<String>>(2)?,
))
})?;
Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
}
pub fn delete_folder(conn: &Connection, folder_id: i64) -> Result<()> {
conn.execute("DELETE FROM folders WHERE id = ?1", params![folder_id])?;
Ok(())
@@ -904,6 +951,44 @@ fn map_image_row(row: &Row<'_>) -> rusqlite::Result<ImageRecord> {
})
}
/// Returns cached tag-cloud entries if the stored hash matches `current_hash`.
pub fn get_tag_cloud_cache(
conn: &Connection,
folder_scope: &str,
current_hash: u64,
) -> Result<Option<String>> {
let result: rusqlite::Result<(i64, String)> = conn.query_row(
"SELECT image_ids_hash, entries_json FROM tag_cloud_cache WHERE folder_scope = ?1",
params![folder_scope],
|row| Ok((row.get(0)?, row.get(1)?)),
);
match result {
Ok((stored_hash, json)) if stored_hash as u64 == current_hash => Ok(Some(json)),
Ok(_) => Ok(None),
Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
Err(e) => Err(e.into()),
}
}
/// Upserts the tag-cloud cache for the given scope.
pub fn set_tag_cloud_cache(
conn: &Connection,
folder_scope: &str,
image_ids_hash: u64,
entries_json: &str,
) -> Result<()> {
conn.execute(
"INSERT INTO tag_cloud_cache (folder_scope, image_ids_hash, entries_json, created_at)
VALUES (?1, ?2, ?3, datetime('now'))
ON CONFLICT(folder_scope) DO UPDATE SET
image_ids_hash = excluded.image_ids_hash,
entries_json = excluded.entries_json,
created_at = excluded.created_at",
params![folder_scope, image_ids_hash as i64, entries_json],
)?;
Ok(())
}
fn ensure_column(conn: &Connection, table: &str, column: &str, definition: &str) -> Result<()> {
let mut stmt = conn.prepare(&format!("PRAGMA table_info({})", table))?;
let mut rows = stmt.query([])?;
src-tauri/src/embedder.rs
+21 -158
View File
@@ -1,173 +1,24 @@
use anyhow::{Context, Result};
use anyhow::Result;
use candle_core::{DType, Device, Tensor};
use candle_nn::VarBuilder;
use candle_transformers::models::clip::{self, ClipModel};
use hf_hub::{api::sync::Api, Repo, RepoType};
use std::io::{Read, Write};
use std::path::{Path, PathBuf};
use std::sync::{Mutex, OnceLock};
use tokenizers::Tokenizer;
static TEXT_SEARCH_EMBEDDER: OnceLock<Mutex<Option<ClipImageEmbedder>>> = OnceLock::new();
/// In-process cache so the disk file is only read/written once per session.
static VOCAB_EMBED_CACHE: OnceLock<Mutex<Option<Vec<Vec<f32>>>>> = OnceLock::new();
/// Default vocabulary bundled with the binary.
pub const DEFAULT_VOCABULARY: &str = include_str!("../data/vocabulary.txt");
/// Embed a text query using a lazily-initialized, cached CLIP embedder.
pub fn embed_text_query(query: &str) -> Result<Vec<f32>> {
with_text_embedder(|e| e.embed_text(query))
}
/// Parse vocabulary text: strip comment lines (starting with `#`) and blank lines.
pub fn parse_vocabulary(text: &str) -> Vec<String> {
text.lines()
.map(|l| l.trim())
.filter(|l| !l.is_empty() && !l.starts_with('#'))
.map(|l| l.to_string())
.collect()
}
/// Load the vocabulary from `{cache_dir}/vocabulary.txt` if it exists,
/// otherwise fall back to the binary-bundled default.
pub fn load_vocabulary(cache_dir: &Path) -> Vec<String> {
let custom_path = cache_dir.join("vocabulary.txt");
if custom_path.exists() {
if let Ok(text) = std::fs::read_to_string(&custom_path) {
let words = parse_vocabulary(&text);
if !words.is_empty() {
println!("Using custom vocabulary from {:?} ({} words)", custom_path, words.len());
return words;
}
}
}
parse_vocabulary(DEFAULT_VOCABULARY)
}
/// Embed the vocabulary, using a disk cache so CLIP is only called when the vocabulary
/// actually changes. Cache file: `{cache_dir}/vocab_embeddings.bin`.
///
/// Format: `[u64 hash][u32 n_words][u32 dim][(n_words × dim) × f32 LE]`
pub fn embed_vocab_cached(vocab: &[String], cache_dir: &Path) -> Result<Vec<Vec<f32>>> {
// In-process cache hit
{
let guard = VOCAB_EMBED_CACHE
.get_or_init(|| Mutex::new(None))
.lock()
.map_err(|_| anyhow::anyhow!("Vocab cache lock poisoned"))?;
if let Some(cached) = guard.as_ref() {
return Ok(cached.clone());
}
}
let vocab_hash = fnv_hash(vocab);
let disk_path = cache_dir.join("vocab_embeddings.bin");
// Try loading from disk
if let Ok(embeddings) = load_disk_cache(&disk_path, vocab_hash) {
let mut guard = VOCAB_EMBED_CACHE
.get_or_init(|| Mutex::new(None))
.lock()
.map_err(|_| anyhow::anyhow!("Vocab cache lock poisoned"))?;
*guard = Some(embeddings.clone());
println!("Vocabulary embeddings loaded from disk cache ({} words).", embeddings.len());
return Ok(embeddings);
}
// Compute embeddings
println!("Computing vocabulary embeddings ({} words) — this is cached after the first run.", vocab.len());
let prompts: Vec<String> = vocab.iter().map(|w| format!("a photo of {}", w)).collect();
let prompt_refs: Vec<&str> = prompts.iter().map(|s| s.as_str()).collect();
let embeddings = with_text_embedder(|e| {
let mut all = Vec::with_capacity(vocab.len());
for chunk in prompt_refs.chunks(64) {
all.extend(e.embed_texts_batch(chunk)?);
}
Ok(all)
})?;
// Save to disk
if let Err(e) = save_disk_cache(&disk_path, vocab_hash, &embeddings) {
eprintln!("Warning: could not write vocab cache: {e}");
}
let mut guard = VOCAB_EMBED_CACHE
.get_or_init(|| Mutex::new(None))
.lock()
.map_err(|_| anyhow::anyhow!("Vocab cache lock poisoned"))?;
*guard = Some(embeddings.clone());
Ok(embeddings)
}
fn fnv_hash(words: &[String]) -> u64 {
let mut h: u64 = 0xcbf29ce484222325;
for w in words {
for b in w.bytes() {
h = h.wrapping_mul(0x100000001b3) ^ (b as u64);
}
h = h.wrapping_mul(0x100000001b3) ^ 0x0A; // newline separator
}
h
}
fn load_disk_cache(path: &Path, expected_hash: u64) -> Result<Vec<Vec<f32>>> {
let mut f = std::fs::File::open(path).context("no cache file")?;
let mut buf = Vec::new();
f.read_to_end(&mut buf)?;
if buf.len() < 16 {
anyhow::bail!("cache too short");
}
let hash = u64::from_le_bytes(buf[0..8].try_into()?);
if hash != expected_hash {
anyhow::bail!("vocab hash mismatch — recomputing");
}
let n = u32::from_le_bytes(buf[8..12].try_into()?) as usize;
let dim = u32::from_le_bytes(buf[12..16].try_into()?) as usize;
let expected_len = 16 + n * dim * 4;
if buf.len() != expected_len {
anyhow::bail!("cache size mismatch");
}
let mut embeddings = Vec::with_capacity(n);
let data = &buf[16..];
for i in 0..n {
let start = i * dim * 4;
let emb: Vec<f32> = data[start..start + dim * 4]
.chunks_exact(4)
.map(|b| f32::from_le_bytes([b[0], b[1], b[2], b[3]]))
.collect();
embeddings.push(emb);
}
Ok(embeddings)
}
fn save_disk_cache(path: &Path, hash: u64, embeddings: &[Vec<f32>]) -> Result<()> {
let n = embeddings.len();
let dim = embeddings.first().map(|e| e.len()).unwrap_or(0);
let mut buf = Vec::with_capacity(16 + n * dim * 4);
buf.extend_from_slice(&hash.to_le_bytes());
buf.extend_from_slice(&(n as u32).to_le_bytes());
buf.extend_from_slice(&(dim as u32).to_le_bytes());
for emb in embeddings {
for &v in emb {
buf.extend_from_slice(&v.to_le_bytes());
}
}
std::fs::write(path, buf)?;
println!("Vocabulary embeddings cached to disk ({n} words, {dim} dims).");
Ok(())
}
fn with_text_embedder<T>(f: impl FnOnce(&ClipImageEmbedder) -> Result<T>) -> Result<T> {
let lock = TEXT_SEARCH_EMBEDDER.get_or_init(|| Mutex::new(None));
let mut guard = lock.lock().map_err(|_| anyhow::anyhow!("Text embedder lock poisoned"))?;
let mut guard = lock
.lock()
.map_err(|_| anyhow::anyhow!("Text embedder lock poisoned"))?;
if guard.is_none() {
println!("Initializing CLIP text embedder...");
*guard = Some(ClipImageEmbedder::new()?);
@@ -319,16 +170,28 @@ fn load_images(paths: &[PathBuf], image_size: usize) -> Result<Tensor> {
Ok(Tensor::stack(&images, 0)?)
}
/// Returns the path that should be fed to the CLIP image embedder for a given media file.
///
/// For videos the thumbnail image is used (because CLIP only understands still images).
/// If a video has no thumbnail yet, an error is returned — the caller should mark the
/// embedding job as failed rather than trying to decode the raw video file.
pub fn embedding_source_path(
path: &str,
thumbnail_path: Option<&str>,
media_kind: &str,
) -> PathBuf {
) -> Result<PathBuf> {
if media_kind == "video" {
thumbnail_path
.map(PathBuf::from)
.unwrap_or_else(|| PathBuf::from(path))
match thumbnail_path {
Some(thumb) => Ok(PathBuf::from(thumb)),
None => Err(anyhow::anyhow!(
"No thumbnail available yet for video '{}' — embedding deferred until thumbnail is generated",
std::path::Path::new(path)
.file_name()
.map(|n| n.to_string_lossy())
.unwrap_or_default()
)),
}
} else {
PathBuf::from(path)
Ok(PathBuf::from(path))
}
}
src-tauri/src/indexer.rs
+60 -19
View File
@@ -1,4 +1,4 @@
use crate::db::{self, DbPool, FolderJobProgress, ImageRecord, IndexedMediaEntry};
use crate::db::{self, DbPool, EmbeddingJob, FolderJobProgress, ImageRecord, IndexedMediaEntry};
use crate::embedder::{embedding_source_path, ClipImageEmbedder};
use crate::media::{probe_video_metadata, MediaTools};
use crate::storage::{detect_storage_profile, RuntimeAdaptiveProfile, StorageProfile};
@@ -539,27 +539,59 @@ fn process_embedding_batch(
let embedder = embedder.as_ref().expect("embedder should be initialized");
let infer_started_at = Instant::now();
let source_paths = jobs
// Resolve the source path for each job. Videos without a thumbnail produce an Err
// here — those jobs are marked failed immediately without going to the embedder.
let source_results: Vec<Result<PathBuf>> = jobs
.iter()
.map(|job| embedding_source_path(&job.path, job.thumbnail_path.as_deref(), &job.media_kind))
.collect::<Vec<_>>();
.collect();
let results = match embedder.embed_images(&source_paths) {
Ok(embeddings) => jobs
.into_iter()
.zip(embeddings.into_iter().map(Ok))
.collect::<Vec<_>>(),
Err(batch_error) => {
eprintln!(
"Embedding batch fallback to per-image mode: {}",
batch_error
);
jobs.into_iter()
.zip(source_paths.into_iter())
.map(|(job, source_path)| (job, embedder.embed_image(&source_path)))
.collect::<Vec<_>>()
// Separate jobs with a valid source from those that fail early (e.g. video with no thumbnail).
let mut embeddable_indices: Vec<usize> = Vec::new();
let mut embeddable_paths: Vec<PathBuf> = Vec::new();
// image_id -> early error message for jobs that cannot be embedded yet
let mut pre_failed: HashMap<i64, String> = HashMap::new();
for (i, (job, result)) in jobs.iter().zip(source_results.into_iter()).enumerate() {
match result {
Ok(path) => {
embeddable_indices.push(i);
embeddable_paths.push(path);
}
Err(e) => {
pre_failed.insert(job.image_id, e.to_string());
}
}
};
}
// Run CLIP only on the jobs that have a valid source image.
// image_id -> embedding result
let mut embed_results: HashMap<i64, Result<Vec<f32>>> = HashMap::new();
if !embeddable_indices.is_empty() {
let embeddable_jobs: Vec<&EmbeddingJob> =
embeddable_indices.iter().map(|&i| &jobs[i]).collect();
match embedder.embed_images(&embeddable_paths) {
Ok(embeddings) => {
for (job, embedding) in embeddable_jobs.iter().zip(embeddings.into_iter()) {
embed_results.insert(job.image_id, Ok(embedding));
}
}
Err(batch_error) => {
eprintln!(
"Embedding batch fallback to per-image mode: {}",
batch_error
);
for (job, source_path) in embeddable_jobs
.into_iter()
.zip(embeddable_paths.into_iter())
{
embed_results.insert(job.image_id, embedder.embed_image(&source_path));
}
}
}
}
let infer_elapsed = infer_started_at.elapsed();
let write_started_at = Instant::now();
@@ -568,7 +600,16 @@ fn process_embedding_batch(
let tx = conn.transaction()?;
let mut updated_images = Vec::new();
for (job, embedding_result) in results {
for job in &jobs {
let embedding_result: Result<Vec<f32>> =
if let Some(err) = pre_failed.remove(&job.image_id) {
Err(anyhow::anyhow!("{}", err))
} else if let Some(r) = embed_results.remove(&job.image_id) {
r
} else {
Err(anyhow::anyhow!("no result for image {}", job.image_id))
};
match embedding_result {
Ok(embedding) => {
vector::upsert_embedding(&tx, job.image_id, &embedding)?;
src-tauri/src/lib.rs
+1
View File
@@ -78,6 +78,7 @@ pub fn run() {
commands::set_worker_paused,
commands::get_worker_states,
commands::get_tag_cloud,
commands::get_failed_embedding_images,
])
.run(tauri::generate_context!())
.expect("error while running tauri application");
src-tauri/src/vector.rs
+21 -11
View File
@@ -78,32 +78,42 @@ pub fn find_similar_image_ids(conn: &Connection, image_id: i64, limit: usize) ->
Ok(ids)
}
/// Returns all stored image embeddings, optionally filtered to one folder.
/// Each embedding is returned as a normalized f32 vector.
pub fn get_all_image_embeddings(conn: &Connection, folder_id: Option<i64>) -> Result<Vec<Vec<f32>>> {
let packed_rows: Vec<Vec<u8>> = match folder_id {
/// Returns all stored image embeddings with their image IDs, optionally filtered to one folder.
/// Each entry is `(image_id, normalized_f32_embedding)`.
pub fn get_all_image_embeddings_with_ids(
conn: &Connection,
folder_id: Option<i64>,
) -> Result<Vec<(i64, Vec<f32>)>> {
let packed_rows: Vec<(i64, Vec<u8>)> = match folder_id {
Some(fid) => {
let mut stmt = conn.prepare(
"SELECT embedding FROM image_vec
"SELECT image_id, embedding FROM image_vec
WHERE image_id IN (SELECT id FROM images WHERE folder_id = ?1)",
)?;
let rows: Vec<Vec<u8>> = stmt
.query_map([fid], |row| row.get::<_, Vec<u8>>(0))?
let rows: Vec<(i64, Vec<u8>)> = stmt
.query_map([fid], |row| {
Ok((row.get::<_, i64>(0)?, row.get::<_, Vec<u8>>(1)?))
})?
.filter_map(|r| r.ok())
.collect();
rows
}
None => {
let mut stmt = conn.prepare("SELECT embedding FROM image_vec")?;
let rows: Vec<Vec<u8>> = stmt
.query_map([], |row| row.get::<_, Vec<u8>>(0))?
let mut stmt = conn.prepare("SELECT image_id, embedding FROM image_vec")?;
let rows: Vec<(i64, Vec<u8>)> = stmt
.query_map([], |row| {
Ok((row.get::<_, i64>(0)?, row.get::<_, Vec<u8>>(1)?))
})?
.filter_map(|r| r.ok())
.collect();
rows
}
};
Ok(packed_rows.iter().map(|b| unpack_f32(b)).collect())
Ok(packed_rows
.into_iter()
.map(|(id, b)| (id, unpack_f32(&b)))
.collect())
}
fn unpack_f32(bytes: &[u8]) -> Vec<f32> {