vipr_api.web.routers package

Subpackages

• vipr_api.web.routers.inference package
  • Submodules
  • vipr_api.web.routers.inference.export module
    • export_cli_yaml()
  • vipr_api.web.routers.inference.tasks module
    • cancel_task()
    • check_celery_backend()
    • get_task_progress()
    • health_check()
    • run_inference_async()
  • Module contents
• vipr_api.web.routers.panpe package
  • Submodules
  • vipr_api.web.routers.panpe.config module
    • get_available_panpe_models()
    • get_panpe_parameters()
  • Module contents
• vipr_api.web.routers.plugins package
  • Submodules
  • vipr_api.web.routers.plugins.config_manager module
    • batch_update_plugin_configs()
    • ensure_plugin_in_config()
    • get_runtime_config_path()
    • remove_plugin_from_config()
    • update_plugin_config()
  • vipr_api.web.routers.plugins.constants module
  • vipr_api.web.routers.plugins.exceptions module
    • PluginAlreadyExistsError
    • PluginConfigError
    • PluginError
    • PluginNotFoundError
    • PluginValidationError
  • vipr_api.web.routers.plugins.plugin_service module
    • batch_delete_runtime_plugins()
    • delete_runtime_plugin()
    • get_plugin_list_from_cement()
  • vipr_api.web.routers.plugins.upload_handlers module
    • handle_plugin_upload()
  • vipr_api.web.routers.plugins.validators module
    • validate_plugin_code()
    • validate_plugin_name()
    • validate_plugin_zip()
  • Module contents
    • batch_delete_plugins()
    • batch_toggle_plugins()
    • delete_plugin()
    • list_plugins()
    • toggle_plugin()
    • upload_plugin()
• vipr_api.web.routers.reflectorch package
  • Submodules
  • vipr_api.web.routers.reflectorch.config module
    • complete_config_with_defaults()
    • get_available_models()
    • get_parameters()
    • get_standard_config()
    • validate_config()
  • Module contents
    • complete_config_with_defaults()

Submodules

vipr_api.web.routers.config_security module

Config security validation utilities for VIPR API.

Provides validation for configuration data against registry and security patterns.

class vipr_api.web.routers.config_security.ConfigSecurityValidator

Bases: object

Validates VIPR configuration for security issues.

static validate_config_content_security(config_dict: dict[str, Any]) → list[str]

Validate config content for dangerous patterns.

Parameters:

config_dict – Configuration dictionary to validate

Returns:

List of validation error messages

static validate_handler_whitelist(config_dict: dict[str, Any]) → list[str]

Validate that all handlers are from allowed plugins.

Uses new config hierarchy: vipr.inference.{step}

Parameters:

config_dict – Configuration dictionary to validate

Returns:

List of validation error messages

static validate_parameter_bounds(config_dict: dict[str, Any]) → list[str]

Validate parameter values are within safe bounds.

Parameters:

config_dict – Configuration dictionary to validate

Returns:

List of validation warnings (not errors)

static validate_registry_compliance(config_dict: dict[str, Any]) → list[str]

Validate config against registry for dynamic hooks/filters.

Uses new config hierarchy: vipr.inference.hooks/filters

Parameters:

config_dict – VIPR configuration dictionary

Returns:

List of validation error messages

vipr_api.web.routers.config_security.validate_config_security(config_dict: dict[str, Any]) → dict[str, list[str]]

Comprehensive config security validation.

Parameters:

config_dict – Configuration dictionary to validate

Returns:

Dictionary with ‘errors’ and ‘warnings’ lists

vipr_api.web.routers.files module

Inference Router for VIPR Framework.

This router provides API endpoints for the inference functions of the VIPR Framework.

class vipr_api.web.routers.files.HDF5Dataset(*, name: str, spectraCount: int)

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

spectraCount: int

class vipr_api.web.routers.files.HDF5Metadata(*, datasets: List[HDF5Dataset], totalSpectra: int, filePath: str)

Bases: BaseModel

datasets: List[HDF5Dataset]

filePath: str

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

totalSpectra: int

class vipr_api.web.routers.files.HDF5PreviewData(*, qRange: List[float], intensityRange: List[float], dataPoints: int, hasQErrors: bool, hasIntensityErrors: bool, q: List[float], I: List[float], dQ: List[float] | None = None, dI: List[float] | None = None)

Bases: BaseModel

I: List[float]

dI: List[float] | None

dQ: List[float] | None

dataPoints: int

hasIntensityErrors: bool

hasQErrors: bool

intensityRange: List[float]

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

q: List[float]

qRange: List[float]

async vipr_api.web.routers.files.fetch_file(file_path: str)

Fetch file content from server based on data_path.

This endpoint handles both relative paths (using VIPR’s resolve_file_path) and absolute paths (like /tmp files) to provide file access for the frontend.

Parameters:

file_path – The path to the file to fetch

Returns:

The file content with appropriate headers

Return type:

FileResponse

Raises:

HTTPException – If file is not found, access denied, or other errors

async vipr_api.web.routers.files.get_hdf5_metadata(file_path: str)

Get metadata from HDF5 file including available datasets and spectrum counts.

This endpoint uses the SpectraReader from the HDF5SpectraReaderDataLoader to extract metadata from HDF5 files.

Parameters:

file_path – The path to the HDF5 file

Returns:

Metadata containing datasets, spectrum counts, and file info

Return type:

HDF5Metadata

Raises:

HTTPException – If file is not found, not HDF5, or metadata extraction fails

async vipr_api.web.routers.files.get_hdf5_preview(request: Dict[str, Any])

Get preview data for a specific dataset/spectrum combination from HDF5 file.

Parameters:

request – Dictionary containing: - filePath: Path to HDF5 file - datasetName: Name of dataset to preview - spectrumIndex: Index of spectrum (optional, for single spectrum preview) - batchProcessing: Whether to preview batch data (optional)

Returns:

Preview data with Q/intensity ranges and metadata

Return type:

HDF5PreviewData

Raises:

HTTPException – If file access fails or preview extraction fails

async vipr_api.web.routers.files.save_uploaded_file(file: UploadFile, content: bytes | None = None) → Path

Saves an uploaded file to a temporary location.

Parameters:

• file – The uploaded file

• content – Optional alternative file content (e.g., sanitized content)

Returns:

Path to the saved temporary file

async vipr_api.web.routers.files.upload_file(file: UploadFile = File(PydanticUndefined))

Validates and saves an uploaded file. Supports both text-based files (CSV, TXT, etc.) and HDF5 files with appropriate validation for each type.

Parameters:

file – The uploaded file (UTF-8 text format for CSV/TXT or binary for HDF5)

Returns:

Validation result with file path on success

Return type:

FileValidationResponse

Raises:

HTTPException – With appropriate status code and detail message on failure

vipr_api.web.routers.streaming module

Generic Streaming Router for VIPR Framework.

This router provides API endpoints for RabbitMQ streaming functionality that can be used across different domains/plugins, not just reflectorch.

async vipr_api.web.routers.streaming.get_consumer_stats(consumer_id: str) → ConsumerStatsResponse

Get statistics for a specific streaming consumer.

Parameters:

consumer_id – Unique identifier of the consumer

Returns:

ConsumerStatsResponse with detailed consumer statistics

Raises:

HTTPException – If consumer not found

async vipr_api.web.routers.streaming.get_consumer_tasks(consumer_id: str, limit: int | None = None, since: str | None = None) → ConsumerTasksResponse

Get task list for a specific streaming consumer.

Parameters:

• consumer_id – Unique identifier of the consumer

• limit – Maximum number of tasks to return (most recent first)

• since – ISO timestamp to filter tasks triggered after this time

Returns:

ConsumerTasksResponse containing task list and metadata

Raises:

HTTPException – If consumer not found

async vipr_api.web.routers.streaming.list_active_consumers() → dict[str, Any]

List all active streaming consumers.

Returns:

Dict containing list of active consumer IDs and summary stats

async vipr_api.web.routers.streaming.start_streaming_prediction(request: VIPRConfigWithStreaming) → StreamingResponse

Start streaming prediction consumer for real-time processing.

This endpoint starts a RabbitMQ consumer that listens for spectral data messages and triggers individual VIPR inference tasks for each subscan message.

Parameters:

request – StreamingStartRequest containing config and RabbitMQ settings

Returns:

StreamingResponse with consumer ID and status

Raises:

HTTPException – If consumer startup fails

async vipr_api.web.routers.streaming.stop_all_streaming_consumers() → dict[str, Any]

Stop all active streaming consumers.

Returns:

Dict containing final statistics for all stopped consumers

Raises:

HTTPException – If operation fails

async vipr_api.web.routers.streaming.stop_streaming_prediction(consumer_id: str) → dict[str, Any]

Stop a running streaming prediction consumer.

Parameters:

consumer_id – Unique identifier of the consumer to stop

Returns:

Dict containing final consumer statistics

Raises:

HTTPException – If consumer not found or stop fails

Module contents

API Router Module for VIPR Integration.

This package contains the various API routers for integrating VIPR functions into the FastAPI application.