122 lines
No EOL
4.7 KiB
Python
122 lines
No EOL
4.7 KiB
Python
"""
|
||
Модель для работы с 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 |