fuck yeah!

This commit is contained in:
Vic Sergeev 2026-06-10 17:33:12 +03:00
parent ccb53d9091
commit da10f5e132
44 changed files with 3260 additions and 448 deletions

13
models/__init__.py Normal file
View file

@ -0,0 +1,13 @@
# Модели для Helioviewer приложения
from models.api_model import HelioviewerAPI
from models.image_model import ImageModel, ImageLayer
from models.timelapse_model import TimelapseModel, TimelapseConfig, TimelapseStatus
__all__ = [
'HelioviewerAPI',
'ImageModel',
'ImageLayer',
'TimelapseModel',
'TimelapseConfig',
'TimelapseStatus'
]

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

122
models/api_model.py Normal file
View file

@ -0,0 +1,122 @@
"""
Модель для работы с Helioviewer API
Single Responsibility: только загрузка данных из API
"""
import requests
from datetime import datetime
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass
from pathlib import Path
@dataclass
class SolarImage:
"""DTO для солнечного снимка"""
source_id: int
date: datetime
wavelength: str
observatory: str
instrument: str
filepath: Optional[Path] = None
metadata: Optional[Dict] = None
class HelioviewerAPI:
"""Клиент для работы с Helioviewer API"""
BASE_URL = "https://api.helioviewer.org/v1/"
# Предустановленные источники (можно расширять)
SOURCES = {
14: {"name": "AIA 335", "wavelength": "335 Å", "observatory": "SDO", "instrument": "AIA", "color": "#FFD700"},
13: {"name": "AIA 304", "wavelength": "304 Å", "observatory": "SDO", "instrument": "AIA", "color": "#FF4500"},
12: {"name": "AIA 211", "wavelength": "211 Å", "observatory": "SDO", "instrument": "AIA", "color": "#00FF00"},
11: {"name": "AIA 193", "wavelength": "193 Å", "observatory": "SDO", "instrument": "AIA", "color": "#00BFFF"},
10: {"name": "AIA 171", "wavelength": "171 Å", "observatory": "SDO", "instrument": "AIA", "color": "#87CEEB"},
9: {"name": "AIA 131", "wavelength": "131 Å", "observatory": "SDO", "instrument": "AIA", "color": "#FF1493"},
8: {"name": "AIA 94", "wavelength": "94 Å", "observatory": "SDO", "instrument": "AIA", "color": "#9400D3"},
4: {"name": "LASCO C2", "wavelength": "White Light", "observatory": "SOHO", "instrument": "LASCO",
"color": "#FFFFFF"},
5: {"name": "LASCO C3", "wavelength": "White Light", "observatory": "SOHO", "instrument": "LASCO",
"color": "#FFFFFF"},
}
@classmethod
def get_available_sources(cls) -> Dict[int, Dict]:
"""Возвращает список доступных источников"""
return cls.SOURCES
@classmethod
def download_image(cls, source_id: int, date: datetime, save_path: Path) -> Optional[Path]:
"""
Скачивает изображение с API Helioviewer
Args:
source_id: ID источника
date: Дата и время снимка
save_path: Путь для сохранения
Returns:
Path к сохраненному файлу или None при ошибке
"""
formatted_date = date.strftime("%Y-%m-%dT%H:%M:%SZ")
url = f"{cls.BASE_URL}getJP2Image/"
params = {'sourceId': source_id, 'date': formatted_date}
try:
response = requests.get(url, params=params, timeout=30)
response.raise_for_status()
# Создаем имя файла
source_info = cls.SOURCES.get(source_id, {})
filename = f"solar_{source_id}_{date.strftime('%Y%m%d_%H%M%S')}.jp2"
filepath = save_path / filename
# Сохраняем
with open(filepath, 'wb') as f:
f.write(response.content)
return filepath
except Exception as e:
print(f"Ошибка скачивания: {e}")
return None
@classmethod
def download_timelapse_images(cls, source_id: int, start_date: datetime,
end_date: datetime, save_path: Path,
progress_callback=None) -> List[Path]:
"""
Скачивает серию изображений для таймлапса
Args:
source_id: ID источника
start_date: Начальная дата
end_date: Конечная дата
save_path: Папка для сохранения
progress_callback: Функция для обновления прогресса
Returns:
Список путей к скачанным файлам
"""
downloaded_files = []
# Генерируем даты (каждый день в 12:00 UTC)
current_date = start_date.replace(hour=12, minute=0, second=0)
delta = end_date - start_date
total_days = delta.days + 1
for i in range(total_days):
if progress_callback:
progress_callback(i + 1, total_days, current_date)
filepath = cls.download_image(source_id, current_date, save_path)
if filepath:
downloaded_files.append(filepath)
# Переходим к следующему дню
from datetime import timedelta
current_date += timedelta(days=1)
return downloaded_files

111
models/image_model.py Normal file
View file

@ -0,0 +1,111 @@
"""
Модель для управления слоями изображений
"""
from PySide6.QtCore import QObject, Signal
from typing import List, Optional, Dict
from pathlib import Path
from dataclasses import dataclass
from datetime import datetime
@dataclass
class ImageLayer:
"""Модель слоя изображения"""
id: int
name: str
filepath: Path
source_id: int
date: datetime
wavelength: str
visible: bool = True
opacity: float = 1.0
image_data: Optional[any] = None
metadata: Optional[Dict] = None
class ImageModel(QObject):
"""Модель для хранения и управления слоями изображений"""
# Сигналы для оповещения View
layer_added = Signal(object) # ImageLayer
layer_removed = Signal(int) # layer_id
layer_visibility_changed = Signal(int, bool) # layer_id, visible
layer_opacity_changed = Signal(int, float) # layer_id, opacity
layer_selected = Signal(int) # layer_id
def __init__(self):
super().__init__()
self._layers: List[ImageLayer] = []
self._next_id = 1
self._selected_layer_id: Optional[int] = None
def add_layer(self, filepath: Path, source_id: int, date: datetime,
wavelength: str, image_data: any, metadata: Dict = None) -> int:
"""Добавляет новый слой"""
layer = ImageLayer(
id=self._next_id,
name=f"{wavelength} - {date.strftime('%Y-%m-%d %H:%M')}",
filepath=filepath,
source_id=source_id,
date=date,
wavelength=wavelength,
image_data=image_data,
metadata=metadata
)
self._layers.append(layer)
self._next_id += 1
self.layer_added.emit(layer)
return layer.id
def remove_layer(self, layer_id: int):
"""Удаляет слой"""
self._layers = [l for l in self._layers if l.id != layer_id]
self.layer_removed.emit(layer_id)
if self._selected_layer_id == layer_id:
self._selected_layer_id = None
def set_layer_visibility(self, layer_id: int, visible: bool):
"""Изменяет видимость слоя"""
for layer in self._layers:
if layer.id == layer_id:
layer.visible = visible
self.layer_visibility_changed.emit(layer_id, visible)
break
def set_layer_opacity(self, layer_id: int, opacity: float):
"""Изменяет прозрачность слоя"""
for layer in self._layers:
if layer.id == layer_id:
layer.opacity = opacity
self.layer_opacity_changed.emit(layer_id, opacity) # ← ЭМИИМ СИГНАЛ
break
def select_layer(self, layer_id: int):
"""Выбирает слой"""
self._selected_layer_id = layer_id
self.layer_selected.emit(layer_id)
def get_selected_layer(self) -> Optional[ImageLayer]:
"""Возвращает выбранный слой"""
if self._selected_layer_id:
for layer in self._layers:
if layer.id == self._selected_layer_id:
return layer
return None
def get_visible_layers(self) -> List[ImageLayer]:
"""Возвращает все видимые слои"""
return [layer for layer in self._layers if layer.visible]
def get_all_layers(self) -> List[ImageLayer]:
"""Возвращает все слои"""
return self._layers.copy()
def clear(self):
"""Очищает все слои"""
layer_ids = [layer.id for layer in self._layers]
self._layers.clear()
for layer_id in layer_ids:
self.layer_removed.emit(layer_id)

227
models/timelapse_model.py Normal file
View file

@ -0,0 +1,227 @@
"""
Модель для управления процессом создания таймлапса
Single Responsibility: только данные и состояние процесса таймлапса
"""
from PySide6.QtCore import QObject, Signal, QThread
from datetime import datetime, timedelta
from pathlib import Path
from typing import List, Optional, Callable
from dataclasses import dataclass, field
from enum import Enum
class TimelapseStatus(Enum):
"""Статус процесса создания таймлапса"""
IDLE = "idle"
PREPARING = "preparing"
DOWNLOADING = "downloading"
PROCESSING = "processing"
CREATING_VIDEO = "creating_video"
COMPLETED = "completed"
FAILED = "failed"
CANCELLED = "cancelled"
@dataclass
class TimelapseConfig:
"""Конфигурация таймлапса"""
source_id: int
start_date: datetime
end_date: datetime
output_path: Path
fps: int = 10
quality: int = 90 # качество видео (0-100)
include_metadata: bool = True
output_format: str = "mp4" # mp4, gif, webm
@dataclass
class TimelapseProgress:
"""Прогресс создания таймлапса"""
status: TimelapseStatus
current: int = 0
total: int = 0
message: str = ""
current_date: Optional[datetime] = None
downloaded_files: List[Path] = field(default_factory=list)
class TimelapseModel(QObject):
"""
Модель для управления созданием таймлапса
Хранит состояние и предоставляет интерфейс для контроллера
"""
# Сигналы для оповещения о изменениях
progress_updated = Signal(TimelapseProgress)
status_changed = Signal(TimelapseStatus)
log_message = Signal(str)
def __init__(self):
super().__init__()
self._current_progress = TimelapseProgress(status=TimelapseStatus.IDLE)
self._config: Optional[TimelapseConfig] = None
self._is_cancelled = False
def configure(self, config: TimelapseConfig) -> bool:
"""
Настраивает таймлапс
Args:
config: Конфигурация таймлапса
Returns:
True если конфигурация валидна
"""
# Валидация параметров
if config.start_date >= config.end_date:
self.log_message.emit("Ошибка: Дата начала должна быть раньше даты окончания")
return False
if config.fps < 1 or config.fps > 60:
self.log_message.emit("Ошибка: FPS должен быть в диапазоне 1-60")
return False
if not config.output_path.parent.exists():
self.log_message.emit(f"Ошибка: Папка {config.output_path.parent} не существует")
return False
self._config = config
self._is_cancelled = False
# Рассчитываем общее количество кадров
delta = config.end_date - config.start_date
total_frames = delta.days + 1
self._current_progress = TimelapseProgress(
status=TimelapseStatus.PREPARING,
total=total_frames,
message=f"Подготовка к созданию таймлапса из {total_frames} кадров"
)
self.log_message.emit(f"Настроен таймлапс: {total_frames} кадров, {config.fps} FPS")
return True
def get_total_frames(self) -> int:
"""Возвращает общее количество кадров"""
if self._config:
delta = self._config.end_date - self._config.start_date
return delta.days + 1
return 0
def update_progress(self, current_frame: int, total_frames: int,
message: str = "", current_date: datetime = None):
"""
Обновляет прогресс создания таймлапса
Args:
current_frame: Текущий кадр
total_frames: Всего кадров
message: Сообщение о прогрессе
current_date: Текущая обрабатываемая дата
"""
self._current_progress.current = current_frame
self._current_progress.total = total_frames
self._current_progress.message = message or self._current_progress.message
self._current_progress.current_date = current_date or self._current_progress.current_date
# Вычисляем процент
if total_frames > 0:
percent = (current_frame / total_frames) * 100
status_text = f"{percent:.1f}%"
self.progress_updated.emit(self._current_progress)
def set_status(self, status: TimelapseStatus, message: str = ""):
"""
Устанавливает статус процесса
Args:
status: Новый статус
message: Сообщение (опционально)
"""
self._current_progress.status = status
if message:
self._current_progress.message = message
self.status_changed.emit(status)
self.progress_updated.emit(self._current_progress)
self.log_message.emit(message or f"Статус: {status.value}")
def add_downloaded_file(self, filepath: Path):
"""Добавляет скачанный файл в список"""
self._current_progress.downloaded_files.append(filepath)
def get_downloaded_files(self) -> List[Path]:
"""Возвращает список скачанных файлов"""
return self._current_progress.downloaded_files.copy()
def cancel(self):
"""Отменяет создание таймлапса"""
self._is_cancelled = True
self.set_status(TimelapseStatus.CANCELLED, "Создание таймлапса отменено пользователем")
def is_cancelled(self) -> bool:
"""Проверяет, отменен ли процесс"""
return self._is_cancelled
def get_config(self) -> Optional[TimelapseConfig]:
"""Возвращает конфигурацию таймлапса"""
return self._config
def reset(self):
"""Сбрасывает состояние модели"""
self._config = None
self._is_cancelled = False
self._current_progress = TimelapseProgress(status=TimelapseStatus.IDLE)
self.progress_updated.emit(self._current_progress)
def generate_date_sequence(self) -> List[datetime]:
"""
Генерирует последовательность дат для таймлапса
Returns:
Список дат для каждого кадра
"""
if not self._config:
return []
dates = []
current = self._config.start_date.replace(hour=12, minute=0, second=0)
while current <= self._config.end_date:
dates.append(current)
current += timedelta(days=1)
return dates
def get_estimated_size(self) -> int:
"""
Оценивает примерный размер видео в байтах
Returns:
Оценочный размер или 0 если невозможно оценить
"""
if not self._config:
return 0
# Приблизительная оценка: ~500KB на кадр для HD видео
total_frames = self.get_total_frames()
estimated_bytes = total_frames * 500 * 1024
# Корректируем в зависимости от FPS и качества
estimated_bytes = estimated_bytes * (self._config.fps / 30) * (self._config.quality / 100)
return int(estimated_bytes)
def get_formatted_estimated_size(self) -> str:
"""Возвращает отформатированную оценку размера"""
size_bytes = self.get_estimated_size()
if size_bytes < 1024:
return f"{size_bytes} B"
elif size_bytes < 1024 * 1024:
return f"{size_bytes / 1024:.1f} KB"
elif size_bytes < 1024 * 1024 * 1024:
return f"{size_bytes / (1024 * 1024):.1f} MB"
else:
return f"{size_bytes / (1024 * 1024 * 1024):.2f} GB"