В настоящее время я работаю с массивом, содержащим категориальные данные. Категории организованы следующим образом: None, zoneA, zoneB Мой массив является мерой датчиков, он сообщает мне, находится ли в любой момент датчик в zoneA, zoneB или нет в зоне.

Моя цель здесь — сгладить эти значения.

Например, датчик может находиться вне зоны a или b в течение периода в 30 измерений, но если это произошло, я хочу, чтобы эти меры были «сглажены».

Пример :

массив [zoneA, zoneA, zoneA, None, None, zoneA, zoneA, None, None, None, zoneA]

должно дать

массив [zoneA, zoneA, zoneA, zoneA, zoneA, zoneA, zoneA, Нет, Нет, Нет, zoneA]

с порогом 2.

В настоящее время я использую итерацию по массивам, но ее вычисление слишком дорого и может занять 1 или 2 минуты вычислений. Существует ли существующий алгоритм для решения этой проблемы?

Мой текущий код :

  def smooth(self, df: pd.DataFrame) -> pd.DataFrame:
    """
    Args:
        df (pd.DataFrame): dataframe with landlot column to smooth.
    Returns:dataframe smoothed
    """
    df_iter = df
    last = "None"
    last_index = 0
    for num, line in df_iter.iterrows():
        if (
                (line.landlot != "None")
                and (line.landlot == last)
                and (num - last_index <= self.delay)
                and (
                df_iter.iloc[(num - 1), df_iter.columns.get_loc("landlot")]
                == "None"
        )
        ):
            df_iter.iloc[
            last_index: (num   1),  # noqa: E203
            df_iter.columns.get_loc("landlot"),
            ] = last
        if line.landlot != "None":
            last = line.landlot
            last_index = num
    return df_iter
 

Реализация Python

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

 class Interpolator:
    def __init__(self, data):
        self.data = data
        self.current_idx = 0
        self.current_nan_region_start = None
        self.result = None
        self.maxgap = 1

    def run(self, maxgap=2):
        # Initialization
        self.result = [None] * len(self.data)
        self.maxgap = maxgap
        self.current_nan_region_start = None
        prev_isnan = 0

        for idx, item in enumerate(self.data):
            isnan = item is None
            self.current_idx = idx
            if isnan:
                if prev_isnan:
                    # Result is already filled with empty data.
                    # Do nothing.
                    continue
                else:
                    self.entered_nan_region()
                    prev_isnan = 1
            else:  # not nan
                if prev_isnan:
                    self.exited_nan_region()
                    prev_isnan = 0
                else:
                    self.continuing_in_categorical_region()

    def entered_nan_region(self):
        self.current_nan_region_start = self.current_idx

    def continuing_in_categorical_region(self):
        self.result[self.current_idx] = self.data[self.current_idx]

    def exited_nan_region(self):

        nan_region_end = self.current_idx - 1
        nan_region_length = nan_region_end - self.current_nan_region_start   1

        # Always copy the empty region endpoint even if gap is not filled
        self.result[self.current_idx] = self.data[self.current_idx]

        if nan_region_length > self.maxgap:
            # Do not interpolate as exceeding maxgap
            return

        if self.current_nan_region_start == 0:
            # Special case. data starts with "None"
            # ->  Cannot interpolate
            return

        if self.data[self.current_nan_region_start - 1] != self.data[self.current_idx]:
            # Do not fill as both ends of missing data
            # region do not have same value
            return

        # Fill the gap
        for idx in range(self.current_nan_region_start, self.current_idx):
            self.result[idx] = self.data[self.current_idx]


def interpolate(data, maxgap=2):
    """
    Interpolate categorical variables over missing
    values (None's).

    Parameters
    ----------
    data: list of objects
        The data to interpolate. Holds
        categorical data, such as 'cat', 'dog'
        or 108. None is handled as missing data.
    maxgap: int
        The maximum gap to interpolate over.
        For example, with maxgap=2, ['car', None,
        None, 'car', None, None, None, 'car']
        would become  ['car', 'car', 'car' 'car',
        None, None None, 'car'].

    Note: Interpolation will only occur on missing
    data regions where both ends contain the same value.
    For example, [1, None, 2, None, 2] will become
    [1, None, 2, 2, 2].
    """

    interpolator = Interpolator(data)
    interpolator.run(maxgap=maxgap)
    return interpolator.result
 

Вот как можно было бы его использовать (код для get_data() ниже):

 data = get_data(k=100)
interpolated_data = interpolate(data)
 

Копирование-вставка реализации Cython

Скорее всего, реализация python достаточно быстрая, так как при размере массива 1000 000 время, необходимое для обработки данных, составляет 0,504 секунды на моем ноутбуке. В любом случае, создание версий Cython — это весело и может дать небольшой дополнительный временной бонус.

Необходимые шаги:

• Скопируйте и вставьте реализацию python в новый файл, называемый fast_categorical_interpolate.pyx
• Создайте setup.py в ту же папку со следующим содержимым:

 from setuptools import setup
from Cython.Build import cythonize

setup(
    ext_modules=cythonize(
        "fast_categorical_interpolate.pyx",
        language_level="3",
    ),
)
 

• Запустите python setup.py build_ext --inplace для создания расширения Cython. Вы увидите что-то вроде fast_categorical_interpolate.cp38-win_amd64.pyd в той же папке.
• Теперь вы можете использовать интерполятор следующим образом:

 import fast_categorical_interpolate as fpi
data = get_data(k=100)
interpolated_data = fpi.interpolate(data)
 

• Конечно, в коде Cython могут быть некоторые оптимизации, которые вы могли бы выполнить, чтобы сделать это еще быстрее, но на моей машине улучшение скорости составило 38% из коробки при N = 1000 000 и 126% при N = 10 000.

Тайминги на моей машине

• При N = 100 (количество элементов в списке) реализация python примерно в 160 раз, а реализация Cython примерно в 250 раз быстрее, чем smooth

 In [8]: timeit smooth(test_df, delay=2)
10.2 ms ± 669 µs per loop (mean ± std. dev. of 7 runs, 10 loops each)
In [9]: timeit interpolate(data)

64.8 µs ± 7.39 µs per loop (mean ± std. dev. of 7 runs, 10000 loops each)

In [10]: timeit fpi.interpolate(data)
41.3 µs ± 4.64 µs per loop (mean ± std. dev. of 7 runs, 10000 loops each)

 

• При N = 10.000 разница во времени составляет примерно от 190x (Python) до 302x (Cython).

 In [5]: timeit smooth(test_df, delay=2)
1.08 s ± 166 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

In [6]: timeit interpolate(data)
5.69 ms ± 852 µs per loop (mean ± std. dev. of 7 runs, 100 loops each)

In [7]: timeit fpi.interpolate(data)
3.57 ms ± 377 µs per loop (mean ± std. dev. of 7 runs, 100 loops each)

 

• При N = 1000 000 реализация python примерно в 210 раз быстрее, а реализация Cython примерно в 287 раз быстрее.

 In [9]: timeit smooth(test_df, delay=2)
1min 45s ± 24.2 s per loop (mean ± std. dev. of 7 runs, 1 loop each)
In [10]: timeit interpolate(data)
504 ms ± 67.2 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)
In [11]: timeit fpi.interpolate(data)
365 ms ± 38 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)
 

Приложение

Создатель тестовых данных get_data()

 import random
random.seed(0)

def get_data(k=100):
    return random.choices(population=[None, "ZoneA", "ZoneB"], weights=[4, 3, 2], k=k)
 

Функциональные и тестовые данные для тестирования smooth()

 import pandas as pd

data = get_data(k=1000)
test_df = pd.DataFrame(dict(landlot=data)).fillna("None")


def smooth(df: pd.DataFrame, delay=2) -> pd.DataFrame:
    """
    Args:
        df (pd.DataFrame): dataframe with landlot column to smooth.
    Returns:dataframe smoothed
    """
    df_iter = df
    last = "None"
    last_index = 0
    for num, line in df_iter.iterrows():
        if (
            (line.landlot != "None")
            and (line.landlot == last)
            and (num - last_index <= delay)
            and (df_iter.iloc[(num - 1), df_iter.columns.get_loc("landlot")] == "None")
        ):
            df_iter.iloc[
                last_index : (num   1),  # noqa: E203
                df_iter.columns.get_loc("landlot"),
            ] = last
        if line.landlot != "None":
            last = line.landlot
            last_index = num
    return df_iter
 

Обратите внимание на «текущий код»

Я думаю, что где-то должна быть какая-то ошибка копирования-вставки, поскольку «текущий код» работает не так, как все. Я заменил self.delay на delay=2 аргумент ключевого слова, чтобы указать максимальный разрыв. Я предполагаю, что так и должно было быть. Даже при этом логика не работала корректно с предоставленными вами данными простого примера.