ПРИЛОЖЕНИЕ 4

ЭТАЛОН ВЫПОЛНЕНИЯ ЛАБОРАТОРНОЙ РАБОТЫ №3 (ЛР3): ЧИСЛЕННЫЙ АНАЛИЗ ДАННЫХ С ИСПОЛЬЗОВАНИЕМ NUMPY

Файл: lab3_numpy.py
Данные: data/spotify_edu_dataset.csv

"""
ЛР3. Численный анализ данных с использованием NumPy (эталонный каркас).

Назначение:
- показать рекомендуемую архитектуру решения;
- продемонстрировать формирование массивов NumPy из CSV;
- обеспечить корректную работу с пропусками (np.nan);
- заложить основу для этапов 2–6 (статистика, маски, корреляция, масштабирование, время).

Требование:
- при использовании эталона обучающийся должен понимать логику каждого шага
  и отразить в отчёте собственные результаты и интерпретации.
"""

from __future__ import annotations

import csv
import time
from typing import Dict, List, Tuple

import numpy as np

# -------------------------
# Константы и настройки
# -------------------------

DATA_PATH = "data/spotify_edu_dataset.csv"

NUMERIC_COLUMNS = [
    "duration_sec",
    "energy",
    "danceability",
    "tempo",
    "views",
]


# -------------------------
# Вспомогательные функции
# -------------------------

def to_float_or_nan(value: str) -> float:
    """
    Преобразует строковое значение в float.
Возвращает np.nan, если значение пустое или некорректное.

    Примеры некорректности:
    - пустая строка
    - нечисловой текст
    - пробелы и т.п.
"""
    if value is None:
        return np.nan

    s = str(value).strip()
    if s == "":
        return np.nan

    try:
        return float(s)
    except ValueError:
        return np.nan


def read_csv_as_dicts(path: str) -> List[Dict[str, str]]:
    """
    Считывает CSV в виде списка словарей.
Ключи словаря – имена столбцов (из заголовка CSV).
    Значения – строки (как в файле).
"""
    rows: List[Dict[str, str]] = []
    with open(path, "r", encoding="utf-8", newline="") as f:
        reader = csv.DictReader(f)
        for row in reader:
            rows.append(row)
    return rows


def build_numeric_arrays(rows: List[Dict[str, str]], columns: List[str]) -> Dict[str, np.ndarray]:
    """
    Извлекает указанные числовые столбцы и формирует для каждого np.ndarray dtype=float.
    Пропуски/ошибки преобразования заменяются на np.nan.
    """
    arrays: Dict[str, np.ndarray] = {}

    for col in columns:
        values = [to_float_or_nan(r.get(col, "")) for r in rows]
        arrays[col] = np.array(values, dtype=float)

    return arrays


def check_array(arr: np.ndarray, name: str) -> None:
    """
    Выводит минимально необходимые проверки:
    - dtype
    - shape
    - число пропусков (np.nan)
    """
    nan_count = int(np.isnan(arr).sum())
    print(f"[CHECK] {name}: dtype={arr.dtype}, shape={arr.shape}, NaN={nan_count}")


# -------------------------
# Этап 2: статистика (заготовки)
# -------------------------

def basic_stats(arr: np.ndarray) -> Dict[str, float]:
    """
    Базовые статистики с игнорированием NaN там, где это необходимо.
    """
    return {
        "mean": float(np.nanmean(arr)),
        "median": float(np.nanmedian(arr)),
        "std": float(np.nanstd(arr)),
        "min": float(np.nanmin(arr)),
        "max": float(np.nanmax(arr)),
    }


# -------------------------
# Этап 3: булевы маски (заготовки)
# -------------------------

def count_where(arr: np.ndarray, condition: np.ndarray) -> int:
    """
    Количество элементов, удовлетворяющих условию (маске).
    """
    # condition – булев массив True/False той же формы, что и arr
    return int(condition.sum())


# -------------------------
# Этап 4: корреляция (заготовки)
# -------------------------

def corr_pearson(x: np.ndarray, y: np.ndarray) -> float:
    """
    Корреляция Пирсона между двумя массивами, с удалением пар NaN.
    """
    mask = ~np.isnan(x) & ~np.isnan(y)
    x_clean = x[mask]
    y_clean = y[mask]

    # Если слишком мало данных, корреляция может быть неустойчивой
    if x_clean.size < 2:
        return float("nan")

    r_matrix = np.corrcoef(x_clean, y_clean)
    return float(r_matrix[0, 1])


# -------------------------
# Этап 5: масштабирование (заготовки)
# -------------------------

def standardize(x: np.ndarray) -> np.ndarray:
    """
    Стандартизация: z = (x - mean) / std (с игнорированием NaN).
    """
    mu = np.nanmean(x)
    sigma = np.nanstd(x)
    return (x - mu) / sigma


def minmax_normalize(x: np.ndarray) -> np.ndarray:
    """
    Min–max нормализация в диапазон [0, 1] (с игнорированием NaN).
    """
    x_min = np.nanmin(x)
    x_max = np.nanmax(x)
    return (x - x_min) / (x_max - x_min)


# -------------------------
# Этап 6: производительность (заготовки)
# -------------------------

def mean_by_loop(arr: np.ndarray) -> float:
    """
    Среднее по циклу Python с игнорированием NaN.
    """
    total = 0.0
    count = 0
    for v in arr:
        if not np.isnan(v):
            total += float(v)
            count += 1
    return total / count if count else float("nan")


def time_call(fn, *args, repeats: int = 5) -> Tuple[float, float]:
    """
    Измеряет время выполнения функции fn(*args).
    Возвращает (минимум, среднее) по нескольким запускам.
    """
    times = []
    for _ in range(repeats):
        t0 = time.time()
        fn(*args)
        t1 = time.time()
        times.append(t1 - t0)
    return min(times), sum(times) / len(times)


# -------------------------
# main
# -------------------------

def main() -> None:
    # ЭТАП 1. Чтение CSV и формирование массивов
    rows = read_csv_as_dicts(DATA_PATH)
    arrays = build_numeric_arrays(rows, NUMERIC_COLUMNS)

    print(f"Загружено записей: {len(rows)}")
    for name, arr in arrays.items():
        check_array(arr, name)

    # Ниже – примеры вызовов для следующих этапов (можно раскомментировать по заданию)

    # ЭТАП 2. Базовая статистика
    # for name, arr in arrays.items():
    #     stats = basic_stats(arr)
    #     print(name, stats)

    # ЭТАП 3. Маски
    # energy = arrays["energy"]
    # n_high_energy = count_where(energy, energy > 0.8)
    # print("energy > 0.8:", n_high_energy)

    # ЭТАП 4. Корреляция
    # r = corr_pearson(arrays["energy"], arrays["danceability"])
    # print("corr(energy, danceability) =", r)

    # ЭТАП 5. Масштабирование
    # z = standardize(arrays["views"])
    # x_norm = minmax_normalize(arrays["views"])
    # print("views standardized: ", z[:5])
    # print("views normalized: ", x_norm[:5])

    # ЭТАП 6. Производительность
    # arr = arrays["duration_sec"]
    # tmin_loop, tavg_loop = time_call(mean_by_loop, arr)
    # tmin_np, tavg_np = time_call(np.nanmean, arr)
    # print("Loop mean time (min/avg):", tmin_loop, tavg_loop)
    # print("NumPy mean time (min/avg):", tmin_np, tavg_np)


if __name__ == "__main__":
    main()