208 lines
8.7 KiB
Python
208 lines
8.7 KiB
Python
"""
|
|
State management utilities for CAE Mesh Generator
|
|
"""
|
|
import threading
|
|
from datetime import datetime
|
|
from typing import Optional, Dict, Any
|
|
from backend.models.data_models import UploadedFile, ProcessingStatus, MeshResult
|
|
|
|
class StateManager:
|
|
"""
|
|
Thread-safe state manager for the CAE Mesh Generator
|
|
"""
|
|
|
|
def __init__(self):
|
|
self._lock = threading.Lock()
|
|
self._current_file: Optional[UploadedFile] = None
|
|
self._processing_status = ProcessingStatus(
|
|
status='IDLE',
|
|
message='Ready to process files'
|
|
)
|
|
self._mesh_result: Optional[MeshResult] = None
|
|
self._session_data: Dict[str, Any] = {}
|
|
|
|
# File management methods
|
|
def set_current_file(self, file: UploadedFile) -> None:
|
|
"""Set the current uploaded file"""
|
|
with self._lock:
|
|
self._current_file = file
|
|
self._processing_status.message = f'File "{file.filename}" uploaded successfully'
|
|
|
|
def get_current_file(self) -> Optional[UploadedFile]:
|
|
"""Get the current uploaded file"""
|
|
with self._lock:
|
|
return self._current_file
|
|
|
|
def clear_current_file(self) -> None:
|
|
"""Clear the current file"""
|
|
with self._lock:
|
|
self._current_file = None
|
|
self._mesh_result = None
|
|
self._processing_status.status = 'IDLE'
|
|
self._processing_status.message = 'Ready to process files'
|
|
self._processing_status.start_time = None
|
|
self._processing_status.end_time = None
|
|
self._processing_status.error_message = None
|
|
|
|
def update_file_status(self, status: str) -> None:
|
|
"""Update the status of the current file"""
|
|
with self._lock:
|
|
if self._current_file:
|
|
self._current_file.status = status
|
|
|
|
# Processing status methods
|
|
def get_processing_status(self) -> ProcessingStatus:
|
|
"""Get the current processing status"""
|
|
with self._lock:
|
|
return ProcessingStatus(
|
|
status=self._processing_status.status,
|
|
message=self._processing_status.message,
|
|
start_time=self._processing_status.start_time,
|
|
end_time=self._processing_status.end_time,
|
|
error_message=self._processing_status.error_message,
|
|
progress_percentage=getattr(self._processing_status, 'progress_percentage', 0.0),
|
|
current_operation=getattr(self._processing_status, 'current_operation', None),
|
|
last_updated=getattr(self._processing_status, 'last_updated', None),
|
|
completed_at=getattr(self._processing_status, 'completed_at', None)
|
|
)
|
|
|
|
def update_processing_status(self, status: ProcessingStatus) -> None:
|
|
"""Update the processing status with a ProcessingStatus object"""
|
|
with self._lock:
|
|
self._processing_status.status = status.status
|
|
self._processing_status.message = status.message
|
|
self._processing_status.start_time = status.start_time
|
|
self._processing_status.end_time = status.end_time
|
|
self._processing_status.error_message = status.error_message
|
|
self._processing_status.progress_percentage = status.progress_percentage
|
|
self._processing_status.current_operation = status.current_operation
|
|
self._processing_status.last_updated = status.last_updated or datetime.now()
|
|
self._processing_status.completed_at = status.completed_at
|
|
|
|
def set_processing_status(self, status: str, message: str) -> None:
|
|
"""Set the processing status"""
|
|
with self._lock:
|
|
self._processing_status.status = status
|
|
self._processing_status.message = message
|
|
|
|
if status == 'PROCESSING':
|
|
self._processing_status.start_time = datetime.now()
|
|
self._processing_status.end_time = None
|
|
self._processing_status.error_message = None
|
|
elif status in ['COMPLETED', 'ERROR']:
|
|
self._processing_status.end_time = datetime.now()
|
|
|
|
def set_processing_error(self, error_message: str) -> None:
|
|
"""Set processing error"""
|
|
with self._lock:
|
|
self._processing_status.status = 'ERROR'
|
|
self._processing_status.error_message = error_message
|
|
self._processing_status.end_time = datetime.now()
|
|
|
|
def start_processing(self, message: str = "Starting mesh generation...") -> None:
|
|
"""Start processing"""
|
|
with self._lock:
|
|
self._processing_status.status = 'PROCESSING'
|
|
self._processing_status.message = message
|
|
self._processing_status.start_time = datetime.now()
|
|
self._processing_status.end_time = None
|
|
self._processing_status.error_message = None
|
|
|
|
# Update file status
|
|
if self._current_file:
|
|
self._current_file.status = 'PROCESSING'
|
|
|
|
def complete_processing(self, message: str = "Processing completed successfully") -> None:
|
|
"""Complete processing"""
|
|
with self._lock:
|
|
self._processing_status.status = 'COMPLETED'
|
|
self._processing_status.message = message
|
|
self._processing_status.end_time = datetime.now()
|
|
self._processing_status.error_message = None
|
|
|
|
# Update file status
|
|
if self._current_file:
|
|
self._current_file.status = 'COMPLETED'
|
|
|
|
# Mesh result methods
|
|
def set_mesh_result(self, result: MeshResult) -> None:
|
|
"""Set the mesh generation result"""
|
|
with self._lock:
|
|
self._mesh_result = result
|
|
|
|
def get_mesh_result(self) -> Optional[MeshResult]:
|
|
"""Get the mesh generation result"""
|
|
with self._lock:
|
|
return self._mesh_result
|
|
|
|
def clear_mesh_result(self) -> None:
|
|
"""Clear the mesh result"""
|
|
with self._lock:
|
|
self._mesh_result = None
|
|
|
|
# Session data methods
|
|
def set_session_data(self, key: str, value: Any) -> None:
|
|
"""Set session data"""
|
|
with self._lock:
|
|
self._session_data[key] = value
|
|
|
|
def get_session_data(self, key: str, default: Any = None) -> Any:
|
|
"""Get session data"""
|
|
with self._lock:
|
|
return self._session_data.get(key, default)
|
|
|
|
def clear_session_data(self) -> None:
|
|
"""Clear all session data"""
|
|
with self._lock:
|
|
self._session_data.clear()
|
|
|
|
# Utility methods
|
|
def is_ready_for_processing(self) -> bool:
|
|
"""Check if system is ready for processing"""
|
|
with self._lock:
|
|
return (self._current_file is not None and
|
|
self._current_file.status == 'UPLOADED' and
|
|
self._processing_status.status in ['IDLE', 'COMPLETED', 'ERROR'])
|
|
|
|
def is_processing(self) -> bool:
|
|
"""Check if currently processing"""
|
|
with self._lock:
|
|
return self._processing_status.status == 'PROCESSING'
|
|
|
|
def get_processing_time(self) -> Optional[float]:
|
|
"""Get processing time in seconds"""
|
|
with self._lock:
|
|
if (self._processing_status.start_time and
|
|
self._processing_status.end_time):
|
|
delta = self._processing_status.end_time - self._processing_status.start_time
|
|
return delta.total_seconds()
|
|
return None
|
|
|
|
def get_system_state(self) -> Dict[str, Any]:
|
|
"""Get complete system state"""
|
|
with self._lock:
|
|
# Calculate values directly without calling other locked methods
|
|
is_ready = (self._current_file is not None and
|
|
self._current_file.status == 'UPLOADED' and
|
|
self._processing_status.status in ['IDLE', 'COMPLETED', 'ERROR'])
|
|
|
|
is_processing = self._processing_status.status == 'PROCESSING'
|
|
|
|
processing_time = None
|
|
if (self._processing_status.start_time and
|
|
self._processing_status.end_time):
|
|
delta = self._processing_status.end_time - self._processing_status.start_time
|
|
processing_time = delta.total_seconds()
|
|
|
|
return {
|
|
'current_file': self._current_file.to_dict() if self._current_file else None,
|
|
'processing_status': self._processing_status.to_dict(),
|
|
'mesh_result': self._mesh_result.to_dict() if self._mesh_result else None,
|
|
'is_ready_for_processing': is_ready,
|
|
'is_processing': is_processing,
|
|
'processing_time': processing_time,
|
|
'session_data': dict(self._session_data)
|
|
}
|
|
|
|
# Global state manager instance
|
|
state_manager = StateManager() |