C# и IoT: разработка Edge приложений с .NET и Azure IoT

namespace EnvironmentSensorModule
{
    using System;
    using System.IO;
    using System.Runtime.InteropServices;
    using System.Runtime.Loader;
    using System.Security.Cryptography.X509Certificates;
    using System.Text;
    using System.Threading;
    using System.Threading.Tasks;
    using Microsoft.Azure.Devices.Client;
    using Microsoft.Azure.Devices.Client.Transport.Mqtt;
    using Newtonsoft.Json;
 
    class Program
    {
        static int counter;
        static readonly Random Rnd = new Random();
 
        static void Main(string[] args)
        {
            Init().Wait();
 
            // Ждём, пока модуль не будет остановлен
            WaitForExit().Wait();
        }
 
        /// <summary>
        /// Инициализация модуля IoT Edge
        /// </summary>
        static async Task Init()
        {
            var moduleClient = await CreateModuleClientAsync();
            await moduleClient.OpenAsync();
            Console.WriteLine("Модуль IoT Edge запущен!");
 
            // Запускаем отправку телеметрии
            var cts = new CancellationTokenSource();
            SendTelemetryAsync(moduleClient, cts.Token);
 
            // Устанавливаем обработчики входящих сообщений
            await moduleClient.SetInputMessageHandlerAsync("control", ControlMessageHandler, moduleClient);
        }
 
        /// <summary>
        /// Создание клиента модуля IoT Edge
        /// </summary>
        static async Task<ModuleClient> CreateModuleClientAsync()
        {
            var transportSettings = new MqttTransportSettings(TransportType.Mqtt_Tcp_Only);
            // Настраиваем MQTT как транспорт
            ITransportSettings[] settings = { transportSettings };
 
            // Создаём клиента из переменных окружения, предоставляемых средой выполнения IoT Edge
            ModuleClient moduleClient = await ModuleClient.CreateFromEnvironmentAsync(settings);
            
            // Настраиваем тайм-ауты соединения
            moduleClient.SetConnectionStatusChangesHandler((status, reason) => 
            {
                Console.WriteLine($"Статус соединения: {status}, причина: {reason}");
            });
 
            return moduleClient;
        }
 
        /// <summary>
        /// Отправка телеметрии с датчиков
        /// </summary>
        static async void SendTelemetryAsync(ModuleClient moduleClient, CancellationToken cancellationToken)
        {
            while (!cancellationToken.IsCancellationRequested)
            {
                try
                {
                    // Симулируем чтение данных с датчика
                    var temperature = 20 + Rnd.NextDouble() * 10;
                    var humidity = 30 + Rnd.NextDouble() * 50;
                    var messageBody = new
                    {
                        temperature = Math.Round(temperature, 2),
                        humidity = Math.Round(humidity, 2),
                        timestamp = DateTime.UtcNow
                    };
 
                    string messageString = JsonConvert.SerializeObject(messageBody);
                    var message = new Message(Encoding.UTF8.GetBytes(messageString));
                    
                    // Добавляем свойства сообщения для маршрутизации
                    message.Properties.Add("sensorType", "environment");
                    message.Properties.Add("deviceLocation", "building1");
                    
                    // Отправляем сообщение
                    await moduleClient.SendEventAsync("output1", message);
                    Console.WriteLine($"Отправлено: {messageString}");
                    
                    await Task.Delay(5000, cancellationToken);
                }
                catch (Exception ex)
                {
                    Console.WriteLine($"Ошибка при отправке телеметрии: {ex.Message}");
                }
            }
        }
 
        /// <summary>
        /// Обработчик входящих управляющих сообщений
        /// </summary>
        static async Task<MessageResponse> ControlMessageHandler(Message message, object userContext)
        {
            var moduleClient = userContext as ModuleClient;
            var messageBytes = message.GetBytes();
            var messageString = Encoding.UTF8.GetString(messageBytes);
            
            Console.WriteLine($"Получено сообщение: {messageString}");
            
            // Здесь может быть логика обработки команд
            // Например, изменение частоты отправки телеметрии
            
            return MessageResponse.Completed;
        }
 
        /// <summary>
        /// Ожидание остановки модуля
        /// </summary>
        static Task WaitForExit()
        {
            var resetEvent = new ManualResetEvent(false);
            AssemblyLoadContext.Default.Unloading += (ctx) => resetEvent.Set();
            Console.CancelKeyPress += (sender, args) => resetEvent.Set();
            return Task.Run(() => resetEvent.WaitOne());
        }
    }
}

using System.Device.I2c;
using Iot.Device.Bmxx80;
using Iot.Device.Bmxx80.PowerMode;
 
// Параметры подключения I2C
var i2cSettings = new I2cConnectionSettings(1, Bme280.DefaultI2cAddress);
using var i2cDevice = I2cDevice.Create(i2cSettings);
using var bme280 = new Bme280(i2cDevice);
 
// Настройка датчика
bme280.SetPowerMode(Bmx280PowerMode.Normal);
 
// Чтение данных
var readResult = await bme280.ReadAsync();
if (readResult.HasValue)
{
    var temperature = readResult.Value.Temperature;
    var humidity = readResult.Value.Humidity;
    var pressure = readResult.Value.Pressure;
    
    Console.WriteLine($"Температура: {temperature.DegreesCelsius:F1}°C");
    Console.WriteLine($"Влажность: {humidity.Percent:F1}%");
    Console.WriteLine($"Давление: {pressure.Hectopascals:F1} гПа");
}

using System.Numerics;
 
// Упрощённая реализация БПФ для анализа вибраций
private double[] PerformFFT(double[] timeData)
{
    int n = timeData.Length;
    Complex[] spectrum = new Complex[n];
    
    // Подготовка данных (применение оконной функции Хэмминга)
    for (int i = 0; i < n; i++)
    {
        double window = 0.54 - 0.46 * Math.Cos(2 * Math.PI * i / (n - 1));
        spectrum[i] = new Complex(timeData[i] * window, 0);
    }
    
    // Выполнение БПФ
    int bits = (int)Math.Log(n, 2);
    for (int j = 0; j < bits; j++)
    {
        for (int i = 0; i < n; i++)
        {
            int swapPos = i ^ (1 << j);
            if (i < swapPos)
            {
                var temp = spectrum[i];
                spectrum[i] = spectrum[swapPos];
                spectrum[swapPos] = temp;
            }
        }
    }
    
    for (int i = 1; i <= bits; i++)
    {
        int m = 1 << i;
        int m2 = m >> 1;
        Complex w = new Complex(1, 0);
        Complex wm = new Complex(Math.Cos(Math.PI / m2), -Math.Sin(Math.PI / m2));
        
        for (int j = 0; j < m2; j++)
        {
            for (int k = j; k < n; k += m)
            {
                Complex t = w * spectrum[k + m2];
                Complex u = spectrum[k];
                spectrum[k] = u + t;
                spectrum[k + m2] = u - t;
            }
            w *= wm;
        }
    }
    
    // Вычисление амплитудного спектра
    double[] magnitudes = new double[n / 2];
    for (int i = 0; i < n / 2; i++)
    {
        magnitudes[i] = Math.Sqrt(spectrum[i].Real * spectrum[i].Real + 
                                  spectrum[i].Imaginary * spectrum[i].Imaginary);
    }
    
    return magnitudes;
}

// Подготовка сообщения
var messageBody = new { processedData = processedValue, status = "OK" };
string messageString = JsonConvert.SerializeObject(messageBody);
var message = new Message(Encoding.UTF8.GetBytes(messageString));
 
// Добавление метаданных для маршрутизации
message.Properties.Add("dataType", "processedSensor");
message.Properties.Add("processingLevel", "L1");
 
// Отправка на указанный выходной порт
await moduleClient.SendEventAsync("output1", message);

await moduleClient.SetInputMessageHandlerAsync("input1", ProcessInputMessage, null);
 
// ...
 
private static async Task<MessageResponse> ProcessInputMessage(Message message, object userContext)
{
    try
    {
        var messageBytes = message.GetBytes();
        var messageString = Encoding.UTF8.GetString(messageBytes);
        
        Console.WriteLine($"Получено сообщение: {messageString}");
        
        // Десериализация и обработка данных
        var sensorData = JsonConvert.DeserializeObject<SensorData>(messageString);
        var processedData = ProcessData(sensorData);
        
        // Создание и отправка нового сообщения с обработанными данными
        await SendProcessedDataAsync(processedData);
        
        // Подтверждение обработки сообщения
        return MessageResponse.Completed;
    }
    catch (Exception ex)
    {
        Console.WriteLine($"Ошибка при обработке входящего сообщения: {ex.Message}");
        return MessageResponse.Abandoned;
    }
}

FROM mcr.microsoft.com/dotnet/sdk:5.0 AS build-env
WORKDIR /app
 
COPY *.csproj ./
RUN dotnet restore
 
COPY . ./
RUN dotnet publish -c Release -o out
 
FROM mcr.microsoft.com/dotnet/runtime:5.0
WORKDIR /app
COPY --from=build-env /app/out ./
 
RUN useradd -ms /bin/bash moduleuser
USER moduleuser
ENTRYPOINT ["dotnet", "EnvironmentSensorModule.dll"]

// Получение Twin для текущего модуля
var moduleTwin = await moduleClient.GetTwinAsync();
var desiredProperties = moduleTwin.Properties.Desired;
 
// Чтение настроек
int samplingRate = desiredProperties["samplingRate"];
string sensorMode = desiredProperties["sensorMode"];
 
// Установка обработчика обновлений Twin
await moduleClient.SetDesiredPropertyUpdateCallbackAsync(OnDesiredPropertiesChanged, null);
 
// ...
 
// Обработчик изменений конфигурации
private static async Task OnDesiredPropertiesChanged(TwinCollection desiredProperties, object userContext)
{
    try
    {
        if (desiredProperties.Contains("samplingRate"))
        {
            int samplingRate = desiredProperties["samplingRate"];
            _samplingRateMs = samplingRate;
            Console.WriteLine($"Обновлена частота опроса: {samplingRate} мс");
        }
        
        // Сохраняем факт обновления в reported properties
        var reportedProperties = new TwinCollection();
        reportedProperties["lastConfigUpdate"] = DateTime.UtcNow.ToString("o");
        reportedProperties["currentSamplingRate"] = _samplingRateMs;
        
        await _moduleClient.UpdateReportedPropertiesAsync(reportedProperties);
    }
    catch (Exception ex)
    {
        Console.WriteLine($"Ошибка при обновлении настроек: {ex.Message}");
    }
}

// Запрос секрета из хранилища Edge
var workloadClient = new WorkloadClient(
    Environment.GetEnvironmentVariable("IOTEDGE_WORKLOADURI"), 
    Environment.GetEnvironmentVariable("IOTEDGE_MODULEGENERATIONID"), 
    Environment.GetEnvironmentVariable("IOTEDGE_MODULEID"));
 
var apiKey = await workloadClient.GetSecretAsync(
    new SecretRequest("externalAPIKeys", "locationServiceKey"));
 
// Теперь можно использовать apiKey для запросов к внешнему сервису

private static async Task ModuleStartupAsync(ModuleClient moduleClient)
{
    try
    {
        // Инициализация компонентов
        _storage = new LocalStorage("data");
        await _storage.InitializeAsync();
        
        // Загрузка и применение последней известной конфигурации
        var config = await _storage.LoadConfigAsync();
        if (config != null)
        {
            // Применить конфигурацию
            ApplyConfig(config);
        }
        
        // Синхронизация состояния с облаком
        var twin = await moduleClient.GetTwinAsync();
        await ProcessDesiredPropertiesAsync(twin.Properties.Desired);
        
        // Отчёт о успешном запуске
        var reportedProperties = new TwinCollection();
        reportedProperties["status"] = "running";
        reportedProperties["startTime"] = DateTime.UtcNow.ToString("o");
        reportedProperties["version"] = Assembly.GetExecutingAssembly().GetName().Version.ToString();
        
        await moduleClient.UpdateReportedPropertiesAsync(reportedProperties);
    }
    catch (Exception ex)
    {
        Console.WriteLine($"Ошибка при инициализации модуля: {ex}");
        // В реальном проекте здесь должна быть более продвинутая обработка ошибок
    }
}
 
private static async Task ModuleShutdownAsync()
{
    try
    {
        // Корректное освобождение ресурсов
        if (_storage != null)
        {
            await _storage.FlushAsync();
            _storage.Dispose();
        }
        
        // Остановка фоновых задач
        _cancellationTokenSource.Cancel();
        
        // Ожидание завершения всех задач
        await Task.WhenAll(_activeTasks.ToArray());
    }
    catch (Exception ex)
    {
        Console.WriteLine($"Ошибка при завершении работы модуля: {ex}");
    }
}

public class MessageBuffer
{
    private readonly ConcurrentQueue<Message> _messageQueue = new ConcurrentQueue<Message>();
    private readonly int _maxBufferSize;
    private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
    
    public MessageBuffer(int maxBufferSize = 1000)
    {
        _maxBufferSize = maxBufferSize;
    }
    
    public async Task EnqueueAsync(Message message)
    {
        await _semaphore.WaitAsync();
        try
        {
            // Если буфер переполнен, удаляем старые сообщения
            while (_messageQueue.Count >= _maxBufferSize)
            {
                _messageQueue.TryDequeue(out _);
            }
            
            _messageQueue.Enqueue(message);
        }
        finally
        {
            _semaphore.Release();
        }
    }
    
    public async Task<IEnumerable<Message>> DequeueAllAsync()
    {
        await _semaphore.WaitAsync();
        try
        {
            var messages = _messageQueue.ToArray();
            _messageQueue.Clear();
            return messages;
        }
        finally
        {
            _semaphore.Release();
        }
    }
    
    public int Count => _messageQueue.Count;
}

private static async Task SendMessagesWithRetryAsync(ModuleClient moduleClient, CancellationToken cancellationToken)
{
    while (!cancellationToken.IsCancellationRequested)
    {
        try
        {
            // Попытка отправить буферизованные сообщения
            if (_messageBuffer.Count > 0)
            {
                var messages = await _messageBuffer.DequeueAllAsync();
                foreach (var message in messages)
                {
                    await moduleClient.SendEventAsync("output1", message);
                    Console.WriteLine($"Отправлено буферизованное сообщение: {Encoding.UTF8.GetString(message.GetBytes())}");
                }
            }
            
            // Чтение новых данных с датчика
            var sensorData = await ReadSensorDataAsync();
            
            // Отправка новых данных
            var messageString = JsonConvert.SerializeObject(sensorData);
            var message = new Message(Encoding.UTF8.GetBytes(messageString));
            
            await moduleClient.SendEventAsync("output1", message);
            Console.WriteLine($"Отправлено: {messageString}");
        }
        catch (Exception ex)
        {
            // При ошибке отправки буферизуем сообщение
            var sensorData = await ReadSensorDataAsync();
            var messageString = JsonConvert.SerializeObject(sensorData);
            var message = new Message(Encoding.UTF8.GetBytes(messageString));
            
            await _messageBuffer.EnqueueAsync(message);
            Console.WriteLine($"Соединение потеряно. Сообщение буферизовано: {messageString}");
            
            // Делаем паузу перед следущей попыткой
            await Task.Delay(TimeSpan.FromSeconds(30), cancellationToken);
        }
        
        await Task.Delay(TimeSpan.FromSeconds(_telemetryIntervalSeconds), cancellationToken);
    }
}

public class PlantHealthPredictor
{
    private readonly PredictionEngine<SensorData, PlantHealthPrediction> _predictionEngine;
    
    public PlantHealthPredictor(string modelPath)
    {
        // Загрузка предварительно обученной модели
        var mlContext = new MLContext();
        var model = mlContext.Model.Load(modelPath, out _);
        
        // Создание движка предсказаний
        _predictionEngine = mlContext.Model.CreatePredictionEngine<SensorData, PlantHealthPrediction>(model);
    }
    
    public PlantHealthPrediction PredictHealth(float soilMoisture, float temperature, float lightLevel)
    {
        var data = new SensorData
        {
            SoilMoisture = soilMoisture,
            Temperature = temperature,
            LightLevel = lightLevel
        };
        
        return _predictionEngine.Predict(data);
    }
}
 
// Классы данных для модели
public class SensorData
{
    public float SoilMoisture { get; set; }
    public float Temperature { get; set; }
    public float LightLevel { get; set; }
}
 
public class PlantHealthPrediction
{
    [ColumnName("PredictedLabel")]
    public string Status { get; set; }
    
    [ColumnName("Score")]
    public float[] Probabilities { get; set; }
}

// Инициализация предиктора при старте модуля
var predictor = new PlantHealthPredictor("model/plant_health_model.zip");
 
// В цикле обработки данных
while (!cancellationToken.IsCancellationRequested)
{
    // Получение данных с датчиков
    var soilMoisture = await _soilMoistureReader.ReadAsync();
    var temperature = await _temperatureReader.ReadAsync();
    var lightLevel = await _lightMeter.ReadAsync();
    
    // Локальное предсказание
    var prediction = predictor.PredictHealth(soilMoisture, temperature, lightLevel);
    
    // Принятие решения на основе предсказания
    if (prediction.Status == "Dehydration" && prediction.Probabilities[0] > 0.85)
    {
        // Автоматическое включение полива
        await _irrigationController.StartIrrigationAsync(duration: TimeSpan.FromMinutes(5));
        
        // Отправка уведомления в облако
        await SendIrrigationNotificationAsync(moduleClient, soilMoisture, prediction);
    }
    
    await Task.Delay(TimeSpan.FromMinutes(15), cancellationToken);
}

public class LocalStateManager
{
    private readonly string _stateFilePath;
    private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
    
    public LocalStateManager(string stateFilePath)
    {
        _stateFilePath = stateFilePath;
        Directory.CreateDirectory(Path.GetDirectoryName(stateFilePath));
    }
    
    public async Task<T> LoadStateAsync<T>() where T : new()
    {
        await _semaphore.WaitAsync();
        try
        {
            if (!File.Exists(_stateFilePath))
                return new T();
                
            var json = await File.ReadAllTextAsync(_stateFilePath);
            return JsonConvert.DeserializeObject<T>(json) ?? new T();
        }
        catch (Exception ex)
        {
            Console.WriteLine($"Ошибка при загрузке состояния: {ex.Message}");
            return new T();
        }
        finally
        {
            _semaphore.Release();
        }
    }
    
    public async Task SaveStateAsync<T>(T state)
    {
        await _semaphore.WaitAsync();
        try
        {
            var json = JsonConvert.SerializeObject(state);
            await File.WriteAllTextAsync(_stateFilePath, json);
        }
        catch (Exception ex)
        {
            Console.WriteLine($"Ошибка при сохранении состояния: {ex.Message}");
        }
        finally
        {
            _semaphore.Release();
        }
    }
}

// В классе модуля
private static int _errorCount = 0;
private static readonly int _maxErrorsBeforeRestart = 5;
 
private static void IncrementErrorCount()
{
    Interlocked.Increment(ref _errorCount);
    if (_errorCount >= _maxErrorsBeforeRestart)
    {
        Console.WriteLine("Достигнут порог ошибок. Перезапуск модуля...");
        // В реальном сценарии здесь было бы корректное завершение
        Environment.Exit(1); // Docker перезапустит контейнер
    }
}

public class RealTimeProcessingPipeline
{
    private readonly BlockingCollection<SensorReading> _readingQueue = new(100);
    private readonly CancellationTokenSource _cts = new();
    private Task[] _processingTasks;
    
    public void Start()
    {
        _processingTasks = new[]
        {
            Task.Run(() => AcquisitionStage(_cts.Token)),
            Task.Run(() => FilteringStage(_cts.Token)),
            Task.Run(() => AnalysisStage(_cts.Token)),
            Task.Run(() => ActionStage(_cts.Token))
        };
        
        Console.WriteLine("Конвейер обработки запущен");
    }
    
    private async Task AcquisitionStage(CancellationToken token)
    {
        while (!token.IsCancellationRequested)
        {
            try
            {
                // Считывание с датчика с фиксированной частотой
                var reading = await _sensorReader.GetReadingAsync();
                reading.Timestamp = DateTime.UtcNow;
                
                _readingQueue.Add(reading, token);
            }
            catch (Exception ex) when (!(ex is OperationCanceledException))
            {
                Console.WriteLine($"Ошибка на этапе сбора данных: {ex.Message}");
            }
            
            await Task.Delay(10, token); // 100Hz выборка
        }
    }
    
    // Остальные этапы конвейера...
}

public class MessagePool
{
    private readonly ConcurrentBag<Message> _pool = new();
    private readonly int _maxPoolSize;
    
    public MessagePool(int maxPoolSize = 1000)
    {
        _maxPoolSize = maxPoolSize;
    }
    
    public Message Rent()
    {
        return _pool.TryTake(out var message) ? message : new Message();
    }
    
    public void Return(Message message)
    {
        message.Reset(); // Очистка полей
        
        if (_pool.Count < _maxPoolSize)
            _pool.Add(message);
    }
}

private ReadOnlySpan<byte> ParsePayload(ReadOnlySpan<byte> data)
{
    // Обработка данных без выделения дополнительной памяти
    var headerSize = BitConverter.ToInt32(data.Slice(0, 4));
    return data.Slice(4 + headerSize);
}

public class SecureStartup
{
    public async Task<bool> VerifyIntegrityAsync()
    {
        var tpm = new TpmDevice();
        await tpm.OpenAsync();
        
        // Проверка PCR регистров, содержащих хеши загрузочного кода
        var pcrValues = await tpm.ReadPcrAsync(new[] { 0, 1, 2, 3 });
        
        // Проверка подписи прошивки
        return VerifySignature(pcrValues);
    }
}

private static async Task<ModuleClient> CreateModuleClientWithMutualAuthAsync()
{
    // Загрузка сертификата клиента
    var clientCert = new X509Certificate2("device-cert.pfx", "password");
    
    // Настройка проверки сертификата сервера
    var options = new ClientOptions
    {
        SslProtocols = System.Security.Authentication.SslProtocols.Tls12,
        ServerCertificateValidationCallback = ValidateServerCertificate
    };
    
    // Создание клиента с взаимной аутентификацией
    return await ModuleClient.CreateFromX509CertificateAsync(
        Environment.GetEnvironmentVariable("IOTEDGE_GATEWAYHOST"),
        Environment.GetEnvironmentVariable("IOTEDGE_DEVICEID"),
        Environment.GetEnvironmentVariable("IOTEDGE_MODULEID"),
        clientCert,
        false,
        TransportType.Amqp_Tcp_Only,
        options);
}

public class EncryptedStorage
{
    private readonly string _keyFile;
    private readonly string _dataFile;
    private byte[] _encryptionKey;
    
    public async Task<T> ReadSecureDataAsync<T>()
    {
        var encryptionKey = await GetOrCreateKeyAsync();
        var encryptedData = await File.ReadAllBytesAsync(_dataFile);
        
        using var aes = Aes.Create();
        // Извлечение IV из начала файла
        var iv = encryptedData.Take(16).ToArray();
        var ciphertext = encryptedData.Skip(16).ToArray();
        
        aes.Key = encryptionKey;
        aes.IV = iv;
        
        using var decryptor = aes.CreateDecryptor();
        using var ms = new MemoryStream(ciphertext);
        using var cs = new CryptoStream(ms, decryptor, CryptoStreamMode.Read);
        using var reader = new StreamReader(cs);
        
        var json = await reader.ReadToEndAsync();
        return JsonConvert.DeserializeObject<T>(json);
    }
    
    // Прочие методы для работы с зашифрованными данными...
}

public class AccessController
{
    private readonly Dictionary<string, HashSet<string>> _permissions = new();
    
    public void SetupPermissions(string moduleId, IEnumerable<string> allowedOperations)
    {
        _permissions[moduleId] = new HashSet<string>(allowedOperations);
    }
    
    public bool CheckPermission(string requesterModuleId, string operation)
    {
        if (!_permissions.TryGetValue(requesterModuleId, out var allowed))
            return false;
            
        return allowed.Contains(operation) || allowed.Contains("*");
    }
}

public class ErpIntegrationModule
{
    private readonly ServiceBusClient _serviceBusClient;
    private readonly ServiceBusSender _sender;
    
    public ErpIntegrationModule(string connectionString, string queueName)
    {
        _serviceBusClient = new ServiceBusClient(connectionString);
        _sender = _serviceBusClient.CreateSender(queueName);
    }
    
    public async Task SendInventoryUpdateAsync(InventoryChange change)
    {
        // Трансформация данных в формат, понятный корпоративной системе
        var erpFormat = TransformToErpFormat(change);
        
        // Создание сообщения для очереди
        var message = new ServiceBusMessage(
            Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(erpFormat)))
        {
            ContentType = "application/json",
            Subject = "inventory-update",
            MessageId = Guid.NewGuid().ToString(),
            CorrelationId = change.TransactionId
        };
        
        // Отправка с гарантированной доставкой
        await _sender.SendMessageAsync(message);
    }
    
    // Метод трансформации данных...
}

public class ErpQueryService
{
    private readonly HttpClient _httpClient;
    private readonly string _apiKey;
    
    public async Task<ProductInfo> GetProductDetailsAsync(string barcode)
    {
        var request = new HttpRequestMessage(HttpMethod.Get, 
            $"/api/products/{barcode}");
        
        request.Headers.Add("X-API-Key", _apiKey);
        
        var response = await _httpClient.SendAsync(request);
        response.EnsureSuccessStatusCode();
        
        var content = await response.Content.ReadAsStringAsync();
        return JsonConvert.DeserializeObject<ProductInfo>(content);
    }
}

public class PowerManager
{
  private readonly BatteryMonitor _batteryMonitor;
  private PowerState _currentState = PowerState.Normal;
  
  public TimeSpan SamplingInterval { get; private set; } = TimeSpan.FromMinutes(5);
  
  public async Task UpdatePowerStateAsync()
  {
      var batteryLevel = await _batteryMonitor.GetLevelAsync();
      
      var newState = batteryLevel switch
      {
          > 70 => PowerState.Normal,
          > 30 => PowerState.Economical,
          > 15 => PowerState.Critical,
          _ => PowerState.Emergency
      };
      
      if (newState != _currentState)
      {
          _currentState = newState;
          
          // Адаптация интервалов опроса датчиков
          SamplingInterval = newState switch
          {
              PowerState.Normal => TimeSpan.FromMinutes(5),
              PowerState.Economical => TimeSpan.FromMinutes(15),
              PowerState.Critical => TimeSpan.FromMinutes(30),
              PowerState.Emergency => TimeSpan.FromHours(1),
              _ => TimeSpan.FromMinutes(5)
          };
          
          // Уведомление о смене режима
          await ReportPowerStateChnageAsync();
      }
  }
  
  // Остальная реализация...
}

public class MeshComputingManager
{
  private readonly Dictionary<string, DeviceCapability> _peerCapabilities = new();
  private readonly HttpClient _httpClient = new();
  
  public async Task<ProcessingResult> DistributedProcessAsync(byte[] rawData, ProcessingRequirements reqs)
  {
      // Найти оптимального исполнителя
      var bestPeer = FindBestExecutor(reqs);
      
      if (bestPeer == null || IsLocal(bestPeer))
      {
          // Выполнить локально
          return await LocalProcessAsync(rawData, reqs);
      }
      
      // Делегировать обработку другому устройству
      var result = await DelegateProcessingAsync(bestPeer, rawData, reqs);
      
      // Обновить информацию о пире
      await UpdatePeerStatusAsync(bestPeer);
      
      return result;
  }
}

public class DiagnosticsManager
{
  private readonly CircularBuffer<LogEntry> _localBuffer = new(capacity: 1000);
  private readonly BlockingCollection<LogEntry> _criticalBuffer = new(capacity: 100);
  
  public void Log(LogLevel level, string message, Exception ex = null)
  {
      var entry = new LogEntry
      {
          Timestamp = DateTime.UtcNow,
          Level = level,
          Message = message,
          Exception = ex?.ToString(),
          DeviceId = _deviceInfo.Id,
          ModuleId = _moduleInfo.Id
      };
      
      // Добавление в локальный буфер
      _localBuffer.Add(entry);
      
      // Критические ошибки сразу отправляем
      if (level >= LogLevel.Error)
      {
          _criticalBuffer.Add(entry);
          TriggerImmediateSend();
      }
  }
}