From dbf0532afdd40e2f97d3b4c46105a5b2b1b2db17 Mon Sep 17 00:00:00 2001 From: Ari Angelo Date: Thu, 30 Jul 2026 15:37:59 +0200 Subject: [PATCH 1/6] docs: rewrite six fabricated module guides from source bucket, dataset, notebook, system, wsi and qupath CLAUDE.md described APIs, CLI commands, classes and behaviour that do not exist in the code (multi-cloud storage, resumable downloads, handler ABCs, fake service methods, wrong QuPath version, etc.). Rewritten from the actual _service.py/_cli.py/__init__.py of each module, following a lean template: purpose, real public API, real CLI commands, non-obvious behaviour, dependencies. Point to source and pyproject rather than restating. --- src/aignostics/bucket/CLAUDE.md | 502 ++------------------- src/aignostics/dataset/CLAUDE.md | 617 +++----------------------- src/aignostics/notebook/CLAUDE.md | 376 ++-------------- src/aignostics/qupath/CLAUDE.md | 616 ++++---------------------- src/aignostics/system/CLAUDE.md | 602 ++++--------------------- src/aignostics/wsi/CLAUDE.md | 713 +++++------------------------- 6 files changed, 430 insertions(+), 2996 deletions(-) diff --git a/src/aignostics/bucket/CLAUDE.md b/src/aignostics/bucket/CLAUDE.md index af3ff2c17..ffbc0ec98 100644 --- a/src/aignostics/bucket/CLAUDE.md +++ b/src/aignostics/bucket/CLAUDE.md @@ -1,460 +1,42 @@ -# CLAUDE.md - Bucket Module - -This file provides comprehensive guidance to Claude Code and human engineers when working with the `bucket` module in this repository. - -## Module Overview - -The bucket module provides enterprise-grade cloud storage operations for the Aignostics Platform, abstracting AWS S3 and Google Cloud Storage with unified interfaces for secure file management in medical imaging workflows. - -### Core Responsibilities - -- **Cloud Storage Abstraction**: Unified interface for S3, GCS, and Azure Blob Storage -- **Secure File Transfer**: Signed URL generation with expiry and access control -- **Large File Handling**: Multipart uploads, chunked downloads, resume capability -- **Data Integrity**: CRC32C/MD5 checksums, automatic retry on corruption -- **Compliance**: HIPAA-compliant storage patterns, audit logging, encryption - -### User Interfaces - -**CLI Commands (`_cli.py`):** - -- `bucket upload` - Upload files or directories to cloud storage -- `bucket download` - Download files from cloud storage -- `bucket list` - List bucket contents -- `bucket delete` - Delete files from bucket -- `bucket sign` - Generate signed URLs - -**GUI Component (`_gui.py`):** - -- Storage browser interface -- Upload/download manager with progress -- Signed URL generator - -**Service Layer (`_service.py`):** - -Core storage operations: - -- S3/GCS client management -- Signed URL generation with security constraints -- Chunked upload/download (1MB upload, 10MB download chunks) -- ETag calculation and verification -- Progress tracking with callbacks - -## Architecture & Design Patterns - -### Service Layer Design - -``` -┌────────────────────────────────────────────┐ -│ Bucket Service │ -│ (High-Level Operations) │ -├────────────────────────────────────────────┤ -│ Storage Abstraction │ -│ ┌─────────┬─────────┬─────────┐ │ -│ │ S3 │ GCS │ Azure │ │ -│ └─────────┴─────────┴─────────┘ │ -├────────────────────────────────────────────┤ -│ Transfer Management Layer │ -│ (Multipart, Chunking, Resumption) │ -├────────────────────────────────────────────┤ -│ Security & Compliance │ -│ (Encryption, Signing, Audit) │ -└────────────────────────────────────────────┘ -``` - -### Storage Patterns - -**Hierarchical Organization:** - -``` -bucket/ -├── organizations/{org_id}/ -│ ├── applications/{app_id}/ -│ │ ├── runs/{run_id}/ -│ │ │ ├── inputs/ -│ │ │ ├── outputs/ -│ │ │ └── metadata.json -│ │ └── versions/ -│ └── datasets/ -└── temp/ # Temporary uploads with TTL -``` - -## Critical Implementation Details - -### Signed URL Generation (`_service.py`) - -**Security-First URL Generation:** - -```python -def generate_signed_url( - bucket: str, - key: str, - operation: str = "GET", - expiry_seconds: int = 3600, - content_type: str = None, - metadata: dict = None, -) -> str: - """Generate time-limited signed URL with security constraints.""" - - # Validate permissions - if not has_permission(bucket, key, operation): - raise PermissionError(f"No {operation} access to {key}") - - # Add security headers - params = { - "Bucket": bucket, - "Key": key, - "ResponseContentDisposition": f"attachment; filename={Path(key).name}", - "ResponseContentType": content_type or "application/octet-stream", - } - - # Add server-side encryption - if operation == "PUT": - params["ServerSideEncryption"] = "AES256" - params["ServerSideEncryptionCustomerAlgorithm"] = "AES256" - - # Generate presigned URL - url = s3_client.generate_presigned_url( - ClientMethod=operation.lower() + "_object", Params=params, ExpiresIn=expiry_seconds - ) - - # Audit log - audit_logger.debug( - f"Generated signed URL", - extra={"operation": operation, "bucket": bucket, "key": key, "expiry": expiry_seconds, "user": current_user.id}, - ) - - return url -``` - -### Upload and Download Management - -**Chunk Size Constants (Actual):** - -```python -UPLOAD_CHUNK_SIZE = 1024 * 1024 # 1MB upload chunks -DOWNLOAD_CHUNK_SIZE = 1024 * 1024 * 10 # 10MB download chunks -ETAG_CHUNK_SIZE = 1024 * 1024 * 100 # 100MB for ETag calculation - - -def upload_file(file_path: Path, bucket: str, key: str, progress_callback: Callable = None) -> str: - """Upload file with chunking (actual implementation pattern).""" - - # Generate signed URL for upload - url = self._get_s3_client().generate_presigned_url( - ClientMethod="put_object", Params={"Bucket": bucket, "Key": key}, ExpiresIn=3600 - ) - - # Upload with chunking - with file_path.open("rb") as f: - uploaded = 0 - while True: - chunk = f.read(UPLOAD_CHUNK_SIZE) # 1MB chunks - if not chunk: - break - - # Upload chunk logic here - uploaded += len(chunk) - if progress_callback: - progress_callback(uploaded, file_path.stat().st_size) - - return url -``` - -### Download with Stream Processing - -**Memory-Efficient Download:** - -```python -def download_file(url: str, output_path: Path, progress_callback: Callable = None) -> None: - """Download file with streaming (actual implementation pattern).""" - - response = requests.get(url, stream=True) - total_size = int(response.headers.get("Content-Length", 0)) - downloaded = 0 - - with output_path.open("wb") as f: - for chunk in response.iter_content(chunk_size=DOWNLOAD_CHUNK_SIZE): # 10MB chunks - if chunk: - f.write(chunk) - downloaded += len(chunk) - - if progress_callback: - progress = DownloadProgress(current_file_downloaded=downloaded, current_file_size=total_size) - progress_callback(progress) -``` - -## Cross-Module Integration - -### Platform Module Integration - -The bucket module integrates tightly with the [platform module](../platform/CLAUDE.md): - -- Uses platform authentication for cloud credentials -- Leverages platform's correlation IDs for tracing -- Shares error handling patterns - -### Application Module Usage - -The [application module](../application/CLAUDE.md) uses bucket for: - -- Uploading WSI files for processing -- Downloading analysis results -- Managing temporary storage during runs - -### WSI Module Coordination - -Works with [WSI module](../wsi/CLAUDE.md) for: - -- Validating image formats before upload -- Streaming large WSI files -- Thumbnail generation and caching - -## Usage Patterns & Best Practices - -### Basic Operations - -```python -from aignostics.bucket import Service - -service = Service() - - -# Upload file with progress -def progress(current, total): - print(f"Upload: {current / total:.1%}") - - -url = service.upload( - file_path=Path("slide.svs"), - bucket="aignostics-data", - key=f"runs/{run_id}/inputs/slide.svs", - progress_callback=progress, -) - -# Generate signed download URL -download_url = service.generate_signed_url( - bucket="aignostics-data", - key=f"runs/{run_id}/outputs/results.json", - operation="GET", - expiry_seconds=7200, # 2 hours -) -``` - -### Advanced Patterns - -**Batch Operations with Parallelization:** - -```python -from concurrent.futures import ThreadPoolExecutor - - -def upload_batch(files: list[Path]) -> list[str]: - """Upload multiple files in parallel.""" - - with ThreadPoolExecutor(max_workers=5) as executor: - futures = [] - for file_path in files: - future = executor.submit( - service.upload, file_path=file_path, bucket="aignostics-data", key=f"batch/{file_path.name}" - ) - futures.append(future) - - # Collect results - urls = [] - for future in futures: - try: - url = future.result(timeout=300) - urls.append(url) - except Exception as e: - logger.error(f"Upload failed: {e}") - urls.append(None) - - return urls -``` - -**Resumable Download:** - -```python -def download_with_resume(bucket: str, key: str, output_path: Path) -> None: - """Download with automatic resume on failure.""" - - # Check for partial download - if output_path.exists(): - resume_offset = output_path.stat().st_size - logger.debug(f"Resuming download from byte {resume_offset}") - else: - resume_offset = 0 - - # Download with range header - response = s3_client.get_object( - Bucket=bucket, Key=key, Range=f"bytes={resume_offset}-" if resume_offset > 0 else None - ) - - # Append to existing file or create new - mode = "ab" if resume_offset > 0 else "wb" - with output_path.open(mode) as f: - for chunk in response["Body"].iter_chunks(): - f.write(chunk) -``` - -## Testing Strategies - -### Unit Testing - -```python -@pytest.fixture -def mock_s3_client(): - """Mock boto3 S3 client.""" - with patch("boto3.client") as mock: - client = MagicMock() - client.generate_presigned_url.return_value = "https://signed.url" - mock.return_value = client - yield client - - -def test_signed_url_generation(mock_s3_client): - """Test secure URL generation.""" - service = Service() - url = service.generate_signed_url(bucket="test", key="file.txt", expiry_seconds=3600) - - assert url.startswith("https://") - mock_s3_client.generate_presigned_url.assert_called_once() -``` - -### Integration Testing - -```python -@pytest.mark.docker -def test_multipart_upload_recovery(): - """Test multipart upload with simulated failure.""" - # Use localstack for S3 simulation - # Interrupt upload midway - # Verify resume capability -``` - -## Operational Requirements - -### Monitoring & Observability - -**Key Metrics:** - -- Upload/download throughput (MB/s) -- Signed URL generation rate -- Multipart upload success rate -- Storage costs by organization -- Failed transfer recovery rate - -**Logging Standards:** - -```python -logger.debug( - "File uploaded", - extra={ - "bucket": bucket, - "key": key, - "size_mb": file_size / (1024 * 1024), - "duration_seconds": duration, - "transfer_rate_mbps": transfer_rate, - }, -) -``` - -### Security & Compliance - -**Data Protection:** - -- Server-side encryption (SSE-S3 or SSE-KMS) -- Client-side encryption for sensitive data -- Access logging for audit trails -- Versioning for data recovery -- Object lifecycle policies for retention - -**HIPAA Compliance:** - -```python -# Ensure PHI data is encrypted -if contains_phi(file_path): - encryption_config = { - "Rules": [{"ApplyServerSideEncryptionByDefault": {"SSEAlgorithm": "aws:kms", "KMSMasterKeyID": KMS_KEY_ID}}] - } -``` - -## Performance Optimization - -### Transfer Optimization - -- Adaptive chunk sizing based on bandwidth -- Parallel part uploads for multipart -- Connection pooling and reuse -- Regional endpoint selection -- Transfer acceleration for cross-region - -### Caching Strategies - -```python -# Local cache for frequently accessed files -CACHE_DIR = Path.home() / ".aignostics" / "cache" -MAX_CACHE_SIZE = 10 * 1024 * 1024 * 1024 # 10GB - - -def get_with_cache(bucket: str, key: str) -> Path: - """Get file with local caching.""" - cache_path = CACHE_DIR / bucket / key - - if cache_path.exists(): - # Verify cache validity - if is_cache_valid(cache_path, bucket, key): - return cache_path - - # Download to cache - download_to_cache(bucket, key, cache_path) - manage_cache_size() - - return cache_path -``` - -## Common Pitfalls & Solutions - -### Large File Timeouts - -**Problem:** Uploads timing out for multi-GB files - -**Solution:** Use multipart upload with smaller chunks and retry logic - -### Signed URL Expiry - -**Problem:** URLs expiring during long operations - -**Solution:** Generate URLs just-in-time or implement refresh mechanism - -### Cross-Region Latency - -**Problem:** Slow transfers across regions - -**Solution:** Use S3 Transfer Acceleration or regional replication - -## Module Dependencies - -### Internal Dependencies - -- `platform` - Authentication and API client -- `utils` - Logging and utilities - -### External Dependencies - -- `boto3` - AWS SDK for S3 operations -- `google-cloud-storage` - GCS client -- `azure-storage-blob` - Azure Blob Storage - -## Future Enhancements - -1. **Intelligent Tiering**: Automatic storage class optimization -2. **Deduplication**: Content-addressable storage -3. **Compression**: Transparent compression for efficiency -4. **Encryption**: Client-side encryption key management -5. **CDN Integration**: CloudFront/Fastly for global distribution - ---- - -*This module provides enterprise-grade cloud storage operations for medical imaging workflows.* +# bucket — cloud storage on the Aignostics Platform + +Upload/find/download/delete objects in the org's Google Cloud Storage bucket. +Used by the `application` module (input upload, result download) and via the +`bucket` CLI. See `../CLAUDE.md` for where this sits in the SDK. + +Storage is GCS only, reached through a **boto3 S3-compatible client** pointed at +`storage.googleapis.com` (`ENDPOINT_URL_DEFAULT`) with HMAC keys. There is no +Azure and no `google-cloud-storage` dependency; `BucketProtocol` (`_settings.py`) +only defines `GS`/`S3`. + +## Public API — `_service.py` +`Service()` takes no args. Credentials, bucket name, and region come from the +platform org settings (via `PlatformService.get_user_info().organization`), not +constructor args. + +- `get_bucket_name()` — the org's bucket; raises `ValueError` if unset. All ops target it. +- `create_signed_upload_url(object_key, bucket_name=None)` / `create_signed_download_url(object_key, bucket_name=None)` — presigned URLs (`put_object`/`get_object`). +- `upload(source_path, destination_prefix, callback=None)` — file or directory (recursive glob). Returns `{"success": [...], "failed": [...]}` of object keys. +- `find(what, what_is_key=False, detail=False, include_signed_urls=False)` — `what` is a list of regex patterns (default `[".*"]`) or exact keys when `what_is_key=True`. +- `download(what, destination, what_is_key=False, progress_callback=None)` → `DownloadResult`. +- `delete(what, what_is_key=False, dry_run=True)` → count deleted. `dry_run=True` is the default. +- `*_static` variants (`find_static`, `download_static`, `delete_static`) just wrap `Service().()` for one-shot calls. + +## CLI — `_cli.py` +`upload`, `find`, `download`, `delete`, `purge`. (No `list`/`sign`.) `purge` +is `delete(what=None)` over the whole bucket. `delete`/`purge` default to +`--dry-run`; pass `--no-dry-run` to actually delete. + +## Behaviour worth knowing +- `find` infers a common S3 `Prefix` from the patterns/keys (`_compute_s3_prefix`) to cut paginator pages on large buckets; invalid regex raises `ValueError` → CLI exits 2. +- Download skip is ETag-based only: if a local file's MD5 (computed in 100MB chunks, `ETAG_CHUNK_SIZE`) equals the object ETag, it is not re-downloaded. No Range/resume, no multipart. +- Upload streams via a presigned PUT in 1MB chunks (`UPLOAD_CHUNK_SIZE`); download streams in 10MB chunks. `callback(bytes, path)` on upload; `progress_callback(DownloadProgress)` on download. +- `upload` CLI `--destination-prefix` templates `{username}` and `{timestamp}`. +- No permission checks, server-side encryption, audit logging, or local cache — do not add docs for those. + +## Dependencies & gotchas +`boto3`/`botocore` (imported lazily inside methods). See `pyproject.toml`. +Requires `aignostics_bucket_hmac_access_key_id` / `_secret_access_key` / +`aignostics_bucket_name` on the org, else `ValueError`. `_settings.py` holds +`region_name` and signed-URL expirations. `PageBuilder` (`_gui.py`) is exported +only when `nicegui` is installed (see `__init__.py`). diff --git a/src/aignostics/dataset/CLAUDE.md b/src/aignostics/dataset/CLAUDE.md index 44f6d4ff5..feaac2a43 100644 --- a/src/aignostics/dataset/CLAUDE.md +++ b/src/aignostics/dataset/CLAUDE.md @@ -1,545 +1,72 @@ -# CLAUDE.md - Dataset Module - -This file provides comprehensive guidance to Claude Code and human engineers when working with the `dataset` module in this repository. - -## Module Overview - -The dataset module handles enterprise-scale medical imaging dataset operations, providing high-performance downloads from the National Cancer Institute's Imaging Data Commons (IDC) and Aignostics Platform with corporate environment support. - -### Core Responsibilities - -- **IDC Integration**: Access to 40+ TB of public cancer imaging data -- **High-Performance Downloads**: s5cmd-based parallel transfers -- **Corporate Support**: Proxy handling, custom certificates, firewall traversal -- **Process Management**: Subprocess lifecycle with graceful cleanup -- **Progress Tracking**: Real-time download monitoring with callbacks - -### User Interfaces - -**CLI Commands (`_cli.py`):** - -IDC commands: - -- `dataset idc browse` - Open browser to explore IDC portal -- `dataset idc indices` - List available indices -- `dataset idc columns` - Show columns for a given index -- `dataset idc collections` - List available collections -- `dataset idc download` - Download dataset from IDC - -Aignostics commands: - -- `dataset aignostics download` - Download proprietary sample datasets - -**GUI Component (`_gui.py`):** - -- Dataset browser interface -- Download manager with progress tracking -- Collection explorer - -**Service Layer (`_service.py`):** - -Core dataset operations: - -- IDC client with proxy support -- s5cmd subprocess management -- Download progress tracking -- Path length validation for Windows -- Process cleanup on exit - -## Architecture & Design Patterns - -### Service Architecture - -``` -┌────────────────────────────────────────────┐ -│ Dataset Service │ -│ (High-Level Operations) │ -├────────────────────────────────────────────┤ -│ IDC Client Layer │ -│ (Modified for Corporate Proxies) │ -├────────────────────────────────────────────┤ -│ Download Engine (s5cmd) │ -│ (Subprocess Management, Progress) │ -├────────────────────────────────────────────┤ -│ Path & Metadata Management │ -│ (DICOM Organization, Layout) │ -└────────────────────────────────────────────┘ -``` - -### Process Management Pattern - -```python -# Global process registry for cleanup -_active_processes: list[subprocess.Popen] = [] - -# Automatic cleanup on exit -atexit.register(_cleanup_processes) -``` - -## Critical Implementation Details - -### IDC Client Enhancement (`third_party/idc_index.py`) - -**Corporate Environment Modifications:** - -```python -class IDCClient: - """Enhanced IDC client with corporate proxy support.""" - - def __init__(self): - # Original IDC initialization - super().__init__() - - # Add proxy support - self.session = requests.Session() - self.session.proxies = {"http": os.environ.get("HTTP_PROXY"), "https": os.environ.get("HTTPS_PROXY")} - - # Custom certificate bundle - if ca_bundle := os.environ.get("REQUESTS_CA_BUNDLE"): - self.session.verify = ca_bundle - - # Retry configuration for flaky networks - retry_strategy = Retry(total=5, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504]) - adapter = HTTPAdapter(max_retries=retry_strategy) - self.session.mount("https://", adapter) -``` - -### High-Performance Download Engine (`_service.py`) - -**s5cmd Integration with Process Management:** - -```python -TARGET_LAYOUT_DEFAULT = "%collection_id/%PatientID/%StudyInstanceUID/%Modality_%SeriesInstanceUID/" - - -def download_dataset( - collection_id: str, output_dir: Path, progress_callback: Callable = None, filters: dict = None -) -> None: - """Download IDC dataset with s5cmd for maximum performance.""" - - # Query IDC for download manifest - idc_client = IDCClient() - manifest = idc_client.get_download_manifest(collection_id=collection_id, filters=filters) - - # Create s5cmd commands file - commands_file = Path(tempfile.mktemp(suffix=".txt")) - with commands_file.open("w") as f: - for item in manifest: - source = f"s3://public-datasets/{item['aws_url']}" - target = output_dir / self._get_target_path(item) - f.write(f"cp {source} {target}\n") - - # Launch s5cmd subprocess - process = subprocess.Popen( - [ - "s5cmd", - "--endpoint-url", - "https://s3.amazonaws.com", - "--concurrent", - "100", # Parallel transfers - "--retry-count", - "3", - "run", - str(commands_file), - ], - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - text=True, - creationflags=SUBPROCESS_CREATION_FLAGS, - ) - - # Track process for cleanup - _active_processes.append(process) - - try: - # Monitor progress - self._monitor_download(process, progress_callback) - finally: - # Ensure cleanup - _active_processes.remove(process) - commands_file.unlink(missing_ok=True) -``` - -### Process Lifecycle Management - -**Graceful Cleanup (from tests):** - -```python -def _terminate_process(process: subprocess.Popen) -> None: - """Terminate subprocess with escalation strategy.""" - try: - logger.warning(f"Terminating subprocess {process.pid}") - process.terminate() - - # Give process time to cleanup - for _ in range(5): - if process.poll() is not None: - return - time.sleep(0.1) - - # Escalate to kill if needed - if process.poll() is None: - logger.warning(f"Force killing subprocess {process.pid}") - process.kill() - process.wait(timeout=5) - - except Exception as e: - logger.exception(f"Error terminating process {process.pid}: {e}") - - -def _cleanup_processes() -> None: - """Clean up all active processes on exit.""" - for process in _active_processes[:]: - if process.poll() is None: # Still running - _terminate_process(process) -``` - -### Progress Tracking Implementation - -**Multi-Threaded Progress Monitoring:** - -```python -def _monitor_download(process: subprocess.Popen, progress_callback: Callable) -> None: - """Monitor s5cmd output for progress updates.""" - - total_files = 0 - completed_files = 0 - failed_files = 0 - current_file = None - - # Parse s5cmd output - for line in process.stdout: - if "ERROR" in line: - failed_files += 1 - logger.error(f"Download error: {line}") - - elif "cp s3://" in line: - # Extract file being processed - match = re.search(r"cp s3://[^\s]+ ([^\s]+)", line) - if match: - current_file = match.group(1) - - elif "OK" in line: - completed_files += 1 - - if progress_callback: - progress = DownloadProgress( - total_files=total_files, - completed_files=completed_files, - failed_files=failed_files, - current_file=current_file, - throughput_mbps=self._calculate_throughput(), - ) - progress_callback(progress) - - # Check final status - return_code = process.wait() - if return_code != 0: - raise RuntimeError(f"s5cmd failed with code {return_code}") -``` - -## Cross-Module Integration - -### Platform Module Coordination - -The dataset module works with [platform module](../platform/CLAUDE.md): - -- Uses platform authentication for private datasets -- Shares correlation IDs for tracing downloads -- Leverages platform's error handling patterns - -### Application Module Usage - -The [application module](../application/CLAUDE.md) uses datasets for: - -- Bulk WSI file acquisition for batch processing -- Training data preparation for ML models -- Reference dataset management - -### WSI Module Integration - -Coordinates with [WSI module](../wsi/CLAUDE.md) for: - -- Validating downloaded DICOM files -- Converting between imaging formats -- Generating dataset previews - -## Usage Patterns & Best Practices - -### Basic Dataset Download - -```python -from aignostics.dataset import Service, IDCClient - -service = Service() -idc = IDCClient() - -# Browse available collections -collections = idc.get_collections() -for collection in collections: - print(f"{collection.id}: {collection.subjects} subjects") - - -# Download specific collection -def progress_callback(progress): - print(f"Downloaded: {progress.completed_files}/{progress.total_files}") - print(f"Throughput: {progress.throughput_mbps:.1f} MB/s") - - -service.download_dataset( - collection_id="TCGA-LUAD", - output_dir=Path("./datasets/lung"), - progress_callback=progress_callback, - filters={ - "Modality": "SM", # Slide Microscopy only - "BodyPartExamined": "LUNG", - }, -) -``` - -### Advanced Patterns - -**Incremental Download with Resume:** - -```python -def download_with_resume(collection_id: str, output_dir: Path): - """Download dataset with resume capability.""" - - # Check existing files - existing_files = set(output_dir.rglob("*.dcm")) - - # Get manifest - idc = IDCClient() - manifest = idc.get_download_manifest(collection_id) - - # Filter already downloaded - to_download = [item for item in manifest if Path(item["target_path"]) not in existing_files] - - logger.debug(f"Resuming download: {len(to_download)} files remaining") - - # Download remaining files - service.download_filtered(to_download, output_dir) -``` - -**Parallel Collection Downloads:** - -```python -from concurrent.futures import ProcessPoolExecutor - - -def download_multiple_collections(collections: list[str]): - """Download multiple collections in parallel.""" - - with ProcessPoolExecutor(max_workers=4) as executor: - futures = {} - - for collection_id in collections: - output_dir = Path(f"./datasets/{collection_id}") - future = executor.submit(service.download_dataset, collection_id=collection_id, output_dir=output_dir) - futures[future] = collection_id - - # Monitor completion - for future in concurrent.futures.as_completed(futures): - collection_id = futures[future] - try: - future.result() - logger.debug(f"Completed: {collection_id}") - except Exception as e: - logger.error(f"Failed {collection_id}: {e}") -``` - -## Testing Strategies - -### Process Management Testing (from `service_test.py`) - -```python -def test_cleanup_processes_terminates_running(): - """Verify subprocess cleanup on exit.""" - mock_process = MagicMock(spec=subprocess.Popen) - mock_process.poll.return_value = None # Still running - - _active_processes.append(mock_process) - _cleanup_processes() - - # Verify termination sequence - mock_process.terminate.assert_called_once() - if mock_process.poll.return_value is None: - mock_process.kill.assert_called_once() - - -def test_graceful_termination(): - """Test graceful process shutdown.""" - mock_process = MagicMock(spec=subprocess.Popen) - mock_process.poll.side_effect = [None, 0] # Terminates gracefully - - _terminate_process(mock_process) - - mock_process.terminate.assert_called_once() - mock_process.kill.assert_not_called() -``` - -### Integration Testing - -```python -@pytest.mark.docker -def test_idc_download_with_proxy(): - """Test IDC download through corporate proxy.""" - # Use squid proxy container - # Verify proxy authentication - # Check certificate handling -``` - -## Operational Requirements - -### Monitoring & Observability - -**Key Metrics:** - -- Download throughput (MB/s, GB/hour) -- Success/failure rates by collection -- s5cmd subprocess health -- Disk space utilization -- Network bandwidth usage - -**Logging Standards:** - -```python -logger.debug( - "Dataset download started", - extra={ - "collection_id": collection_id, - "estimated_size_gb": size_gb, - "output_directory": str(output_dir), - "filters": filters, - }, -) - -logger.error( - "Download failed", - extra={ - "collection_id": collection_id, - "error": str(e), - "completed_files": completed, - "failed_files": failed, - "duration_seconds": duration, - }, -) -``` - -### Performance Optimization - -**Network Optimization:** - -- Concurrent transfers (default: 100) -- Adaptive concurrency based on bandwidth -- Regional endpoint selection -- Connection pooling and reuse - -**Storage Optimization:** - -```python -# Path length validation for Windows -PATH_LENGTH_MAX = 260 - - -def validate_path_length(path: Path) -> bool: - """Ensure path compatibility with Windows.""" - if len(str(path)) > PATH_LENGTH_MAX: - # Shorten path using hash - return shorten_path(path) - return path -``` - -### Error Recovery - -**Automatic Retry with Exponential Backoff:** - -```python -def download_with_retry(manifest_item: dict, max_retries: int = 3): - """Download single file with retry logic.""" - - for attempt in range(max_retries): - try: - download_file(manifest_item) - return - except Exception as e: - wait_time = 2**attempt # Exponential backoff - logger.warning(f"Retry {attempt + 1}/{max_retries} after {wait_time}s") - time.sleep(wait_time) - - raise RuntimeError(f"Failed after {max_retries} attempts") -``` - -## Common Pitfalls & Solutions - -### Corporate Proxy Issues - -**Problem:** SSL verification failures behind proxy - -**Solution:** - -```bash -export HTTPS_PROXY=http://proxy:8080 -export REQUESTS_CA_BUNDLE=/path/to/ca-bundle.crt -export AWS_CA_BUNDLE=/path/to/ca-bundle.crt -``` - -### Large Dataset Storage - -**Problem:** Running out of disk space mid-download - -**Solution:** - -```python -def check_disk_space(required_gb: float, path: Path): - """Pre-flight disk space check.""" - free_gb = shutil.disk_usage(path).free / (1024**3) - if free_gb < required_gb * 1.1: # 10% buffer - raise IOError(f"Insufficient space: {free_gb:.1f}GB < {required_gb:.1f}GB") -``` - -### Windows Path Length - -**Problem:** Path too long errors on Windows - -**Solution:** Use layout with shorter paths or enable long path support - -## Module Dependencies - -### Internal Dependencies - -- `platform` - Authentication and API client -- `utils` - Logging and process utilities -- `wsi` - Image validation - -### External Dependencies - -- `s5cmd` - High-performance S3 client -- `idc-index-data` - IDC metadata -- `pydicom` - DICOM file handling -- `duckdb` - Local dataset indexing - -### Modified Third-Party - -- `third_party/idc_index.py` - Enhanced IDC client with proxy support - -## Future Enhancements - -1. **Streaming Downloads**: Process files during download -2. **Smart Caching**: Local dataset cache management -3. **Bandwidth Throttling**: Rate limiting for shared networks -4. **Compression**: On-the-fly compression for storage efficiency -5. **P2P Transfer**: Distributed dataset sharing within organization - -## Performance Considerations - -**Optimization Targets:** - -- Parallel downloads using s5cmd with configurable concurrency -- Chunked transfers for large files -- Automatic retry on failures -- Efficient memory usage through streaming - ---- - -*This module enables high-throughput downloads of cancer imaging datasets from public repositories.* +# dataset — download imaging datasets from IDC and Aignostics + +Downloads public cancer imaging data from the NCI Imaging Data Commons (IDC) +and proprietary sample datasets from Aignostics buckets. CLI + GUI wrap a +static-method `Service`. See `../CLAUDE.md` for where this module sits in the SDK. + +## Public API (`_service.py`) +`Service` (subclass of `utils.BaseService`) exposes static methods: +- `download_idc(source, target, target_layout=TARGET_LAYOUT_DEFAULT, dry_run=False) -> int` — + synchronous IDC download; returns count of matched identifier types. +- `download_with_queue(queue, source, target, target_layout, dry_run)` — same as + above but runs the actual download in a subprocess and pushes float progress + (0.0–1.0) onto a `multiprocessing.Queue`; used by the GUI. +- `download_aignostics(source_url, destination_directory, download_progress_callable=None) -> Path` — + streams one bucket object to a folder via a platform signed URL (`requests`, + 8192-byte chunks); callback gets `(len(chunk), total_size, filename)`. + +`__init__` also re-exports `IDCClient` (from `aignostics.third_party.idc_index`) +and, when `nicegui` is importable, `PageBuilder`. + +## CLI (`_cli.py`) +Root `dataset` with two sub-Typers: +- `dataset idc browse` — open the IDC portal in a browser. +- `dataset idc indices` — list `client.indices_overview` keys. +- `dataset idc columns [--index sm_instance_index]` — list columns of an index. +- `dataset idc query [SQL] [--indices ...]` — run a SQL query via `client.sql_query`. +- `dataset idc download SOURCE [TARGET] [--target-layout] [--dry-run]` — calls + `Service.download_idc`. +- `dataset aignostics download SOURCE_URL [DEST]` — calls `Service.download_aignostics` + with a rich progress bar. + +There is no `collections` subcommand. + +## Behaviour worth knowing +- **`source` matching**: `source` is a comma-separated list of IDs. Each ID list + is matched, in order, against `collection_id`, `PatientID`, `StudyInstanceUID`, + `SeriesInstanceUID`, then `SOPInstanceUID` (last uses `sm_instance_index`). + A `download_from_selection` call fires per matched column. Zero matches across + all columns raises `ValueError`. +- **How IDC bytes actually move**: `download_idc` calls + `IDCClient.download_from_selection(..., use_s5cmd_sync=True)` directly. + `download_with_queue` instead builds a small `python -c` script and runs it in a + `subprocess.Popen` (or `sys.executable --exec-script` under PyInstaller) so + progress can be scraped. s5cmd is invoked by idc-index internally — this module + never calls the `s5cmd` binary itself. +- **Progress scraping**: `Service._capture_progress_output(process, queue, base_progress=0.5)` + reads subprocess stderr one char at a time and matches `r"Downloading data:\s+(\d+)%"`, + scaling the percentage into `[base_progress, 0.99]` and pushing `1.0` on completion. + No throughput/ETA calculation. +- **Subprocess cleanup**: module-level `_active_processes` list + `atexit`-registered + `_cleanup_processes()`; `_terminate_process()` does terminate → 0.5s grace → kill. +- **Exit codes**: `ValueError` (bad input) → exit 2; any other exception → exit 1. +- **Target dir must already exist** — `download_idc`/`download_with_queue` raise + `ValueError` if it does not (CLI also enforces `exists=True`). `download_aignostics` + creates the destination directory. +- `TARGET_LAYOUT_DEFAULT` and `PATH_LENGTH_MAX = 260` are defined in both + `_service.py` and `_cli.py`. `PATH_LENGTH_MAX` is currently unused — there is no + path-shortening helper. + +## `IDCClient` (`../third_party/idc_index.py`) +Vendored + patched copy of the `idc-index` client. Use the +`IDCClient.client()` classmethod (cached per-class instance), not `IDCClient()`. +Patched to run behind corporate proxies and to retry transient HTTP failures via +`tenacity` (`_requests_get_with_retry`, 4 attempts, exponential jitter, retries +connection/timeout errors and HTTP 5xx). Requires the `s5cmd` binary on PATH (or +bundled with the package) — `__init__` raises `FileNotFoundError` otherwise. + +## Dependencies & gotchas +Deps (`idc-index`, `s5cmd`, `pandas`, `requests`, …) and extras: see +`pyproject.toml`. `Service.info`/`health` are stubs (return `{}` / `UP`). GUI code +in `_gui.py` is imported only when `find_spec("nicegui")` succeeds (`__init__.py`). +`download_aignostics` signs the source URL via `platform.generate_signed_url`. diff --git a/src/aignostics/notebook/CLAUDE.md b/src/aignostics/notebook/CLAUDE.md index 9af048f6f..41aa9b411 100644 --- a/src/aignostics/notebook/CLAUDE.md +++ b/src/aignostics/notebook/CLAUDE.md @@ -1,339 +1,37 @@ -# CLAUDE.md - Notebook Module - -This file provides comprehensive guidance to Claude Code and human engineers when working with the `notebook` module in this repository. - -## Module Overview - -The notebook module provides interactive Marimo notebook functionality for the Aignostics Platform, enabling data exploration and analysis in a reactive notebook environment. - -### Core Responsibilities - -- **Marimo Integration**: Embedded Marimo notebook server management -- **Process Management**: Lifecycle control of notebook server subprocess -- **Health Monitoring**: Server health checks and status reporting -- **Interactive Analysis**: Reactive notebook environment for data exploration - -### User Interfaces - -**Service Layer (`_service.py`):** - -The service manages the Marimo notebook server: - -- Server lifecycle (start/stop/restart) -- Health monitoring -- Process management with graceful shutdown - -**GUI Component (`_gui.py`):** - -- Embedded notebook interface within the main GUI -- Server status display -- Launch controls - -**No CLI**: This module is GUI-only for interactive use - -## Architecture & Design Patterns - -### Process Management Pattern - -```python -class _Runner: - """Manages Marimo server subprocess.""" - - _marimo_server: Popen[str] | None = None - _monitor_thread: Thread | None = None - _server_ready: Event = Event() - - def __init__(self): - atexit.register(self.stop) # Cleanup on exit -``` - -### Server Lifecycle - -``` -Start → Monitor Output → Extract URL → Ready - ↓ - Health Check - ↓ - Stop → Cleanup -``` - -## Critical Implementation Details - -### Marimo Server Management - -**Server Startup:** - -```python -MARIMO_SERVER_STARTUP_TIMEOUT = 60 # seconds - - -def start(self, notebook_path: Path, host: str, port: int): - """Start Marimo server subprocess.""" - - cmd = [sys.executable, "-m", "marimo", "run", str(notebook_path), "--host", host, "--port", str(port), "--headless"] - - self._marimo_server = Popen(cmd, stdout=PIPE, stderr=STDOUT, text=True, creationflags=SUBPROCESS_CREATION_FLAGS) - - # Monitor thread watches for server URL - self._monitor_thread = Thread(target=self._monitor_output, daemon=True) -``` - -**URL Extraction Pattern:** - -```python -def _monitor_output(self): - """Monitor Marimo output for server URL.""" - - url_pattern = r"http://\S+:\d+" - - for line in self._marimo_server.stdout: - self._output += line - - if match := re.search(url_pattern, line): - self._server_url = match.group() - self._server_ready.set() -``` - -### Health Monitoring - -```python -def health(self) -> Health: - """Check server and monitor thread health.""" - - components = { - "marimo_server": Health(status=Health.Code.UP if self.is_marimo_server_running() else Health.Code.DOWN), - "monitor_thread": Health(status=Health.Code.UP if self.is_monitor_thread_alive() else Health.Code.DOWN), - } - - return Health(status=Health.Code.UP, components=components) -``` - -## Usage Patterns - -### Starting a Notebook Session - -```python -from aignostics.notebook import Service -from pathlib import Path - -service = Service() - -# Start Marimo server -notebook = Path("analysis.marimo.py") -server_url = service.start_notebook(notebook_path=notebook, host="127.0.0.1", port=8080) - -# Server URL available after startup -print(f"Notebook running at: {server_url}") - -# Check health -health = service.health() -print(f"Server status: {health.status}") - -# Stop when done -service.stop_notebook() -``` - -### Integration with GUI - -The notebook module integrates with the main GUI launchpad: - -```python -# In GUI context -from aignostics.notebook._gui import create_notebook_interface - - -def setup_notebook_tab(ui): - """Add notebook tab to GUI.""" - - with ui.tab("Notebook"): - create_notebook_interface() -``` - -## Dependencies & Integration - -### Internal Dependencies - -- `utils` - Base service, logging, subprocess flags -- `constants` - Default notebook configuration - -### External Dependencies - -- `marimo` - Reactive notebook framework (required) - -### Integration Points - -- **GUI Module**: Embedded in main launchpad interface -- **Application Module**: Can launch notebooks for result analysis -- **Dataset Module**: Interactive data exploration after downloads - -## Operational Requirements - -### Process Management - -- **Startup timeout**: 60 seconds -- **Graceful shutdown**: Registered with `atexit` -- **Output monitoring**: Separate daemon thread -- **Resource cleanup**: Automatic on exit - -### Monitoring & Observability - -**Key Metrics:** - -- Server startup time -- Active notebook sessions -- Memory usage per session -- CPU utilization - -**Logging Patterns:** - -```python -logger.debug("Starting Marimo server", extra={"notebook": str(notebook_path), "host": host, "port": port}) - -logger.warning("Server startup timeout", extra={"timeout": MARIMO_SERVER_STARTUP_TIMEOUT, "output": self._output}) -``` - -## Common Pitfalls & Solutions - -### Port Conflicts - -**Problem:** Port already in use - -**Solution:** - -```python -def find_free_port(start=8080, end=9000): - """Find available port.""" - import socket - - for port in range(start, end): - with socket.socket() as s: - if s.connect_ex(("127.0.0.1", port)) != 0: - return port - raise RuntimeError("No free ports") -``` - -### Server Startup Failures - -**Problem:** Marimo server fails to start - -**Solution:** - -```python -# Check Marimo installation -if not find_spec("marimo"): - raise ImportError("Marimo not installed: pip install marimo") - -# Verify notebook file exists -if not notebook_path.exists(): - raise FileNotFoundError(f"Notebook not found: {notebook_path}") -``` - -### Zombie Processes - -**Problem:** Server process not cleaned up - -**Solution:** - -```python -def cleanup_zombie_processes(): - """Kill any lingering Marimo processes.""" - import psutil - - for proc in psutil.process_iter(["pid", "name", "cmdline"]): - if "marimo" in proc.info["cmdline"]: - proc.terminate() -``` - -## Development Guidelines - -### Adding New Notebook Templates - -1. Create template in `notebooks/` directory -2. Use `.marimo.py` extension -3. Include metadata header: - - ```python - # /// script - # requires-python = ">=3.10" - # dependencies = ["aignostics", "marimo", "pandas"] - # /// - ``` - -### Extending Server Management - -```python -class ExtendedRunner(_Runner): - """Extended runner with additional features.""" - - def restart(self): - """Restart server.""" - self.stop() - self.start(self._last_notebook, self._last_host, self._last_port) - - def get_notebooks(self) -> list[Path]: - """List available notebooks.""" - notebook_dir = get_user_data_directory() / "notebooks" - return list(notebook_dir.glob("*.marimo.py")) -``` - -## Testing Strategies - -### Unit Testing - -```python -@pytest.fixture -def mock_marimo_server(): - """Mock Marimo subprocess.""" - with patch("subprocess.Popen") as mock: - process = MagicMock() - process.stdout = iter(["Starting server...", "http://localhost:8080"]) - process.poll.return_value = None - mock.return_value = process - yield mock - - -def test_server_startup(mock_marimo_server): - """Test server starts and extracts URL.""" - service = Service() - url = service.start_notebook(Path("test.marimo.py")) - assert url == "http://localhost:8080" -``` - -### Integration Testing - -```python -@pytest.mark.integration -def test_real_marimo_server(): - """Test with actual Marimo server.""" - service = Service() - - # Create test notebook - notebook = Path("test.marimo.py") - notebook.write_text("import marimo as mo\nmo.md('Test')") - - try: - url = service.start_notebook(notebook) - assert requests.get(url).status_code == 200 - finally: - service.stop_notebook() - notebook.unlink() -``` - -## Performance Notes - -### Resource Usage - -- **Memory**: ~100-500MB per notebook session -- **CPU**: Minimal when idle, spikes during cell execution -- **Network**: Local server only (default 127.0.0.1) - -### Optimization Opportunities - -1. Connection pooling for multiple notebooks -2. Shared kernel for resource efficiency -3. Notebook preloading for faster startup -4. Output caching for repeated computations - ---- - -*This module enables interactive data analysis through reactive Marimo notebooks integrated with the Aignostics Platform.* +# notebook — embedded Marimo notebook server + +Launches a local [Marimo](https://marimo.io/) server running a single fixed +notebook, embedded in the GUI for exploring application results. GUI-only, no +CLI. See `../CLAUDE.md` for placement in the SDK. + +The whole module is gated in `__init__.py` on +`find_spec("marimo") and find_spec("nicegui")` — if either is missing it exports +nothing and is silently skipped (no error). + +## Public API — `_service.py` +- Module singleton `runner: _Runner | None` + `_get_runner()` (lazy init). `_Runner` owns the subprocess, monitor thread, and URL state. +- `Service(BaseService)` is a thin facade over the singleton: + - `start()` — no args; starts the server (if not already running) and returns its URL. + - `stop()`, `health()`, `is_marimo_server_running()`, `is_monitor_thread_alive()`. + +## GUI — `_gui.py` +`PageBuilder(BasePageBuilder)` registers two pages: +- `/notebook` — landing card with a launch link. +- `/notebook/{run_id}` — calls `Service().start()` and embeds `server_url` in an + `