Какую программу/функцию мне использовать для запуска сверточной нейронной сети, чтобы обучать данные имитации движения на основании параметров, которые их сгенерировали?

Question 1

Я перегружен литературой, и каждая примерная задача, которую я вижу с использованием Pytorch или TensorFlow, отличается от моей. У меня есть N симуляций данных на сетке — это матрицы широт и долгот. Сами данные непрерывные, но я разбил их на квадраты на сетке. Они были сгенерированы из модели движения с M параметрами. Я хотел бы обучить параметры, которые смоделировали каждый набор данных, на соответствующих смоделированных наборах данных.

Если A_i — это матрицы с широтами в строках, долготами в столбцах и значениями в виде числа случаев в этом квадрате сетки, а P_i — это векторы, содержащие параметры, которые сгенерировали каждую A_i, как я могу программно запустить CNN? Цель заключается в способности предсказать параметры, которые сгенерировали данный тестовый набор.

Я изучал руководства по PyTorch, но большинство задач, которые я вижу в Интернете, отличаются от моей, особенно в двух аспектах:

Мои выходные данные — это счет. Я также могу сделать их долями от общего числа, и с удовольствием использовал бы функцию активации, которая дает результат между 0 и 1. Большинство задач, которые я вижу, — это задачи идентификации.
У меня нет причин использовать “цветовые каналы”.

Заранее благодарю.

Question 2

Я думаю, что лучше начать с простого и развиваться оттуда. Работайте над созданием базовой модели, работающей с вашими данными. Я рекомендую создавать и обучать модель с нуля.

Вам не обязательно нужно использовать CNN, даже если ваши данные по природе пространственные. Если данные достаточно простые, то традиционные ML модели (в сочетании с инженерией признаков) могут работать достаточно хорошо. Я продолжу с CNN, так как вы об этом спрашивали.

CNN будет обрабатывать пакеты $A_i$ формата (размер пакета, каналы=1, ширина сетки, ширина сетки) и выдавать набор предсказаний для этого пакета формата (размер пакета, M). Другими словами, модель будет обучена оценивать $P_i$, при условии что ей был предоставлен $A_i$.

Рассмотрим пример ниже. Модель, состоящая из $M=2$ параметров, генерирует такое распределение пространственных счетов:

Я случайным образом генерирую 5000 таких пар $(A_i, P_i)$, где сначала случайным образом выбираю параметры модели, а затем генерирую соответствующий $A_i$.

Я использовал 80% для обучения, а оставшуюся часть для валидации. Сначала матрицы конвертируются из массивов в тензоры PyTorch, а затем оборачиваются в DataLoader, который выдает batch_size образца за раз.

Пакеты последовательно передаются модели, и модель обучается на основе ошибки между $P_i^{predicted}$ и $P_i^{actual}$.

Ниже приведены результаты простой модели, которую я натренировал. Я написал ее вокруг задачи, которую вы описали, так что вы, возможно, сможете применить ее к своим собственным данным.

Воспроизводимый пример

Импорт и настройка модели широта-долгота:

import numpy as np
import matplotlib.pyplot as plt

#
# Создаем сетку значений широты и долготы
#
grid_width = 24
lat_grid, long_grid = np.meshgrid(*[np.linspace(-0.2, 0.8, num=grid_width)]*2)

#
# Выбираем случайный набор параметров и видим результат модели
#
def sample_model(pA, pB, lat_grid, long_grid):
    spatial_pattern = np.sinc(9*pA*lat_grid + 2*pB*long_grid) + np.cos(pB*long_grid) 
    counts = (spatial_pattern - spatial_pattern.min()).__mul__(50).round().astype(int)
    return counts

# np.random.seed(6)
pA = np.random.uniform(0, 0.5)
pB = np.random.uniform(0.5, 1)
Pi = np.array([pA, pB])

# Получаем значения на решетке результата модели
Ai = sample_model(pA, pB, lat_grid, long_grid)

# Визуализируем Ai/счеты по сетке широта-долгота
def view_sample(Ai, Pi, lat_grid, long_grid):
    f, ax = plt.subplots(figsize=(5.5, 4.5), layout="tight")

    mappable = ax.scatter(lat_grid, long_grid, c=Ai, marker="s", s=25, cmap='magma')
    ax.set(
        xlabel="долгота",
        ylabel="широта",
        title=f'$A_i$ (счет по сетке) из модели с параметрами $P_i$=({Pi[0]:.1f}, {Pi[1]:.1f})'
    )
    f.colorbar(mappable, label="счет")

    ax.spines[:].set_visible(False)

    plt.show()

view_sample(Ai, Pi, lat_grid, long_grid)

Создаем и делим датасет из 5000 пар $(A_i, P_i)$:

#
# Генерация n_samples (симуляций) пар признаков-целей (Ai, Pi)
# Ai: признаки для образца i (счеты по сетке формата grid_width x grid_width)
# Pi: параметры модели, которые сгенерировали Ai (вектор длиной n_parameters)
#
n_samples = 5_000 # так же, как ваши "N симуляции"
n_parameters = 2  # так же, как ваши параметры модели "M"

P_all = np.column_stack([
    np.random.uniform(0, 0.5, size=n_samples),
    np.random.uniform(0.5, 1, size=n_samples),
])
A_all = np.stack([sample_model(pA, pB, lat_grid, long_grid) for pA, pB in P_all])

print('Форма P_all:', P_all.shape, '| Форма A_all:', A_all.shape)

#
# Разделение на обучающую и валидационную выборки
# Перемешивание и выделение 80% для обучающей выборки, остальное для валидации
#
train_size = int(0.8 * n_samples)
print('Размеры обучающей/валидационной выборок:', f'{train_size}/{n_samples - train_size}')

shuffled_ixs = np.random.permutation(n_samples)
P_train, A_train = [data[shuffled_ixs][:train_size] for data in (P_all, A_all)]
P_val, A_val = [data[shuffled_ixs][train_size:] for data in (P_all, A_all)]

# Визуализируем обучающий образец
sample_idx = np.random.randint(train_size)
view_sample(A_train[sample_idx], P_train[sample_idx], lat_grid, long_grid)

Определяем маленькую CNN:

import torch
from torch import nn
from torch.utils.data import TensorDataset, DataLoader

#
# Определение простой CNN, которая принимает (B, 1, размер_сетки, размер_сетки) и выдает (B, 2 параметра)
#
def create_model(n_output_parameters, hidden_size=16):
    model = nn.Sequential(
        # Нормализация входного набора
        nn.BatchNorm2d(num_features=1),

        # Обучение hidden_size из 3x3 ядер, которые обнаруживают признаки
        nn.Conv2d(in_channels=1, out_channels=hidden_size, kernel_size=3, padding='same'),
        nn.ReLU(),
        nn.BatchNorm2d(hidden_size),

        # Уменьшение пространственных размеров вдвое, с сохранением максимальных значений на квадрат
        nn.MaxPool2d(2),

        # Усреднение пространственных размеров до одного скаляра и удаление лишних размеров
        nn.AdaptiveAvgPool2d(1),
        nn.Flatten(start_dim=1),

        # Преобразование входной формы (B, hidden_size) в (B, n_output_parameters)
        nn.Linear(in_features=hidden_size, out_features=hidden_size),
        nn.ReLU(),
        nn.Linear(in_features=hidden_size, out_features=n_output_parameters),
    )
    return model

# Тестирование форм с фиктивным пакетом из 64 образцов
dummy_batch = torch.rand(64, 1, grid_width, grid_width)

cnn_model = create_model(n_parameters)
output = cnn_model(dummy_batch)
print('Форма тестового входа:', dummy_batch.shape, '-> форма выхода:', output.shape, end='\n\n')

# Сообщение о размере CNN
print(
    'CNN имеет',
    sum(p.numel() for p in cnn_model.parameters() if p.requires_grad),
    'тренируемых параметров'
)

Подготовка данных для модели:

#
# Подготовка данных для модели
# 1. Преобразование в тензоры
# 2. Добавление размерности каналов
# 3. Оборачивание в DataLoader для получения пакетных образцов
#

# Преобразование в тензоры
P_train_t, P_val_t, A_train_t, A_val_t = [
    torch.tensor(array).float() for array in (P_train, P_val, A_train, A_val)
]

# CNN ожидает входной формат (пакет/образцы, каналы, высота, ширина)
# У нас сейчас (B, H, W), поэтому добавляем одинокую размерность каналов, чтобы
# получить (B, C=1, H, W)
A_train_t, A_val_t = [tensor.unsqueeze(dim=1) for tensor in (A_train_t, A_val_t)]

# Датасеты для обучающей и валидационной выборок, возвращающие (Ai, Pi) пару для каждого индекса.
# Иначе говоря, они соединяют Ai и Pi каждого образца вместе.
train_dataset = TensorDataset(A_train_t, P_train_t)
val_dataset = TensorDataset(A_val_t, P_val_t)

Ai, Pi = train_dataset[0] # возвращает Ai и Pi для образца 0
print('Образец 0 (A0, P0) из train_dataset:', Ai.shape, ',', Pi.shape)

# Пакетирование
batch_size = 32
train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)

Обучение модели и отчет о процессе:

#
# Обучение и оценка cnn_model
#

# Для воспроизводимости
np.random.seed(0)
torch.manual_seed(0)

cnn_model = create_model(n_parameters)
optimizer = torch.optim.Adam(cnn_model.parameters())
loss_function = nn.MSELoss()

#
# Цикл обучения
#
n_epochs = 12

# Используется для записи потерей и метрик
from collections import defaultdict
metrics_dict = defaultdict(list)

for epoch in range(n_epochs):

    cnn_model.train()
    for minibatch in train_loader:
        A_minibatch, P_minibatch = minibatch

        P_predicted = cnn_model(A_minibatch)
        loss = loss_function(P_predicted, P_minibatch)

        # Шаг оптимизатора
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
    # / конец эпохи

    #
    # Оценка cnn_model за каждую эпоху
    #
    cnn_model.eval()
    with torch.no_grad():
        P_predicted_train = cnn_model(A_train_t)
        P_predicted_val = cnn_model(A_val_t)

    train_loss = loss_function(P_predicted_train, P_train_t).item()
    val_loss = loss_function(P_predicted_val, P_val_t).item()
    val_mae = nn.L1Loss()(P_predicted_val, P_val_t).item()

    metrics_dict['epoch'].append(epoch + 1)
    metrics_dict['train_loss'].append(train_loss)
    metrics_dict['val_loss'].append(val_loss)
    metrics_dict['val_mae'].append(val_mae)

    #
    # Отчет о результатах
    #
    print(
        f'[эпоха {epoch + 1:2d}/{n_epochs}]',
        f'[потеря на обучении: {train_loss:5.3f} | потеря на валидации: {val_loss:5.3f}]',
        f'[MAE на валидации: {val_mae:5.3f}]'
    )

Построение кривой обучения:

#
# Построение кривой обучения
#
f, ax = plt.subplots(figsize=(8, 2.5), layout="tight")

start_ep = 2
epochs = metrics_dict['epoch'][start_ep:]
ax.plot(epochs, metrics_dict['train_loss'][start_ep:], color="tab:red", label="потеря на обучении")
ax.plot(epochs, metrics_dict['val_loss'][start_ep:], color="tab:green", label="потеря на валидации")

ax2 = ax.twinx()
ax2.plot(epochs, metrics_dict['val_mae'][start_ep:], color="tab:green", linestyle=":", label="MAE на валидации")

# Форматирование
f.legend(loc="upper right", framealpha=1, shadow=True)
ax.set(xlabel="эпоха", ylabel="потеря", title="Кривая обучения")
ax.set_xticks(epochs)
ax2.set_ylabel('MAE')

Сравнение предсказанных и фактических параметров модели:

f, axs = plt.subplots(figsize=(6, 2.5), layout="tight", ncols=2)

ax = axs[0]
ax.scatter(P_val_t[:, 0], P_predicted_val[:, 0], marker=".", alpha=0.3, label="выборка на валидации")
ax.plot([0, 0.5], [0, 0.5], color="tab:red", label="y=x")
ax.set(xlabel="истинное pA", ylabel="предсказанное pA")
ax.legend(framealpha=0, fontsize=9)

ax = axs[1]
ax.scatter(P_val_t[:, 1], P_predicted_val[:, 1], marker=".", alpha=0.3)
ax.plot([0.5, 1], [0.5, 1], color="tab:red")
ax.set(xlabel="истинное pB", ylabel="предсказанное pB")

f.suptitle('Истинные параметры против предсказанных параметров')

Больше визуализации для одного образца, на этот раз с фокусировкой на $A_i$:

#
# Сравнение фактических (Ai, Pi) с предсказанным Pi моделью (и Ai, который в результате получается)
#
Ai, Pi = train_dataset[np.random.randint(train_size)]
print('Фактические Ai и Pi:')
view_sample(Ai, Pi, lat_grid, long_grid)

with torch.no_grad():
    Pi_pred = cnn_model(Ai.unsqueeze(0)).numpy().ravel()
print('Фактическое vs предсказанное Pi:', Pi.tolist(), '|', Pi_pred)

Ai_model = sample_model(*Pi_pred, lat_grid, long_grid)
print('Ai с использованием предсказанного Pi моделью:')
view_sample(Ai_model, Pi_pred, lat_grid, long_grid)

# Визуализация ошибки
Ai_mismatch = Ai_model - Ai.numpy()

f, ax = plt.subplots(figsize=(5.5, 4.5), layout="tight")
mappable = ax.scatter(lat_grid, long_grid, c=Ai_mismatch, marker="s", s=25, cmap='magma')
ax.set(xlabel="долгота", ylabel="широта", title="Ошибка Ai")

f.colorbar(mappable, label="несоответствие счетов")
ax.spines[:].set_visible(False)
plt.show()

Вопрос или проблема

Воспроизводимый пример

Ответ или решение

Постановка задачи

Решение с использованием PyTorch

Заключение