Общее задание

src/data_loader.py

@@ -0,0 +1,35 @@
+import csv
+from src.data_parser import parse_observation_line
+
+def load_observations(file_path: str) -> list:
+    """Загружает наблюдения из CSV-файла"""
+    try:
+        observations = []
+        with open(file_path, 'r', encoding='utf-8') as file:
+            reader = csv.reader(file)
+            next(reader)  # Пропускаем заголовок
+            for row in reader:
+                line = ','.join(row)
+                observations.append(parse_observation_line(line))
+        return observations
+    except FileNotFoundError:
+        print(f"Ошибка: Файл {file_path} не найден")
+        return []
+
+def load_stations(file_path: str) -> dict:
+    """Загружает справочник станций из CSV-файла"""
+    try:
+        stations = {}
+        with open(file_path, 'r', encoding='utf-8') as file:
+            reader = csv.DictReader(file)
+            for row in reader:
+                station_id = row['station_id']
+                stations[station_id] = {
+                    "name": row['station_name'],
+                    "latitude": float(row['latitude']),
+                    "longitude": float(row['longitude'])
+                }
+        return stations
+    except FileNotFoundError:
+        print(f"Ошибка: Файл {file_path} не найден")
+        return {}

src/data_parser.py

@@ -0,0 +1,19 @@
+def parse_observation_line(line: str) -> dict:
+    """Парсит строку из CSV-файла с наблюдениями"""
+    parts = line.strip().split(',')
+
+    station_id = parts[0]
+    date = parts[1]
+
+    # Преобразуем числовые значения, пустые строки заменяем на None
+    temperature = float(parts[2]) if parts[2] and parts[2] != '' else None
+    precipitation = float(parts[3]) if parts[3] and parts[3] != '' else None
+    wind_speed = float(parts[4]) if parts[4] and parts[4] != '' else None
+
+    return {
+        "station_id": station_id,
+        "date": date,
+        "temperature": temperature,
+        "precipitation": precipitation,
+        "wind_speed": wind_speed
+    }

src/filters.py

@@ -0,0 +1,45 @@
+def filter_by_date_range(observations: list, start_date: str, end_date: str) -> list:
+    """Фильтрует наблюдения по диапазону дат"""
+    filtered = []
+    for obs in observations:
+        if start_date <= obs['date'] <= end_date:
+            filtered.append(obs)
+    return filtered
+
+
+def clean_observations(observations: list) -> list:
+    """Очищает наблюдения от некорректных значений"""
+    cleaned = []
+
+    for obs in observations:
+        # Проверяем температуру
+        temp = obs['temperature']
+        if temp is not None and (temp < -50 or temp > 50):
+            continue
+
+        # Проверяем осадки
+        precip = obs['precipitation']
+        if precip is not None and precip < 0:
+            continue
+
+        # Проверяем скорость ветра
+        wind = obs['wind_speed']
+        if wind is not None and wind < 0:
+            continue
+
+        # Если все проверки пройдены, добавляем запись
+        cleaned_obs = obs.copy()
+        cleaned.append(cleaned_obs)
+
+    return cleaned
+
+def classify_wind_speed(speed: float) -> str:
+    """Классифицирует скорость ветра"""
+    if speed < 1:
+        return "штиль"
+    elif 1 <= speed < 5:
+        return "слабый"
+    elif 5 <= speed <= 10:
+        return "умеренный"
+    else:
+        return "сильный"

src/main.py

@@ -0,0 +1,73 @@
+import os
+from src.data_loader import load_observations, load_stations
+from src.filters import filter_by_date_range, clean_observations
+from src.statistics import (calculate_daily_stats, find_extreme_stations,
+                            add_wind_category)
+from src.report_generator import generate_report
+
+
+def main():
+    DATA_DIR = "../data"
+    REPORTS_DIR = "../reports"
+    START_DATE = "2024-01-01"
+    END_DATE = "2024-12-31"
+
+    os.makedirs(REPORTS_DIR, exist_ok=True)
+
+    print("Загрузка данных о станциях...")
+    stations = load_stations(f"{DATA_DIR}/stations.csv")
+
+    print("Загрузка данных о наблюдениях...")
+    observations = load_observations(f"{DATA_DIR}/observations.csv")
+
+    if not observations:
+        print("Нет данных для обработки!")
+        return
+
+    print(f"Фильтрация данных за период {START_DATE} - {END_DATE}...")
+    filtered_obs = filter_by_date_range(observations, START_DATE, END_DATE)
+
+    print("Очистка данных...")
+    cleaned_obs = clean_observations(filtered_obs)
+
+    print("Добавление категорий ветра...")
+    obs_with_wind = add_wind_category(cleaned_obs)
+
+    print("Вычисление дневной статистики...")
+    daily_stats = calculate_daily_stats(obs_with_wind)
+
+    print("Поиск экстремальных станций...")
+    top_avg_temp = find_extreme_stations(obs_with_wind, stations, "avg_temp", 3)
+    top_max_wind = find_extreme_stations(obs_with_wind, stations, "max_wind", 3)
+
+    print("Формирование отчета...")
+    output_path = f"{REPORTS_DIR}/weather_report.txt"
+    generate_report(obs_with_wind, stations, daily_stats, top_avg_temp, output_path)
+
+    print("\n" + "=" * 50)
+    print("РЕЗУЛЬТАТЫ АНАЛИЗА")
+    print("=" * 50)
+
+    print("\nТоп-3 станции по средней температуре:")
+    for i, (name, value) in enumerate(top_avg_temp, 1):
+        print(f"  {i}. {name}: {value:.2f}°C")
+
+    print("\nТоп-3 станции по максимальной скорости ветра:")
+    for i, (name, value) in enumerate(top_max_wind, 1):
+        # Определяем категорию ветра для каждого значения
+        if value < 1:
+            category = "штиль"
+        elif 1 <= value < 5:
+            category = "слабый"
+        elif 5 <= value <= 10:
+            category = "умеренный"
+        else:
+            category = "сильный"
+        print(f"  {i}. {name}: {value:.1f} м/с ({category})")
+
+    print(f"\nОтчет сохранен: {output_path}")
+    print("=" * 50)
+
+
+if __name__ == "__main__":
+    main()

src/report_generator.py

@@ -0,0 +1,47 @@
+def generate_report(observations: list, stations: dict, daily_stats: dict,
+                    extreme_stations: list, output_path: str) -> None:
+    """Формирует текстовый отчет"""
+
+    # Общее количество наблюдений
+    total_observations = len(observations)
+
+    # Диапазон дат
+    dates = [obs['date'] for obs in observations]
+    if dates:
+        min_date = min(dates)
+        max_date = max(dates)
+    else:
+        min_date = max_date = "Нет данных"
+
+    # Список экстремальных станций
+    extreme_stations_str = "\n".join([f"  - {name}: {value:.2f}" for name, value in extreme_stations])
+
+    # Средняя температура за весь период
+    temperatures = [obs['temperature'] for obs in observations if obs['temperature'] is not None]
+    avg_temperature = sum(temperatures) / len(temperatures) if temperatures else 0
+
+    # Количество дней с сильным ветром
+    strong_wind_days = sum(1 for obs in observations
+                           if obs.get('wind_category') == "сильный")
+
+    # Формируем отчет
+    report = f"""ОТЧЕТ ПО ПОГОДНЫМ НАБЛЮДЕНИЯМ
+{'=' * 50}
+
+Общее количество обработанных наблюдений: {total_observations}
+
+Диапазон дат: {min_date} - {max_date}
+
+Список экстремальных станций (топ-3):
+{extreme_stations_str}
+
+Средняя температура за весь период: {avg_temperature:.2f}°C
+
+Количество дней с сильным ветром: {strong_wind_days}
+
+{'=' * 50}
+"""
+
+    # Сохраняем отчет
+    with open(output_path, 'w', encoding='utf-8') as file:
+        file.write(report)

src/statistics.py

@@ -0,0 +1,94 @@
+from src.filters import classify_wind_speed
+
+def calculate_daily_stats(observations: list) -> dict:
+    """Вычисляет дневную статистику по наблюдениям"""
+    daily_data = {}
+
+    for obs in observations:
+        date = obs['date']
+
+        if date not in daily_data:
+            daily_data[date] = {
+                'temperatures': [],
+                'total_precip': 0,
+                'max_wind': 0
+            }
+
+        # Добавляем температуру (игнорируем None)
+        if obs['temperature'] is not None:
+            daily_data[date]['temperatures'].append(obs['temperature'])
+
+        # Добавляем осадки
+        if obs['precipitation'] is not None:
+            daily_data[date]['total_precip'] += obs['precipitation']
+
+        # Обновляем максимальный ветер
+        if obs['wind_speed'] is not None:
+            daily_data[date]['max_wind'] = max(daily_data[date]['max_wind'], obs['wind_speed'])
+
+    # Формируем результат
+    result = {}
+    for date, data in daily_data.items():
+        avg_temp = sum(data['temperatures']) / len(data['temperatures']) if data['temperatures'] else None
+        result[date] = {
+            'avg_temp': avg_temp,
+            'total_precip': data['total_precip'],
+            'max_wind': data['max_wind']
+        }
+
+    return result
+
+
+def find_extreme_stations(observations: list, stations: dict, metric: str, top_n: int = 3) -> list:
+    """Находит топ-N станций по заданной метрике"""
+    station_metrics = {}
+
+    # Группируем наблюдения по станциям
+    for obs in observations:
+        station_id = obs['station_id']
+        if station_id not in station_metrics:
+            station_metrics[station_id] = {
+                'temperatures': [],
+                'wind_speeds': []
+            }
+
+        if obs['temperature'] is not None:
+            station_metrics[station_id]['temperatures'].append(obs['temperature'])
+
+        if obs['wind_speed'] is not None:
+            station_metrics[station_id]['wind_speeds'].append(obs['wind_speed'])
+
+    # Вычисляем метрики для каждой станции
+    results = []
+    for station_id, metrics in station_metrics.items():
+        if station_id not in stations:
+            continue
+
+        station_name = stations[station_id]['name']
+
+        if metric == 'avg_temp':
+            if metrics['temperatures']:
+                value = sum(metrics['temperatures']) / len(metrics['temperatures'])
+                results.append((station_name, value))
+        elif metric == 'max_wind':
+            if metrics['wind_speeds']:
+                value = max(metrics['wind_speeds'])
+                results.append((station_name, value))
+
+    # Сортируем и берем top_n
+    results.sort(key=lambda x: x[1], reverse=True)
+    return results[:top_n]
+
+
+
+def add_wind_category(observations: list) -> list:
+    """Добавляет категорию ветра к наблюдениям"""
+    new_observations = []
+    for obs in observations:
+        obs_copy = obs.copy()
+        if obs_copy['wind_speed'] is not None:
+            obs_copy['wind_category'] = classify_wind_speed(obs_copy['wind_speed'])
+        else:
+            obs_copy['wind_category'] = None
+        new_observations.append(obs_copy)
+    return new_observations