""" Модель для работы с 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