diff --git a/spitec/callbacks/callbacks.py b/spitec/callbacks/callbacks.py
index 09ecff5..9a8ff0f 100644
--- a/spitec/callbacks/callbacks.py
+++ b/spitec/callbacks/callbacks.py
@@ -20,11 +20,11 @@
def set_data_folder():
platform = sys.platform
- folder = Path("data")
+ folder = Path(__file__).parent.parent / "data"
if platform == "linux":
- folder = Path("/var/spitec/data")
+ folder = Path(__file__).parent.parent / "data"
elif platform == "win32":
- folder = Path("data")
+ folder = Path(__file__).parent.parent / "data"
folder.mkdir(parents=True, exist_ok=True)
return folder
diff --git a/spitec/callbacks/figure.py b/spitec/callbacks/figure.py
index 8bdb2dd..92a94f7 100644
--- a/spitec/callbacks/figure.py
+++ b/spitec/callbacks/figure.py
@@ -1,615 +1,688 @@
-from spitec.view.visualization import *
-from spitec.processing.data_processing import *
-from spitec.processing.data_products import DataProducts
-from spitec.processing.trajectorie import Trajectorie
-from spitec.processing.site_processing import *
from datetime import datetime, timezone
-import numpy as np
-import plotly.express as px
from pathlib import Path
+from typing import Optional
+
+import numpy as np
import pandas as pd
+import plotly.express as px
+import plotly.graph_objects as go
+
+from spitec.view.visualization import (
+ create_site_map_with_points,
+ create_fig_for_map,
+ create_site_map_with_tag,
+ create_site_map_with_trajectories,
+ create_site_data,
+ PointColor,
+ ProjectionType,
+)
+from spitec.processing.data_processing import (
+ get_namelatlon_arrays,
+ get_el_az,
+ retrieve_data,
+)
+from spitec.processing.data_products import DataProducts
+from spitec.processing.trajectorie import Trajectorie
+from spitec.processing.site_processing import (
+ Site,
+ Coordinate,
+ Sat,
+)
+
def create_map_with_points(
- site_coords: dict[Site, dict[Coordinate, float]],
+ site_coords: Optional[dict[Site, dict[Coordinate, float]]],
projection_value: ProjectionType,
show_names_site: bool,
- region_site_names: dict[str, int],
- site_data_store: dict[str, int],
- relayout_data: dict[str, float],
+ region_site_names: Optional[dict[str, int]],
+ site_data_store: Optional[dict[str, int]],
+ relayout_data: Optional[dict[str, float]],
scale_map_store: float,
- new_points: dict[str, dict[str, str | float]],
+ new_points: Optional[dict[str, dict[str, str | float]]],
) -> go.Figure:
+ """Создает или обновляет карту с точками станций."""
site_map_points = create_site_map_with_points()
site_map = create_fig_for_map(site_map_points)
- # Смена проекции
if projection_value != site_map.layout.geo.projection.type:
site_map.update_layout(geo=dict(projection_type=projection_value))
if site_coords is not None:
site_array, lat_array, lon_array = get_namelatlon_arrays(site_coords)
- # Показать\скрыть имена станций
- configure_show_site_names(show_names_site, site_data_store, site_map, site_array)
+ configure_show_site_names(
+ show_names_site, site_data_store, site_map, site_array
+ )
- colors = np.array([PointColor.SILVER.value] * site_array.shape[0])
+ colors = np.array(
+ [PointColor.SILVER.value] * site_array.shape[0]
+ )
- # Добавление данных
site_map.data[0].lat = lat_array
site_map.data[0].lon = lon_array
site_map.data[0].marker.color = colors
- # Смена цвета точек
- _change_points_on_map(region_site_names, site_data_store, site_map)
+ _change_points_color_on_map(
+ region_site_names, site_data_store, site_map
+ )
- # Добавление точек пользователя
if new_points is not None:
- _add_new_points(site_map, new_points)
- # Смена маштаба
+ _add_new_points_to_map(site_map, new_points)
+
if relayout_data is not None:
- _change_scale_map(
+ _change_map_scale_and_center(
site_map, relayout_data, scale_map_store, projection_value
)
return site_map
-def _add_new_points(
- site_map: go.Figure,
- new_points: dict[str, dict[str, str | float]],
+
+def _add_new_points_to_map(
+ site_map: go.Figure,
+ new_points: dict[str, dict[str, str | float]],
) -> None:
+ """Добавляет пользовательские точки на карту."""
for name, value in new_points.items():
- # Создаем объект для отрисовки точек пользователя
- site_map_point = create_site_map_with_tag(10, value["marker"], name)
+ site_map_point = create_site_map_with_tag(
+ size=10,
+ marker_symbol=value["marker"],
+ name=name
+ )
site_map_point.lat = [value["lat"]]
site_map_point.lon = [value["lon"]]
site_map_point.marker.color = value["color"]
-
- # Добавляем на карту
site_map.add_trace(site_map_point)
+
def configure_show_site_names(
- show_names_site: bool,
- site_data_store: dict[str, int],
- site_map: go.Figure,
- site_array: NDArray
- ) -> None:
- # Показать\скрыть имена станций
+ show_names_site: bool,
+ site_data_store: Optional[dict[str, int]],
+ site_map: go.Figure,
+ site_array: np.ndarray,
+) -> None:
+ """Настраивает отображение имен станций на карте."""
+ sites_to_display_names = []
+ if site_data_store:
+ sites_to_display_names = list(site_data_store.keys())
+
if show_names_site:
site_map.data[0].text = [site.upper() for site in site_array]
else:
- if site_data_store is not None:
- sites_name_lower = list(site_data_store.keys())
- else:
- sites_name_lower = []
site_map.data[0].text = [
- site.upper() if site in sites_name_lower else ""
- for site in site_array
- ]
+ site.upper() if site in sites_to_display_names else ""
+ for site in site_array
+ ]
site_map.data[0].customdata = [
- site.upper() if site not in sites_name_lower else ""
- for site in site_array
- ]
+ site.upper() if site not in sites_to_display_names else ""
+ for site in site_array
+ ]
site_map.data[0].hoverinfo = "text"
site_map.data[0].hovertemplate = (
- "%{customdata} (%{lat}, %{lon})"
- )
+ "%{customdata} (%{lat}, %{lon})"
+ )
+
-def _change_scale_map(
+def _change_map_scale_and_center(
site_map: go.Figure,
relayout_data: dict[str, float],
scale_map_store: float,
projection_value: ProjectionType,
) -> None:
- # Меняем маштаб
- if relayout_data.get("geo.projection.scale", None) is not None:
- scale = relayout_data.get("geo.projection.scale")
- else:
- scale = scale_map_store
+ """Изменяет масштаб и центр карты."""
+ scale = relayout_data.get("geo.projection.scale", scale_map_store)
+
+ common_params = dict(
+ projection=dict(
+ rotation=dict(
+ lon=relayout_data.get("geo.projection.rotation.lon", 0)
+ ),
+ scale=scale,
+ ),
+ center=dict(
+ lon=relayout_data.get("geo.center.lon", 0),
+ lat=relayout_data.get("geo.center.lat", 0),
+ ),
+ )
+
if projection_value != ProjectionType.ORTHOGRAPHIC.value:
- site_map.update_layout(
- geo=dict(
- projection=dict(
- rotation=dict(
- lon=relayout_data.get("geo.projection.rotation.lon", 0)
- ),
- scale=scale,
- ),
- center=dict(
- lon=relayout_data.get("geo.center.lon", 0),
- lat=relayout_data.get("geo.center.lat", 0),
- ),
- )
- )
+ site_map.update_layout(geo=common_params)
else:
- site_map.update_layout(
- geo=dict(
- projection=dict(
- rotation=dict(
- lon=relayout_data.get(
- "geo.projection.rotation.lon", 0
- ),
- lat=relayout_data.get(
- "geo.projection.rotation.lat", 0
- ),
- ),
- scale=scale,
- )
+ ortho_params = dict(
+ projection=dict(
+ rotation=dict(
+ lon=relayout_data.get("geo.projection.rotation.lon", 0),
+ lat=relayout_data.get("geo.projection.rotation.lat", 0),
+ ),
+ scale=scale,
)
)
+ site_map.update_layout(geo=ortho_params)
+
-def _change_points_on_map(
- region_site_names: dict[str, int],
- site_data_store: dict[str, int],
+def _change_points_color_on_map(
+ region_site_names: Optional[dict[str, int]],
+ site_data_store: Optional[dict[str, int]],
site_map: go.Figure,
) -> None:
- # Меняем цвета точек на карте
+ """Изменяет цвета точек на карте в зависимости от их статуса."""
colors = site_map.data[0].marker.color.copy()
- if region_site_names is not None:
+ if region_site_names:
for idx in region_site_names.values():
colors[idx] = PointColor.GREEN.value
- if site_data_store is not None:
+ if site_data_store:
for idx in site_data_store.values():
colors[idx] = PointColor.RED.value
site_map.data[0].marker.color = colors
-def _get_objs_trajectories(
- local_file: Path,
- site_data_store: dict[str, int], # все выбранные точки
- site_coords: dict[Site, dict[Coordinate, float]],
- sat: Sat,
- hm: float,
- ) -> list[Trajectorie]:
- list_trajectorie: list[Trajectorie] = []
- _, lat_array, lon_array = get_namelatlon_arrays(site_coords)
-
- # Заполняем список с объектами Trajectorie
- for name, idx in site_data_store.items():
- traj = Trajectorie(name, sat, np.radians(lat_array[idx]), np.radians(lon_array[idx]))
- list_trajectorie.append(traj)
-
- # Извлекаем значения el и az по станциям
- site_names = list(site_data_store.keys())
- site_azimuth, site_elevation, is_satellite = get_el_az(local_file, site_names, sat)
-
- # Добавлем долгату и широту для точек траекторий
- for traj in list_trajectorie:
- if not is_satellite[traj.site_name]:
+def _get_trajectory_objects(
+ processed_file_path: Path,
+ selected_sites_data: dict[str, int],
+ site_coordinates: dict[Site, dict[Coordinate, float]],
+ satellite: Sat,
+ height_m: float,
+) -> list[Trajectorie]:
+ """Подготавливает список объектов Trajectorie для выбранных станций."""
+ trajectory_objects: list[Trajectorie] = []
+ _, lat_array, lon_array = get_namelatlon_arrays(site_coordinates)
+
+ for name, idx in selected_sites_data.items():
+ traj = Trajectorie(
+ name,
+ satellite,
+ np.radians(lat_array[idx]),
+ np.radians(lon_array[idx])
+ )
+ trajectory_objects.append(traj)
+
+ site_names_list = list(selected_sites_data.keys())
+ site_azimuth, site_elevation, is_satellite_data_available = get_el_az(
+ processed_file_path, site_names_list, satellite
+ )
+
+ for traj in trajectory_objects:
+ if not is_satellite_data_available.get(traj.site_name, False):
traj.sat_exist = False
continue
- traj.add_trajectory_points(
- site_azimuth[traj.site_name][traj.sat_name][DataProducts.azimuth],
- site_elevation[traj.site_name][traj.sat_name][DataProducts.elevation],
- site_azimuth[traj.site_name][traj.sat_name][DataProducts.time],
- hm
- )
- return list_trajectorie
+
+ az_data = site_azimuth.get(traj.site_name, {}).get(traj.sat_name, {})
+ el_data = site_elevation.get(traj.site_name, {}).get(traj.sat_name, {})
-def _find_time(times: NDArray, target_time: datetime, look_more = True):
- exact_match_idx = np.where(times == target_time)[0]
- exact_time = False
+ if not (az_data and el_data):
+ traj.sat_exist = False
+ continue
- if exact_match_idx.size > 0:
- # Если точное совпадение найдено
- exact_time = True
- return exact_match_idx[0], exact_time
+ traj.add_trajectory_points(
+ az_data[DataProducts.azimuth],
+ el_data[DataProducts.elevation],
+ az_data[DataProducts.time],
+ height_m,
+ )
+ return trajectory_objects
+
+
+def _find_time_index(
+ times_array: np.ndarray,
+ target_time: datetime,
+ find_later_time: bool = True,
+) -> tuple[int, bool]:
+ """Находит индекс точного или ближайшего времени в массиве."""
+ exact_match_indices = np.where(times_array == target_time)[0]
+ is_exact_match = False
+
+ if exact_match_indices.size > 0:
+ is_exact_match = True
+ return exact_match_indices[0], is_exact_match
+
+ if find_later_time:
+ candidate_times_indices = np.where(times_array > target_time)[0]
+ if candidate_times_indices.size > 0:
+ return candidate_times_indices[0], is_exact_match
else:
- # Если точного совпадения нет, ищем ближайшее большее/меньшее время
- nearest_other_idx = -1
- if look_more:
- other_times = np.where(times > target_time)[0]
- if other_times.size > 0:
- nearest_other_idx = other_times[0]
- else:
- other_times = np.where(times < target_time)[0]
- if other_times.size > 0:
- nearest_other_idx = other_times[-1]
+ candidate_times_indices = np.where(times_array < target_time)[0]
+ if candidate_times_indices.size > 0:
+ return candidate_times_indices[-1], is_exact_match
+
+ return -1, is_exact_match
- if nearest_other_idx == -1:
- return -1, exact_time
-
- return nearest_other_idx, exact_time
-
def create_map_with_trajectories(
- site_map: go.Figure,
- local_file: str,
- site_data_store: dict[str, int], # все выбранные точки
- site_coords: dict[Site, dict[Coordinate, float]],
- sat: Sat,
- data_colors: dict[Site, str],
- time_value: list[int],
- hm: float,
- sip_tag_time_dict: dict,
- all_select_sip_tag: list[dict],
- new_trajectory: dict[str, dict[str, float | str]]
+ site_map: go.Figure,
+ processed_file: Optional[str],
+ selected_sites_data: Optional[dict[str, int]],
+ site_coordinates: Optional[dict[Site, dict[Coordinate, float]]],
+ satellite: Optional[Sat],
+ trajectory_colors: dict[Site, str],
+ time_range_hours: list[int],
+ height_m: Optional[float],
+ current_sip_tag_time: Optional[dict],
+ all_selected_sip_tags: Optional[list[dict]],
+ new_manual_trajectories: Optional[dict[str, dict[str, float | str]]],
) -> go.Figure:
-
- if sat is None or local_file is None or \
- site_coords is None or site_data_store is None or \
- len(site_data_store) == 0 or hm is None or \
- site_map.layout.geo.projection.type != ProjectionType.ORTHOGRAPHIC.value:
+ """Добавляет траектории и SIP-теги на карту."""
+ if not all([
+ satellite, processed_file, site_coordinates,
+ selected_sites_data, height_m
+ ]) or not selected_sites_data:
return site_map
- local_file_path = Path(local_file)
+ if site_map.layout.geo.projection.type != ProjectionType.ORTHOGRAPHIC.value:
+ return site_map
+
+ processed_file_path = Path(processed_file)
- new_trajectory_objs, new_trajectory_colors = _get_objs_new_trajectories(
- new_trajectory
+ manual_traj_objs, manual_traj_colors = _get_manual_trajectory_objects(
+ new_manual_trajectories
)
- # Создаем список с объектом Trajectorie
- trajectory_objs: list[Trajectorie] = _get_objs_trajectories(
- local_file_path,
- site_data_store,
- site_coords,
- sat,
- hm,
+ station_traj_objs = _get_trajectory_objects(
+ processed_file_path,
+ selected_sites_data,
+ site_coordinates,
+ satellite,
+ height_m,
)
- limit_start, limit_end = _create_limit_xaxis(time_value, local_file_path)
+
+ time_limit_start, time_limit_end = _create_datetime_limits_for_xaxis(
+ time_range_hours, processed_file_path
+ )
+
+ all_trajectory_objects = manual_traj_objs + station_traj_objs
+
+ num_manual_trajectories = len(manual_traj_objs)
- new_trajectory_objs.extend(trajectory_objs)
- for i, traj in enumerate(new_trajectory_objs):
- if not traj.sat_exist: # данных по спутнику нет
+ for i, traj in enumerate(all_trajectory_objects):
+ if not traj.sat_exist:
continue
- # Определяем цвет траектории
- if traj.site_name in data_colors.keys():
- curtent_color = data_colors[traj.site_name]
- elif i < len(new_trajectory_objs) - len(trajectory_objs):
- curtent_color = new_trajectory_colors[i]
+ current_color: str
+ if traj.site_name in trajectory_colors:
+ current_color = trajectory_colors[traj.site_name]
+ elif i < num_manual_trajectories:
+ current_color = manual_traj_colors[i]
else:
- curtent_color = "black"
+ current_color = "black" # Цвет по умолчанию
- # Ищем ближайщие индексы времени
- traj.idx_start_point, _ = _find_time(traj.times, limit_start)
- traj.idx_end_point, _ = _find_time(traj.times, limit_end, False)
- if traj.traj_lat[traj.idx_start_point] is None:
+ traj.idx_start_point, _ = _find_time_index(traj.times, time_limit_start)
+ traj.idx_end_point, _ = _find_time_index(
+ traj.times, time_limit_end, find_later_time=False
+ )
+
+ if traj.idx_start_point != -1 and traj.traj_lat[traj.idx_start_point] is None:
traj.idx_start_point += 3
- if traj.traj_lat[traj.idx_end_point] is None:
+ if traj.idx_end_point != -1 and traj.traj_lat[traj.idx_end_point] is None:
traj.idx_end_point -= 3
- if traj.idx_start_point >= traj.idx_end_point or \
- traj.idx_start_point == -1 or \
- traj.idx_end_point == -1:
- # если не нашли, или нашли неверно
- continue
+ if (traj.idx_start_point == -1 or
+ traj.idx_end_point == -1 or
+ traj.idx_start_point >= traj.idx_end_point):
+ continue
- # создаем объекты для отрисовки траектории
- site_map_trajs, site_map_end_trajs = _create_trajectory(curtent_color, traj, traj.site_name)
- site_map.add_traces([site_map_trajs, site_map_end_trajs])
-
- if ( sip_tag_time_dict is not None and sip_tag_time_dict["time"] is not None and \
- (len(sip_tag_time_dict["time"]) == 8 or len(sip_tag_time_dict["time"]) == 19) ) or \
- (all_select_sip_tag is not None):
- site_map = _add_sip_tags(
- site_map,
- local_file_path,
- trajectory_objs,
- data_colors,
- all_select_sip_tag,
- sip_tag_time_dict
+ map_trajectory_trace, map_end_trace = _create_trajectory_traces(
+ current_color, traj, traj.site_name
+ )
+ site_map.add_traces([map_trajectory_trace, map_end_trace])
+
+ should_add_sip_tags = (
+ current_sip_tag_time is not None and
+ current_sip_tag_time.get("time") and
+ (len(current_sip_tag_time["time"]) in [8, 19])
+ ) or (all_selected_sip_tags is not None)
+
+ if should_add_sip_tags:
+ site_map = _add_sip_tags_to_map(
+ site_map=site_map,
+ processed_file_path=processed_file_path,
+ station_trajectory_objects=station_traj_objs,
+ trajectory_colors=trajectory_colors,
+ all_selected_sip_tags=all_selected_sip_tags,
+ current_sip_tag_time=current_sip_tag_time,
)
return site_map
-def _get_objs_new_trajectories(
- new_trajectory: dict[str, dict[str, float | str]]
- ) -> list[list[Trajectorie], list[str]]:
- new_trajectory_objs = []
- new_trajectory_colors = []
- if new_trajectory is not None:
- for name, data in new_trajectory.items():
- trajectory = Trajectorie(name, None, None, None)
- datetime_array = pd.to_datetime(data["times"])
- trajectory.times = np.array(datetime_array)
- trajectory.traj_lat = np.array(data["traj_lat"], dtype=object)
- trajectory.traj_lon = np.array(data["traj_lon"], dtype=object)
- trajectory.traj_hm = np.array(data["traj_hm"], dtype=object)
- new_trajectory_colors.append(data["color"])
- new_trajectory_objs.append(trajectory)
- return new_trajectory_objs, new_trajectory_colors
-
-def _create_trajectory(
- current_color: str,
- traj: Trajectorie,
- name: str = None
- ) -> list[go.Scattergeo]:
- site_map_trajs = create_site_map_with_trajectories() # создаем объект для отрисовки траектории
- site_map_end_trajs = create_site_map_with_tag(name=name) # создаем объект для отрисовки конца траектории
-
- # Устанавливаем точки траектории
- site_map_trajs.lat = traj.traj_lat[traj.idx_start_point:traj.idx_end_point:3]
- site_map_trajs.lon = traj.traj_lon[traj.idx_start_point:traj.idx_end_point:3]
- site_map_trajs.marker.color = current_color
-
- # Устанавливаем координаты последней точки
- site_map_end_trajs.lat = [traj.traj_lat[traj.idx_end_point]]
- site_map_end_trajs.lon = [traj.traj_lon[traj.idx_end_point]]
- site_map_end_trajs.marker.color = current_color
-
- return site_map_trajs, site_map_end_trajs
-
-def _add_sip_tags(
- site_map: go.Figure,
- local_file: Path,
- trajectory_objs: list[Trajectorie],
- data_colors: dict[Site, str],
- all_select_st: list[dict],
- sip_tag_time_dict: dict
- ):
- if all_select_st is None:
- all_select_sip_tag = []
- else:
- all_select_sip_tag = all_select_st.copy()
-
- if sip_tag_time_dict is not None:
- if len(sip_tag_time_dict["time"]) == 8:
- sip_tag_time = sip_tag_time_dict["time"]
- current_date = local_file.stem # Получаем '2024-01-01'
- sip_tag_time_dict["time"] = f"{current_date} {sip_tag_time}"
- sip_tag_time_dict["coords"] = []
- all_select_sip_tag.append(sip_tag_time_dict)
-
- for i, sip_data in enumerate(all_select_sip_tag):
- tag_lat = []
- tag_lon = []
- tag_color = []
- for traj in trajectory_objs:
- if not traj.sat_exist: # данных по спутнику нет
+
+def _get_manual_trajectory_objects(
+ new_trajectory_data: Optional[dict[str, dict[str, float | str]]],
+) -> tuple[list[Trajectorie], list[str]]:
+ """Создает объекты Trajectorie из данных ручного ввода."""
+ manual_trajectory_objects = []
+ manual_trajectory_colors = []
+ if new_trajectory_data is None:
+ return manual_trajectory_objects, manual_trajectory_colors
+
+ for name, data in new_trajectory_data.items():
+ trajectory = Trajectorie(name, None, None, None)
+ datetime_array = pd.to_datetime(data["times"])
+ trajectory.times = np.array(datetime_array)
+ trajectory.traj_lat = np.array(data["traj_lat"], dtype=object)
+ trajectory.traj_lon = np.array(data["traj_lon"], dtype=object)
+ trajectory.traj_hm = np.array(data["traj_hm"], dtype=object)
+ trajectory.sat_exist = True
+ manual_trajectory_colors.append(str(data["color"]))
+ manual_trajectory_objects.append(trajectory)
+ return manual_trajectory_objects, manual_trajectory_colors
+
+
+def _create_trajectory_traces(
+ color: str,
+ trajectory: Trajectorie,
+ name: Optional[str] = None,
+) -> tuple[go.Scattergeo, go.Scattergeo]:
+ """Создает графические объекты Plotly для траектории и ее конца."""
+ trajectory_trace = create_site_map_with_trajectories()
+ trajectory_trace.lat = trajectory.traj_lat[
+ trajectory.idx_start_point:trajectory.idx_end_point:3
+ ]
+ trajectory_trace.lon = trajectory.traj_lon[
+ trajectory.idx_start_point:trajectory.idx_end_point:3
+ ]
+ trajectory_trace.marker.color = color
+
+ end_point_trace = create_site_map_with_tag(name=name)
+ end_point_trace.lat = [trajectory.traj_lat[trajectory.idx_end_point]]
+ end_point_trace.lon = [trajectory.traj_lon[trajectory.idx_end_point]]
+ end_point_trace.marker.color = color
+
+ return trajectory_trace, end_point_trace
+
+
+def _add_sip_tags_to_map(
+ site_map: go.Figure,
+ processed_file_path: Path,
+ station_trajectory_objects: list[Trajectorie],
+ trajectory_colors: dict[Site, str],
+ all_selected_sip_tags: Optional[list[dict]],
+ current_sip_tag_time: Optional[dict],
+) -> go.Figure:
+ active_sip_tags = []
+ if all_selected_sip_tags:
+ active_sip_tags.extend(all_selected_sip_tags)
+
+ if current_sip_tag_time and current_sip_tag_time.get("time"):
+ time_str = current_sip_tag_time["time"]
+ if len(time_str) == 8:
+ current_date_str = processed_file_path.stem
+ current_sip_tag_time["time"] = f"{current_date_str} {time_str}"
+ current_sip_tag_time["coords"] = []
+ active_sip_tags.append(current_sip_tag_time)
+
+ for i, sip_data in enumerate(active_sip_tags):
+ tag_lats: list[Optional[float]] = []
+ tag_lons: list[Optional[float]] = []
+ tag_point_colors: list[str] = []
+
+ target_datetime = convert_str_to_datetime(str(sip_data["time"]))
+
+ for traj in station_trajectory_objects:
+ if not traj.sat_exist:
continue
- if isinstance(sip_data["time"], str):
- tag_time = convert_time(sip_data["time"])
- else:
- tag_time = sip_data["time"]
-
- # получаем индекс метки времени
- sip_tag_idx, exact_time = _find_time(traj.times, tag_time)
-
- idx_start_point = traj.idx_start_point
- idx_end_point = traj.idx_end_point
- if idx_start_point == -1:
- idx_start_point = sip_tag_idx + 1
-
- if idx_end_point == -1:
- idx_end_point = sip_tag_idx - 1
-
- if exact_time and traj.traj_lat[sip_tag_idx] is not None:
- if sip_tag_idx >= idx_start_point and sip_tag_idx <= idx_end_point:
- tag_lat.append(traj.traj_lat[sip_tag_idx])
- tag_lon.append(traj.traj_lon[sip_tag_idx])
- if all_select_st is not None and \
- i < len(all_select_st) and \
- all_select_st[i]["site"] == traj.site_name:
- all_select_st[i]["lat"] = np.radians(traj.traj_lat[sip_tag_idx])
- all_select_st[i]["lon"] = np.radians(traj.traj_lon[sip_tag_idx])
+ sip_tag_idx, is_exact_match = _find_time_index(
+ traj.times, target_datetime
+ )
+
+ valid_start = traj.idx_start_point != -1
+ valid_end = traj.idx_end_point != -1
+
+ if is_exact_match and sip_tag_idx != -1 and \
+ traj.traj_lat[sip_tag_idx] is not None and \
+ traj.traj_lon[sip_tag_idx] is not None and \
+ (not valid_start or sip_tag_idx >= traj.idx_start_point) and \
+ (not valid_end or sip_tag_idx <= traj.idx_end_point):
- if sip_tag_time_dict is not None:
- sip_tag_time_dict["coords"].append({
- "site": traj.site_name,
- "lat": np.radians(traj.traj_lat[sip_tag_idx]),
- "lon": np.radians(traj.traj_lon[sip_tag_idx])
- }
- )
+ lat_rad = np.radians(traj.traj_lat[sip_tag_idx])
+ lon_rad = np.radians(traj.traj_lon[sip_tag_idx])
- if sip_data["color"] is None:
- tag_color.append(data_colors[traj.site_name])
+ tag_lats.append(traj.traj_lat[sip_tag_idx])
+ tag_lons.append(traj.traj_lon[sip_tag_idx])
+
+ if all_selected_sip_tags and i < len(all_selected_sip_tags) and \
+ all_selected_sip_tags[i].get("site") == traj.site_name:
+ all_selected_sip_tags[i]["lat"] = lat_rad
+ all_selected_sip_tags[i]["lon"] = lon_rad
+
+ if current_sip_tag_time and sip_data is current_sip_tag_time:
+ current_sip_tag_time["coords"].append({
+ "site": traj.site_name,
+ "lat": lat_rad,
+ "lon": lon_rad,
+ })
+
+ if sip_data.get("color"):
+ tag_point_colors.append(str(sip_data["color"]))
+ elif traj.site_name in trajectory_colors:
+ tag_point_colors.append(trajectory_colors[traj.site_name])
else:
- tag_color = sip_data["color"]
+ tag_point_colors.append("purple")
- site_map_tags = create_site_map_with_tag(10, sip_data["marker"], sip_data["name"])
- site_map_tags.lat = tag_lat
- site_map_tags.lon = tag_lon
- site_map_tags.marker.color = tag_color
-
- site_map.add_trace(site_map_tags)
+ if tag_lats:
+ map_sip_tags_trace = create_site_map_with_tag(
+ size=10,
+ marker_symbol=sip_data["marker"],
+ name=sip_data["name"],
+ )
+ map_sip_tags_trace.lat = tag_lats
+ map_sip_tags_trace.lon = tag_lons
+ if len(set(tag_point_colors)) == 1:
+ map_sip_tags_trace.marker.color = tag_point_colors[0]
+ else:
+ map_sip_tags_trace.marker.color = tag_point_colors
+
+ site_map.add_trace(map_sip_tags_trace)
+
return site_map
-def create_site_data_with_values(
- site_data_store: dict[str, int],
- sat: Sat,
- data_types: str,
- local_file: str,
- time_value: list[int],
- shift: float,
- sip_tag_time_dict: dict,
- all_select_sip_tag: list[dict],
+
+def create_site_data_plot_with_values(
+ selected_sites_data: Optional[dict[str, int]],
+ satellite: Optional[Sat],
+ data_type_name: str,
+ processed_file: Optional[str],
+ time_range_hours: list[int],
+ y_axis_shift_value: Optional[float],
+ current_sip_tag_time: Optional[dict],
+ all_selected_sip_tags: Optional[list[dict]],
) -> go.Figure:
- site_data = create_site_data()
-
- if site_data_store is not None and site_data_store:
- local_file_path = Path(local_file)
- if sip_tag_time_dict is not None and \
- sip_tag_time_dict["time"] is not None and \
- (len(sip_tag_time_dict["time"]) == 8 or len(sip_tag_time_dict["time"]) == 19):
- sip_tag_time = sip_tag_time_dict["time"]
- if len(sip_tag_time) == 8:
- current_date = local_file_path.stem # Получаем '2024-01-01'
- sip_tag_datetime = datetime.strptime(f"{current_date} {sip_tag_time}", "%Y-%m-%d %H:%M:%S")
- elif len(sip_tag_time) == 19:
- sip_tag_datetime = datetime.strptime(sip_tag_time, "%Y-%m-%d %H:%M:%S")
- sip_tag_datetime = sip_tag_datetime.replace(tzinfo=timezone.utc)
- add_sip_tag_line(site_data, sip_tag_datetime)
-
- if all_select_sip_tag is not None and len(all_select_sip_tag) != 0:
- for tag in all_select_sip_tag:
-
- if isinstance(tag["time"], str):
- tag_time = convert_time(tag["time"])
- else:
- tag_time = tag["time"]
+ """Создает график данных со станций"""
+ site_data_plot = create_site_data()
+
+ if not selected_sites_data or not processed_file:
+ return site_data_plot
+
+ processed_file_path = Path(processed_file)
- add_sip_tag_line(site_data, tag_time, tag["color"])
+ if (current_sip_tag_time and
+ current_sip_tag_time.get("time") and
+ len(current_sip_tag_time["time"]) in [8, 19]):
- # Определяем тип данных
- dataproduct = _define_data_type(data_types)
- # Определяем размер сдвига
- if shift is None or shift == 0:
- shift = -1
- if dataproduct in [DataProducts.dtec_2_10, DataProducts.roti, DataProducts.dtec_10_20]:
- shift = -0.5
- # Добавляем данные
- _add_lines(
- site_data,
- list(site_data_store.keys()),
- sat,
- dataproduct,
- local_file_path,
- shift,
+ time_str = current_sip_tag_time["time"]
+ if len(time_str) == 8:
+ current_date_str = processed_file_path.stem
+ sip_datetime_str = f"{current_date_str} {time_str}"
+ else:
+ sip_datetime_str = time_str
+
+ sip_tag_dt = convert_str_to_datetime(sip_datetime_str)
+ add_vertical_line_to_plot(site_data_plot, sip_tag_dt)
+
+ if all_selected_sip_tags:
+ for tag_data in all_selected_sip_tags:
+ tag_dt = convert_str_to_datetime(str(tag_data["time"]))
+ add_vertical_line_to_plot(
+ site_data_plot, tag_dt, str(tag_data.get("color"))
+ )
+
+ target_data_product = _get_data_product_enum(data_type_name)
+
+ if y_axis_shift_value is None or y_axis_shift_value == 0:
+ y_axis_shift = -1.0
+ if target_data_product in [
+ DataProducts.dtec_2_10, DataProducts.roti, DataProducts.dtec_10_20
+ ]:
+ y_axis_shift = -0.5
+ else:
+ y_axis_shift = y_axis_shift_value
+
+ _add_data_lines_to_plot(
+ site_data_plot,
+ list(selected_sites_data.keys()),
+ satellite,
+ target_data_product,
+ processed_file_path,
+ y_axis_shift,
+ )
+
+ if site_data_plot.data:
+ time_limit_start, time_limit_end = _create_datetime_limits_for_xaxis(
+ time_range_hours, processed_file_path
+ )
+ site_data_plot.update_layout(
+ xaxis=dict(range=[time_limit_start, time_limit_end])
)
- if len(site_data.data) > 0:
- # Ограничиваем вывод данных по времени
- limit = _create_limit_xaxis(time_value, local_file_path)
- site_data.update_layout(xaxis=dict(range=[limit[0], limit[1]]))
- return site_data
-
-def add_sip_tag_line(
- site_data: go.Figure,
- sip_tag_datetime: datetime,
- color: str = "darkblue"
- ) -> None:
- site_data.add_shape(
+ return site_data_plot
+
+
+def add_vertical_line_to_plot(
+ plot_figure: go.Figure,
+ datetime_value: datetime,
+ line_color: str = "darkblue",
+) -> None:
+ """Добавляет вертикальную линию на график в указанное время."""
+ plot_figure.add_shape(
type="line",
- x0=sip_tag_datetime,
- x1=sip_tag_datetime,
+ x0=datetime_value,
+ x1=datetime_value,
y0=0,
- y1=1,
- yref="paper",
+ y1=1,
+ yref="paper", # Относительно высоты графика
line=dict(
- color=color,
- width=1,
- dash="dash"
- )
+ color=line_color,
+ width=1,
+ dash="dash",
+ ),
)
-def convert_time(point_x: str) -> datetime:
- x_time = datetime.strptime(
- point_x, "%Y-%m-%d %H:%M:%S"
- ).replace(tzinfo=timezone.utc)
-
- return x_time
-
-
-def _define_data_type(data_types: str) -> DataProducts:
- # Определяем тип данных
- dataproduct = DataProducts.dtec_2_10
- for name_data in DataProducts.__members__:
- if data_types == name_data:
- dataproduct = DataProducts.__members__[name_data]
- break
- return dataproduct
-
-
-def _add_lines(
- site_data: go.Figure,
- sites_name: list[Site],
- sat: Sat,
- dataproduct: DataProducts,
- local_file: Path,
- shift: float,
+
+def convert_str_to_datetime(datetime_str: str) -> datetime:
+ """Конвертирует строку времени в объект datetime с UTC."""
+ # Поддерживает форматы "ГГГГ-ММ-ДД ЧЧ:ММ:СС" и "ЧЧ:ММ:СС" (добавляя текущую дату)
+ try:
+ dt_object = datetime.strptime(datetime_str, "%Y-%m-%d %H:%M:%S")
+ except ValueError:
+ raise ValueError(f"Invalid datetime format: {datetime_str}")
+
+ return dt_object.replace(tzinfo=timezone.utc)
+
+
+def _get_data_product_enum(data_product_name: str) -> DataProducts:
+ """Возвращает член Enum DataProducts по его имени."""
+ try:
+ return DataProducts[data_product_name]
+ except KeyError:
+ # Возвращаем значение по умолчанию или вызываем ошибку,
+ return DataProducts.dtec_2_10
+
+
+def _add_data_lines_to_plot(
+ plot_figure: go.Figure,
+ site_names: list[Site],
+ satellite: Optional[Sat],
+ data_product: DataProducts,
+ processed_file_path: Path,
+ y_axis_shift: float,
) -> None:
- # Получем все возможные цвета
- colors = px.colors.qualitative.Plotly
- # Ивлекаем данные
- site_data_tmp, is_satellite = retrieve_data(
- local_file, sites_name, sat, dataproduct
+ """Добавляет линии данных (например, TEC) для каждой станции на график."""
+ color_palette = px.colors.qualitative.Plotly
+
+ site_names_str = [str(name) for name in site_names]
+
+ processed_data, is_satellite_available = retrieve_data(
+ processed_file_path, site_names_str, satellite, data_product
)
- scatters = []
- for i, name in enumerate(sites_name):
- if sat is None or not is_satellite[name]: # Если у станции нет спутника
- sat_tmp = list(site_data_tmp[name].keys())[0]
-
- vals = site_data_tmp[name][sat_tmp][dataproduct]
- times = site_data_tmp[name][sat_tmp][DataProducts.time]
- vals_tmp = np.zeros_like(vals)
-
- # Рисуем прямую серую линию
- scatters.append(
- go.Scatter(
- x=times,
- y=vals_tmp + shift * (i + 1),
- customdata=vals_tmp,
- mode="markers",
- name=name.upper(),
- hoverinfo="text",
- hovertemplate="%{x}, %{customdata}",
- marker=dict(
- size=2,
- color = "gray",
- ),
+
+ scatter_traces = []
+ for i, site_name_str in enumerate(site_names_str):
+ if satellite is None or not is_satellite_available.get(site_name_str, False):
+ if site_name_str not in processed_data or not processed_data[site_name_str]:
+ continue
+
+ first_available_sat = list(processed_data[site_name_str].keys())[0]
+ data_values = processed_data[site_name_str][first_available_sat][data_product]
+ time_values = processed_data[site_name_str][first_available_sat][DataProducts.time]
+ # Отображаем как нулевые значения серой линией
+ display_values = np.zeros_like(data_values)
+ marker_color = "gray"
+ marker_size = 2
+ else: # Данные по указанному спутнику есть
+ # Конвертация в градусы, если это углы
+ if data_product in [DataProducts.azimuth, DataProducts.elevation]:
+ data_values = np.degrees(
+ processed_data[site_name_str][satellite.name][data_product]
)
- )
- else: # Если у станции есть спутник
- # Если azimuth или elevation переводим в градусы
- if (
- dataproduct == DataProducts.azimuth
- or dataproduct == DataProducts.elevation
- ):
- vals = np.degrees(site_data_tmp[name][sat][dataproduct])
else:
- vals = site_data_tmp[name][sat][dataproduct]
-
- times = site_data_tmp[name][sat][DataProducts.time]
-
- # Определяем цвет данных на графике
- idx_color = i if i < len(colors) else i - len(colors)*(i // len(colors))
- # Рисуем данные
- scatters.append(
- go.Scatter(
- x=times,
- y=vals + shift * (i + 1),
- customdata=vals,
- mode="markers",
- name=name.upper(),
- hoverinfo="text",
- hovertemplate="%{x}, %{customdata}",
- marker=dict(
- size=3,
- color = colors[idx_color],
- ),
- )
+ data_values = processed_data[site_name_str][satellite.name][data_product]
+
+ time_values = processed_data[site_name_str][satellite.name][DataProducts.time]
+ display_values = data_values
+ marker_color = color_palette[i % len(color_palette)]
+ marker_size = 3
+
+ scatter_traces.append(
+ go.Scatter(
+ x=time_values,
+ y=display_values + y_axis_shift * (i + 1),
+ customdata=display_values,
+ mode="markers",
+ name=site_name_str.upper(),
+ hoverinfo="text",
+ hovertemplate="%{x}, %{customdata:.2f}",
+ marker=dict(
+ size=marker_size,
+ color=marker_color,
+ ),
)
- site_data.add_traces(scatters)
+ )
+ plot_figure.add_traces(scatter_traces)
- # Настраиваем ось y для отображения имен станций
- site_data.layout.yaxis.tickmode = "array"
- site_data.layout.yaxis.tickvals = [
- shift * (i + 1) for i in range(len(sites_name))
+ # Настройка оси Y для отображения имен станций
+ plot_figure.layout.yaxis.tickmode = "array"
+ plot_figure.layout.yaxis.tickvals = [
+ y_axis_shift * (i + 1) for i in range(len(site_names_str))
]
- site_data.layout.yaxis.ticktext = list(map(str.upper, sites_name))
+ plot_figure.layout.yaxis.ticktext = [name.upper() for name in site_names_str]
-def _create_limit_xaxis(
- time_value: list[int], local_file: Path
-) -> tuple[datetime]:
- # Переводим целые значения времени в datetime
- date = local_file.stem # Получаем '2024-01-01'
- date = datetime.strptime(date, '%Y-%m-%d')
+def _create_datetime_limits_for_xaxis(
+ time_range_hours: list[int],
+ processed_file_path: Path,
+) -> tuple[datetime, datetime]:
+ """Создает начальную и конечную метки времени для оси X."""
+ date_str = processed_file_path.stem
+ base_date = datetime.strptime(date_str, '%Y-%m-%d')
- hour_start_limit = 23 if time_value[0] == 24 else time_value[0]
- minute_start_limit = 59 if time_value[0] == 24 else 0
- second_start_limit = 59 if time_value[0] == 24 else 0
+ start_hour_val = time_range_hours[0]
+ end_hour_val = time_range_hours[1]
- hour_end_limit = 23 if time_value[1] == 24 else time_value[1]
- minute_end_limit = 59 if time_value[1] == 24 else 0
- second_end_limit = 59 if time_value[1] == 24 else 0
+ # Обработка случая, когда час указан как 24 (конец дня)
+ start_hour = 23 if start_hour_val == 24 else start_hour_val
+ start_minute = 59 if start_hour_val == 24 else 0
+ start_second = 59 if start_hour_val == 24 else 0
+
+ end_hour = 23 if end_hour_val == 24 else end_hour_val
+ end_minute = 59 if end_hour_val == 24 else 0
+ end_second = 59 if end_hour_val == 24 else 0
+
start_limit = datetime(
- date.year,
- date.month,
- date.day,
- hour=hour_start_limit,
- minute=minute_start_limit,
- second=second_start_limit,
+ base_date.year, base_date.month, base_date.day,
+ hour=start_hour, minute=start_minute, second=start_second,
tzinfo=timezone.utc,
)
end_limit = datetime(
- date.year,
- date.month,
- date.day,
- hour=hour_end_limit,
- minute=minute_end_limit,
- second=second_end_limit,
+ base_date.year, base_date.month, base_date.day,
+ hour=end_hour, minute=end_minute, second=end_second,
tzinfo=timezone.utc,
)
- return (start_limit, end_limit)
\ No newline at end of file
+
+ return start_limit, end_limit
\ No newline at end of file