Простой цифровой фильтр на базе пузырькового алгоритма сортировки на языке microPython используя микроконтроллер ESP8266.

В данном уроке мы рассмотри простой способ фильтрации полученных данных.

Предположим вы делаете измерение некой величины, при чем устройство работает в условиях помех: самое простое это измерения напряжения в бортовой сети машины. При измерении может пройти помеха в виде не верной величины, и нам нужно ее убрать.

Самое простое, это сделать N измерений, поместить их в массив, отсортировать и убрать первые и последние значения. Если не вдаваться в глубокую теорию, то как раз первые и последние значения в массиве будут максимально отдалены от истинных значений. Количество значений в массиве и количество элементов которые убираем из массива индивидуально и зависит от вида и условий измерений. У нас будет выборка из 20 измерений и будем убирать по 3 значения слева и справа массива.

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

Программа реализации пузырькового алгоритма сортировки:

import utime




list = [10.0, 10.2, 10.1, 11.0, 10.3, 9.99, 9.98, 10.1, 10.2, 9.97,10.0, 10.2, 10.1, 11.0, 10.3, 9.99, 9.98, 10.1, 10.2, 9.87]
print (list)

def summa (num, z):
   for i in range(z):
     list.pop(0)
     list.pop()
   sum=0
   for i in range(len(num) - 1):
       sum +=list[i]
   sum /= (len(list) - 1)
   return sum

def bubble_sort(num):
     # Устанавливаем sw в True, для запуска цикл минимум один раз 
    sw = True
    while sw:
         sw = False
         for i in range(len(num) - 1):
             if num[i] > num[i + 1]:
               # Меняем элементы
               num[i], num[i + 1] = num[i + 1], num[i]
               # Устанавливаем sw в True для следующей итерации
               sw = True

# запуск алгоритма
t1 = utime.ticks_us()
bubble_sort(list)
print (list)
rez=summa (list, 3)
t2 = utime.ticks_us()
temp = (t2 - t1)
print (list)
print('Sum_bubble= %f' % rez)
print('T_bubble= %uus' % temp)

Что же мы написали:

t1 = utime.ticks_us()

Мы фиксируем время начала измерения, в конце выполнения всего алгоритма мы фиксируем время окончания измерения, и высчитываем время которое ушло на выполнение данного алгоритма:

t2 = utime.ticks_us()
temp = (t2 - t1)

Простой цифровой фильтр на базе алгоритма сортировки выборкой на языке microPython используя микроконтроллер ESP8266.

import utime

list = [10.0, 10.2, 10.1, 11.0, 10.3, 9.99, 9.98, 10.1, 10.2, 9.97,10.0, 10.2, 10.1, 11.0, 10.3, 9.99, 9.98, 10.1, 10.2, 9.87]
print (list)

def selection_sort(num):

# Значение i соответствует кол-ву отсортированных значений
for i in range(len(num)):
# Исходно считаем наименьшим первый элемент
index = i
# Этот цикл перебирает несортированные элементы
for j in range(i + 1, len(num)):
if num[j] < num[index]:
index = j
# Самый маленький элемент меняем с первым в списке
num[i], num[index] = num[index], num[i]

t1 = utime.ticks_us()
selection_sort(list)
rez1=summa (list, 3)
t2 = utime.ticks_us()
temp1 = (t2 – t1)
#print(‘Sum_selection=%f’ % rez1)
#print(‘T_selection=%uus’ % temp1)

Простой цифровой фильтр на базе алгоритма сортировки вставками на языке microPython используя микроконтроллер ESP8266.

import utime

list = [10.0, 10.2, 10.1, 11.0, 10.3, 9.99, 9.98, 10.1, 10.2, 9.97,10.0, 10.2, 10.1, 11.0, 10.3, 9.99, 9.98, 10.1, 10.2, 9.87]
print (list)

def insertion_sort(num):
# Сортировку начинаем со второго элемента, т.к. считается, что первый элемент уже отсортирован
for i in range(1, len(num)):
insert = num[i]
# Сохраняем ссылку на индекс предыдущего элемента
j = i – 1
# Элементы отсортированного сегмента перемещаем вперёд, если они больше
# элемента для вставки
while j >= 0 and num[j] > insert:
num[j + 1] = num[j]
j -= 1
# Вставляем элемент
num[j + 1] = insert

t1 = utime.ticks_us()
insertion_sort(list)
rez2=summa (list, 3)
t2 = utime.ticks_us()
temp2 = (t2 – t1)
#print(‘Sum_selection=%f’ % rez2)
#print(‘T_selection=%uus’ % temp2)

import utime

list = [10.0, 10.2, 10.1, 11.0, 10.3, 9.99, 9.98, 10.1, 10.2, 9.97,10.0, 10.2, 10.1, 11.0, 10.3, 9.99, 9.98, 10.1, 10.2, 9.87]
print (list)

# Если левый потомок корня — допустимый индекс, а элемент больше,
# чем текущий наибольший, обновляем наибольший элемент
if left_child < heap_size and num[left_child] > num[largest]:
largest = left_child

# То же самое для правого потомка корня
if right_child < heap_size and num[right_child] > num[largest]:
largest = right_child

# Если наибольший элемент больше не корневой, они меняются местами
if largest != root_index:
num[root_index], num[largest] = num[largest], num[root_index]
# Heapify the new root element to ensure it’s the largest
heapify(num, heap_size, largest)

def heap_sort(num):
n = len(num)

# Создаём Max Heap из списка
# Второй аргумент означает остановку алгоритма перед элементом -1, т.е.
# перед первым элементом списка
# 3-й аргумент означает повторный проход по списку в обратном направлении,
# уменьшая счётчик i на 1
for i in range(n, -1, -1):
heapify(num, n, i)

# Перемещаем корень Max Heap в конец списка
for i in range(n – 1, 0, -1):
num[i], num[0] = num[0], num[i]
heapify(num, i, 0)

def merge(left_list, right_list):
sorted_list = []
left_list_index = right_list_index = 0

# Длина списков часто используется, поэтому создадим переменные для удобства
left_list_length, right_list_length = len(left_list), len(right_list)

for _ in range(left_list_length + right_list_length):
if left_list_index < left_list_length and right_list_index < right_list_length:
# Сравниваем первые элементы в начале каждого списка
# Если первый элемент левого подсписка меньше, добавляем его
# в отсортированный массив
if left_list[left_list_index] <= right_list[right_list_index]:
sorted_list.append(left_list[left_list_index])
left_list_index += 1
# Если первый элемент правого подсписка меньше, добавляем его
# в отсортированный массив
else:
sorted_list.append(right_list[right_list_index])
right_list_index += 1

# Если достигнут конец левого списка, элементы правого списка
# добавляем в конец результирующего списка
elif left_list_index == left_list_length:
sorted_list.append(right_list[right_list_index])
right_list_index += 1
# Если достигнут конец правого списка, элементы левого списка
# добавляем в отсортированный массив
elif right_list_index == right_list_length:
sorted_list.append(left_list[left_list_index])
left_list_index += 1

return sorted_list

def merge_sort(num):
# Возвращаем список, если он состоит из одного элемента
if len(num) <= 1:
return num

# Для того чтобы найти середину списка, используем деление без остатка
# Индексы должны быть integer
mid = len(num) // 2

# Сортируем и объединяем подсписки
left_list = merge_sort(num[:mid])
right_list = merge_sort(num[mid:])

# Объединяем отсортированные списки в результирующий
return merge(left_list, right_list)

#—
def partition(num, low, high):
# Выбираем средний элемент в качестве опорного
# Также возможен выбор первого, последнего
# или произвольного элементов в качестве опорного
pivot = num[(low + high) // 2]
i = low – 1
j = high + 1
while True:
i += 1
while num[i] < pivot:
i += 1

j -= 1
while num[j] > pivot:
j -= 1

if i >= j:
return j

# Если элемент с индексом i (слева от опорного) больше, чем
# элемент с индексом j (справа от опорного), меняем их местами
num[i], num[j] = num[j], num[i]

def quick_sort(num):
# Создадим вспомогательную функцию, которая вызывается рекурсивно
def _quick_sort(items, low, high):
if low < high:
# This is the index after the pivot, where our lists are split
split_index = partition(items, low, high)
_quick_sort(items, low, split_index)
_quick_sort(items, split_index + 1, high)

_quick_sort(num, 0, len(num) – 1)

#–
def summa (num, z):
for i in range(z):
list.pop(0)
list.pop()
sum=0
for i in range(len(num) – 1):
sum +=list[i]
sum /= (len(list)-1)
return sum

# Смотрим что вышло
t1 = utime.ticks_us()
bubble_sort(list)
#print(list)
rez=summa (list, 3)
#print(list)
t2 = utime.ticks_us()
temp = (t2 – t1)
#print(‘Sum_bublle=%f’ % rez)
#print(‘T_bublle=%uus’ % temp)

list = [10.0, 10.2, 10.1, 11.0, 10.3, 9.99, 9.98, 10.1, 10.2, 9.97,10.0, 10.2, 10.1, 11.0, 10.3, 9.99, 9.98, 10.1, 10.2, 9.87]
t1 = utime.ticks_us()
selection_sort(list)
rez1=summa (list, 3)
t2 = utime.ticks_us()
temp1 = (t2 – t1)
#print(‘Sum_selection=%f’ % rez1)
#print(‘T_selection=%uus’ % temp1)

list = [10.0, 10.2, 10.1, 11.0, 10.3, 9.99, 9.98, 10.1, 10.2, 9.97,10.0, 10.2, 10.1, 11.0, 10.3, 9.99, 9.98, 10.1, 10.2, 9.87]
t1 = utime.ticks_us()
insertion_sort(list)
rez2=summa (list, 3)
t2 = utime.ticks_us()
temp2 = (t2 – t1)
#print(‘Sum_selection=%f’ % rez2)
#print(‘T_selection=%uus’ % temp2)

list = [10.0, 10.2, 10.1, 11.0, 10.3, 9.99, 9.98, 10.1, 10.2, 9.97,10.0, 10.2, 10.1, 11.0, 10.3, 9.99, 9.98, 10.1, 10.2, 9.87]
t1 = utime.ticks_us()
heap_sort(list)
rez3=summa (list, 3)
t2 = utime.ticks_us()
temp3 = (t2 – t1)
#print(‘Sum_heap=%f’ % rez3)
#print(‘T_heap=%uus’ % temp3)

list = [10.0, 10.2, 10.1, 11.0, 10.3, 9.99, 9.98, 10.1, 10.2, 9.97,10.0, 10.2, 10.1, 11.0, 10.3, 9.99, 9.98, 10.1, 10.2, 9.87]
t1 = utime.ticks_us()
list=merge_sort(list)
rez4=summa (list, 3)
t2 = utime.ticks_us()
temp4 = (t2 – t1)
#print(‘Sum_merge=%f’ % rez4)
#print(‘T_merge=%uus’ % temp4)

list = [10.0, 10.2, 10.1, 11.0, 10.3, 9.99, 9.98, 10.1, 10.2, 9.97,10.0, 10.2, 10.1, 11.0, 10.3, 9.99, 9.98, 10.1, 10.2, 9.87]
t1 = utime.ticks_us()
quick_sort(list)
rez5=summa (list, 3)
t2 = utime.ticks_us()
temp5 = (t2 – t1)
#print(‘Sum_quick=%f’ % rez5)
#print(‘T_quick=%uus’ % temp5)

print(‘T_bublle=%uus T_selection=%uus T_insertion=%uus T_heap=%uus T_merge=%uus T_quick=%uus’ % (temp, temp1, temp2, temp3, temp4, temp5))
print(‘Sum_bublle=%f Sum_selection=%f Sum_insertion=%f Sum_heap=%f Sum_merge=%f Sum_quick=%f’ % (rez, rez1, rez2, rez3, rez4, rez5))

import utime
import machine, time

def summa (num, z):
for i in range(z):
list.pop(0)
list.pop()
sum=0
for i in range(len(num) — 1):
sum +=list[i]
sum /= (len(list) — 1)
return sum

def bubble_sort(num):
# Устанавливаем sw в True, для запуска цикл минимум один раз
sw = True
while sw:
sw = False
for i in range(len(num) — 1):
if num[i] > num[i + 1]:
# Меняем элементы
num[i], num[i + 1] = num[i + 1], num[i]
# Устанавливаем sw в True для следующей итерации
sw = True

def selection_sort(num):
# Значение i соответствует кол-ву отсортированных значений
for i in range(len(num)):
# Исходно считаем наименьшим первый элемент
index = i
# Этот цикл перебирает несортированные элементы
for j in range(i + 1, len(num)):
if num[j] < num[index]:
index = j
# Самый маленький элемент меняем с первым в списке
num[i], num[index] = num[index], num[i]

def insertion_sort(num):
# Сортировку начинаем со второго элемента, т.к. считается, что первый элемент уже отсортирован
for i in range(1, len(num)):
insert = num[i]
# Сохраняем ссылку на индекс предыдущего элемента
j = i — 1
# Элементы отсортированного сегмента перемещаем вперёд, если они больше
# элемента для вставки
while j >= 0 and num[j] > insert:
num[j + 1] = num[j]
j -= 1
# Вставляем элемент
num[j + 1] = insert

def heapify(num, heap_size, root_index):
# Индекс наибольшего элемента считаем корневым индексом
largest = root_index
left_child = (2 * root_index) + 1
right_child = (2 * root_index) + 2
# Если левый потомок корня — допустимый индекс, а элемент больше,
# чем текущий наибольший, обновляем наибольший элемент
if left_child < heap_size and num[left_child] > num[largest]:
largest = left_child
# То же самое для правого потомка корня
if right_child < heap_size and num[right_child] > num[largest]:
largest = right_child
# Если наибольший элемент больше не корневой, они меняются местами
if largest != root_index:
num[root_index], num[largest] = num[largest], num[root_index]
# Heapify the new root element to ensure it’s the largest
heapify(num, heap_size, largest)

def heap_sort(num):
n = len(num)
# Создаём Max Heap из списка
# Второй аргумент означает остановку алгоритма перед элементом -1, т.е.
# перед первым элементом списка
# 3-й аргумент означает повторный проход по списку в обратном направлении,
# уменьшая счётчик i на 1
for i in range(n, -1, -1):
heapify(num, n, i)
# Перемещаем корень Max Heap в конец списка
for i in range(n — 1, 0, -1):
num[i], num[0] = num[0], num[i]
heapify(num, i, 0)

def merge(left_list, right_list):
sorted_list = []
left_list_index = right_list_index = 0
# Длина списков часто используется, поэтому создадим переменные для удобства
left_list_length, right_list_length = len(left_list), len(right_list)
for _ in range(left_list_length + right_list_length):
if left_list_index < left_list_length and right_list_index < right_list_length:
# Сравниваем первые элементы в начале каждого списка
# Если первый элемент левого подсписка меньше, добавляем его
# в отсортированный массив
if left_list[left_list_index] <= right_list[right_list_index]:
sorted_list.append(left_list[left_list_index])
left_list_index += 1
# Если первый элемент правого подсписка меньше, добавляем его
# в отсортированный массив
else:
sorted_list.append(right_list[right_list_index])
right_list_index += 1
# Если достигнут конец левого списка, элементы правого списка
# добавляем в конец результирующего списка
elif left_list_index == left_list_length:
sorted_list.append(right_list[right_list_index])
right_list_index += 1
# Если достигнут конец правого списка, элементы левого списка
# добавляем в отсортированный массив
elif right_list_index == right_list_length:
sorted_list.append(left_list[left_list_index])
left_list_index += 1
return sorted_list

def merge_sort(num):
# Возвращаем список, если он состоит из одного элемента
if len(num) <= 1:
return num
# Для того чтобы найти середину списка, используем деление без остатка
# Индексы должны быть integer
mid = len(num) // 2
# Сортируем и объединяем подсписки
left_list = merge_sort(num[:mid])
right_list = merge_sort(num[mid:])
# Объединяем отсортированные списки в результирующий
return merge(left_list, right_list)

def partition(numbers, low, high):
# Выбираем средний элемент в качестве опорного
# Также возможен выбор первого, последнего
# или произвольного элементов в качестве опорного
pivot = numbers[high]
item = low — 1
for i in range(low, high):
if numbers[i] <= pivot:
item = item + 1
(numbers[item], numbers[i]) = (numbers[i], numbers[item])
(numbers[item + 1], numbers[high]) = (numbers[high], numbers[item + 1])
return item + 1

def quick_sort(numbers, low, high):
if low < high:
pivot = partition(numbers, low, high)
quick_sort(numbers, low, pivot — 1)
quick_sort(numbers, pivot + 1, high)

# Смотрим что вышло
time.sleep_ms(100)
list = [10.0, 10.2, 10.1, 11.0, 10.3, 9.99, 9.98, 10.1, 10.2, 9.97,10.0, 10.2, 10.1, 11.0, 10.3, 9.99, 9.98, 10.1, 10.2, 9.87]
t1 = utime.ticks_us()
bubble_sort(list)
rez=summa (list, 3)
t2 = utime.ticks_us()
temp = (t2 — t1)

time.sleep_ms(100)
list = [10.0, 10.2, 10.1, 11.0, 10.3, 9.99, 9.98, 10.1, 10.2, 9.97,10.0, 10.2, 10.1, 11.0, 10.3, 9.99, 9.98, 10.1, 10.2, 9.87]
t1 = utime.ticks_us()
selection_sort(list)
rez1=summa (list, 3)
t2 = utime.ticks_us()
temp1 = (t2 — t1)

time.sleep_ms(100)
list = [10.0, 10.2, 10.1, 11.0, 10.3, 9.99, 9.98, 10.1, 10.2, 9.97,10.0, 10.2, 10.1, 11.0, 10.3, 9.99, 9.98, 10.1, 10.2, 9.87]
t1 = utime.ticks_us()
insertion_sort(list)
rez2=summa (list, 3)
t2 = utime.ticks_us()
temp2 = (t2 — t1)

time.sleep_ms(100)
list = [10.0, 10.2, 10.1, 11.0, 10.3, 9.99, 9.98, 10.1, 10.2, 9.97,10.0, 10.2, 10.1, 11.0, 10.3, 9.99, 9.98, 10.1, 10.2, 9.87]
t1 = utime.ticks_us()
heap_sort(list)
rez3=summa (list, 3)
t2 = utime.ticks_us()
temp3 = (t2 — t1)

time.sleep_ms(100)
list = [10.0, 10.2, 10.1, 11.0, 10.3, 9.99, 9.98, 10.1, 10.2, 9.97,10.0, 10.2, 10.1, 11.0, 10.3, 9.99, 9.98, 10.1, 10.2, 9.87]
t1 = utime.ticks_us()
list=merge_sort(list)
rez4=summa (list, 3)
t2 = utime.ticks_us()
temp4 = (t2 — t1)

time.sleep_ms(100)
list = [10.0, 10.2, 10.1, 11.0, 10.3, 9.99, 9.98, 10.1, 10.2, 9.97,10.0, 10.2, 10.1, 11.0, 10.3, 9.99, 9.98, 10.1, 10.2, 9.87]
t1 = utime.ticks_us()
quick_sort(list,0, len(list)-1)
rez5=summa (list, 3)
t2 = utime.ticks_us()
temp5 = (t2 — t1)

time.sleep_ms(100)
print(‘T_bublle=%uus; T_selection=%uus; T_insertion=%uus; T_heap=%uus; T_merge=%uus; T_quick=%uus’ % (temp, temp1, temp2, temp3, temp4, temp5))
time.sleep_ms(100)
print(‘Sum_bublle=%f; Sum_selection=%f; Sum_insertion=%f; Sum_heap=%f; Sum_merge=%f; Sum_quick=%f’ % (rez, rez1, rez2, rez3, rez4, rez5))