Основы фреймворка Combine для ML в Swift

Привет, Хабр!

Combine — это фреймворк, представленный Apple в 2019 году, предназначенный для работы с асинхронными потоками данных на платформе Swift. Основная задача Combine заключается в том, чтобы упростить и унифицировать обработку асинхронных событий, таких как сетевые запросы, таймеры, уведомления и пользовательские действия.

Фреймворк является частью экосистемы Swift и доступен начиная с iOS 13, macOS 10.15, watchOS 6.0 и tvOS 13. Если вы работаете с более ранними версиями, обновление Xcode до версии 11 или выше автоматом предоставит вам доступ к Combine.

В этой статье мы рассмотрим основы этого замечательного фреймворка.

После установки Xcode вы сможете использовать Combine в своих проектах, импортировав его следующим образом:

import Combine

Создание и комбинирование потоков данных с помощью Combine

В Combine всё начинается с Publisher — сущности, которая эмитирует значения с течением времени. Это может быть что угодно: массив чисел, результат сетевого запроса, или события пользовательского интерфейса.

Простейший пример Publisher — это массив, преобразованный в Publisher с помощью .publisher:

import Combine  let numbers = [1, 2, 3, 4, 5].publisher  numbers.sink { value in     print("Received value: \(value)") }

Массив numbers превращается в Publisher, который выдает каждое число последовательно. Метод sink — это Subscriber, который «подписывается» на Publisher и выводит каждый полученный элемент.

Если хочется создать собственный Publisher, можно использовать PassthroughSubject или CurrentValueSubject:

import Combine  let subject = PassthroughSubject<String, Never>()  subject.sink { value in     print("Received value: \(value)") }  subject.send("Hello") subject.send("Combine")

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

Когда дело доходит до объединения данных из разных источников, есть несколько операторов: merge, combineLatest и zip.

• merge объединяет несколько Publisher-ов в один, эмитируя значения по мере их поступления:

import Combine  let publisher1 = PassthroughSubject<Int, Never>() let publisher2 = PassthroughSubject<Int, Never>()  let merged = publisher1.merge(with: publisher2)  merged.sink { value in     print("Merged value: \(value)") }  publisher1.send(1) publisher2.send(2) publisher1.send(3)

Результат работы merge будет последовательность: 1, 2, 3, т.е. значения от обоих Publisher-ов объединяются в один поток.

• combineLatest выдает пары значений от обоих Publisher-ов, когда оба источника данных предоставили хотя бы одно значение:

import Combine  let publisherA = PassthroughSubject<String, Never>() let publisherB = PassthroughSubject<String, Never>()  let combined = publisherA.combineLatest(publisherB)  combined.sink { value in     print("Combined value: \(value)") }  publisherA.send("Hello") publisherB.send("World") // Вывод: "Combined value: ("Hello", "World")" publisherA.send("Hi") // Вывод: "Combined value: ("Hi", "World")"

combineLatest ждет данные от обоих Publisher-ов и затем выдает их в виде пары, каждый раз, когда поступает новое значение от одного из них.

• zip работает подобно combineLatest, но выдает пары только тогда, когда обе стороны готовы передать по одному значению:

import Combine  let pub1 = PassthroughSubject<Int, Never>() let pub2 = PassthroughSubject<String, Never>()  let zipped = pub1.zip(pub2)  zipped.sink { value in     print("Zipped value: \(value)") }  pub1.send(1) pub2.send("One") // Вывод: "Zipped value: (1, "One")" pub1.send(2) pub2.send("Two") // Вывод: "Zipped value: (2, "Two")"

Этот оператор хорош, когда необходимо синхронизировать данные из разных источников.

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

import Combine  // Создаем два Publisher-а: один для ввода пользователя, другой для предсказаний let userInput = PassthroughSubject<String, Never>() let prediction = PassthroughSubject<String, Never>()  // Комбинируем данные, используя combineLatest let combinedData = userInput.combineLatest(prediction)  combinedData.sink { user, predicted in     print("User input: \(user), Model prediction: \(predicted)") }  // Эмитируем значения userInput.send("User says: Hello") prediction.send("Model predicts: Hi") // Вывод: "User input: User says: Hello, Model prediction: Hi"

combineLatest позволяет связать ввод пользователя с результатами работы ML-модели.

Трансформация данных в потоках

Оператор map является основным инструментом для трансформации данных в Combine. Он позволяет взять данные, выдаваемые Publisher-ом, и преобразовать их в новую форму, изменяя каждый элемент потока.

Представим, что есть поток чисел, и хочется умножить каждое из них на 2. Это легко реализуется с помощью map:

import Combine  let numbers = [1, 2, 3, 4, 5].publisher  let multipliedNumbers = numbers.map { $0 * 2 }  multipliedNumbers.sink { value in     print("Transformed value: \(value)") }

Результатом выполнения этого кода будет вывод:

Transformed value: 2 Transformed value: 4 Transformed value: 6 Transformed value: 8 Transformed value: 10

Оператор map подходит для предобработки данных перед их передачей в ML модель. Например, если нужно нормализовать входные данные перед подачей их в модель:

let rawInputs = [0.5, 0.75, 1.0, 1.25].publisher  let normalizedInputs = rawInputs.map { $0 / 2.0 }  normalizedInputs.sink { value in     print("Normalized value: \(value)") }

Оператор filter позволяет отфильтровывать значения, пропуская только те, которые удовлетворяют определенным условиям.

Рассмотрим пример, где нужно оставить в потоке только четные числа:

let numbers = [1, 2, 3, 4, 5, 6].publisher  let evenNumbers = numbers.filter { $0 % 2 == 0 }  evenNumbers.sink { value in     print("Filtered value: \(value)") }

Вывод:

Filtered value: 2 Filtered value: 4 Filtered value: 6

В контексте ML, filter может быть полезен для удаления выбросов или данных, не соответствующих определенным критериям. Например, если нужно исключить из входных данных все значения, превышающие допустимый предел:

let rawInputs = [0.5, 2.5, 1.0, 3.5].publisher  let validInputs = rawInputs.filter { $0 <= 2.0 }  validInputs.sink { value in     print("Valid input: \(value)") }

flatMap — это оператор, который раскрывает новый Publisher на основе каждого значения, выдаваемое исходным Publisher-ом.

Рассмотрим пример с сетевым запросом, который возвращает данные на основе пользовательского ввода:

import Foundation  let userInput = PassthroughSubject<String, Never>() let searchResults = userInput.flatMap { query in     URLSession.shared.dataTaskPublisher(for: URL(string: "https://api.example.com/search?q=\(query)")!)         .map { $0.data }         .decode(type: [String].self, decoder: JSONDecoder())         .catch { _ in Just([]) }  // Обрабатываем возможные ошибки }  searchResults.sink { results in     print("Search results: \(results)") }  userInput.send("apple")

flatMap используется для запуска сетевого запроса на основе пользовательского ввода. Каждый раз, когда пользователь вводит новый запрос, flatMap раскрывает новый Publisher, который затем обрабатывает результаты и возвращает их.

Примеры использования

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

let rawInputs = [1.5, 0.9, 2.5, 3.7, 1.8].publisher  let processedInputs = rawInputs     .filter { $0 < 3.0 } // Оставляем только допустимые значения     .map { $0 / 3.0 } // Нормализуем данные  processedInputs.sink { value in     print("Processed input: \(value)") }

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

let modelPredictions = [0.2, 0.8, 1.0, 0.4].publisher  let filteredPredictions = modelPredictions     .filter { $0 > 0.5 } // Оставляем только уверенные предсказания     .map { $0 * 100.0 } // Преобразуем в проценты  filteredPredictions.sink { value in     print("Filtered prediction: \(value)%") }

Управление ошибками и отложенная обработка данных

В Combine ошибки обрабатываются с помощью операторов catch и retry.

Оператор catch позволяет перехватывать ошибки, возникающие в потоке данных, и предоставлять альтернативный Publisher для продолжения работы.

Пример:

import Combine  struct MyError: Error {}  let faultyPublisher = Fail<Int, MyError>(error: MyError())  let safePublisher = faultyPublisher     .catch { _ in Just(0) } // В случае ошибки возвращаем значение 0     .sink { value in         print("Received value: \(value)")     }  // Вывод: "Received value: 0"

ИспользуемFail для эмуляции ошибки, а затем обработали ее с помощью catch, который возвращает безопасное значение.

Оператор retry позволяет автоматически повторять выполнение потока данных, если в нем произошла ошибка.

Пример:

import Combine  var attemptCount = 0  let retryingPublisher = Deferred {     attemptCount += 1     return attemptCount < 3 ? Fail(error: URLError(.badServerResponse)) : Just("Success") } .retry(3) // Повторяем до трех раз .sink(     receiveCompletion: { completion in         switch completion {         case .finished:             print("Completed successfully")         case .failure(let error):             print("Failed with error: \(error)")         }     },     receiveValue: { value in         print("Received value: \(value)")     } )  // Вывод: "Received value: Success"

Здесь retry пытается выполнить поток данных до трех раз. На третьей попытке поток завершается успешно.

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

debounce — это оператор, который откладывает обработку событий до тех пор, пока не пройдет определенное время без поступления новых данных.

Пример:

import Combine import Foundation  let searchTextPublisher = PassthroughSubject<String, Never>()  let debouncedSearch = searchTextPublisher     .debounce(for: .milliseconds(500), scheduler: RunLoop.main)     .sink { value in         print("Search query: \(value)")     }  searchTextPublisher.send("S") searchTextPublisher.send("Sw") searchTextPublisher.send("Swi") searchTextPublisher.send("Swif") searchTextPublisher.send("Swift") // Через 500 мс после последнего ввода: "Search query: Swift"

Здесь debounce откладывает отправку данных до тех пор, пока пользователь не закончит ввод текста, избегая излишних запросов.

throttle позволяет пропускать события только через определенные интервалы времени, ограничивая частоту их обработки.

Пример:

import Combine import Foundation  let eventPublisher = PassthroughSubject<Void, Never>()  let throttledEvents = eventPublisher     .throttle(for: .seconds(1), scheduler: RunLoop.main, latest: true)     .sink {         print("Event received")     }  for _ in 1...5 {     eventPublisher.send()     Thread.sleep(forTimeInterval: 0.3) // Эмуляция частого срабатывания события }  // Вывод: "Event received" будет напечатано только один раз в секунду

throttle в этом примере пропускает только одно событие в секунду, игнорируя остальные, что позволяет эффективно управлять нагрузкой.

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

import Combine import Foundation  enum SensorError: Error {     case sensorFailure }  let sensorDataPublisher = PassthroughSubject<Int, SensorError>() let serverDataPublisher = PassthroughSubject<Int, URLError>()  let combinedPublisher = Publishers.CombineLatest(sensorDataPublisher, serverDataPublisher)     .retry(2) // Пытаемся заново в случае ошибки     .debounce(for: .milliseconds(300), scheduler: RunLoop.main) // Избегаем частой обработки данных     .catch { _ in Just((0, 0)) } // В случае ошибки возвращаем безопасное значение     .sink { sensorData, serverData in         print("Sensor: \(sensorData), Server: \(serverData)")     }  sensorDataPublisher.send(100) serverDataPublisher.send(200)  // Эмулируем ошибку сенсора sensorDataPublisher.send(completion: .failure(.sensorFailure))  serverDataPublisher.send(300)

Используем CombineLatest для объединения данных от сенсоров и сервера, обрабатываем ошибки с помощью retry, избегаем частой обработки данных с помощью debounce, и, наконец, безопасно обрабатываем ошибки с помощью catch.

Подробнее с библиотекой Combine можно ознакомиться здесь.

---

Актуальна тема подготовки данных в Pandas? Приходите на открытый урок 19 августа, который пройдет в рамках специализации Machine Learning.

В результате вебинара вы узнаете о методах подготовки данных и научитесь чистить данные при помощи библиотеки Pandas. Записывайтесь на урок по ссылке.