vipr_api.web.models package

Submodules

vipr_api.web.models.progress module

Progress tracking models for VIPR API.

class vipr_api.web.models.progress.CeleryExecutionMode(value)

Bases: str, Enum

Celery execution modes.

ASYNC = 'async'
EAGER = 'eager'
class vipr_api.web.models.progress.CeleryHealthStatus(value)

Bases: str, Enum

Celery health check status.

HEALTHY = 'healthy'
UNAVAILABLE = 'unavailable'
class vipr_api.web.models.progress.HealthCheckResponse(*, celery_mode: CeleryExecutionMode, can_execute: bool, redis_required: bool, redis_available: bool | None = None, status: CeleryHealthStatus, error: str | None = None)

Bases: BaseModel

Response model for health check endpoint.

can_execute: bool
celery_mode: CeleryExecutionMode
error: str | None
model_config: ClassVar[ConfigDict] = {}

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

redis_available: bool | None
redis_required: bool
status: CeleryHealthStatus
class vipr_api.web.models.progress.TaskProgressResponse(*, task_id: str, status: CeleryState, message: str, progress: Dict[str, Any] | None = None, result: Dict[str, Any] | None = None, ui_data: Dict[str, Any] | None = None, error: str | None = None)

Bases: BaseModel

Response model for task progress endpoints.

error: str | None
message: str
model_config: ClassVar[ConfigDict] = {}

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

progress: Dict[str, Any] | None
result: Dict[str, Any] | None
status: CeleryState
task_id: str
ui_data: Dict[str, Any] | None

vipr_api.web.models.reflectometry module

Reflectorch-specific models for the VIPR framework.

This module contains models specific to reflectometry data processing using reflectorch.

class vipr_api.web.models.reflectometry.AvailableModel(*, name: str, max_num_layers: int, param_ranges: ModelParameterRanges)

Bases: BaseModel

Information about an available reflectometry model.

max_num_layers: int
model_config: ClassVar[ConfigDict] = {}

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

name: str
param_ranges: ModelParameterRanges
class vipr_api.web.models.reflectometry.BoundWidthRanges(*, thicknesses: list[float], roughnesses: list[float], slds: list[float], r_scale: list[float] | None = None, q_shift: list[float] | None = None, log10_background: list[float] | None = None)

Bases: BaseModel

Bound width ranges for subprior intervals.

log10_background: list[float] | None
model_config: ClassVar[ConfigDict] = {}

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

q_shift: list[float] | None
r_scale: list[float] | None
roughnesses: list[float]
slds: list[float]
thicknesses: list[float]
class vipr_api.web.models.reflectometry.ModelParameterRanges(*, thicknesses: list[float], roughnesses: list[float], slds: list[float])

Bases: BaseModel

Parameter ranges for a reflectometry model.

model_config: ClassVar[ConfigDict] = {}

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

roughnesses: list[float]
slds: list[float]
thicknesses: list[float]
class vipr_api.web.models.reflectometry.ParameterBounds(*, thicknesses: list[float], roughnesses: list[float], slds: list[float], r_scale: list[float] | None = None, q_shift: list[float] | None = None, log10_background: list[float] | None = None)

Bases: BaseModel

Reflectometry-specific parameter bounds for model configuration.

log10_background: list[float] | None
model_config: ClassVar[ConfigDict] = {}

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

q_shift: list[float] | None
r_scale: list[float] | None
roughnesses: list[float]
slds: list[float]
thicknesses: list[float]
class vipr_api.web.models.reflectometry.ReflectometryModelParameters(*, max_layers: int, parameter_bounds: ParameterBounds, bound_width_ranges: BoundWidthRanges, requires_q_resolution: bool)

Bases: BaseModel

Reflectometry-specific model parameters.

bound_width_ranges: BoundWidthRanges
max_layers: int
model_config: ClassVar[ConfigDict] = {}

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

parameter_bounds: ParameterBounds
requires_q_resolution: bool

vipr_api.web.models.streaming module

Pydantic models for generic streaming functionality.

This module contains data models for RabbitMQ streaming endpoints that can be used across different domains/plugins, not just reflectorch.

class vipr_api.web.models.streaming.ConsumerStatsInfo(*, total_tasks_triggered: int, is_consuming: bool, last_message_time: str | None = None)

Bases: BaseModel

Consumer statistics information for tasks response.

is_consuming: bool
last_message_time: str | None
model_config: ClassVar[ConfigDict] = {}

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

total_tasks_triggered: int
class vipr_api.web.models.streaming.ConsumerStatsResponse(*, consumer_id: str, is_consuming: bool, messages_received: int, tasks_triggered: int, processing_errors: int, start_time: str | None = None, last_message_time: str | None = None, uptime_seconds: float | None = None, configuration: dict[str, str])

Bases: BaseModel

Consumer statistics response.

configuration: dict[str, str]
consumer_id: str
is_consuming: bool
last_message_time: str | None
messages_received: int
model_config: ClassVar[ConfigDict] = {}

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

processing_errors: int
start_time: str | None
tasks_triggered: int
uptime_seconds: float | None
class vipr_api.web.models.streaming.ConsumerTasksResponse(*, consumer_id: str, total_tasks_available: int, tasks: list[StreamingTask], consumer_stats: ConsumerStatsInfo)

Bases: BaseModel

Response containing consumer tasks and metadata.

consumer_id: str
consumer_stats: ConsumerStatsInfo
model_config: ClassVar[ConfigDict] = {}

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

tasks: list[StreamingTask]
total_tasks_available: int
class vipr_api.web.models.streaming.RabbitMQConfig(*, rabbitmq_url: str = 'amqp://localhost:5672/', exchange_name: str = 'vipr.experiments', routing_pattern: str = 'reflectometry.growth.#')

Bases: BaseModel

Configuration for RabbitMQ connection.

exchange_name: str
model_config: ClassVar[ConfigDict] = {}

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

rabbitmq_url: str
routing_pattern: str
class vipr_api.web.models.streaming.StreamingConfig(*, rabbitmq_config: RabbitMQConfig)

Bases: BaseModel

Streaming configuration wrapper.

model_config: ClassVar[ConfigDict] = {}

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

rabbitmq_config: RabbitMQConfig
class vipr_api.web.models.streaming.StreamingResponse(*, consumer_id: str, status: str, message: str)

Bases: BaseModel

Response from streaming operations.

consumer_id: str
message: str
model_config: ClassVar[ConfigDict] = {}

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

status: str
class vipr_api.web.models.streaming.StreamingTask(*, task_id: str, triggered_at: str, growth_step: int, scan_id: str, csv_file_path: str, routing_key: str)

Bases: BaseModel

Individual streaming task information.

csv_file_path: str
growth_step: int
model_config: ClassVar[ConfigDict] = {}

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

routing_key: str
scan_id: str
task_id: str
triggered_at: str
class vipr_api.web.models.streaming.VIPRConfigWithStreaming(*, config_name: str | None = None, vipr: VIPRInference, streaming: StreamingConfig | None = None)

Bases: BaseModel

Complete VIPR configuration including streaming capabilities.

config_name: str | None
model_config: ClassVar[ConfigDict] = {}

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

streaming: StreamingConfig | None
vipr: VIPRInference

Module contents