Files
rustytorch/demos/server/src/slidescope_service.rs
T
2026-03-04 00:08:42 +00:00

753 lines
23 KiB
Rust

//! SlideScope pathology service implementation
//!
//! This module provides the main service interface for SlideScope digital
//! pathology processing. It handles slide import, NMF stain unmixing,
//! tile pyramid serving, and job management.
use std::collections::HashMap;
use std::num::NonZeroUsize;
use std::path::Path;
use std::sync::Arc;
use std::time::{Instant, SystemTime, UNIX_EPOCH};
use image::GenericImageView;
use lru::LruCache;
use tokio::sync::{RwLock, broadcast, mpsc};
use uuid::Uuid;
use rtx_slidescope::{
MultiLayerPyramid, NmfProcessingConfig, ProcessingConfig, SlidescopeProcessor,
};
use slidescope_shared::{
DziMetadata, ImageFormat, JobProgress, JobStatus, NmfConfig, NmfResult, SlideFilter,
SlideMetadata, SlideStatus, SlidescopeStatus, StainType, TileLayer, TileRequest, TileResponse,
jobs::JobState,
};
use crate::error::{ServerError, ServerResult};
/// SlideScope service configuration
#[derive(Debug, Clone)]
pub struct SlidescopeServiceConfig {
/// Tile size for pyramids
pub tile_size: u32,
/// JPEG quality for tiles (0-100)
pub jpeg_quality: u8,
/// Maximum tiles in LRU cache
pub max_cached_tiles: usize,
/// Maximum memory for tile cache in MB
pub max_cache_memory_mb: usize,
/// Default NMF components
pub default_nmf_components: usize,
/// Maximum NMF iterations
pub max_nmf_iterations: usize,
/// NMF convergence tolerance
pub nmf_tolerance: f32,
/// Use GPU for NMF
pub use_gpu: bool,
}
impl Default for SlidescopeServiceConfig {
fn default() -> Self {
Self {
tile_size: 256,
jpeg_quality: 85,
max_cached_tiles: 10000,
max_cache_memory_mb: 512,
default_nmf_components: 2,
max_nmf_iterations: 200,
nmf_tolerance: 1e-4,
use_gpu: true,
}
}
}
impl SlidescopeServiceConfig {
/// Fast configuration for testing
#[must_use]
pub fn fast() -> Self {
Self {
tile_size: 128,
jpeg_quality: 75,
max_cached_tiles: 1000,
max_cache_memory_mb: 64,
default_nmf_components: 2,
max_nmf_iterations: 50,
nmf_tolerance: 1e-3,
use_gpu: false,
}
}
}
/// Tile cache key
#[derive(Debug, Clone, Hash, PartialEq, Eq)]
struct TileCacheKey {
slide_id: String,
layer: TileLayer,
level: u8,
x: u32,
y: u32,
}
/// Processing job for the job queue (reserved for async job manager)
#[allow(dead_code)]
#[derive(Debug)]
struct ProcessingJob {
job_id: String,
slide_id: String,
config: NmfConfig,
}
/// SlideScope handle containing active state
struct SlidescopeHandle {
/// Workspace path
workspace: String,
/// Slide metadata registry
slides: HashMap<String, SlideMetadata>,
/// Loaded pyramids (slide_id -> layer -> pyramid)
pyramids: HashMap<String, MultiLayerPyramid>,
/// Tile cache
tile_cache: LruCache<TileCacheKey, Vec<u8>>,
/// Processing results
results: HashMap<String, NmfResult>,
/// Job statuses
jobs: HashMap<String, JobStatus>,
/// Start time
start_time: Instant,
}
impl SlidescopeHandle {
/// Create a new handle
fn new(workspace: String, cache_size: usize) -> Self {
let cache_size = NonZeroUsize::new(cache_size).unwrap_or(NonZeroUsize::new(10000).unwrap());
Self {
workspace,
slides: HashMap::new(),
pyramids: HashMap::new(),
tile_cache: LruCache::new(cache_size),
results: HashMap::new(),
jobs: HashMap::new(),
start_time: Instant::now(),
}
}
/// Get current status
fn status(&self) -> SlidescopeStatus {
let pending_count = self
.jobs
.values()
.filter(|j| j.state == JobState::Queued)
.count();
let running_job = self.jobs.values().find(|j| j.state == JobState::Running);
SlidescopeStatus {
initialized: true,
gpu_available: rtx_slidescope::gpu_available(),
gpu_name: rtx_slidescope::gpu_device_name(),
slide_count: self.slides.len(),
pending_jobs: pending_count,
active_job: running_job.map(|j| j.id.clone()),
workspace: Some(self.workspace.clone()),
uptime_secs: self.start_time.elapsed().as_secs(),
}
}
}
/// SlideScope pathology service
///
/// This service manages slide import, NMF stain unmixing,
/// tile pyramid generation and serving.
pub struct SlidescopeService {
/// Service configuration
config: SlidescopeServiceConfig,
/// Active handle
handle: Arc<RwLock<Option<SlidescopeHandle>>>,
/// Job sender
job_tx: mpsc::UnboundedSender<ProcessingJob>,
/// Progress broadcast
progress_tx: broadcast::Sender<JobProgress>,
}
impl SlidescopeService {
/// Create a new SlideScope service
#[must_use]
pub fn new(config: SlidescopeServiceConfig) -> Self {
let (job_tx, _job_rx) = mpsc::unbounded_channel();
let (progress_tx, _) = broadcast::channel(100);
Self {
config,
handle: Arc::new(RwLock::new(None)),
job_tx,
progress_tx,
}
}
/// Create a service with default configuration
#[must_use]
pub fn with_defaults() -> Self {
Self::new(SlidescopeServiceConfig::default())
}
/// Subscribe to progress updates
pub fn subscribe_progress(&self) -> broadcast::Receiver<JobProgress> {
self.progress_tx.subscribe()
}
/// Initialize the service with a workspace path
///
/// Creates the workspace directory and initializes the tile cache.
pub async fn initialize(&self, workspace: &str) -> ServerResult<()> {
let mut handle_guard = self.handle.write().await;
if handle_guard.is_some() {
return Err(ServerError::AlreadyInitialized);
}
// Create workspace directory
let workspace_path = Path::new(workspace);
std::fs::create_dir_all(workspace_path)
.map_err(|e| ServerError::internal(format!("Failed to create workspace: {}", e)))?;
let handle = SlidescopeHandle::new(workspace.to_string(), self.config.max_cached_tiles);
tracing::info!(
"SlideScope service initialized with workspace: {}",
workspace
);
*handle_guard = Some(handle);
Ok(())
}
/// Reset the service
pub async fn reset(&self) -> ServerResult<()> {
let mut handle_guard = self.handle.write().await;
*handle_guard = None;
Ok(())
}
/// Check if initialized
pub async fn is_initialized(&self) -> bool {
self.handle.read().await.is_some()
}
/// Import a slide from a file path
pub async fn import_slide(&self, path: &str) -> ServerResult<SlideMetadata> {
let mut handle_guard = self.handle.write().await;
let handle = handle_guard.as_mut().ok_or(ServerError::NotInitialized)?;
// Load the image
let image_path = Path::new(path);
if !image_path.exists() {
return Err(ServerError::not_found(format!("File not found: {}", path)));
}
let image = image::open(image_path)
.map_err(|e| ServerError::internal(format!("Failed to load image: {}", e)))?;
let (width, height) = image.dimensions();
let _format = detect_image_format(path);
let name = image_path
.file_stem()
.and_then(|s| s.to_str())
.unwrap_or("unknown")
.to_string();
let id = Uuid::new_v4().to_string();
let created_at = current_timestamp();
// Generate pyramid
tracing::info!(
"Generating tile pyramid for {} ({}x{})",
name,
width,
height
);
let processor = SlidescopeProcessor::with_defaults()
.map_err(|e| ServerError::internal(format!("Failed to create processor: {}", e)))?;
let pyramid = processor
.generate_pyramid(&image)
.map_err(|e| ServerError::internal(format!("Failed to generate pyramid: {}", e)))?;
// Create multi-layer pyramid with original only
let mut multi_pyramid = MultiLayerPyramid::new(width, height);
multi_pyramid.add_layer(TileLayer::Original, pyramid);
// Create metadata
let metadata = SlideMetadata {
id: id.clone(),
name: name.clone(),
path: path.to_string(),
width,
height,
status: SlideStatus::Imported,
stain_type: None,
thumbnail_path: None,
pyramid_path: None,
created_at,
processed_at: None,
};
// Store in memory
handle.slides.insert(id.clone(), metadata.clone());
handle.pyramids.insert(id.clone(), multi_pyramid);
tracing::info!(
"Imported slide {} ({} tiles)",
id,
handle
.pyramids
.get(&id)
.map(|p| p.layers().len())
.unwrap_or(0)
);
Ok(metadata)
}
/// List slides with optional filter
pub async fn list_slides(
&self,
filter: Option<SlideFilter>,
) -> ServerResult<Vec<SlideMetadata>> {
let handle_guard = self.handle.read().await;
let handle = handle_guard.as_ref().ok_or(ServerError::NotInitialized)?;
let slides: Vec<SlideMetadata> = handle
.slides
.values()
.filter(|s| {
if let Some(ref f) = filter {
if let Some(ref status) = f.status {
if s.status != *status {
return false;
}
}
if let Some(ref name_filter) = f.name_contains {
if !s.name.contains(name_filter) {
return false;
}
}
}
true
})
.cloned()
.collect();
Ok(slides)
}
/// Get a slide by ID
pub async fn get_slide(&self, slide_id: &str) -> ServerResult<SlideMetadata> {
let handle_guard = self.handle.read().await;
let handle = handle_guard.as_ref().ok_or(ServerError::NotInitialized)?;
handle
.slides
.get(slide_id)
.cloned()
.ok_or_else(|| ServerError::not_found(format!("Slide not found: {}", slide_id)))
}
/// Get a tile from a slide
pub async fn get_tile(&self, request: TileRequest) -> ServerResult<TileResponse> {
let mut handle_guard = self.handle.write().await;
let handle = handle_guard.as_mut().ok_or(ServerError::NotInitialized)?;
let cache_key = TileCacheKey {
slide_id: request.slide_id.clone(),
layer: request.layer,
level: request.z,
x: request.x,
y: request.y,
};
// Check cache first
if let Some(data) = handle.tile_cache.get(&cache_key) {
return Ok(TileResponse {
data: data.clone(),
width: self.config.tile_size,
height: self.config.tile_size,
});
}
// Get from pyramid
let pyramid = handle.pyramids.get(&request.slide_id).ok_or_else(|| {
ServerError::not_found(format!("Slide not loaded: {}", request.slide_id))
})?;
let tile_data = pyramid
.get_tile(request.layer, request.z, request.x, request.y)
.ok_or_else(|| {
ServerError::not_found(format!(
"Tile not found: {}/{}/{}/{}",
request.layer, request.z, request.x, request.y
))
})?;
// Cache the tile
handle.tile_cache.put(cache_key, tile_data.clone());
Ok(TileResponse {
data: tile_data.clone(),
width: self.config.tile_size,
height: self.config.tile_size,
})
}
/// Get DZI metadata for a slide layer
pub async fn get_dzi(&self, slide_id: &str, layer: TileLayer) -> ServerResult<DziMetadata> {
let handle_guard = self.handle.read().await;
let handle = handle_guard.as_ref().ok_or(ServerError::NotInitialized)?;
let pyramid = handle
.pyramids
.get(slide_id)
.ok_or_else(|| ServerError::not_found(format!("Slide not loaded: {}", slide_id)))?;
pyramid
.get_dzi_metadata(layer)
.ok_or_else(|| ServerError::not_found(format!("Layer not found: {:?}", layer)))
}
/// Queue a slide for NMF processing
pub async fn queue_processing(
&self,
slide_id: &str,
config: NmfConfig,
) -> ServerResult<JobStatus> {
let mut handle_guard = self.handle.write().await;
let handle = handle_guard.as_mut().ok_or(ServerError::NotInitialized)?;
// Verify slide exists
if !handle.slides.contains_key(slide_id) {
return Err(ServerError::not_found(format!(
"Slide not found: {}",
slide_id
)));
}
let job_id = Uuid::new_v4().to_string();
let status = JobStatus::new_queued(job_id.clone(), slide_id.to_string());
handle.jobs.insert(job_id.clone(), status.clone());
// Queue for processing
let job = ProcessingJob {
job_id: job_id.clone(),
slide_id: slide_id.to_string(),
config,
};
let _ = self.job_tx.send(job);
Ok(status)
}
/// Process a slide with NMF (synchronous, for simplicity)
pub async fn process_nmf(&self, slide_id: &str, config: NmfConfig) -> ServerResult<NmfResult> {
let start_time = Instant::now();
// Get the slide image
let slide_path = {
let handle_guard = self.handle.read().await;
let handle = handle_guard.as_ref().ok_or(ServerError::NotInitialized)?;
let slide = handle
.slides
.get(slide_id)
.ok_or_else(|| ServerError::not_found(format!("Slide not found: {}", slide_id)))?;
slide.path.clone()
};
// Update status to processing
{
let mut handle_guard = self.handle.write().await;
let handle = handle_guard.as_mut().ok_or(ServerError::NotInitialized)?;
if let Some(slide) = handle.slides.get_mut(slide_id) {
slide.status = SlideStatus::Processing;
}
}
// Load image and process
let image = image::open(&slide_path)
.map_err(|e| ServerError::internal(format!("Failed to load image: {}", e)))?;
// Create processor config
let _nmf_config = NmfProcessingConfig {
n_components: config.n_components,
max_iterations: config.max_iterations,
tolerance: config.tolerance,
use_gpu: config.use_gpu,
random_seed: config.random_seed,
};
// Process with GPU or CPU
let processor_config = ProcessingConfig::default();
let processor = SlidescopeProcessor::new(processor_config)
.map_err(|e| ServerError::internal(format!("Failed to create processor: {}", e)))?;
let nmf_result = processor
.process_nmf(&image)
.map_err(|e| ServerError::processing(format!("NMF processing failed: {}", e)))?;
let processing_time_ms = start_time.elapsed().as_millis() as u64;
// Store the multi-layer pyramid
{
let mut handle_guard = self.handle.write().await;
let handle = handle_guard.as_mut().ok_or(ServerError::NotInitialized)?;
// Update slide status
if let Some(slide) = handle.slides.get_mut(slide_id) {
slide.status = SlideStatus::Processed;
slide.stain_type = Some(StainType::HE);
slide.processed_at = Some(current_timestamp());
}
// Store pyramid
handle
.pyramids
.insert(slide_id.to_string(), nmf_result.multi_pyramid);
}
// Convert stain matrix to RGB triplet vectors
let stain_vectors: Vec<[f32; 3]> = nmf_result
.stain_matrix
.as_slice()
.unwrap_or(&[])
.chunks_exact(3)
.map(|chunk| [chunk[0], chunk[1], chunk[2]])
.collect();
// Create result
let result = NmfResult {
slide_id: slide_id.to_string(),
stain_images: vec![], // Stored as in-memory pyramids
composite_path: None,
reconstruction_error: nmf_result.stain_matrix.iter().sum::<f32>().abs()
/ nmf_result.stain_matrix.len() as f32, // Simple metric
iterations: config.n_components, // NMF components used
processing_time_ms,
gpu_memory_mb: None,
stain_vectors: if stain_vectors.is_empty() {
None
} else {
Some(stain_vectors)
},
};
// Store result
{
let mut handle_guard = self.handle.write().await;
let handle = handle_guard.as_mut().ok_or(ServerError::NotInitialized)?;
handle.results.insert(slide_id.to_string(), result.clone());
}
tracing::info!(
"NMF processing complete for {} in {}ms",
slide_id,
processing_time_ms
);
Ok(result)
}
/// Get job status
pub async fn job_status(&self, job_id: &str) -> ServerResult<JobStatus> {
let handle_guard = self.handle.read().await;
let handle = handle_guard.as_ref().ok_or(ServerError::NotInitialized)?;
handle
.jobs
.get(job_id)
.cloned()
.ok_or_else(|| ServerError::not_found(format!("Job not found: {}", job_id)))
}
/// Get processing result for a slide
pub async fn get_result(&self, slide_id: &str) -> ServerResult<NmfResult> {
let handle_guard = self.handle.read().await;
let handle = handle_guard.as_ref().ok_or(ServerError::NotInitialized)?;
handle
.results
.get(slide_id)
.cloned()
.ok_or_else(|| ServerError::not_found(format!("Result not found: {}", slide_id)))
}
/// Get current service status
pub async fn status(&self) -> ServerResult<SlidescopeStatus> {
let handle_guard = self.handle.read().await;
match handle_guard.as_ref() {
Some(handle) => Ok(handle.status()),
None => Ok(SlidescopeStatus {
initialized: false,
gpu_available: rtx_slidescope::gpu_available(),
gpu_name: rtx_slidescope::gpu_device_name(),
slide_count: 0,
pending_jobs: 0,
active_job: None,
workspace: None,
uptime_secs: 0,
}),
}
}
/// Delete a slide
pub async fn delete_slide(&self, slide_id: &str) -> ServerResult<()> {
let mut handle_guard = self.handle.write().await;
let handle = handle_guard.as_mut().ok_or(ServerError::NotInitialized)?;
// Remove from memory
handle.slides.remove(slide_id);
handle.pyramids.remove(slide_id);
handle.results.remove(slide_id);
// Clear related cache entries
let keys_to_remove: Vec<_> = handle
.tile_cache
.iter()
.filter(|(k, _)| k.slide_id == slide_id)
.map(|(k, _)| k.clone())
.collect();
for key in keys_to_remove {
handle.tile_cache.pop(&key);
}
Ok(())
}
}
// Helper functions
fn current_timestamp() -> i64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_secs() as i64
}
fn detect_image_format(path: &str) -> ImageFormat {
let lower = path.to_lowercase();
if lower.ends_with(".tiff") || lower.ends_with(".tif") {
ImageFormat::Tiff
} else if lower.ends_with(".png") {
ImageFormat::Png
} else if lower.ends_with(".jpg") || lower.ends_with(".jpeg") {
ImageFormat::Jpeg
} else if lower.ends_with(".svs") {
ImageFormat::Svs
} else if lower.ends_with(".ndpi") {
ImageFormat::Ndpi
} else {
ImageFormat::Tiff // Default
}
}
// Helper functions reserved for future persistence layer
#[allow(dead_code)]
fn format_to_string(format: &ImageFormat) -> &'static str {
match format {
ImageFormat::Tiff => "tiff",
ImageFormat::BigTiff => "bigtiff",
ImageFormat::Svs => "svs",
ImageFormat::Ndpi => "ndpi",
ImageFormat::Png => "png",
ImageFormat::Jpeg => "jpeg",
}
}
#[allow(dead_code)]
fn parse_slide_status(s: &str) -> SlideStatus {
match s {
"imported" => SlideStatus::Imported,
"processing" => SlideStatus::Processing,
"processed" => SlideStatus::Processed,
"failed" => SlideStatus::Failed,
_ => SlideStatus::Imported,
}
}
#[allow(dead_code)]
fn parse_stain_type(s: &str) -> StainType {
match s {
"H&E" | "HE" => StainType::HE,
"IHC-DAB" | "IhcDab" => StainType::IhcDab,
_ => StainType::Custom,
}
}
#[allow(dead_code)]
fn parse_image_format(s: &str) -> ImageFormat {
match s {
"tiff" => ImageFormat::Tiff,
"bigtiff" => ImageFormat::BigTiff,
"svs" => ImageFormat::Svs,
"ndpi" => ImageFormat::Ndpi,
"png" => ImageFormat::Png,
"jpeg" => ImageFormat::Jpeg,
_ => ImageFormat::Tiff,
}
}
#[cfg(test)]
mod tests {
use super::*;
use tempfile::tempdir;
async fn create_test_service() -> (SlidescopeService, tempfile::TempDir) {
let dir = tempdir().unwrap();
let service = SlidescopeService::new(SlidescopeServiceConfig::fast());
service
.initialize(dir.path().to_str().unwrap())
.await
.unwrap();
(service, dir)
}
#[tokio::test]
async fn test_service_creation() {
let service = SlidescopeService::with_defaults();
assert!(!service.is_initialized().await);
}
#[tokio::test]
async fn test_initialize() {
let (service, _dir) = create_test_service().await;
assert!(service.is_initialized().await);
}
#[tokio::test]
async fn test_double_initialize_fails() {
let (service, dir) = create_test_service().await;
let result = service.initialize(dir.path().to_str().unwrap()).await;
assert!(result.is_err());
}
#[tokio::test]
async fn test_status() {
let (service, _dir) = create_test_service().await;
let status = service.status().await.unwrap();
assert!(status.initialized);
assert_eq!(status.slide_count, 0);
}
#[tokio::test]
async fn test_reset() {
let (service, _dir) = create_test_service().await;
assert!(service.is_initialized().await);
service.reset().await.unwrap();
assert!(!service.is_initialized().await);
}
#[tokio::test]
async fn test_list_slides_empty() {
let (service, _dir) = create_test_service().await;
let slides = service.list_slides(None).await.unwrap();
assert!(slides.is_empty());
}
}