feat: Implement search and filter functionality in MedTrackerApp
- Added DataFilter class for managing filtering and searching of medical data. - Introduced SearchFilterWidget for UI controls related to search and filters. - Integrated search and filter features into MedTrackerApp, allowing users to filter data by date range, medicine status, and pathology scores. - Implemented quick filters for common use cases (last week, last month, high symptoms). - Enhanced data loading and display logic to accommodate filtered data. - Added error handling for data loading issues. - Updated UIManager to reflect filter status in the application. - Improved entry validation in add_new_entry method to ensure data integrity.
This commit is contained in:
@@ -0,0 +1,325 @@
|
||||
"""Auto-save functionality for TheChart application."""
|
||||
|
||||
import threading
|
||||
from collections.abc import Callable
|
||||
from datetime import datetime
|
||||
from typing import Any
|
||||
|
||||
|
||||
class AutoSaveManager:
|
||||
"""Manages automatic saving of user data at regular intervals."""
|
||||
|
||||
def __init__(
|
||||
self, save_callback: Callable[[], None], interval_minutes: int = 5, logger=None
|
||||
) -> None:
|
||||
"""
|
||||
Initialize auto-save manager.
|
||||
|
||||
Args:
|
||||
save_callback: Function to call for saving data
|
||||
interval_minutes: Minutes between auto-saves (default: 5)
|
||||
logger: Logger instance for debugging
|
||||
"""
|
||||
self.save_callback = save_callback
|
||||
self.interval_seconds = interval_minutes * 60
|
||||
self.logger = logger
|
||||
self._auto_save_enabled = False
|
||||
self._save_thread: threading.Thread | None = None
|
||||
self._stop_event = threading.Event()
|
||||
self._last_save_time: datetime | None = None
|
||||
self._data_modified = False
|
||||
|
||||
def enable_auto_save(self) -> None:
|
||||
"""Enable automatic saving."""
|
||||
if self._auto_save_enabled:
|
||||
return
|
||||
|
||||
self._auto_save_enabled = True
|
||||
self._stop_event.clear()
|
||||
self._save_thread = threading.Thread(target=self._auto_save_loop, daemon=True)
|
||||
self._save_thread.start()
|
||||
|
||||
if self.logger:
|
||||
interval_minutes = self.interval_seconds / 60
|
||||
self.logger.info(
|
||||
f"Auto-save enabled with {interval_minutes:.1f} minute intervals"
|
||||
)
|
||||
|
||||
def disable_auto_save(self) -> None:
|
||||
"""Disable automatic saving."""
|
||||
if not self._auto_save_enabled:
|
||||
return
|
||||
|
||||
self._auto_save_enabled = False
|
||||
self._stop_event.set()
|
||||
|
||||
if self._save_thread and self._save_thread.is_alive():
|
||||
self._save_thread.join(timeout=2.0)
|
||||
|
||||
if self.logger:
|
||||
self.logger.info("Auto-save disabled")
|
||||
|
||||
def mark_data_modified(self) -> None:
|
||||
"""Mark that data has been modified and needs saving."""
|
||||
self._data_modified = True
|
||||
|
||||
def force_save(self) -> None:
|
||||
"""Force an immediate save if data has been modified."""
|
||||
if self._data_modified:
|
||||
try:
|
||||
self.save_callback()
|
||||
self._last_save_time = datetime.now()
|
||||
self._data_modified = False
|
||||
|
||||
if self.logger:
|
||||
self.logger.debug("Force save completed successfully")
|
||||
except Exception as e:
|
||||
if self.logger:
|
||||
self.logger.error(f"Force save failed: {e}")
|
||||
|
||||
def get_last_save_time(self) -> datetime | None:
|
||||
"""Get the timestamp of the last successful save."""
|
||||
return self._last_save_time
|
||||
|
||||
def is_enabled(self) -> bool:
|
||||
"""Check if auto-save is currently enabled."""
|
||||
return self._auto_save_enabled
|
||||
|
||||
def has_unsaved_changes(self) -> bool:
|
||||
"""Check if there are unsaved changes."""
|
||||
return self._data_modified
|
||||
|
||||
def _auto_save_loop(self) -> None:
|
||||
"""Main auto-save loop running in background thread."""
|
||||
while not self._stop_event.wait(self.interval_seconds):
|
||||
if self._data_modified:
|
||||
try:
|
||||
self.save_callback()
|
||||
self._last_save_time = datetime.now()
|
||||
self._data_modified = False
|
||||
|
||||
if self.logger:
|
||||
self.logger.debug("Auto-save completed successfully")
|
||||
|
||||
except Exception as e:
|
||||
if self.logger:
|
||||
self.logger.error(f"Auto-save failed: {e}")
|
||||
|
||||
def set_interval(self, minutes: int) -> None:
|
||||
"""
|
||||
Change the auto-save interval.
|
||||
|
||||
Args:
|
||||
minutes: New interval in minutes (minimum 1, maximum 60)
|
||||
"""
|
||||
if not 1 <= minutes <= 60:
|
||||
raise ValueError("Auto-save interval must be between 1 and 60 minutes")
|
||||
|
||||
old_interval = self.interval_seconds / 60
|
||||
self.interval_seconds = minutes * 60
|
||||
|
||||
if self.logger:
|
||||
self.logger.info(
|
||||
f"Auto-save interval changed from {old_interval:.1f} "
|
||||
f"to {minutes} minutes"
|
||||
)
|
||||
|
||||
# Restart auto-save with new interval if it was running
|
||||
if self._auto_save_enabled:
|
||||
self.disable_auto_save()
|
||||
self.enable_auto_save()
|
||||
|
||||
def cleanup(self) -> None:
|
||||
"""Clean up resources when shutting down."""
|
||||
self.disable_auto_save()
|
||||
|
||||
# Perform final save if there are unsaved changes
|
||||
if self._data_modified:
|
||||
if self.logger:
|
||||
self.logger.info("Performing final save on cleanup")
|
||||
self.force_save()
|
||||
|
||||
|
||||
class BackupManager:
|
||||
"""Manages automatic backup creation for data files."""
|
||||
|
||||
def __init__(
|
||||
self, data_file_path: str, backup_directory: str = "backups", logger=None
|
||||
):
|
||||
"""
|
||||
Initialize backup manager.
|
||||
|
||||
Args:
|
||||
data_file_path: Path to the main data file
|
||||
backup_directory: Directory to store backups
|
||||
logger: Logger instance for debugging
|
||||
"""
|
||||
self.data_file_path = data_file_path
|
||||
self.backup_directory = backup_directory
|
||||
self.logger = logger
|
||||
self._ensure_backup_directory()
|
||||
|
||||
def _ensure_backup_directory(self) -> None:
|
||||
"""Create backup directory if it doesn't exist."""
|
||||
import os
|
||||
|
||||
os.makedirs(self.backup_directory, exist_ok=True)
|
||||
|
||||
def create_backup(self, backup_type: str = "manual") -> str | None:
|
||||
"""
|
||||
Create a backup of the data file.
|
||||
|
||||
Args:
|
||||
backup_type: Type of backup ("manual", "auto", "daily")
|
||||
|
||||
Returns:
|
||||
Path to created backup file, or None if backup failed
|
||||
"""
|
||||
import os
|
||||
import shutil
|
||||
from datetime import datetime
|
||||
|
||||
if not os.path.exists(self.data_file_path):
|
||||
if self.logger:
|
||||
self.logger.warning("Cannot create backup: data file doesn't exist")
|
||||
return None
|
||||
|
||||
try:
|
||||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
base_name = os.path.splitext(os.path.basename(self.data_file_path))[0]
|
||||
backup_filename = f"{base_name}_backup_{backup_type}_{timestamp}.csv"
|
||||
backup_path = os.path.join(self.backup_directory, backup_filename)
|
||||
|
||||
shutil.copy2(self.data_file_path, backup_path)
|
||||
|
||||
if self.logger:
|
||||
self.logger.info(f"Backup created: {backup_path}")
|
||||
|
||||
return backup_path
|
||||
|
||||
except Exception as e:
|
||||
if self.logger:
|
||||
self.logger.error(f"Backup creation failed: {e}")
|
||||
return None
|
||||
|
||||
def cleanup_old_backups(self, keep_count: int = 10) -> None:
|
||||
"""
|
||||
Remove old backup files, keeping only the most recent ones.
|
||||
|
||||
Args:
|
||||
keep_count: Number of backup files to keep
|
||||
"""
|
||||
import glob
|
||||
import os
|
||||
|
||||
try:
|
||||
backup_pattern = os.path.join(self.backup_directory, "*_backup_*.csv")
|
||||
backup_files = glob.glob(backup_pattern)
|
||||
|
||||
if len(backup_files) <= keep_count:
|
||||
return
|
||||
|
||||
# Sort by modification time (newest first)
|
||||
backup_files.sort(key=os.path.getmtime, reverse=True)
|
||||
|
||||
# Remove old files
|
||||
files_to_remove = backup_files[keep_count:]
|
||||
for file_path in files_to_remove:
|
||||
os.remove(file_path)
|
||||
if self.logger:
|
||||
self.logger.debug(f"Removed old backup: {file_path}")
|
||||
|
||||
if self.logger:
|
||||
self.logger.info(f"Cleaned up {len(files_to_remove)} old backup files")
|
||||
|
||||
except Exception as e:
|
||||
if self.logger:
|
||||
self.logger.error(f"Backup cleanup failed: {e}")
|
||||
|
||||
def restore_from_backup(self, backup_path: str) -> bool:
|
||||
"""
|
||||
Restore data from a backup file.
|
||||
|
||||
Args:
|
||||
backup_path: Path to the backup file to restore
|
||||
|
||||
Returns:
|
||||
True if restoration was successful, False otherwise
|
||||
"""
|
||||
import os
|
||||
import shutil
|
||||
|
||||
if not os.path.exists(backup_path):
|
||||
if self.logger:
|
||||
self.logger.error(f"Backup file doesn't exist: {backup_path}")
|
||||
return False
|
||||
|
||||
try:
|
||||
# Create a backup of current data before restoring
|
||||
current_backup = self.create_backup("pre_restore")
|
||||
|
||||
# Restore from backup
|
||||
shutil.copy2(backup_path, self.data_file_path)
|
||||
|
||||
if self.logger:
|
||||
self.logger.info(f"Successfully restored from backup: {backup_path}")
|
||||
if current_backup:
|
||||
self.logger.info(f"Previous data backed up to: {current_backup}")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
if self.logger:
|
||||
self.logger.error(f"Restore from backup failed: {e}")
|
||||
return False
|
||||
|
||||
def list_backups(self) -> list[dict[str, Any]]:
|
||||
"""
|
||||
List all available backup files with their details.
|
||||
|
||||
Returns:
|
||||
List of dictionaries containing backup file information
|
||||
"""
|
||||
import glob
|
||||
import os
|
||||
from datetime import datetime
|
||||
|
||||
backup_pattern = os.path.join(self.backup_directory, "*_backup_*.csv")
|
||||
backup_files = glob.glob(backup_pattern)
|
||||
|
||||
backups = []
|
||||
for backup_path in backup_files:
|
||||
try:
|
||||
stat = os.stat(backup_path)
|
||||
backups.append(
|
||||
{
|
||||
"path": backup_path,
|
||||
"filename": os.path.basename(backup_path),
|
||||
"size": stat.st_size,
|
||||
"created": datetime.fromtimestamp(stat.st_mtime),
|
||||
"type": self._extract_backup_type(backup_path),
|
||||
}
|
||||
)
|
||||
except Exception as e:
|
||||
if self.logger:
|
||||
self.logger.warning(f"Error reading backup file {backup_path}: {e}")
|
||||
|
||||
# Sort by creation time (newest first)
|
||||
backups.sort(key=lambda x: x["created"], reverse=True)
|
||||
return backups
|
||||
|
||||
def _extract_backup_type(self, backup_path: str) -> str:
|
||||
"""Extract backup type from filename."""
|
||||
import os
|
||||
|
||||
filename = os.path.basename(backup_path)
|
||||
if "_backup_auto_" in filename:
|
||||
return "auto"
|
||||
elif "_backup_daily_" in filename:
|
||||
return "daily"
|
||||
elif "_backup_manual_" in filename:
|
||||
return "manual"
|
||||
elif "_backup_pre_restore_" in filename:
|
||||
return "pre_restore"
|
||||
else:
|
||||
return "unknown"
|
||||
Reference in New Issue
Block a user