IThub

SQL: язык, которому 50 лет, но он не устарелSQL изобрели в IBM в 1974 году. С тех пор появились NoSQL, NewSQL, GraphQL, временны́е базы данных, документные хранилища. Но SQL не умер — он стал стандартом для большинства задач работы с данными. Реляционные СУБД (PostgreSQL, MySQL, SQLite, MS SQL, Oracle) хранят данные в большинстве корпоративных систем мира. И даже "NoSQL" системы (ClickHouse, DuckDB, BigQuery) используют SQL-диалект. Знание SQL — это инвестиция с гарантированной отдачей для любого разработчика. Архитектура запроса: как PostgreSQL исполняет SQLПонимание этого даёт инсайт, почему одни запросы быстрые, а другие — нет: Текст запроса ↓ [Parser] — проверка синтаксиса ↓ [Rewriter] — разворачивание Views, правила ↓ [Planner/Optimizer] — КЛЮЧЕВОЙ ЭТАП! - Оценка стоимости разных планов - Выбор порядка JOIN-ов - Выбор алгоритма соединения (Hash Join, Nested Loop, Merge Join) - Решение: использовать индекс или seq scan ↓ [Executor] — выполнение выбранного плана ↓ Результат Планировщик работает на основе статистики (pg_statistics). Устаревшая статистика → неоптимальный план → медленный запрос. Поэтому важен ANALYZE или автовакуум. EXPLAIN ANALYZE: видим что происходит-- Всегда используйте ANALYZE для реального времени (но он выполняет запрос!) -- Для SELECT это безопасно. Для DML используйте ROLLBACK: -- BEGIN; EXPLAIN ANALYZE UPDATE ...; ROLLBACK; EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT) SELECT u.name, COUNT(o.id) as order_count, SUM(o.total) as total_sum FROM users u JOIN orders o ON o.user_id = u.id WHERE u.created_at > '2024-01-01' AND o.status = 'completed' GROUP BY u.id, u.name ORDER BY total_sum DESC LIMIT 20; -- Типичный вывод и как его читать: /* Limit (cost=1250.45..1250.50 rows=20 width=48) (actual time=45.231..45.234 rows=20 loops=1) -> Sort (cost=1250.45..1253.95 rows=1400 width=48) (actual time=45.228..45.231 rows=20 loops=1) Sort Key: (sum(o.total)) DESC Sort Method: top-N heapsort Memory: 27kB -> HashAggregate (cost=1190.23..1204.73 rows=1400 width=48) (actual time=44.123..45.012 rows=1823 loops=1) Group Key: u.id, u.name Batches: 1 Memory Usage: 657kB -> Hash Join (cost=485.30..1148.73 rows=8300 width=24) (actual time=2.341..38.201 rows=9843 loops=1) Hash Cond: (o.user_id = u.id) Buffers: shared hit=342 read=891 ← ВАЖНО! 891 блоков с диска! -> Seq Scan on orders o (cost=0.00..456.23 rows=12800 width=16) ↑ SEQ SCAN на большой таблице = тревожный сигнал! Filter: ((status)::text = 'completed'::text) Rows Removed by Filter: 23456 -> Hash (cost=423.55..423.55 rows=4940 width=16) (actual time=2.103..2.103 rows=4892 loops=1) -> Index Scan using idx_users_created on users u Index Cond: (created_at > '2024-01-01'::date) Planning Time: 0.523 ms Execution Time: 45.789 ms ← Реальное время выполнения */ Что искать в EXPLAIN:Сигнал Что значит Решение Seq Scan на большой таблице Нет индекса или не используется Добавить индекс Rows Removed by Filter: N (N >> результата) Фильтр работает после scan Индекс на колонку фильтра shared read: N (N > 1000) Много чтений с диска Индекс, увеличить shared_buffers Nested Loop при большом N Плохой алгоритм JOIN Статистика, индексы, rewrite Sort без using index Сортировка в памяти/диске Индекс на ORDER BY колонку Индексы: типы и когда применятьB-Tree (по умолчанию)Подходит для: =, <, >, BETWEEN, LIKE 'prefix%', ORDER BY, диапазоны дат. -- Обычный индекс CREATE INDEX idx_orders_user_id ON orders(user_id); -- Составной индекс (порядок важен!) -- Покрывает: WHERE user_id = X AND status = Y -- Покрывает: WHERE user_id = X (только первая колонка) -- НЕ покрывает: WHERE status = Y (без user_id) CREATE INDEX idx_orders_user_status ON orders(user_id, status); -- Частичный индекс (только для подмножества строк) -- Гораздо меньше, работает быстрее для частых запросов с фильтром CREATE INDEX idx_orders_active ON orders(created_at) WHERE status IN ('pending', 'processing'); -- Индекс с включёнными колонками (covering index) -- SELECT user_id, total FROM orders WHERE status = 'completed' -- будет выполнен только из индекса, без обращения к таблице! CREATE INDEX idx_orders_status_covering ON orders(status) INCLUDE (user_id, total); Hash индексТолько для = (равенство). Быстрее B-Tree для равенства, но нет диапазонов: CREATE INDEX idx_sessions_token ON sessions USING HASH (token); -- Отлично для: WHERE token = 'abc123' (авторизация) GIN (Generalized Inverted Index)Для массивов, JSONB, полнотекстового поиска, pg_trgm: -- Полнотекстовый поиск CREATE INDEX idx_articles_fts ON articles USING GIN (to_tsvector('russian', title || ' ' || body)); -- Поиск по JSONB CREATE INDEX idx_devices_meta ON devices USING GIN (metadata jsonb_path_ops); -- Запрос: WHERE metadata @> '{"type": "sensor"}' -- pg_trgm для LIKE '%substring%' (иначе seq scan!) CREATE EXTENSION IF NOT EXISTS pg_trgm; CREATE INDEX idx_products_name_trgm ON products USING GIN (name gin_trgm_ops); -- Запрос: WHERE name ILIKE '%насос%' BRIN (Block Range INdex)Для очень больших таблиц с монотонно возрастающими данными (временны́е метки): -- Таблица телеметрии: 10 миллиардов строк -- B-Tree индекс займёт 200 ГБ -- BRIN займёт 1 МБ! (хранит мин/макс по блокам) CREATE INDEX idx_telemetry_time_brin ON telemetry USING BRIN (measured_at) WITH (pages_per_range = 128); -- Работает только если данные ФИЗИЧЕСКИ упорядочены по времени -- (INSERT в хронологическом порядке) Оконные функции: SQL нового уровняОконные функции — одна из самых мощных возможностей SQL, которую многие не знают. -- Задача: для каждого заказа показать его номер в последовательности -- заказов этого клиента и общее количество заказов клиента SELECT id, user_id, created_at, total, -- Номер строки в партиции (по каждому user_id отдельно) ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY created_at) AS order_number, -- Ранг (при одинаковых значениях — одинаковый ранг, следующий пропускается) RANK() OVER (PARTITION BY user_id ORDER BY total DESC) AS rank_by_total, -- Dense Rank (без пропусков) DENSE_RANK() OVER (PARTITION BY user_id ORDER BY total DESC) AS dense_rank, -- Количество строк в партиции COUNT(*) OVER (PARTITION BY user_id) AS total_orders, -- Нарастающая сумма SUM(total) OVER (PARTITION BY user_id ORDER BY created_at ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS running_total, -- Скользящее среднее (последние 3 заказа) AVG(total) OVER (PARTITION BY user_id ORDER BY created_at ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) AS moving_avg_3, -- Предыдущее и следующее значение LAG(total, 1) OVER (PARTITION BY user_id ORDER BY created_at) AS prev_order_total, LEAD(total, 1) OVER (PARTITION BY user_id ORDER BY created_at) AS next_order_total, -- Процент от общей суммы клиента ROUND(total / SUM(total) OVER (PARTITION BY user_id) * 100, 2) AS pct_of_customer_total, -- Процентиль PERCENT_RANK() OVER (PARTITION BY user_id ORDER BY total) AS percentile FROM orders ORDER BY user_id, created_at; Практический пример: анализ телеметрии-- Обнаружение аномалий: значения > avg + 2*stddev WITH stats AS ( SELECT device_id, AVG(temperature) AS avg_temp, STDDEV(temperature) AS std_temp FROM telemetry WHERE measured_at > NOW() - INTERVAL '7 days' GROUP BY device_id ), windowed AS ( SELECT t.*, s.avg_temp, s.std_temp, -- Z-score (t.temperature - s.avg_temp) / NULLIF(s.std_temp, 0) AS z_score, -- Скользящее среднее за 5 измерений AVG(t.temperature) OVER ( PARTITION BY t.device_id ORDER BY t.measured_at ROWS BETWEEN 4 PRECEDING AND CURRENT ROW ) AS moving_avg_5, -- Предыдущее значение (для расчёта скорости изменения) LAG(t.temperature) OVER ( PARTITION BY t.device_id ORDER BY t.measured_at ) AS prev_temp, LAG(t.measured_at) OVER ( PARTITION BY t.device_id ORDER BY t.measured_at ) AS prev_time FROM telemetry t JOIN stats s ON s.device_id = t.device_id WHERE t.measured_at > NOW() - INTERVAL '24 hours' ) SELECT device_id, measured_at, temperature, ROUND(z_score::numeric, 2) AS z_score, ROUND(moving_avg_5::numeric, 2) AS moving_avg, -- Скорость изменения (°C/мин) ROUND( (temperature - prev_temp) / NULLIF(EXTRACT(EPOCH FROM (measured_at - prev_time)) / 60.0, 0) , 2) AS rate_per_min, CASE WHEN ABS(z_score) > 3 THEN 'КРИТИЧЕСКАЯ АНОМАЛИЯ' WHEN ABS(z_score) > 2 THEN 'Аномалия' ELSE 'Норма' END AS status FROM windowed WHERE ABS(z_score) > 2 ORDER BY ABS(z_score) DESC; CTE: читаемые и повторно используемые запросы-- CTE (Common Table Expression) — именованные подзапросы -- Делают сложные запросы читаемыми WITH -- Шаг 1: активные устройства за последние 24 часа active_devices AS ( SELECT DISTINCT device_id FROM telemetry WHERE measured_at > NOW() - INTERVAL '24 hours' ), -- Шаг 2: статистика по каждому устройству device_stats AS ( SELECT t.device_id, COUNT(*) AS reading_count, AVG(t.temperature) AS avg_temp, MAX(t.temperature) AS max_temp, MIN(t.temperature) AS min_temp, SUM(CASE WHEN t.fault THEN 1 ELSE 0 END) AS fault_count FROM telemetry t INNER JOIN active_devices ad ON ad.device_id = t.device_id WHERE t.measured_at > NOW() - INTERVAL '24 hours' GROUP BY t.device_id ), -- Шаг 3: ранжирование по количеству аварий ranked AS ( SELECT *, RANK() OVER (ORDER BY fault_count DESC) AS fault_rank FROM device_stats ) -- Финальный запрос SELECT r.device_id, d.name, d.location, r.reading_count, ROUND(r.avg_temp::numeric, 2) AS avg_temp, r.max_temp, r.fault_count, r.fault_rank, CASE WHEN r.fault_count > 10 THEN '🔴 Требует внимания' ELSE '🟢 OK' END AS status FROM ranked r JOIN devices d ON d.id = r.device_id ORDER BY r.fault_rank; Рекурсивные CTE: для деревьев и графов-- Дерево категорий оборудования WITH RECURSIVE category_tree AS ( -- Базовый случай: корневые категории SELECT id, name, parent_id, 1 AS depth, name::text AS path FROM categories WHERE parent_id IS NULL UNION ALL -- Рекурсивный шаг: дочерние категории SELECT c.id, c.name, c.parent_id, ct.depth + 1, ct.path || ' > ' || c.name FROM categories c INNER JOIN category_tree ct ON ct.id = c.parent_id ) SELECT depth, REPEAT(' ', depth - 1) || name AS name_indented, path FROM category_tree ORDER BY path; -- Результат: -- Оборудование -- Насосное оборудование -- Центробежные насосы -- Шестерённые насосы -- Нагреватели -- Ленточные Транзакции и ACID-- Пример транзакции: перевод средств -- ACID: Atomicity, Consistency, Isolation, Durability BEGIN; -- Несколько операций — или все, или ничего! UPDATE accounts SET balance = balance - 1000 WHERE id = 1; UPDATE accounts SET balance = balance + 1000 WHERE id = 2; INSERT INTO transactions (from_id, to_id, amount, created_at) VALUES (1, 2, 1000, NOW()); -- Проверка (если не OK — откатываем всё) DO $$ DECLARE balance DECIMAL; BEGIN SELECT balance INTO balance FROM accounts WHERE id = 1; IF balance < 0 THEN RAISE EXCEPTION 'Недостаточно средств!'; END IF; END; $$; COMMIT; -- Всё OK, фиксируем -- или ROLLBACK; -- Если что-то пошло не так -- Уровни изоляции транзакций: -- READ UNCOMMITTED: видит незафиксированные данные (грязное чтение) -- READ COMMITTED: видит только зафиксированные (по умолчанию в PG) -- REPEATABLE READ: повторное чтение даёт тот же результат -- SERIALIZABLE: полная изоляция, как последовательное выполнение SET TRANSACTION ISOLATION LEVEL REPEATABLE READ; Партиционирование: для больших таблиц-- Партиционирование таблицы телеметрии по месяцам CREATE TABLE telemetry ( id BIGSERIAL, device_id INT NOT NULL, measured_at TIMESTAMPTZ NOT NULL, temperature FLOAT, pressure FLOAT, current FLOAT ) PARTITION BY RANGE (measured_at); -- Создаём партиции по месяцам CREATE TABLE telemetry_2024_01 PARTITION OF telemetry FOR VALUES FROM ('2024-01-01') TO ('2024-02-01'); CREATE TABLE telemetry_2024_02 PARTITION OF telemetry FOR VALUES FROM ('2024-02-01') TO ('2024-03-01'); -- ... и так далее -- Индексы создаются на каждой партиции отдельно CREATE INDEX ON telemetry_2024_01 (device_id, measured_at); CREATE INDEX ON telemetry_2024_02 (device_id, measured_at); -- Автоматическое создание партиций (pg_partman) -- SELECT partman.create_parent('public.telemetry', 'measured_at', -- 'native', 'monthly'); -- Преимущества: -- 1. Partition pruning: запрос за январь сканирует только telemetry_2024_01 -- 2. Быстрое удаление старых данных: DROP TABLE telemetry_2023_01 -- 3. Параллельное сканирование разных партиций N+1 проблема: самая частая ошибка-- N+1: вместо одного запроса делаем N+1 -- Типичная ошибка при работе с ORM -- ❌ ПЛОХО (в Python/PHP коде): -- users = db.execute("SELECT * FROM users LIMIT 100") -- for user in users: -- orders = db.execute("SELECT * FROM orders WHERE user_id = ?", user.id) -- Итого: 1 + 100 = 101 запрос! -- ✅ ХОРОШО: один JOIN SELECT u.id, u.name, u.email, COUNT(o.id) AS order_count, SUM(o.total) AS total_spent FROM users u LEFT JOIN orders o ON o.user_id = u.id GROUP BY u.id, u.name, u.email LIMIT 100; -- ✅ ХОРОШО: два запроса с IN (для сложных случаев) -- users = db.execute("SELECT * FROM users LIMIT 100") -- user_ids = [u.id for u in users] -- orders = db.execute("SELECT * FROM orders WHERE user_id = ANY(?)", user_ids) -- Итого: 2 запроса! Практические паттерны оптимизации-- 1. UPSERT (INSERT или UPDATE если существует) INSERT INTO device_status (device_id, status, updated_at) VALUES (1, 'online', NOW()) ON CONFLICT (device_id) DO UPDATE SET status = EXCLUDED.status, updated_at = EXCLUDED.updated_at; -- 2. Batch INSERT (вместо N отдельных INSERT) INSERT INTO telemetry (device_id, measured_at, temperature) VALUES (1, '2024-01-01 10:00', 25.3), (1, '2024-01-01 10:01', 25.4), (2, '2024-01-01 10:00', 22.1) -- До 1000 строк в одном запросе — намного быстрее! -- 3. COPY для массовой загрузки (самый быстрый способ) -- \COPY telemetry FROM '/data/telemetry.csv' CSV HEADER -- 4. Materialized View для сложных агрегатов CREATE MATERIALIZED VIEW daily_device_summary AS SELECT device_id, DATE(measured_at) AS day, AVG(temperature) AS avg_temp, MAX(temperature) AS max_temp, COUNT(*) AS readings FROM telemetry GROUP BY device_id, DATE(measured_at); CREATE UNIQUE INDEX ON daily_device_summary(device_id, day); -- Обновление (можно конкурентно, без блокировки SELECT) REFRESH MATERIALIZED VIEW CONCURRENTLY daily_device_summary; -- 5. EXPLAIN сначала, оптимизировать потом! -- Никогда не оптимизируйте наугад. Всегда смотрите план. PostgreSQL: важные настройки performance-- Ключевые параметры postgresql.conf для production: -- Память (зависит от RAM сервера): -- shared_buffers = 25% RAM (напр. 4GB для 16GB) -- effective_cache_size = 75% RAM -- work_mem = 64MB (для сортировок и hash join) -- maintenance_work_mem = 1GB (для VACUUM, CREATE INDEX) -- Диск (для SSD): -- random_page_cost = 1.1 (вместо 4.0) -- effective_io_concurrency = 200 -- Параллелизм: -- max_parallel_workers_per_gather = 4 -- max_worker_processes = 8 -- Checkpoint: -- checkpoint_completion_target = 0.9 -- wal_buffers = 64MB -- Проверка текущих настроек: SELECT name, setting, unit, context FROM pg_settings WHERE name IN ('shared_buffers', 'work_mem', 'max_connections'); -- Статистика медленных запросов (pg_stat_statements): CREATE EXTENSION IF NOT EXISTS pg_stat_statements; SELECT substring(query, 1, 80) AS query_short, calls, ROUND(total_exec_time::numeric / 1000, 2) AS total_sec, ROUND(mean_exec_time::numeric, 2) AS mean_ms, rows FROM pg_stat_statements ORDER BY total_exec_time DESC LIMIT 20; ЗаключениеSQL — не просто "язык запросов", это язык управления данными. Понимание планировщика, правильное использование индексов, оконные функции и CTE — это инструменты, которые превращают "работает" в "работает быстро и масштабируется". Практический совет: запустите pg_stat_statements на вашем production-сервере прямо сейчас. Посмотрите топ-20 медленных запросов. С вероятностью 80% — там найдётся очевидная оптимизация, которая ускорит приложение в разы. Инвестируйте в "Use The Index, Luke" (use-the-index-luke.com) — лучшее бесплатное руководство по индексам SQL. И всегда: EXPLAIN ANALYZE перед любой "оптимизацией".

PCB-дизайн: мост между схемой и устройствомОтличная схема в плохом PCB-дизайне — это источник помех, нестабильная работа, проблемы с EMC и перегрев. Хороший PCB-дизайн — это такая же инженерная дисциплина, как схемотехника. Современные инструменты доступны бесплатно (KiCad), производство быстрое и дешёвое (JLCPCB, PCBWay, OSHPark — 5 плат за $2 с доставкой за 2 недели). Барьер для входа в PCB-разработку никогда не был ниже. Но количество "тонких мест" не уменьшилось. Сегодня — практика без воды. Инструменты: что выбратьKiCad (бесплатно, открытый исходный код)Версия 7/8 — функционально близка к коммерческим решениям Отличный Schematic Editor и PCB Editor SPICE-симуляция, 3D-просмотр Огромная библиотека компонентов, активное сообщество Рекомендация: для большинства проектов вполне достаточно Altium DesignerПромышленный стандарт в телекоме и аэрокосмосе $8 000+/год лицензия Нельзя просто взять и попробовать CircuitMaker (бесплатная версия Altium) — сильно урезана Eagle (Autodesk)Бесплатно до 2 слоёв и 80 см² Хорошая экосистема библиотек SparkFun/Adafruit Интегрируется с Fusion 360 EasyEDA / LCEDAБраузерный, бесплатный Прямая интеграция с JLCPCB и их библиотекой компонентов Быстрый старт для простых проектов Процесс разработки: от идеи до платы1. Спецификация → 2. Схема → 3. Выбор корпуса → 4. Разводка → 5. DRC/проверки → 6. Генерация Gerber → 7. Заказ → 8. Пайка → 9. Отладка Шаг 1: Схема (Schematic)Правила хорошей схемы: Читается слева направо: сигнал течёт от входа к выходу VCC сверху, GND снизу: стандартная конвенция Все пины обозначены: нет "hanging pins" без назначения Развязочные конденсаторы: рядом с каждой микросхемой на схеме (не просто в угол!) Комментарии: номиналы, допуски, критичные параметры Обязательные компоненты для питания: - Входной конденсатор: электролит 100 мкФ/16В (bulk capacitor) - Bypass конденсатор: керамика 100 нФ X7R рядом с каждым VCC пином IC - Bypass конденсатор: 10 нФ дополнительно для высокочастотных IC - Ferrite bead (если нужна изоляция аналоговой и цифровой земли) Шаг 2: Footprint и 3D-моделиВыбор правильного корпуса компонента критически важен: Проверьте datasheet производителя — landing pattern (рекомендуемый footprint) IPC-7351 — стандарт land pattern для SMD Предпочитайте компоненты из основных серий (0402, 0603, 0805 — легко заказать) Слои PCB: понимание стекаДвухслойная PCB (2-layer):─── Copper Top (компоненты, сигналы) ─── Core (диэлектрик FR4, 1.6 мм) ─── Copper Bottom (земля, сигналы) Дёшево ($2–5 за 10 плат), достаточно для большинства низкочастотных проектов. Четырёхслойная PCB (4-layer):─── Layer 1: Copper Top (сигналы, компоненты) ─── Prepreg (диэлектрик) ─── Layer 2: Ground Plane (сплошная земля!) ─── Core ─── Layer 3: Power Plane (питание) ─── Prepreg ─── Layer 4: Copper Bottom (сигналы) Дороже (~$15–30 за 10 плат), но: Слой земли под каждым сигнальным слоем — контролируемый импеданс Чистое питание (мало помех) Лучшая EMC Обязательна при F > 50 МГц или быстрых фронтах Земляной полигон (Ground Plane): основа всегоЭто самое важное правило PCB-дизайна. Сплошной медный полигон на слое GND: Почему это важно: Низкоиндуктивный путь возврата тока для каждого сигнала Экранирование сигнальных слоёв Тепловая масса для компонентов Референс для импеданса сигналов ПРАВИЛЬНО: земля под каждым сигнальным трэком ─Signal──────────────────────────────── Layer 1 ─────────────────────────────────────── Layer 2 (GND plane) Ток сигнала течёт по трэку, ток возврата — прямо под ним по плоскости (минимальная петля!) НЕПРАВИЛЬНО: нет плоскости, возврат по произвольному пути ─Signal──────────────────────────────── Layer 1 ──────────GND wire──────────────────── Layer 2 Ток возврата ищет произвольный путь → большая петля → EMI! Критические правила полигона:Не разрезайте плоскость без необходимости! ❌ Плохо: прорезь делит плоскость на два острова ───────────────────────────────────────────────── ╔═══════════════════╗ ← Прорезь! ─────────╝ ╚──────────────────── Ток возврата вынужден огибать прорезь → большая петля → EMI ✅ Хорошо: полигон цельный ───────────────────────────────────────────────────────────── (никаких прорезей без веской причины) ───────────────────────────────────────────────────────────── Via stitching — соединение полигонов между слоями: Размещайте заземляющие виа равномерно по всей плате (через каждые 1–2 см). Это снижает индуктивность плоскости. Импеданс трэков: для высокоскоростных сигналовПри частоте выше ~100 МГц или временах нарастания фронта <2нс — трэки нужно рассматривать как длинные линии. Импеданс трэка должен совпадать с импедансом источника и нагрузки (обычно 50 Ом для одиночного трэка или 100 Ом для дифференциальной пары). Формула для микрополосковой линии (Microstrip):Трэк на поверхностном слое над плоскостью земли: Z0 ≈ (87 / √(εr + 1.41)) × ln(5.98 × h / (0.8 × w + t)) Где: εr = диэлектрическая проницаемость (FR4: 4.2–4.5) h = расстояние от трэка до плоскости, мкм w = ширина трэка, мкм t = толщина меди, мкм (стандарт 1oz = 35 мкм) Для FR4, 4-слойная плата, h=200 мкм: 50 Ом → w ≈ 450 мкм (0.45 мм) 75 Ом → w ≈ 200 мкм (0.20 мм) Практически: используйте онлайн-калькуляторы (Saturn PCB Toolkit, Polar Si9000) или параметр stackup от производителя платы (JLCPCB публикует точные параметры своего FR4). Дифференциальные парыUSB, LVDS, HDMI, Ethernet, SerDes — все используют дифференциальные пары. Правила трассировки: Правила дифференциальных пар: 1. Одинаковая длина обоих трэков (skew < 5 мил/пс сигнала) 2. Одинаковое расстояние между трэками по всей длине (coupling) 3. Расстояние между трэками пары: 2-3 толщины диэлектрика 4. Без прямых углов (45° или радиусы) 5. Пересечение плоскости GND: только перпендикулярно, не вдоль щели USB FS (12 Мбит/с): Z_diff = 90 Ом, зазор 150 мкм USB HS (480 Мбит/с): Z_diff = 90 Ом, контроль длины ±0.1 мм Ethernet 100M: Z_diff = 100 Ом через трансформатор LVDS: Z_diff = 100 Ом Декупплинг конденсаторы: где и какиеСтратегия декупплинга (от источника питания к IC): [Источник] → [100 мкФ electrolytic] → [10 мкФ MLCC] → [100 нФ MLCC] → [IC] (bulk, далеко) (medium, ближе) (bypass, вплотную) Расположение на плате: ┌────────────────────────────────────┐ │ ┌──────┐ │ │ │ IC │ ← 100нФ вплотную к VCC пину │ └──────┘ ← 10нФ рядом │ ... │ │ [100мкФ] │ └────────────────────────────────────┘ Расстояние: - 100 нФ: ≤ 1 мм от VCC пина IC - 10 нФ: ≤ 3 мм - 100 мкФ: ≤ 10 мм НЕПРАВИЛЬНО: конденсатор в угол платы далеко от IC Эффективность падает экспоненциально с расстоянием! Выбор диэлектрика конденсатора:Диэлектрик Применение Зависимость от V/T X7R Bypass, фильтры (100 пФ – 10 мкФ) Умеренная X5R Bulk bypass (1 мкФ – 47 мкФ) Значительная при V C0G/NP0 Точные цепи, LC-фильтры Минимальная Y5V Не использовать в серьёзных проектах Огромная (-80%!) Важно: MLCC конденсатор 10 мкФ X5R 6.3В при напряжении 5В теряет 60% ёмкости из-за DC bias! Проверяйте даташит. Тепловой дизайн на PCBТепловое сопротивление медной области: R_th = L / (λ × A) = L / (380 Вт/(м·К) × ширина × толщина) Для FR4 (плохой теплопроводник, λ=0.3 Вт/(м·К)): Тепло течёт ПО МЕДИ, не через диэлектрик! Рекомендации: 1. Thermal vias под горячими компонентами: ┌──────────────────────────┐ │ IC (рассеивает 2 Вт) │ │ ●●●●●●●●●●●●●●●●●●● │ ← via array к Cu plane └──────────────────────────┘ Диаметр via: 0.3 мм, шаг: 0.8–1.0 мм 2. Copper pour (медный полигон) рядом с горячим компонентом 3. Radiator pad: обнажённая медь сверху для конвекции Расчёт температуры: T_junction = T_ambient + P × (R_th_jc + R_th_board + R_th_air) ↑ Это то, на что влияет PCB-дизайн EMC: электромагнитная совместимостьПлохой PCB-дизайн — главная причина проблем с EMC-сертификацией. Три правила EMC для PCB:1. Минимизировать площадь токовых петель: Высокочастотный ток протекает: Source → трэк → нагрузка → возврат по плоскости GND (под трэком) Площадь петли = длина трэка × расстояние до плоскости Уменьшить расстояние = 4-слойка со сплошной GND плоскостью 2. Разделить аналоговую и цифровую землю правильно: Популярный МИФ: "нужно разделить AGND и DGND полностью" РЕАЛЬНОСТЬ: один сплошной полигон GND, аналоговые компоненты в одном углу, цифровые в другом. Соединяйте земли в ОДНОЙ точке под ADC/DAC. Разрезать полигон почти никогда не нужно и часто вредно! 3. Развязка питания IC: Уже разобрали выше — 100нФ вплотную к каждому VCC пину. Дополнительные меры EMC:Ferrite bead в линии питания шумных цифровых блоков Common-mode фильтры на интерфейсных линиях Guard ring (защитное кольцо) вокруг аналоговых блоков Минимизировать длину высокочастотных трэков Правила трассировки: шпаргалкаШирина трэков: - Питание (до 1А): 0.5 мм - Питание (до 2А): 1.0 мм - Питание (до 5А): 2.5 мм - Сигналы: 0.15–0.25 мм (минимум производства: обычно 0.1 мм) Зазоры: - Сигнал-сигнал: ≥ 0.15 мм (производственный минимум) - 100В AC: ≥ 1 мм (по воздуху), 2 мм (по поверхности) - 250В AC: ≥ 2 мм / 4 мм Переходные отверстия (via): - Стандарт: диаметр 0.6 мм (отверстие 0.3 мм) - Micro via: 0.2 мм — дорого, только если необходимо - Технологические отверстия (крепёжные): нет меди, 3.2 мм (под M3) Углы трэков: ✅ 45° (стандарт) ✅ Радиусы (лучше для высоких частот) ❌ 90° прямые углы (устарелая проблема, но лучше избегать) Подготовка к производству: Gerber файлыНабор файлов для производства: Gerber файлы: .GTL - Top Copper (верхний слой меди) .GBL - Bottom Copper .GTS - Top Solder Mask (маска верхнего слоя) .GBS - Bottom Solder Mask .GTO - Top Silkscreen (маркировка) .GBO - Bottom Silkscreen .GKO - Board Outline (контур платы) .GM1..N - Inner layers (внутренние слои, если есть) Drill файл: .DRL или .XLN - координаты и размеры отверстий BOM (Bill of Materials): .CSV - список компонентов с номиналами, производителем, part number Pick and Place: .CSV - координаты и ориентация SMD компонентов (для PCBA) JLCPCB: заказ платы и PCBA-сборкиJLCPCB — наиболее популярный среди разработчиков производитель: Параметры стандартного заказа: Количество: 5 штук Слои: 2 Размер: ≤ 100×100 мм Цена: $2 + доставка Срок: 2 дня производство + 1-2 недели доставка PCBA (сборка компонентов): Выбор "SMT Assembly" при оформлении заказа Загрузить: Gerber + BOM + Pick&Place CSV Компоненты: из их склада (Basic Parts бесплатно; Extended Parts - $3/тип) Нюанс: минимальный заказ PCBA - 2 платы, некоторые компоненты не доступны Типичные ошибки новичков1. Слишком тонкие трэки питания Трэк 0.15 мм, ток 500мА → нагрев, падение напряжения, деградация. 2. Конденсаторы декупплинга далеко от IC Декупплинг работает только при минимальной индуктивности пути. 3. Разрезанный земляной полигон Трэки проходят через полигон, создавая "острова" — петли, помехи. 4. Несоответствие footprint реальному корпусу Проверьте в 3D-просмотре ДО заказа! Footprint 0402 vs 0603 — разные! 5. Нет тестовых точек Как отлаживать плату без точек для щупа? Добавьте testpad на каждый критичный сигнал. 6. Не проверена DRC (Design Rule Check) KiCad/EasyEDA имеют встроенную проверку. Всегда запускайте перед экспортом. ЗаключениеPCB-дизайн — навык, который приходит с практикой. Сделайте свою первую плату, закажите, спаяйте, найдите ошибки, сделайте вторую лучше. Итерационный процесс. KiCad 8 — отличная бесплатная точка входа. Пройдите официальные туториалы на kicandhw.io. Изучите IPC-2221 (Generic Standard on Printed Board Design) — документ объёмный, но содержит ответы на большинство вопросов по правилам разводки. Инвестируйте в понимание физики: как ток возвращается к источнику, что такое импеданс трэка, как работает декупллинг. С этим пониманием большинство решений по разводке становятся очевидными.

ESP32: почему он стал стандартом IoTESP32 от Espressif Systems вышел в 2016 году и быстро стал самым популярным Wi-Fi/BT чипом для IoT. Причины: 240 МГц Xtensa LX6 (два ядра!) — серьёзная вычислительная мощь Wi-Fi 802.11 b/g/n + Bluetooth 4.2/BLE — встроено в один чип 520 КБ SRAM + внешняя Flash — достаточно для реальных приложений Богатая периферия: 18 каналов ADC, 2 DAC, 3 UART, 2 SPI, 2 I2C, I2S, CAN, Touch, Hall Цена $2–5 (модуль ESP32-WROOM-32) FreeRTOS в основе SDK — готовая RTOS "из коробки" Семейство ESP32 сегодня: ESP32 — оригинал, Xtensa LX6 240 МГц, Wi-Fi + BT Classic + BLE ESP32-S2 — одно ядро, USB OTG, нет BT, дешевле ESP32-S3 — два ядра, USB OTG, AI-расширения (ML) ESP32-C3 — RISC-V одно ядро, Wi-Fi + BLE, ультрадешёвый (~$1) ESP32-C6 — RISC-V, Wi-Fi 6, BLE 5, Thread/Zigbee (Matter) ESP32-H2 — только BLE 5 + Thread (802.15.4), без Wi-Fi Архитектура: два ядра и их назначениеESP32 имеет два ядра Xtensa LX6 с именами PRO_CPU (ядро 0) и APP_CPU (ядро 1): PRO_CPU (Protocol CPU, Core 0): - Wi-Fi/Bluetooth стек (работает здесь) - Системные задачи FreeRTOS - Обработка прерываний от периферии APP_CPU (Application CPU, Core 1): - Ваш прикладной код - Бизнес-логика - Задачи реального времени приложения При использовании Arduino framework — код в loop() выполняется на APP_CPU. В ESP-IDF — вы явно указываете ядро при создании задачи: // ESP-IDF: создание задач с привязкой к ядру void wifi_task(void *pvParam) { // Эта задача работает на PRO_CPU — ближе к WiFi-стеку while (1) { // Сетевые операции vTaskDelay(pdMS_TO_TICKS(100)); } } void sensor_task(void *pvParam) { // Эта задача на APP_CPU — изолирована от WiFi-шумов while (1) { float adc_val = read_adc(); run_pid(adc_val); vTaskDelay(pdMS_TO_TICKS(10)); } } void app_main(void) { // PRO_CPU (Core 0) — WiFi, сетевые задачи xTaskCreatePinnedToCore(wifi_task, "WiFi", 4096, NULL, 5, NULL, 0); // APP_CPU (Core 1) — приложение xTaskCreatePinnedToCore(sensor_task, "Sensor", 4096, NULL, 4, NULL, 1); } Важно: WiFi-стек использует PRO_CPU интенсивно во время передачи. Задачи реального времени лучше держать на APP_CPU чтобы WiFi не вызывал джиттер. WiFi: три режима работыStation Mode (STA) — подключение к роутеру// Arduino framework #include <WiFi.h> const char* SSID = "MyNetwork"; const char* PASSWORD = "MyPassword"; void wifi_connect() { WiFi.mode(WIFI_STA); WiFi.begin(SSID, PASSWORD); // Статический IP (не DHCP) — обязательно для production! IPAddress local_ip(192, 168, 1, 200); IPAddress gateway(192, 168, 1, 1); IPAddress subnet(255, 255, 255, 0); IPAddress dns1(192, 168, 1, 1); WiFi.config(local_ip, gateway, subnet, dns1); Serial.print("Подключение к WiFi"); uint32_t timeout = millis() + 30000; // 30 секунд таймаут while (WiFi.status() != WL_CONNECTED) { if (millis() > timeout) { Serial.println("\nОшибка подключения! Перезагрузка..."); ESP.restart(); } delay(500); Serial.print("."); } Serial.printf("\nПодключено! IP: %s, RSSI: %d dBm\n", WiFi.localIP().toString().c_str(), WiFi.RSSI()); } // Мониторинг соединения в loop(): void check_wifi_reconnect() { static uint32_t lastCheck = 0; if (millis() - lastCheck < 5000) return; lastCheck = millis(); if (WiFi.status() != WL_CONNECTED) { Serial.println("WiFi потерян, переподключение..."); WiFi.disconnect(); WiFi.begin(SSID, PASSWORD); } } Access Point Mode (AP) — ESP32 как точка доступаvoid start_access_point() { WiFi.mode(WIFI_AP); // SSID, пароль, канал, скрытый?, макс клиентов WiFi.softAP("ESP32-Config", "setup12345", 6, false, 4); // Настройка IP точки доступа IPAddress ap_ip(192, 168, 4, 1); IPAddress ap_netmask(255, 255, 255, 0); WiFi.softAPConfig(ap_ip, ap_ip, ap_netmask); Serial.printf("AP запущен: %s, IP: %s\n", WiFi.softAPSSID().c_str(), WiFi.softAPIP().toString().c_str()); } STA+AP (одновременно!) — для конфигурации устройстваvoid start_sta_ap_mode() { WiFi.mode(WIFI_AP_STA); // AP для настройки (пока устройство не настроено) WiFi.softAP("ESP32-Setup"); // STA для рабочего подключения if (has_credentials()) { WiFi.begin(saved_ssid, saved_password); } } BLE: Bluetooth Low Energy в деталяхBLE в ESP32 реализован через NimBLE (более лёгкий стек, рекомендуется) или Bluedroid. GATT Server — ESP32 как BLE периферия#include <NimBLEDevice.h> // UUID сервисов и характеристик (генерируйте свои на uuidgenerator.net) #define SERVICE_UUID "4fafc201-1fb5-459e-8fcc-c5c9c331914b" #define TEMPERATURE_UUID "beb5483e-36e1-4688-b7f5-ea07361b26a8" #define CONTROL_UUID "beb5483e-36e1-4688-b7f5-ea07361b26a9" NimBLEServer* pServer = nullptr; NimBLECharacteristic* pTempChar = nullptr; NimBLECharacteristic* pCtrlChar = nullptr; bool deviceConnected = false; // Callbacks для событий подключения class ServerCallbacks : public NimBLEServerCallbacks { void onConnect(NimBLEServer* pServer, ble_gap_conn_desc* desc) override { deviceConnected = true; Serial.printf("BLE: клиент подключён, addr: %s\n", NimBLEAddress(desc->peer_ota_addr).toString().c_str()); // Обновляем параметры соединения для лучшей производительности pServer->updateConnParams(desc->conn_handle, 6, 6, 0, 100); } void onDisconnect(NimBLEServer* pServer) override { deviceConnected = false; Serial.println("BLE: клиент отключился"); // Перезапускаем рекламу NimBLEDevice::startAdvertising(); } }; // Callbacks для записи характеристики управления class ControlCallbacks : public NimBLECharacteristicCallbacks { void onWrite(NimBLECharacteristic* pChar) override { std::string value = pChar->getValue(); if (value.length() > 0) { uint8_t command = value[0]; Serial.printf("BLE: команда получена: 0x%02X\n", command); switch (command) { case 0x01: relay_on(); break; case 0x00: relay_off(); break; } } } }; void ble_init() { NimBLEDevice::init("ESP32-Sensor"); NimBLEDevice::setMTU(185); // Увеличиваем MTU для больших пакетов pServer = NimBLEDevice::createServer(); pServer->setCallbacks(new ServerCallbacks()); // Создаём сервис NimBLEService* pService = pServer->createService(SERVICE_UUID); // Характеристика температуры (только чтение + уведомления) pTempChar = pService->createCharacteristic( TEMPERATURE_UUID, NIMBLE_PROPERTY::READ | NIMBLE_PROPERTY::NOTIFY ); pTempChar->setValue(0.0f); // Характеристика управления (запись) pCtrlChar = pService->createCharacteristic( CONTROL_UUID, NIMBLE_PROPERTY::WRITE | NIMBLE_PROPERTY::WRITE_NR // NR = No Response (быстрее) ); pCtrlChar->setCallbacks(new ControlCallbacks()); pService->start(); // Настройка рекламы NimBLEAdvertising* pAdv = NimBLEDevice::getAdvertising(); pAdv->addServiceUUID(SERVICE_UUID); pAdv->setScanResponse(true); pAdv->setMinPreferred(0x06); pAdv->start(); Serial.println("BLE запущен, ожидаем подключения..."); } // Обновление данных температуры (вызывать периодически) void ble_update_temperature(float temperature) { if (!deviceConnected) return; // Отправляем float как 4 байта uint8_t data[4]; memcpy(data, &temperature, 4); pTempChar->setValue(data, 4); pTempChar->notify(); // Push уведомление подключённому клиенту } NVS: хранение настроек во FlashNVS (Non-Volatile Storage) — key-value хранилище в Flash памяти ESP32. Пережи вает перезагрузки и обновления прошивки: #include <Preferences.h> // Arduino framework Preferences prefs; struct DeviceConfig { char mqtt_host[64]; uint16_t mqtt_port; char device_id[32]; float setpoint; bool auto_mode; }; DeviceConfig config; void config_load_defaults() { strlcpy(config.mqtt_host, "192.168.1.100", sizeof(config.mqtt_host)); config.mqtt_port = 1883; strlcpy(config.device_id, "esp32_001", sizeof(config.device_id)); config.setpoint = 25.0f; config.auto_mode = true; } bool config_load() { prefs.begin("config", true); // true = read-only if (!prefs.isKey("mqtt_host")) { prefs.end(); return false; // Первый запуск — нет сохранённых настроек } prefs.getString("mqtt_host", config.mqtt_host, sizeof(config.mqtt_host)); config.mqtt_port = prefs.getUShort("mqtt_port", 1883); prefs.getString("device_id", config.device_id, sizeof(config.device_id)); config.setpoint = prefs.getFloat("setpoint", 25.0f); config.auto_mode = prefs.getBool("auto_mode", true); prefs.end(); return true; } void config_save() { prefs.begin("config", false); // false = read-write prefs.putString("mqtt_host", config.mqtt_host); prefs.putUShort("mqtt_port", config.mqtt_port); prefs.putString("device_id", config.device_id); prefs.putFloat("setpoint", config.setpoint); prefs.putBool("auto_mode", config.auto_mode); prefs.end(); Serial.println("Конфигурация сохранена в NVS"); } void config_reset() { prefs.begin("config", false); prefs.clear(); prefs.end(); Serial.println("NVS очищен, перезагрузка..."); ESP.restart(); } // Использование: void setup() { if (!config_load()) { Serial.println("Первый запуск, загружаем дефолты"); config_load_defaults(); config_save(); } Serial.printf("MQTT: %s:%d\n", config.mqtt_host, config.mqtt_port); } OTA: обновление прошивки по воздухуESP32: глубокое погружение#include <ArduinoOTA.h> #include <Update.h> // ===== ПРОСТОЕ OTA ЧЕРЕЗ Arduino IDE ===== void ota_init_arduino() { ArduinoOTA.setHostname("esp32-gateway-001"); ArduinoOTA.setPassword("ota_secret_password"); ArduinoOTA.onStart([]() { String type = ArduinoOTA.getCommand() == U_FLASH ? "прошивку" : "файловую систему"; Serial.printf("OTA: начало обновления %s\n", type.c_str()); // Останавливаем критичные задачи перед обновлением mqtt_stop(); modbus_stop(); }); ArduinoOTA.onEnd([]() { Serial.println("OTA: обновление завершено, перезагрузка..."); }); ArduinoOTA.onProgress([](unsigned int progress, unsigned int total) { static uint8_t last_pct = 0; uint8_t pct = progress * 100 / total; if (pct != last_pct && pct % 10 == 0) { Serial.printf("OTA: %u%%\n", pct); last_pct = pct; } }); ArduinoOTA.onError([](ota_error_t error) { const char* errors[] = {"Auth Failed", "Begin Failed", "Connect Failed", "Receive Failed", "End Failed"}; Serial.printf("OTA Ошибка[%u]: %s\n", error, error <= 4 ? errors[error] : "Unknown"); }); ArduinoOTA.begin(); } // В loop() добавить: // ArduinoOTA.handle(); // ===== HTTP OTA: скачивание прошивки с сервера ===== #include <HTTPUpdate.h> #include <WiFiClientSecure.h> void ota_update_from_server(const char* server_url) { Serial.printf("OTA: загрузка с %s\n", server_url); WiFiClient client; // Или WiFiClientSecure для HTTPS (настройте сертификат!) httpUpdate.setLedPin(LED_BUILTIN, LOW); // Callback прогресса httpUpdate.onProgress([](int current, int total) { Serial.printf("OTA: %d/%d bytes (%d%%)\n", current, total, current * 100 / total); }); t_httpUpdate_return ret = httpUpdate.update(client, server_url); switch (ret) { case HTTP_UPDATE_FAILED: Serial.printf("OTA ошибка (%d): %s\n", httpUpdate.getLastError(), httpUpdate.getLastErrorString().c_str()); break; case HTTP_UPDATE_NO_UPDATES: Serial.println("OTA: нет обновлений"); break; case HTTP_UPDATE_OK: Serial.println("OTA: успех, перезагрузка..."); break; } } // Проверка обновлений по расписанию: void check_for_updates() { static uint32_t lastCheck = 0; const uint32_t CHECK_INTERVAL = 3600000UL; // 1 час if (millis() - lastCheck < CHECK_INTERVAL) return; lastCheck = millis(); // Проверяем версию на сервере String server_version = http_get_json("/api/firmware/version")["version"]; if (server_version != FIRMWARE_VERSION) { Serial.printf("Доступна новая версия: %s (текущая: %s)\n", server_version.c_str(), FIRMWARE_VERSION); ota_update_from_server("http://server/firmware/latest.bin"); } } Deep Sleep: энергосбережениеESP32 в active mode потребляет ~80–240 мА. В deep sleep — 10 мкА! #include <esp_sleep.h> #include <esp_wifi.h> // Типы пробуждения: // - Таймер (RTC таймер) // - GPIO (кнопка, сигнал) // - Touch (сенсорные входы) // - ULP (Ultra-Low Power co-processor) // - UART (RXD0) // - BT (в режиме light sleep) void go_to_deep_sleep(uint32_t sleep_seconds) { Serial.printf("Уходим в сон на %u секунд...\n", sleep_seconds); Serial.flush(); // Закрываем WiFi перед сном (экономит время пробуждения) WiFi.disconnect(true); WiFi.mode(WIFI_OFF); // Настраиваем пробуждение по таймеру esp_sleep_enable_timer_wakeup((uint64_t)sleep_seconds * 1000000ULL); // Пробуждение от GPIO4 (нажатие кнопки) esp_sleep_enable_ext0_wakeup(GPIO_NUM_4, 0); // 0 = LOW уровень // Входим в deep sleep esp_deep_sleep_start(); // Код после этой строки не выполнится! } void setup() { Serial.begin(115200); // Определяем причину пробуждения esp_sleep_wakeup_cause_t cause = esp_sleep_get_wakeup_cause(); switch (cause) { case ESP_SLEEP_WAKEUP_TIMER: Serial.println("Пробуждение: таймер"); send_sensor_data(); // Отправляем данные и снова в сон break; case ESP_SLEEP_WAKEUP_EXT0: Serial.println("Пробуждение: кнопка"); // Пользователь нажал кнопку — полная работа full_operation_mode(); return; case ESP_SLEEP_WAKEUP_UNDEFINED: Serial.println("Первый запуск или reset"); first_boot_setup(); break; default: Serial.printf("Причина: %d\n", cause); } // Снова в сон через 60 секунд go_to_deep_sleep(60); } // RTC Memory: данные переживают deep sleep! RTC_DATA_ATTR int boot_count = 0; RTC_DATA_ATTR float last_temperature = 0.0f; RTC_DATA_ATTR uint32_t error_count = 0; void setup_with_rtc_memory() { boot_count++; Serial.printf("Загрузка #%d, последняя T=%.1f°C\n", boot_count, last_temperature); // Читаем датчик, сохраняем в RTC memory last_temperature = read_temperature(); go_to_deep_sleep(300); // 5 минут } ADC: правильная работа с АЦПАЦП ESP32 имеет репутацию "неточного". Это правда — и вот почему и как с этим работать: #include <esp_adc/adc_oneshot.h> #include <esp_adc/adc_cali.h> #include <esp_adc/adc_cali_scheme.h> // Калиброванный ADC на ESP-IDF (точность ±5мВ вместо ±50мВ) adc_oneshot_unit_handle_t adc1_handle; adc_cali_handle_t cali_handle; void adc_init_calibrated() { // Инициализация ADC adc_oneshot_unit_init_cfg_t init_config = { .unit_id = ADC_UNIT_1, .ulp_mode = ADC_ULP_MODE_DISABLE, }; ESP_ERROR_CHECK(adc_oneshot_new_unit(&init_config, &adc1_handle)); // Настройка канала (GPIO34 = ADC1 Channel 6) adc_oneshot_chan_cfg_t chan_config = { .bitwidth = ADC_BITWIDTH_12, .atten = ADC_ATTEN_DB_12, // 0-3.3В диапазон }; ESP_ERROR_CHECK(adc_oneshot_config_channel(adc1_handle, ADC_CHANNEL_6, &chan_config)); // Калибровка (Line Fitting или Curve Fitting) adc_cali_line_fitting_config_t cali_config = { .unit_id = ADC_UNIT_1, .atten = ADC_ATTEN_DB_12, .bitwidth = ADC_BITWIDTH_12, }; ESP_ERROR_CHECK(adc_cali_create_scheme_line_fitting(&cali_config, &cali_handle)); } float adc_read_voltage_mv() { int raw; ESP_ERROR_CHECK(adc_oneshot_read(adc1_handle, ADC_CHANNEL_6, &raw)); int voltage_mv; ESP_ERROR_CHECK(adc_cali_raw_to_voltage(cali_handle, raw, &voltage_mv)); return (float)voltage_mv; } // Oversampling для повышения точности (16x → +2 бита) float adc_read_averaged(int samples = 64) { int64_t sum = 0; for (int i = 0; i < samples; i++) { int raw; adc_oneshot_read(adc1_handle, ADC_CHANNEL_6, &raw); sum += raw; delayMicroseconds(100); } int avg_raw = sum / samples; int voltage_mv; adc_cali_raw_to_voltage(cali_handle, avg_raw, &voltage_mv); return (float)voltage_mv; } // Важные ограничения ADC ESP32: // - GPIO36, 37, 38, 39: только вход, без pullup/pulldown в кристалле // - ADC2 нельзя использовать одновременно с WiFi! // - Нелинейность вблизи 0В и 3.3В — оставайтесь в диапазоне 100мВ..3.1В // - Для точных измерений: внешний АЦП MCP3208 по SPI Практический проект: промышленный IoT узел// Полная архитектура ESP32 IoT узла #include <Arduino.h> #include <WiFi.h> #include <PubSubClient.h> #include <ArduinoJson.h> #include <Preferences.h> // ===== КОНФИГУРАЦИЯ ===== #define FIRMWARE_VERSION "1.2.3" #define DEVICE_MODEL "ESP32-IoT-Node" #define PUBLISH_INTERVAL_MS 5000 #define WATCHDOG_TIMEOUT_MS 30000 // ===== ГЛОБАЛЬНОЕ СОСТОЯНИЕ ===== struct State { bool wifi_connected = false; bool mqtt_connected = false; float temperature = 0; float humidity = 0; float pressure = 0; uint32_t uptime_sec = 0; uint32_t publish_count = 0; uint32_t error_count = 0; }; State state; WiFiClient wifiClient; PubSubClient mqtt(wifiClient); // ===== МНОГОЗАДАЧНОСТЬ ===== QueueHandle_t sensorQueue; SemaphoreHandle_t stateMutex; // Задача: чтение датчиков (Core 1) void task_sensors(void *pv) { for (;;) { // Здесь: читаем датчики float t = 25.0 + random(-10, 10) / 10.0; // Имитация float h = 50.0 + random(-5, 5) / 10.0; // Обновляем состояние через мьютекс xSemaphoreTake(stateMutex, portMAX_DELAY); state.temperature = t; state.humidity = h; xSemaphoreGive(stateMutex); vTaskDelay(pdMS_TO_TICKS(1000)); } } // Задача: MQTT публикация (Core 0, рядом с WiFi) void task_mqtt(void *pv) { for (;;) { if (!mqtt.connected()) { if (WiFi.isConnected()) { if (mqtt.connect("esp32-node", "user", "pass", "nodes/esp32-001/status", 1, true, "{\"online\":false}")) { mqtt.publish("nodes/esp32-001/status", "{\"online\":true}", true); mqtt.subscribe("nodes/esp32-001/commands"); } } } mqtt.loop(); // Публикация данных static uint32_t lastPublish = 0; if (millis() - lastPublish >= PUBLISH_INTERVAL_MS) { lastPublish = millis(); StaticJsonDocument<256> doc; xSemaphoreTake(stateMutex, portMAX_DELAY); doc["temperature"] = state.temperature; doc["humidity"] = state.humidity; doc["uptime"] = state.uptime_sec; doc["errors"] = state.error_count; doc["version"] = FIRMWARE_VERSION; doc["rssi"] = WiFi.RSSI(); doc["free_heap"] = ESP.getFreeHeap(); xSemaphoreGive(stateMutex); char payload[256]; serializeJson(doc, payload); mqtt.publish("nodes/esp32-001/telemetry", payload); state.publish_count++; } vTaskDelay(pdMS_TO_TICKS(10)); } } // Задача: watchdog и uptime (Core 1) void task_system(void *pv) { for (;;) { state.uptime_sec++; // Проверка heap (memory leak detection) if (ESP.getFreeHeap() < 10000) { Serial.println("КРИТИЧНО: мало памяти! Перезагрузка..."); ESP.restart(); } // Heartbeat LED digitalWrite(LED_BUILTIN, !digitalRead(LED_BUILTIN)); vTaskDelay(pdMS_TO_TICKS(1000)); } } void setup() { Serial.begin(115200); Serial.printf("\nESP32 IoT Node v%s\n", FIRMWARE_VERSION); Serial.printf("Chip: %s, Rev: %d, Cores: %d\n", ESP.getChipModel(), ESP.getChipRevision(), ESP.getChipCores()); // Инициализация stateMutex = xSemaphoreCreateMutex(); sensorQueue = xQueueCreate(10, sizeof(float)); // WiFi WiFi.mode(WIFI_STA); WiFi.begin("SSID", "PASSWORD"); // MQTT mqtt.setServer("192.168.1.100", 1883); mqtt.setBufferSize(1024); // Запускаем задачи на разных ядрах xTaskCreatePinnedToCore(task_sensors, "Sensors", 4096, NULL, 3, NULL, 1); xTaskCreatePinnedToCore(task_mqtt, "MQTT", 8192, NULL, 4, NULL, 0); xTaskCreatePinnedToCore(task_system, "System", 2048, NULL, 1, NULL, 1); Serial.println("Задачи запущены"); } void loop() { // loop() работает на Core 1 с низким приоритетом // Можно использовать для некритичных задач или оставить пустым vTaskDelay(pdMS_TO_TICKS(1000)); } Выбор инструментария: Arduino vs ESP-IDFКритерий Arduino Framework ESP-IDF (native) Порог входа Низкий Высокий Документация Обширная, много примеров Официальная, полная Производительность Достаточная Максимальная Доступ к периферии Через библиотеки Прямой Размер бинарника Больше Меньше RTOS Доступен (FreeRTOS через задачи) Нативный Время разработки Быстрее Медленнее Production-ready Да (если делать правильно) Да Рекомендация Прототипы, несложные задачи Серийное производство ЗаключениеESP32 — один из лучших выборов для промышленных IoT-узлов с умеренными требованиями к реальному времени. Двухъядерность позволяет изолировать WiFi-стек от прикладного кода, богатая периферия закрывает большинство интерфейсных задач, встроенный FreeRTOS — для многозадачности. Ключевые принципы надёжного ESP32-устройства: статический IP вместо DHCP, watchdog timer, NVS для конфигурации, OTA для обновлений, RTC memory для данных через sleep, мониторинг heap и перезагрузка при критичном уровне. ESP32 — это не замена промышленному ПЛК. Но как edge-узел сбора данных, шлюз протоколов или умный датчик — идеальный выбор.

Зачем Docker: "работает на моей машине"История, знакомая каждому разработчику. Приложение работает на вашем ноутбуке. Выкладываете на сервер — падает. Отличия: версия Python 3.9 vs 3.11, разные системные библиотеки, другая переменная PATH, конфликт зависимостей с другим приложением. Docker решает эту проблему радикально: упаковывает приложение вместе со всем окружением — операционной системой, библиотеками, конфигурацией. Контейнер запускается одинаково везде: на ноутбуке разработчика, в CI/CD, на production-сервере, в облаке. Дополнительные бонусы: Изоляция: одно приложение не мешает другому Воспроизводимость: одинаковая среда у всей команды Быстрое развёртывание: docker pull + docker run вместо часа установки Масштабирование: запустить 10 экземпляров так же легко, как 1 Чистота: удалить контейнер = никаких следов на хосте Основные концепцииImage (образ): Слоёная файловая система со всем необходимым. Неизменяемый шаблон. Хранится в Registry (Docker Hub, GitHub Container Registry, ваш собственный). Container (контейнер): Запущенный экземпляр образа. Изолированный процесс с собственной файловой системой, сетью, PID-пространством. Контейнеры ephemeral — данные исчезают при удалении (если нет Volume). Volume (том): Постоянное хранилище данных, переживает удаление контейнера. Network (сеть): Изолированная виртуальная сеть. Контейнеры в одной сети видят друг друга по имени. Dockerfile: Инструкции для сборки образа. Каждая инструкция — новый слой. Registry: Хранилище образов. Docker Hub — публичный. Можно развернуть свой (Harbor, Nexus). Dockerfile: пишем правильноБазовый пример (Python приложение):# Начинаем с официального образа Python # ВАЖНО: всегда указывайте точную версию, не :latest! FROM python:3.11-slim # Метаданные LABEL maintainer="your@email.com" LABEL version="1.0" LABEL description="Industrial IoT Gateway" # Устанавливаем рабочую директорию WORKDIR /app # Копируем ТОЛЬКО файлы зависимостей первыми! # Слои кешируются — если requirements.txt не изменился, # pip install не запустится при следующей сборке COPY requirements.txt . # Устанавливаем зависимости RUN pip install --no-cache-dir -r requirements.txt # Копируем исходный код (изменяется чаще — поэтому последним) COPY src/ ./src/ COPY config/ ./config/ # Создаём непривилегированного пользователя (безопасность!) RUN groupadd -r appuser && useradd -r -g appuser appuser RUN chown -R appuser:appuser /app USER appuser # Переменные окружения ENV PYTHONUNBUFFERED=1 \ PYTHONDONTWRITEBYTECODE=1 \ LOG_LEVEL=INFO # Открываем порт (документация, не публикует сам по себе) EXPOSE 8080 # Healthcheck: Docker проверяет живость контейнера HEALTHCHECK --interval=30s --timeout=10s --start-period=15s --retries=3 \ CMD curl -f http://localhost:8080/health || exit 1 # Команда запуска CMD ["python", "-m", "uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", "8080"] Многоэтапная сборка (Multi-stage build):Критически важна для production. Финальный образ не содержит инструментов сборки (gcc, make, pip), что уменьшает размер и поверхность атаки: # ===== ЭТАП 1: Сборка ===== FROM python:3.11 AS builder WORKDIR /build # Устанавливаем зависимости для сборки RUN apt-get update && apt-get install -y --no-install-recommends \ gcc \ libffi-dev \ && rm -rf /var/lib/apt/lists/* COPY requirements.txt . # Устанавливаем в отдельную директорию RUN pip install --no-cache-dir --prefix=/install -r requirements.txt # ===== ЭТАП 2: Production образ ===== FROM python:3.11-slim AS production # Копируем только установленные пакеты из builder COPY --from=builder /install /usr/local WORKDIR /app # Только runtime зависимости ОС RUN apt-get update && apt-get install -y --no-install-recommends \ curl \ && rm -rf /var/lib/apt/lists/* \ && apt-get clean # Безопасность RUN groupadd -r app && useradd -r -g app -d /app -s /sbin/nologin app COPY --chown=app:app src/ ./src/ USER app ENV PYTHONUNBUFFERED=1 HEALTHCHECK --interval=30s --timeout=5s CMD curl -f http://localhost:8080/health CMD ["python", "src/main.py"] # Результат: образ ~180MB вместо ~900MB! .dockerignore — что НЕ копировать в образ:# .dockerignore **/__pycache__ **/*.pyc **/*.pyo .git .gitignore .venv venv *.env .env.* tests/ docs/ *.md .github/ node_modules/ dist/ *.log .DS_Store # Секреты — никогда в образ! *.key *.pem *secret* config.local.* Docker Compose: многосервисные приложенияDocker Compose — инструмент для запуска нескольких связанных контейнеров. Полный пример: IoT-платформа# docker-compose.yml version: '3.9' # Общие настройки через YAML anchors (DRY) x-common-env: &common-env TZ: Europe/Moscow LOG_LEVEL: ${LOG_LEVEL:-INFO} x-restart-policy: &restart-policy restart: unless-stopped services: # ===== MQTT БРОКЕР ===== mosquitto: image: eclipse-mosquitto:2.0.18 <<: *restart-policy volumes: - ./mosquitto/config:/mosquitto/config:ro - mosquitto_data:/mosquitto/data - mosquitto_logs:/mosquitto/log ports: - "1883:1883" # MQTT - "9001:9001" # WebSocket networks: - iot_network healthcheck: test: ["CMD", "mosquitto_sub", "-t", "$$SYS/#", "-C", "1", "-i", "healthcheck"] interval: 30s timeout: 10s retries: 3 # ===== БАЗА ДАННЫХ ВРЕМЕННЫХ РЯДОВ ===== influxdb: image: influxdb:2.7 <<: *restart-policy environment: <<: *common-env DOCKER_INFLUXDB_INIT_MODE: setup DOCKER_INFLUXDB_INIT_USERNAME: ${INFLUX_USER:-admin} DOCKER_INFLUXDB_INIT_PASSWORD: ${INFLUX_PASSWORD:?INFLUX_PASSWORD required} DOCKER_INFLUXDB_INIT_ORG: factory DOCKER_INFLUXDB_INIT_BUCKET: telemetry DOCKER_INFLUXDB_INIT_RETENTION: 30d DOCKER_INFLUXDB_INIT_ADMIN_TOKEN: ${INFLUX_TOKEN:?INFLUX_TOKEN required} volumes: - influxdb_data:/var/lib/influxdb2 - influxdb_config:/etc/influxdb2 ports: - "8086:8086" networks: - iot_network - monitoring_network healthcheck: test: ["CMD", "curl", "-f", "http://localhost:8086/health"] interval: 30s timeout: 10s retries: 5 start_period: 30s # ===== GRAFANA ===== grafana: image: grafana/grafana:10.3.1 <<: *restart-policy depends_on: influxdb: condition: service_healthy environment: <<: *common-env GF_SECURITY_ADMIN_USER: ${GF_ADMIN_USER:-admin} GF_SECURITY_ADMIN_PASSWORD: ${GF_ADMIN_PASSWORD:?required} GF_SERVER_ROOT_URL: http://localhost:3000 GF_SMTP_ENABLED: "true" GF_SMTP_HOST: ${SMTP_HOST:-localhost:25} volumes: - grafana_data:/var/lib/grafana - ./grafana/provisioning:/etc/grafana/provisioning:ro - ./grafana/dashboards:/var/lib/grafana/dashboards:ro ports: - "3000:3000" networks: - monitoring_network user: "472" # grafana user # ===== IOT ШЛЮЗ (наш сервис) ===== gateway: build: context: . dockerfile: Dockerfile target: production args: BUILD_DATE: ${BUILD_DATE:-unknown} GIT_COMMIT: ${GIT_COMMIT:-unknown} <<: *restart-policy depends_on: mosquitto: condition: service_healthy influxdb: condition: service_healthy environment: <<: *common-env MQTT_HOST: mosquitto # Имя сервиса = DNS-имя внутри сети! MQTT_PORT: 1883 INFLUX_URL: http://influxdb:8086 INFLUX_TOKEN: ${INFLUX_TOKEN} INFLUX_ORG: factory INFLUX_BUCKET: telemetry MODBUS_PORT: /dev/ttyUSB0 # Реальный порт с хоста volumes: - ./config:/app/config:ro - gateway_logs:/app/logs devices: - "/dev/ttyUSB0:/dev/ttyUSB0" # Проброс USB-устройства ports: - "8080:8080" networks: - iot_network - monitoring_network deploy: resources: limits: cpus: '1.0' memory: 512M reservations: cpus: '0.25' memory: 128M # ===== NGINX: реверс-прокси ===== nginx: image: nginx:1.25-alpine <<: *restart-policy depends_on: - grafana - gateway volumes: - ./nginx/nginx.conf:/etc/nginx/nginx.conf:ro - ./nginx/ssl:/etc/nginx/ssl:ro # TLS сертификаты ports: - "80:80" - "443:443" networks: - monitoring_network # ===== ТОМА ===== volumes: mosquitto_data: mosquitto_logs: influxdb_data: influxdb_config: grafana_data: gateway_logs: # ===== СЕТИ ===== networks: iot_network: # Для IoT-компонентов driver: bridge monitoring_network: # Для мониторинга и UI driver: bridge .env файл для Compose:# .env — НЕ коммитить в git! INFLUX_USER=admin INFLUX_PASSWORD=super_secret_influx_pass INFLUX_TOKEN=my-super-secret-token-change-this GF_ADMIN_USER=admin GF_ADMIN_PASSWORD=super_secret_grafana_pass LOG_LEVEL=INFO BUILD_DATE=2024-03-15 GIT_COMMIT=abc123 Docker: ключевые команды# ===== ОБРАЗЫ ===== docker build -t myapp:1.0 . # Собрать образ docker build --no-cache -t myapp:1.0 . # Без кеша docker images # Список образов docker rmi myapp:1.0 # Удалить образ docker image prune # Удалить неиспользуемые # История слоёв (анализ размера): docker history myapp:1.0 # ===== КОНТЕЙНЕРЫ ===== docker run -d \ --name gateway \ -p 8080:8080 \ -e MQTT_HOST=192.168.1.100 \ -v $(pwd)/config:/app/config:ro \ --restart unless-stopped \ myapp:1.0 docker ps # Запущенные контейнеры docker ps -a # Все (включая остановленные) docker logs gateway -f # Логи в реальном времени docker logs gateway --tail 100 # Последние 100 строк docker exec -it gateway bash # Войти внутрь контейнера docker stop gateway # Остановить docker start gateway # Запустить docker restart gateway # Перезапустить docker rm gateway # Удалить (сначала stop) docker stats # Потребление ресурсов # ===== COMPOSE ===== docker compose up -d # Запустить все сервисы docker compose up -d gateway # Запустить только gateway docker compose down # Остановить и удалить контейнеры docker compose down -v # + удалить тома (ОСТОРОЖНО!) docker compose logs -f # Логи всех сервисов docker compose logs -f gateway # Логи конкретного сервиса docker compose ps # Статус сервисов docker compose pull # Обновить образы docker compose build --no-cache # Пересобрать docker compose restart gateway # Перезапустить сервис docker compose exec gateway bash # Войти в контейнер # ===== REGISTRY ===== docker tag myapp:1.0 ghcr.io/myorg/myapp:1.0 docker push ghcr.io/myorg/myapp:1.0 docker pull ghcr.io/myorg/myapp:1.0 # ===== ОЧИСТКА ===== docker system prune -a # Удалить ВСЁ неиспользуемое docker volume prune # Удалить неиспользуемые тома Volumes и persistence: не теряем данные# Типы монтирования: services: app: volumes: # 1. Named Volume (рекомендуется для данных БД) - db_data:/var/lib/postgresql/data # 2. Bind Mount (для конфигов и разработки) - ./config:/app/config:ro # :ro = read-only - ./src:/app/src # Для hot-reload при разработке # 3. tmpfs (только в RAM, для временных данных) - type: tmpfs target: /tmp tmpfs: size: 100M volumes: db_data: driver: local # Для production: внешние тома # external: true # name: prod_db_data Backup томов:# Резервная копия тома influxdb_data docker run --rm \ -v influxdb_data:/source:ro \ -v $(pwd)/backups:/backup \ alpine:3 \ tar czf /backup/influxdb_$(date +%Y%m%d).tar.gz -C /source . # Восстановление docker run --rm \ -v influxdb_data:/target \ -v $(pwd)/backups:/backup:ro \ alpine:3 \ tar xzf /backup/influxdb_20240315.tar.gz -C /target Оптимизация размера образа# ❌ ПЛОХО: большой образ FROM ubuntu:22.04 RUN apt-get update RUN apt-get install -y python3 RUN apt-get install -y python3-pip RUN pip3 install flask COPY app.py . CMD ["python3", "app.py"] # Размер: ~480 МБ, 5 лишних слоёв # ✅ ХОРОШО: оптимизированный образ FROM python:3.11-slim RUN pip install --no-cache-dir flask COPY app.py . CMD ["python", "app.py"] # Размер: ~85 МБ, чистая сборка # ✅ ЕЩЁ ЛУЧШЕ: alpine (минимальный дистрибутив) FROM python:3.11-alpine # Некоторые C-расширения нужно собирать RUN apk add --no-cache gcc musl-dev && \ pip install --no-cache-dir flask && \ apk del gcc musl-dev COPY app.py . CMD ["python", "app.py"] # Размер: ~45 МБ Правило минимума слоёв для apt/apk: # Объединяйте RUN команды в одну для минимизации слоёв RUN apt-get update && \ apt-get install -y --no-install-recommends \ curl \ libpq5 \ && rm -rf /var/lib/apt/lists/* \ && apt-get clean # ОДИН слой вместо нескольких, и сразу очистка кеша Безопасность Docker1. Никогда не запускать от root:# Создаём непривилегированного пользователя RUN addgroup -S appgroup && adduser -S appuser -G appgroup USER appuser 2. Read-only файловая система:services: app: read_only: true # Файловая система только для чтения tmpfs: - /tmp # Разрешаем запись только в tmpfs - /var/run 3. Ограничение capabilities:services: app: cap_drop: - ALL # Убираем ВСЕ capabilities cap_add: - NET_BIND_SERVICE # Добавляем только необходимые 4. Сканирование образов на уязвимости:# Trivy — бесплатный сканер (Aqua Security) docker run --rm aquasec/trivy image myapp:1.0 # Или встроенный Docker Scout docker scout cves myapp:1.0 5. Secrets — не в переменных окружения production:# docker-compose.yml с Docker secrets services: app: secrets: - db_password - api_key environment: DB_PASSWORD_FILE: /run/secrets/db_password # Читаем из файла! secrets: db_password: file: ./secrets/db_password.txt # или external: true для Swarm/K8s Healthcheck и зависимости между сервисамиservices: postgres: image: postgres:15 healthcheck: test: ["CMD-SHELL", "pg_isready -U ${POSTGRES_USER}"] interval: 10s timeout: 5s retries: 5 start_period: 30s # Даём время на инициализацию app: depends_on: postgres: condition: service_healthy # Ждём пока postgres healthy! redis: condition: service_healthy Entrypoint-скрипт для ожидания зависимостей:#!/bin/sh # entrypoint.sh set -e echo "Ожидание готовности базы данных..." until nc -z -w5 ${DB_HOST} ${DB_PORT}; do echo "База данных недоступна, ждём..." sleep 2 done echo "База данных готова!" # Запуск миграций python manage.py migrate --no-input # Запуск приложения exec "$@" COPY entrypoint.sh /entrypoint.sh RUN chmod +x /entrypoint.sh ENTRYPOINT ["/entrypoint.sh"] CMD ["gunicorn", "app:application"] Docker в CI/CD (GitHub Actions)# .github/workflows/docker.yml name: Build and Deploy on: push: tags: ['v*.*.*'] jobs: build-push: runs-on: ubuntu-latest permissions: contents: read packages: write steps: - uses: actions/checkout@v4 - name: Set up Docker Buildx uses: docker/setup-buildx-action@v3 - name: Login to GitHub Container Registry uses: docker/login-action@v3 with: registry: ghcr.io username: ${{ github.actor }} password: ${{ secrets.GITHUB_TOKEN }} - name: Build and push uses: docker/build-push-action@v5 with: context: . platforms: linux/amd64,linux/arm64 # Multi-arch! push: true tags: | ghcr.io/${{ github.repository }}:${{ github.ref_name }} ghcr.io/${{ github.repository }}:latest cache-from: type=gha cache-to: type=gha,mode=max build-args: | GIT_COMMIT=${{ github.sha }} BUILD_DATE=${{ github.event.head_commit.timestamp }} deploy: needs: build-push runs-on: ubuntu-latest environment: production steps: - name: Deploy to server uses: appleboy/ssh-action@v1 with: host: ${{ secrets.SERVER_HOST }} username: deploy key: ${{ secrets.SERVER_SSH_KEY }} script: | cd /opt/iot-platform # Обновляем образ docker compose pull gateway # Обновляем с нулевым downtime docker compose up -d --no-deps gateway # Ждём healthcheck sleep 15 docker compose ps gateway | grep -q "healthy" || exit 1 echo "Деплой успешен!" Заключение: когда Docker, когда нетИспользуйте Docker: Приложения с множеством зависимостей Многосервисные приложения (backend + БД + брокер + мониторинг) CI/CD с гарантированной воспроизводимостью Несколько приложений на одном сервере Нужна возможность быстрого масштабирования Не обязательно Docker: Простые скрипты без зависимостей Приложения с прямым доступом к оборудованию (хотя --device помогает) Жёсткое реальное время (latency контейнера ~1 мкс, но это не ноль) Команда незнакома с Docker — сначала обучение, потом внедрение Docker — это не серебряная пуля, но инвестиция в него окупается быстро. Начните с docker-compose.yml для локальной разработки — уже это даст ощутимый результат: одна команда docker compose up поднимает весь стек вместо часа настройки.

Почему RS-485, а не что-то современноеКаждый год появляются новые промышленные протоколы: EtherCAT, PROFINET, IO-Link, TSN. Но RS-485 не умирает. По данным IHS Markit, ежегодно продаётся более 1 миллиарда чипов RS-485. Новые установки продолжают использовать этот интерфейс. Причины живучести просты: Дешевизна: кабель — витая пара $0.1/м, трансивер MAX485 — $0.3 Надёжность: дифференциальный сигнал устойчив к помехам, работает на расстояниях до 1200 м Простота: понять и реализовать RS-485 можно за один день Совместимость: поддерживается абсолютно всеми промышленными устройствами Modbus RTU, BACnet MS/TP, DMX512, DALI — всё это работает поверх RS-485. Если вы занимаетесь промышленной автоматизацией, знание RS-485 обязательно. Сравнение: RS-232 vs RS-485Параметр RS-232 RS-485 Тип сигнала Однополярный, ±3–15В Дифференциальный, ±200мВ–5В Количество устройств 1:1 (точка-точка) 1:32 без репитеров (до 247 с) Максимальное расстояние 15 м 1200 м Скорость До 115 200 бод (практически) До 10 Мбит/с (при короткой линии) Устойчивость к помехам Низкая Высокая Дуплекс Полный (отдельные TX/RX) Полу (одна пара) или полный (2 пары) Применение Отладка, локальные устройства Промышленные сети, длинные линии Физический уровень: как работает дифференциальный сигналRS-485 использует дифференциальную пару проводников A и B: Состояние MARK (логическая 1, рецессивное): A > B: разность (A-B) = +200мВ...+5В Состояние SPACE (логическая 0, доминантное): B > A: разность (B-A) = +200мВ...+5В, то есть (A-B) = -200мВ...-5В Типичное напряжение при передаче: A ≈ +3.5В, B ≈ -3.5В → разность = +7В (гарантированная "1") A ≈ -3.5В, B ≈ +3.5В → разность = -7В (гарантированная "0") Устойчивость к синфазным помехам: Помеха +5В добавляется на ОБА провода: A = +3.5 + 5 = +8.5В, B = -3.5 + 5 = +1.5В Разность = +8.5 - 1.5 = +7В — сигнал не изменился! Пороги приёмника: если разность (A-B) > +200мВ — принимает "1"; если (A-B) < -200мВ — принимает "0". Диапазон ±200мВ — мёртвая зона (неопределённость). Выбор трансивера RS-485Бюджетные (для начала):MAX485 / MAX485E (Maxim) Самый популярный, $0.3–0.5 Полудуплекс, 2.5 Мбит/с Нет защиты от ESD (добавьте TVS-диоды!) Нет защиты от перегрева Питание 5В SP3485 (Sipex/Exar) Клон MAX485, питание 3.3В Совместимость с STM32, ESP32 напрямую (5В-tolerant входы) Профессиональные (для промышленности):MAX3485 / MAX3488 Расширенный диапазон ESD: ±15 кВ (HBM) Работает от 3.3В SN65HVD1780 (Texas Instruments) Встроенная защита от отказа шины (failsafe) ESD: ±16 кВ IEC 61000-4-2 Для жёстких промышленных условий ADM2587E (Analog Devices) Встроенная гальваническая изоляция 2500 VRMS Изолированный DC-DC для питания изолированной стороны Для применений с разным заземлением узлов Схема подключения: правильно и неправильноМинимальная схема (Arduino + MAX485):Arduino MAX485 RS-485 шина TX ─────────── DI RX ─────────── RO D2 ────┬────── DE A ──── Шина A └────── RE_ B ──── Шина B GND ─────────── GND GND ── Общий провод (обязательно!) 5V ──────────── Vcc Резисторы на шине: A ──[120 Ом]── B ← Терминатор на каждом конце шины Pullup/Pulldown для определённого состояния в паузах: +5В ──[560 Ом]── A ← Pull-up B ──[560 Ом]── GND ← Pull-down Ключевые правила:✅ ПРАВИЛЬНО: [Устройство A]──────[Устройство B]──────[Устройство C] [Term 120Ом] [Term 120Ом] Строго линейная шина, терминаторы только на концах ❌ НЕПРАВИЛЬНО — "звезда": [Устройство A] │ [Уст.B]──[Hub]──[Уст.C] │ [Устройство D] Отражения на каждом разветвлении разрушат сигнал! ❌ НЕПРАВИЛЬНО — терминаторы не там: [Term]──[Уст.A]──[Уст.B]──[Term]──[Уст.C] Терминатор посередине создаёт проблемы! Допустимые ответвления (stub):Короткие отводы от основной шины допустимы при условии: Длина stub < λ/10, где λ — длина волны на рабочей скорости При 9600 бод: λ ≈ 12 км → stub до 1200 м (практически неограничен) При 115200 бод: stub не более 1 м При 1 Мбит/с: stub не более 15 см! Управление направлением передачиRS-485 в полудуплексном режиме требует переключения между передачей и приёмом через сигнал DE/RE: // Arduino: управление направлением через GPIO #define RS485_DE_RE_PIN 2 #define RS485_BAUDRATE 9600 void rs485_init() { Serial.begin(RS485_BAUDRATE); pinMode(RS485_DE_RE_PIN, OUTPUT); rs485_receive_mode(); // По умолчанию — приём } void rs485_receive_mode() { digitalWrite(RS485_DE_RE_PIN, LOW); // DE=0, RE=0 → приём } void rs485_transmit_mode() { digitalWrite(RS485_DE_RE_PIN, HIGH); // DE=1, RE=1 → передача } void rs485_send(uint8_t *data, uint8_t len) { rs485_transmit_mode(); Serial.write(data, len); Serial.flush(); // ЖДЁМ пока все байты уйдут в UART TX буфер! // После flush() данные ещё в UART, нужно дождаться физической передачи // Расчёт времени: N_байт × 10_бит / baudrate uint32_t delay_us = (uint32_t)len * 10 * 1000000UL / RS485_BAUDRATE + 100; delayMicroseconds(delay_us); rs485_receive_mode(); } Аппаратное управление DE/RE (лучше!):На STM32 USART имеет аппаратный сигнал DE для RS-485. Переключение происходит автоматически — с точностью до такта, без программных задержек: // STM32 HAL: аппаратное управление DE через USART // В CubeMX: USART → Mode = Asynchronous, Hardware Flow Control = RS-485 Driver Enable // CubeMX настроит: // huart2.AdvancedInit.AdvFeatureInit = UART_ADVFEATURE_DE_INIT; // huart2.AdvancedInit.DEPolarity = UART_DE_POLARITY_HIGH; // huart2.AdvancedInit.DEAssertionTime = 16; // тактов предвключения // huart2.AdvancedInit.DEDeassertionTime = 16; // тактов послевыключения // После этого просто передаём — DE управляется автоматически! HAL_UART_Transmit(&huart2, data, len, 100); // STM32 сам поднял DE перед передачей и снял после! Расчёт нагрузки на шинуRS-485 трансивер создаёт нагрузку на шину. Стандарт RS-485 определяет "единицу нагрузки" (Unit Load, UL) = 12 кОм. Драйвер должен обеспечить минимум 32 UL. Это значит: максимум 32 "классических" устройства на шине. Современные трансиверы с низким потреблением имеют 1/8 UL или 1/4 UL: Тип трансивера Нагрузка Устройств на шине Стандартный (MAX485) 1 UL 32 1/2 UL (MAX3430) 0.5 UL 64 1/4 UL (MAX3471) 0.25 UL 128 1/8 UL (MAX3491) 0.125 UL 256 Также нагрузку создают терминирующие резисторы: 2 × 120 Ом = 60 Ом = 200 UL (!) — это доминирующая нагрузка Учитывайте это при расчёте суммарной нагрузки Кабель: выбор и прокладкаТребования к кабелю RS-485:Обязательно: Витая пара (не просто два провода!) Волновое сопротивление 120 Ом (терминируется парными резисторами 120 Ом) Рекомендуемые типы кабелей: КВВГЭ 1×2×0.75 — отечественный, экранированная витая пара Belden 9842 — американский стандарт, 120 Ом, двойной экран LiYCY 2×0.5 мм² — гибкий, для подвижных установок Cat5e / Cat6 — работает! (120 Ом, но без промышленной изоляции) Сечение проводника:Падение напряжения на кабеле: ΔU = 2 × R_кабеля × I_нагрузка R = ρ × L / S = 0.0175 (Ом·мм²/м) × 1000 м / 0.5 мм² = 35 Ом При токе утечки 100 мА: ΔU = 2 × 35 × 0.1 = 7В — это уже критично! Для длинных линий выбирайте кабель 1.0 мм² и более. Экранирование:Правила заземления экрана: ✅ Заземлять в ОДНОЙ точке — предотвращает контурные токи Обычно: на стороне мастера/ПЛК ❌ Заземлять с ОБОИХ концов — контурный ток протекает по экрану! При разных потенциалах земли создаёт синфазные помехи. Исключение: при частотах > 100 кГц экран заземляют с обоих концов (через конденсатор 10 нФ с одной стороны). Практика: полный пример Modbus RTU на Pythonimport serial import struct import time from typing import Optional class RS485Master: """ Мастер RS-485 с ручным управлением DE/RE через GPIO (для Raspberry Pi). Или с автоматическим через RTS (для USB-RS485 адаптеров). """ def __init__(self, port: str, baudrate: int = 9600, use_rts: bool = True, rts_level: bool = True): """ port: '/dev/ttyUSB0', 'COM3' и т.д. use_rts: использовать RTS для управления DE/RE (USB-адаптеры) rts_level: уровень RTS при передаче (True = HIGH) """ self.ser = serial.Serial( port = port, baudrate = baudrate, bytesize = serial.EIGHTBITS, parity = serial.PARITY_NONE, stopbits = serial.STOPBITS_ONE, timeout = 0.1 ) if use_rts: self.ser.rts = not rts_level # Начальное состояние — приём self.use_rts = use_rts self.rts_level = rts_level # Время передачи одного символа (для задержки после отправки) self.char_time = 10.0 / baudrate # 10 бит на символ def _tx_enable(self): if self.use_rts: self.ser.rts = self.rts_level time.sleep(0.0001) # 100 мкс предвключение def _rx_enable(self): if self.use_rts: # Ждём физической передачи последнего байта time.sleep(self.char_time) self.ser.rts = not self.rts_level def _crc16(self, data: bytes) -> int: crc = 0xFFFF for byte in data: crc ^= byte for _ in range(8): crc = (crc >> 1) ^ 0xA001 if crc & 1 else crc >> 1 return crc def send_raw(self, data: bytes): """Отправка сырых данных""" self.ser.reset_input_buffer() # Очищаем входной буфер перед отправкой self._tx_enable() self.ser.write(data) self.ser.flush() self._rx_enable() def recv_raw(self, expected_len: int, timeout: float = 0.5) -> Optional[bytes]: """Приём данных с таймаутом""" deadline = time.time() + timeout buf = b'' while time.time() < deadline: chunk = self.ser.read(expected_len - len(buf)) buf += chunk if len(buf) >= expected_len: break time.sleep(0.001) return buf if len(buf) == expected_len else None def modbus_read_registers(self, slave: int, func: int, start_addr: int, count: int) -> Optional[list]: """ Чтение Holding (FC=3) или Input (FC=4) регистров. Возвращает список значений или None при ошибке. """ # Формируем запрос request = struct.pack('>BBHH', slave, func, start_addr, count) crc = self._crc16(request) request += struct.pack('<H', crc) # CRC little-endian! self.send_raw(request) # Ожидаемый размер ответа: addr(1)+fc(1)+byte_count(1)+data(count×2)+crc(2) expected = 5 + count * 2 response = self.recv_raw(expected) if response is None: return None # Таймаут # Проверка CRC recv_crc = struct.unpack('<H', response[-2:])[0] calc_crc = self._crc16(response[:-2]) if recv_crc != calc_crc: return None # CRC ошибка # Проверка Exception if response[1] & 0x80: exc_code = response[2] print(f"Modbus Exception: slave={slave}, code={exc_code}") return None # Распаковываем данные byte_count = response[2] values = list(struct.unpack(f'>{count}H', response[3:3+byte_count])) return values def modbus_write_register(self, slave: int, addr: int, value: int) -> bool: """Запись одного Holding регистра (FC=6)""" request = struct.pack('>BBHH', slave, 6, addr, value) crc = self._crc16(request) request += struct.pack('<H', crc) self.send_raw(request) # Ответ = эхо запроса (8 байт) response = self.recv_raw(8) if response is None: return False recv_crc = struct.unpack('<H', response[-2:])[0] return recv_crc == self._crc16(response[:-2]) def close(self): self.ser.close() # ===== ПРИМЕР ИСПОЛЬЗОВАНИЯ ===== def demo_poll_vfd(): """Опрос частотника по Modbus RTU через RS-485""" master = RS485Master('/dev/ttyUSB0', baudrate=9600, use_rts=True) try: print("Опрос частотника (адрес 1)...") while True: # Читаем 6 Input регистров: статус, частота, ток, напряжение, мощность, fault regs = master.modbus_read_registers(slave=1, func=4, start_addr=0, count=6) if regs is None: print("Нет ответа от устройства") else: status = regs[0] freq_hz = regs[1] / 100.0 curr_a = regs[2] / 10.0 volts = regs[3] power_kw = regs[4] / 10.0 fault = regs[5] running = bool(status & 0x0001) faulted = bool(status & 0x0008) print(f"{'▶ ' if running else '⏹'} " f"f={freq_hz:.1f}Гц " f"I={curr_a:.1f}А " f"U={volts}В " f"P={power_kw:.1f}кВт " f"{'🔴АВАРИЯ' if faulted else ''}") if faulted: print(f"Код аварии: {fault}") time.sleep(1.0) except KeyboardInterrupt: print("Остановлено") finally: master.close() demo_poll_vfd() Диагностика: осциллограф и мультиметрИзмерения мультиметром:Линия A-B без сигнала (все устройства молчат): Должно быть: A > B на 200мВ+ (если есть pull-up/down) Плохо: A = B (неопределённое состояние — нужны резисторы смещения) Во время передачи: Осциллограф: чёткие уровни ±3..4В, без выбросов Плохо: размытые фронты → слишком длинная линия или нет терминаторов Плохо: выбросы >±7В → нет снаббера или плохое заземление Измерение дифференциального сигнала: Щуп A → канал 1, Щуп B → канал 2 Включить Math: CH1 - CH2 Должен быть чёткий прямоугольник ±(3..5)В Типичные осциллограммы проблем:Нет терминаторов: ┌──────┐ │ │ ← Нормальный фронт │ │╲ ← Отражение (заброс) ────┘ └─╲────── └── ← Повторное отражение Слишком длинный stub: ┌──────┐ │ ╲___/ ← Паразитное колебание после фронта ────┘ ──── Хорошая линия: ┌──────────┐ ────┘ └──── ← Чёткие фронты без выбросов Гальваническая развязка: когда обязательнаВ промышленных установках "земля" в разных точках может иметь разный потенциал — десятки и даже сотни вольт. Без развязки: Контурный ток по GND-проводу шины RS-485 разрушает трансиверы Помехи от силового оборудования проникают в логику Один неисправный узел выводит из строя всю сеть Когда нужна обязательно: Устройства питаются от разных источников Расстояние между устройствами > 50 м Рядом с шиной есть мощные электродвигатели или сварочное оборудование Разные здания или разные распределительные щиты Варианты реализации: Вариант 1: Изолированный трансивер (ADM2587E, ISO3082) UART ──[Изолированный трансивер]── RS-485 шина Встроенная изоляция 2500 VRMS, нет внешних компонентов Вариант 2: Оптопара + отдельный трансивер UART_TX ──[HCPL2630]── MAX485 ── RS-485 шина UART_RX ──[HCPL2630]──┘ Дешевле, но нужен изолированный DC-DC конвертер питания Вариант 3: Цифровой изолятор + трансивер UART ──[ISO7720]── MAX485 ── RS-485 шина ISO7720: 5 кВ изоляция, 100 Мбит/с, без светодиодов (долговечнее оптопар) Число устройств больше 32: репитеры и конвертерыЕсли нужно более 32 устройств или протяжённость линии превышает допустимую: Репитер RS-485: [Сегмент 1: 32 уст., 600 м]──[Репитер]──[Сегмент 2: 32 уст., 600 м] Репитер переусиливает и переформирует сигнал. Каждый сегмент — отдельная нагрузка. Популярные репитеры: - ADAM-4510 (Advantech): изолированный, DIN-рейка - Moxa MB-9000: с диагностикой - Homemade: MAX487 + MAX487 с DE/RE управлением Защита от грозовых разрядов и ESDПри длинных линиях между зданиями — молниезащита обязательна: Схема защиты RS-485 линии: RS-485 Шина ──[Предохранитель]──[GDT или MOV]──[TVS-диод]──[Трансивер] A P6KE6.8CA Уровни защиты: 1. GDT (газоразрядник) или MOV: ограничивают до 100..500В за наносекунды 2. TVS-диод P6KE6.8CA: ограничивает до ±6.8В, рассеивает 600 Вт импульсно 3. Последовательный резистор 10..22 Ом: ограничивает ток через TVS Готовые модули молниезащиты RS-485: - MTL5000: промышленный барьер - Phoenix Contact TRABTECH: DIN-рейка - УЗИП-485 (отечественный): Microsemi/аналоги Частые вопросы и ответыQ: Можно ли использовать обычный кабель UTP Cat5e вместо специального? A: Можно, и это работает на практике. Cat5e имеет волновое сопротивление 100 Ом (термinator нужен 100 Ом), паразитные ёмкости немного хуже. До 300м/115200 бод — без проблем. Но в промышленной среде с помехами — лучше экранированный кабель. Q: Нужен ли третий провод GND? A: Для корректной работы трансивера входная синфазная помеха не должна выходить за пределы -7В...+12В (Vcm). При длинных линиях и разных заземлениях это нарушается. GND-провод удерживает синфазное напряжение в пределах допустимого. Включайте GND во все промышленные установки. Q: У устройства нет RS-485, только UART. Как подключить? A: Добавить внешний трансивер MAX485/SP3485 + резистор 300 Ом на DE/RE от GPIO. Q: Что делать если устройства разных производителей не видят друг друга? A: Проверить: 1) Скорость/parity/stopbits совпадают. 2) Полярность A/B (иногда производители маркируют наоборот). 3) Адреса устройств уникальны. 4) Нет конфликта адресов. ЗаключениеRS-485 — это не устаревший протокол, а надёжный, проверенный инструмент для промышленных приложений. Правильная линейная топология с терминаторами, экранированная витая пара, правильное управление DE/RE и гальваническая развязка там, где нужно — всё это обеспечит годы надёжной работы. Вложите время в понимание физического уровня: осциллограф + хорошая книга по физике передачи сигналов. Большинство проблем RS-485 решаются на физическом уровне, а не в программе.

Силовая электроника: между схемотехникой и энергетикойСиловая электроника — область, где электроника управляет реальной мощностью: сотнями ампер, тысячами вольт, мегаваттами. Это регуляторы яркости, частотные преобразователи, зарядные станции для электромобилей, солнечные инверторы, промышленные нагреватели. Ключевое отличие от малосигнальной электроники: КПД критически важен. 99% КПД в источнике 100 кВт означает 1 кВт тепла на радиаторах — и это допустимо. 90% КПД — уже 10 кВт тепловых потерь, требующих серьёзного охлаждения. Тиристор (SCR): управляемый диодТиристор (Silicon Controlled Rectifier, SCR) — четырёхслойный PNPN прибор. Включается коротким импульсом на управляющий электрод (Gate), выключается только при уменьшении тока ниже тока удержания (обычно — переходом через ноль в сети AC). Анод (A) ─── P ─── N ─── P ─── N ─── Катод (K) │ Управляющий электрод (G) Характеристики: Включается: короткий импульс IG > IGT (обычно 10–100 мА) Выключается: ток анода < IH (ток удержания), обычно при переходе AC через ноль Прямое напряжение в открытом состоянии: 1–3В (значительные потери!) Применение: однофазные и трёхфазные выпрямители, регуляторы мощности Фазовое управление тиристоромОсновной метод управления мощностью с тиристором в сети AC: Угол включения α=0°: Полная мощность (100%) Угол включения α=90°: Половинная мощность (~50%) Угол включения α=150°: Малая мощность (~6%) Угол включения α=180°: Минимальная мощность (0%) Средняя мощность ≈ P_max × (1 + cos α) / 2 Схема фазового управления на Arduino/STM32: // Детектор перехода через ноль (Zero-Crossing Detector) // Подключён к INT0 (PD2) через оптопару // При каждом переходе через ноль — прерывание volatile bool zero_cross = false; volatile uint8_t power_pct = 50; // 0-100% // Прерывание от детектора нуля ISR(INT0_vect) // AVR / адаптируй под STM32/ESP32 { zero_cross = true; } void setup() { // Выход на оптотиристор/оптосимистор (через ограничительный резистор ~300Ом) pinMode(9, OUTPUT); digitalWrite(9, LOW); // Прерывание INT0 по фронту (или обоим — зависит от схемы) attachInterrupt(0, zero_cross_ISR, RISING); // Таймер 1: генерирует прерывание через N микросекунд после нуля // Для сети 50 Гц: полупериод = 10 000 мкс // Угол 90° = 5 000 мкс задержка } void zero_cross_ISR() { zero_cross = true; } void loop() { if (zero_cross) { zero_cross = false; // Рассчитываем задержку включения // power_pct = 100 → delay = 0 мкс (включить немедленно) // power_pct = 50 → delay = 5000 мкс (угол 90°) // power_pct = 0 → delay = 10000 мкс (не включать) uint32_t delay_us = (100 - power_pct) * 100; // 0-10000 мкс if (power_pct > 0 && power_pct < 100) { delayMicroseconds(delay_us); // Короткий импульс управления (100-200 мкс достаточно) digitalWrite(9, HIGH); delayMicroseconds(100); digitalWrite(9, LOW); } else if (power_pct >= 100) { digitalWrite(9, HIGH); // Постоянно включено } else { digitalWrite(9, LOW); // Постоянно выключено } } } Проблемы фазового управления: Генерирует гармоники в сети (помехи) Вызывает мигание освещения Создаёт радиопомехи (EMI) Решение для нагревателей: управление по полным полупериодам (Burst Firing) Симистор (TRIAC): для двунаправленного управленияСимистор = два тиристора включённых встречно-параллельно. Проводит ток в ОБОИХ направлениях — идеален для управления нагрузкой переменного тока без выпрямления. MT2 │ ── P ─┤ ── N ─┤── Gate (G) ── P ─┤ ── N ─┤ │ MT1 Управление по нулю (Zero-Crossing Control / Burst Firing)Вместо фазового управления — включаем нагреватель на N полных периодов из M: Мощность 33%: ██░░██░░██░░██░░ (1 из 3 периодов включён) Мощность 50%: ████░░░░████░░░░ (2 из 4 периодов) Мощность 75%: ██████░░██████░░ (3 из 4 периодов) Мощность 100%: ████████████████ (все периоды) Преимущества: нет гармоник, нет EMI, нет щелчков в контакторах. Недостатки: медленнее регулирование (минимальный шаг — полпериода = 10 мс). Оптимально для: промышленные нагреватели, печи сопротивления, ИК-нагреватели. // Burst Firing контроллер class BurstController { private: uint8_t window_size; // Размер окна в полупериодах (например, 20 = 200мс) uint8_t on_count; // Сколько периодов включено uint8_t current_period; // Счётчик текущего периода public: BurstController(uint8_t window = 20) : window_size(window), on_count(0), current_period(0) {} void setPower(float power_pct) { on_count = (uint8_t)(power_pct / 100.0f * window_size + 0.5f); on_count = constrain(on_count, 0, window_size); } // Вызвать при каждом переходе через ноль bool onZeroCross() { bool turn_on = (current_period < on_count); current_period = (current_period + 1) % window_size; return turn_on; } }; BurstController burster(20); // Окно 20 полупериодов = 200 мс void zero_cross_handler() { bool should_be_on = burster.onZeroCross(); digitalWrite(TRIAC_PIN, should_be_on ? HIGH : LOW); } Твёрдотельное реле (SSR): простое решениеSSR (Solid-State Relay) — готовый модуль с тиристором/симистором и оптической развязкой внутри. Управление: 3–32В DC сигнал (совместимо с Arduino/ПЛК), нагрузка: до 40А/480В AC. Выбор SSR: Параметр Рекомендация Ток нагрузки Выбирать с запасом ×2 (40А SSR для 20А нагрузки) Тип управления DC Control (3-32V) для ПЛК, AC Control (90-280V) для PID-регулятора с выходом AC Тип коммутации Zero-Cross для нагревателей, Random Fire для двигателей и трансформаторов Напряжение нагрузки 24-380В AC (проверьте соответствие!) Охлаждение ОБЯЗАТЕЛЬНО радиатор! 0.5°C/Вт для алюминиевого радиатора Тепловой расчёт SSR: def calculate_ssr_heatsink(load_current_a: float, ambient_temp_c: float = 40.0, max_case_temp_c: float = 80.0) -> dict: """ Расчёт требуемого теплового сопротивления радиатора для SSR. SSR: прямое напряжение ~1.2В (Fotek, Crydom) """ # Тепловыделение SSR vf = 1.2 # В, прямое падение на симисторе p_loss = vf * load_current_a # Вт # Тепловое сопротивление корпус-радиатор (junction-to-case): ~0.5°C/Вт rth_jc = 0.5 # °C/W (из datasheet) # Максимальная температура p-n перехода обычно 125°C t_junction_max = 125.0 # Требуемое тепловое сопротивление радиатор-воздух # T_ambient + P × (Rth_jc + Rth_heatsink) = T_case_max rth_heatsink = (max_case_temp_c - ambient_temp_c) / p_loss - rth_jc # Размер алюминиевого радиатора (грубая оценка): # R_th ≈ 50 / (площадь_см²) для вертикального расположения heatsink_area_cm2 = 50.0 / rth_heatsink if rth_heatsink > 0 else float('inf') return { 'load_current_a': load_current_a, 'power_loss_w': round(p_loss, 1), 'rth_heatsink': round(rth_heatsink, 2), 'heatsink_area_cm2': round(heatsink_area_cm2, 0), 'safe_operation': rth_heatsink > 0, } # Пример: SSR 25А нагрузки при T_окр=40°C result = calculate_ssr_heatsink(25.0) print(f"Потери: {result['power_loss_w']} Вт") print(f"Требуется радиатор: {result['heatsink_area_cm2']} см²") # Результат: 30 Вт потерь, нужен радиатор ~150 см² IGBT: для высоких частот и больших мощностейIGBT (Insulated Gate Bipolar Transistor) — гибрид MOSFET и биполярного транзистора. Управляется напряжением (как MOSFET), но имеет низкое напряжение насыщения при больших токах (как биполярный). Где используется: Частотные преобразователи (инвертор моста) ИБП и стабилизаторы Сварочные аппараты Индукционные нагреватели Зарядные станции для электромобилей Ключевые характеристики при выборе IGBT:Vce_max — максимальное напряжение коллектор-эмиттер (выбирать ×2 от напряжения шины!) Ic_max — максимальный ток (с учётом теплового сопротивления!) Vce(sat) — напряжение насыщения (потери в открытом состоянии) Eoff/Eon — энергия переключения (потери на коммутацию, растут с частотой!) toff — время выключения (ограничивает максимальную частоту ШИМ) Расчёт потерь IGBT: def calculate_igbt_losses(vce_sat: float, ic_rms: float, e_on_j: float, e_off_j: float, fsw_hz: float, duty: float) -> dict: """ Расчёт потерь IGBT. vce_sat: напряжение насыщения, В (из datasheet при Ic и Tj) ic_rms: действующий ток коллектора, А e_on_j: энергия включения, Дж (из datasheet) e_off_j: энергия выключения, Дж fsw_hz: частота коммутации, Гц duty: скважность ШИМ (0..1) """ # Потери проводимости (conduction losses) # P_cond = Vce_sat × Ic_avg ic_avg = ic_rms * duty # Упрощение для прямоугольного тока p_cond = vce_sat * ic_avg # Потери переключения (switching losses) # P_sw = (E_on + E_off) × fsw p_sw = (e_on_j + e_off_j) * fsw_hz # Суммарные потери p_total = p_cond + p_sw return { 'conduction_w': round(p_cond, 2), 'switching_w': round(p_sw, 2), 'total_w': round(p_total, 2), 'efficiency_pct': round((1 - p_total / (vce_sat * ic_rms + p_total)) * 100, 1), } # Пример: IGBT 1200В/50А в инверторе 400В # Частотник 11 кВт: Idc ≈ 30А, fsw=8кГц, d=0.8 losses = calculate_igbt_losses( vce_sat = 2.0, # В при 125°C ic_rms = 30.0, # А e_on_j = 0.8e-3, # 0.8 мДж e_off_j = 1.2e-3, # 1.2 мДж fsw_hz = 8000, # 8 кГц duty = 0.8 ) print(f"Потери проводимости: {losses['conduction_w']} Вт") print(f"Потери переключения: {losses['switching_w']} Вт") print(f"Итого: {losses['total_w']} Вт") # При 6 IGBT в трёхфазном мосте: ×6 = итоговые потери инвертора Снабберные цепи: защита от перенапряженийПри выключении IGBT/тиристора индуктивная нагрузка создаёт выброс напряжения: V_spike = L × dI/dt. Без защиты — мгновенная смерть транзистора. RC-снаббер (для симистора):Нагрузка │ [Симистор] │ ───────── ← RC снаббер параллельно симистору │ │ [R ~47 Ом] [C ~47 нФ, 630В] │ │ ───────── │ GND Расчёт RC-снаббера: def calculate_rc_snubber(load_inductance_h: float, switch_current_a: float, supply_voltage_v: float) -> dict: """ Расчёт RC-снаббера для тиристора/симистора. Критерий: выброс напряжения ≤ 2 × Vsupply """ import math # Пиковое напряжение без снаббера # V_peak ≈ V_supply + I × sqrt(L/C_parasitic) # С снаббером ограничиваем до 2×V_supply v_max = 2 * supply_voltage_v # Ёмкость снаббера (минимальная для ограничения выброса) # C ≥ L × I² / (V_max - V_supply)² delta_v = v_max - supply_voltage_v c_min = load_inductance_h * switch_current_a**2 / delta_v**2 c_snubber = c_min * 2 # Запас × 2, нормализуем до стандартного ряда E12 # Стандартный ряд конденсаторов (нФ) e12 = [10, 12, 15, 18, 22, 27, 33, 39, 47, 56, 68, 82] c_nf = c_snubber * 1e9 c_standard_nf = min(e12, key=lambda x: abs(x - c_nf)) # Если нет в ряду — берём следующий больший for val in sorted(e12): if val >= c_nf: c_standard_nf = val break c_actual = c_standard_nf * 1e-9 # Сопротивление снаббера # R ≈ sqrt(L/C) для критического затухания r_critical = math.sqrt(load_inductance_h / c_actual) r_snubber = r_critical # Или немного больше для надёжности # Мощность резистора # P = 0.5 × C × V² × f (для каждого переключения) # Для AC 50Гц: f = 100 (2 перехода через ноль) p_resistor = 0.5 * c_actual * supply_voltage_v**2 * 100 return { 'C_nF': c_standard_nf, 'C_voltage': f"{int(v_max * 1.5 / 100) * 100}В", # Округляем до стандарта 'R_ohm': round(r_snubber, 0), 'R_watts': round(p_resistor * 2, 1), # Запас × 2 } # Пример: управление нагревателем 5 кВт через симистор # Нагреватель = почти чисто активная нагрузка, но есть монтажная индуктивность ~10 мкГн result = calculate_rc_snubber(10e-6, 22.7, 220) print(f"Снаббер: R={result['R_ohm']} Ом/{result['R_watts']} Вт, " f"C={result['C_nF']} нФ/{result['C_voltage']}") Трёхфазное управление нагревателямиДля трёхфазных печей и нагревателей используют 3 SSR (по одному на каждую фазу) или специализированные трёхфазные тиристорные модули: L1 ──[SSR_A]──┐ L2 ──[SSR_B]──┼── Нагреватели (треугольник или звезда) L3 ──[SSR_C]──┘ ↑ ↑ ↑ Сигналы управления с ПЛК/контроллера Управление: все три SSR получают одинаковый сигнал (одновременно вкл/выкл) → симметричная нагрузка, нет перекоса фаз Или: поочерёдное включение фаз (phase rotation) → снижает пиковый ток пуска Алгоритм ПИД для температуры печи с трёхфазным нагревателем: class FurnaceController: """Контроллер температуры трёхфазной печи""" def __init__(self, kp=5.0, ki=0.1, kd=2.0, max_power=100.0): self.pid = PIDController(kp, ki, kd, ts=1.0, out_min=0, out_max=max_power) self.setpoint = 0.0 # Защиты self.max_temp = 1200.0 # °C максимум печи self.fault = False def control_cycle(self, temp_actual: float) -> dict: """Один цикл управления (вызывать каждую секунду)""" # Защита по превышению температуры if temp_actual > self.max_temp: self.fault = True if self.fault: return {'ssr_output': 0.0, 'fault': True, 'temp': temp_actual} # ПИД self.pid.set_setpoint(self.setpoint) power_pct = self.pid.compute(temp_actual) # Мощность → количество периодов включения (Burst Fire) # Окно 10 полупериодов = 100 мс on_periods = int(power_pct / 100 * 10 + 0.5) return { 'ssr_output': power_pct, 'on_periods': on_periods, # из 10 'setpoint': self.setpoint, 'temp_actual': temp_actual, 'error': self.setpoint - temp_actual, 'fault': False, } Тепловой расчёт: как не сжечь компонентыТепловая цепь аналогична электрической: Температура ↔ Напряжение Мощность потерь ↔ Ток Тепловое сопротивление Rth ↔ Электрическое сопротивление T_junction = T_ambient + P_loss × (Rth_j-c + Rth_c-hs + Rth_hs-a) Где: Rth_j-c — тепловое сопротивление кристалл→корпус (из datasheet) Rth_c-hs — корпус→радиатор (зависит от термопасты, ~0.1–0.5 °C/Вт) Rth_hs-a — радиатор→воздух (зависит от размера и обдува) def thermal_check(p_loss_w: float, t_ambient_c: float, rth_jc: float, rth_chs: float, rth_hsa: float, t_junction_max: float = 125.0) -> dict: """ Проверка теплового режима силового прибора. """ rth_total = rth_jc + rth_chs + rth_hsa t_junction = t_ambient_c + p_loss_w * rth_total t_case = t_ambient_c + p_loss_w * rth_hsa # Температура корпуса margin = t_junction_max - t_junction safe = margin > 10.0 # Запас минимум 10°C return { 'T_junction_c': round(t_junction, 1), 'T_case_c': round(t_case, 1), 'margin_c': round(margin, 1), 'safe': safe, 'warning': not safe, } # Проверка SSR 30А: result = thermal_check( p_loss_w = 1.2 * 25, # 1.2В × 25А = 30 Вт t_ambient_c = 40.0, rth_jc = 0.5, # из datasheet Fotek SSR-40DA rth_chs = 0.2, # хорошая термопаста rth_hsa = 1.5, # алюминиевый радиатор 150 см² ) print(f"Температура перехода: {result['T_junction_c']}°C") print(f"Запас: {result['margin_c']}°C — {'OK' if result['safe'] else 'ОПАСНО!'}") Типичные ошибки в силовой электроникеНет снаббера на индуктивной нагрузке → выброс напряжения → смерть тиристора SSR без радиатора → перегрев за несколько минут при токе > 5А Управляющий сигнал без оптической развязки → 220В на Arduino/ПЛК = конец Не проверена полярность тиристора → не переключается или горит сразу Фазовое управление на трансформаторную нагрузку → насыщение сердечника IGBT с Vce_max = Vsupply → первый же выброс убивает → минимум ×2 запас ЗаключениеСиловая электроника — область, где цена ошибки высока: сгоревший IGBT, пожар, травма. Всегда работайте с полной изоляцией от сети, используйте изолирующие трансформаторы при разработке, не экономьте на снабберах и радиаторах. Для начала: освойте управление нагревателем через SSR и ПИД-регулятор — это самая распространённая и безопасная задача. Потом — изучите теорию тиристорного управления. После этого — IGBT в H-мостах для двигателей. И только с хорошей теоретической базой — трёхфазные инверторы. Измерительный осциллограф с изолированными щупами и клещи-амперметр — ваши обязательные инструменты в этой области. Без возможности видеть что происходит на осциллографе — работать в силовой электронике вслепую.

Python — второй язык каждого инженераMatlab стоит дорого. LabVIEW — ещё дороже. Excel мощный, но у него есть потолок. Python — бесплатный, открытый, с огромной экосистемой библиотек для инженерных задач. И с каждым годом он глубже проникает в промышленность. Инженер-электронщик использует Python для: анализа данных с измерительных приборов, автоматизации рутинных расчётов, создания отчётов, обработки сигналов с АЦП, управления лабораторным оборудованием (VISA/PyVISA), прототипирования алгоритмов перед переносом на микроконтроллер. Специалист АСУ ТП — для: работы с Modbus/OPC UA, парсинга логов ПЛК, автоматического тестирования, интеграции различных систем. NumPy: числа быстроNumPy — фундамент научного Python. Массивы NumPy в 10–100 раз быстрее списков Python для математических операций. import numpy as np import time # ===== БАЗОВЫЕ ОПЕРАЦИИ ===== # Создание массивов t = np.linspace(0, 10, 1000) # 1000 точек от 0 до 10 f = np.arange(0, 50, 0.1) # От 0 до 50 с шагом 0.1 zeros = np.zeros((3, 4)) # Матрица 3×4 из нулей eye = np.eye(3) # Единичная матрица 3×3 # Синтетический сигнал (для теста) freq_signal = 50.0 # Гц freq_noise = 200.0 # Гц (помеха) sample_rate = 1000.0 # Гц t = np.arange(0, 1, 1/sample_rate) # 1 секунда данных signal_clean = 2.0 * np.sin(2 * np.pi * freq_signal * t) noise = 0.5 * np.sin(2 * np.pi * freq_noise * t) noise += 0.2 * np.random.randn(len(t)) # Белый шум signal_noisy = signal_clean + noise # ===== СКОРОСТЬ ===== def python_rms(data: list) -> float: return (sum(x**2 for x in data) / len(data)) ** 0.5 def numpy_rms(data: np.ndarray) -> float: return np.sqrt(np.mean(data**2)) # Сравнение скорости: data_list = list(signal_noisy) data_arr = np.array(data_list) t0 = time.time(); python_rms(data_list); t_py = time.time() - t0 t0 = time.time(); numpy_rms(data_arr); t_np = time.time() - t0 print(f"Python: {t_py*1000:.2f} мс, NumPy: {t_np*1000:.3f} мс, " f"Ускорение: {t_py/t_np:.0f}x") # ===== ИНЖЕНЕРНЫЕ РАСЧЁТЫ ===== def calculate_power_factor(voltage: np.ndarray, current: np.ndarray, sample_rate: float) -> dict: """ Расчёт коэффициента мощности из осциллограмм тока и напряжения. """ # RMS значения V_rms = np.sqrt(np.mean(voltage**2)) I_rms = np.sqrt(np.mean(current**2)) # Активная мощность (среднее произведение) P = np.mean(voltage * current) # Полная мощность S = V_rms * I_rms # Коэффициент мощности pf = P / S if S > 0 else 0 # Реактивная мощность Q = np.sqrt(max(0, S**2 - P**2)) return { 'V_rms': round(V_rms, 2), 'I_rms': round(I_rms, 3), 'P_kw': round(P / 1000, 2), 'Q_kvar': round(Q / 1000, 2), 'S_kva': round(S / 1000, 2), 'PF': round(abs(pf), 3), } # Пример использования: # Генерируем тестовые сигналы 220В 50Гц, ток 10А с φ=30° t = np.linspace(0, 0.04, 400) # 2 периода V = 220 * np.sqrt(2) * np.sin(2 * np.pi * 50 * t) I = 10 * np.sqrt(2) * np.sin(2 * np.pi * 50 * t - np.radians(30)) power = calculate_power_factor(V, I, 10000) print(f"P={power['P_kw']} кВт, Q={power['Q_kvar']} квар, cos(φ)={power['PF']}") # Ожидаем: cos(30°) ≈ 0.866 Scipy: анализ сигналовfrom scipy import signal, fft import numpy as np # ===== FFT: СПЕКТРАЛЬНЫЙ АНАЛИЗ ===== def analyze_spectrum(data: np.ndarray, sample_rate: float) -> dict: """ Анализ спектра сигнала через FFT. Используется для диагностики вибраций, качества электроэнергии. """ n = len(data) # Оконная функция (Hanning) для уменьшения спектральных утечек window = np.hanning(n) data_windowed = data * window # FFT spectrum = np.abs(fft.rfft(data_windowed)) freqs = fft.rfftfreq(n, 1.0/sample_rate) # Нормировка (учёт оконной функции) spectrum = spectrum / (n / 2) # THD (Total Harmonic Distortion) — для качества сетевого напряжения # Находим основную частоту (50 Гц) fundamental_idx = np.argmin(np.abs(freqs - 50.0)) fundamental_amp = spectrum[fundamental_idx] # Гармоники 2-я...7-я harmonic_power = sum( spectrum[np.argmin(np.abs(freqs - 50.0 * n))]**2 for n in range(2, 8) ) thd = np.sqrt(harmonic_power) / fundamental_amp * 100 # % # Топ-5 пиков спектра peak_indices = np.argsort(spectrum)[-10:][::-1] top_peaks = [(round(freqs[i], 1), round(spectrum[i], 4)) for i in peak_indices] return { 'freqs': freqs, 'spectrum': spectrum, 'thd_pct': round(thd, 2), 'top_peaks': top_peaks[:5], 'rms': round(np.sqrt(np.mean(data**2)), 4), } # ===== ФИЛЬТРАЦИЯ СИГНАЛОВ ===== def design_lowpass_filter(cutoff_hz: float, sample_rate: float, order: int = 4) -> tuple: """ Проектирование фильтра нижних частот Баттерворта. Используется для сглаживания зашумлённых данных датчиков. """ nyquist = sample_rate / 2 normalized_cutoff = cutoff_hz / nyquist b, a = signal.butter(order, normalized_cutoff, btype='low', analog=False) return b, a def apply_filter(data: np.ndarray, b: np.ndarray, a: np.ndarray, zero_phase: bool = True) -> np.ndarray: """ Применение фильтра к сигналу. zero_phase=True: filtfilt (нет фазового сдвига, требует данных полностью) zero_phase=False: lfilter (реального времени, есть фазовый сдвиг) """ if zero_phase: return signal.filtfilt(b, a, data) # Двупроходной (офлайн-обработка) else: return signal.lfilter(b, a, data) # Однопроходной (онлайн-обработка) # Пример: фильтрация зашумлённого датчика температуры sample_rate = 100.0 # 100 Гц t = np.arange(0, 10, 1/sample_rate) # Реальная температура (медленно меняется) true_temp = 75.0 + 5.0 * np.sin(2 * np.pi * 0.1 * t) # 0.1 Гц # С шумом (50Гц помеха от сети + белый шум) noisy_temp = true_temp + 2.0 * np.sin(2 * np.pi * 50 * t) + \ 0.5 * np.random.randn(len(t)) # Фильтр НЧ с частотой среза 1 Гц (убираем всё выше 1 Гц) b, a = design_lowpass_filter(cutoff_hz=1.0, sample_rate=sample_rate) filtered_temp = apply_filter(noisy_temp, b, a) print(f"Шум до фильтрации: {np.std(noisy_temp - true_temp):.3f}°C") print(f"Шум после фильтра: {np.std(filtered_temp - true_temp):.3f}°C") # ===== КОРРЕЛЯЦИЯ И ОБНАРУЖЕНИЕ СИГНАЛА ===== def find_pattern_in_signal(signal_data: np.ndarray, pattern: np.ndarray) -> list[int]: """ Поиск паттерна в сигнале через кросс-корреляцию. Применение: нахождение пакетов в потоке данных, обнаружение событий. """ correlation = np.correlate(signal_data, pattern, mode='valid') threshold = 0.8 * np.max(np.abs(correlation)) peaks, _ = signal.find_peaks(correlation, height=threshold, distance=len(pattern)) return list(peaks) Pandas: анализ промышленных данныхimport pandas as pd import numpy as np from datetime import datetime, timedelta # ===== ЗАГРУЗКА И ОЧИСТКА ДАННЫХ ===== def load_plc_log(filepath: str) -> pd.DataFrame: """ Загрузка и нормализация лога ПЛК. Типичный формат: CSV с временной меткой и значениями тегов. """ df = pd.read_csv(filepath, parse_dates=['timestamp'], index_col='timestamp') # Нормализация имён колонок df.columns = df.columns.str.lower().str.replace(' ', '_').str.replace('.', '_') # Приведение типов numeric_cols = ['temperature', 'pressure', 'current', 'flow'] for col in numeric_cols: if col in df.columns: df[col] = pd.to_numeric(df[col], errors='coerce') bool_cols = ['running', 'fault', 'alarm'] for col in bool_cols: if col in df.columns: df[col] = df[col].astype(bool, errors='ignore') # Удаление дубликатов df = df[~df.index.duplicated(keep='first')] # Сортировка по времени df = df.sort_index() # Интерполяция пропущенных значений (не более 5 пропусков подряд) df[numeric_cols] = df[numeric_cols].interpolate( method='time', limit=5, limit_direction='forward' ) return df def analyze_production_data(df: pd.DataFrame) -> dict: """ Анализ производственных данных: KPI, простои, отклонения. """ results = {} # ===== ДОСТУПНОСТЬ ОБОРУДОВАНИЯ ===== if 'running' in df.columns: total_time = (df.index[-1] - df.index[0]).total_seconds() / 3600 # часы running_time = df['running'].mean() * total_time results['availability'] = { 'total_hours': round(total_time, 1), 'running_hours': round(running_time, 1), 'availability_pct': round(df['running'].mean() * 100, 1), } # ===== АНАЛИЗ ПРОСТОЕВ ===== if 'running' in df.columns: # Нахождение периодов простоя running_changes = df['running'].astype(int).diff() stop_times = df.index[running_changes == -1] # Моменты остановки start_times = df.index[running_changes == 1] # Моменты пуска downtimes = [] for stop in stop_times: # Найти следующий пуск после остановки next_start = start_times[start_times > stop] if len(next_start) > 0: duration = (next_start[0] - stop).total_seconds() / 60 # минуты downtimes.append({'stop': stop, 'start': next_start[0], 'duration_min': round(duration, 1)}) if downtimes: dt_df = pd.DataFrame(downtimes) results['downtimes'] = { 'count': len(dt_df), 'total_min': round(dt_df['duration_min'].sum(), 1), 'avg_min': round(dt_df['duration_min'].mean(), 1), 'max_min': round(dt_df['duration_min'].max(), 1), 'longest_stop': dt_df.loc[dt_df['duration_min'].idxmax(), 'stop'].isoformat(), } # ===== СТАТИСТИКА ПАРАМЕТРОВ ===== numeric_cols = df.select_dtypes(include=np.number).columns.tolist() if numeric_cols: stats = df[numeric_cols].describe() results['parameters'] = stats.to_dict() # ===== ОБНАРУЖЕНИЕ ВЫБРОСОВ (метод IQR) ===== outliers = {} for col in numeric_cols: Q1 = df[col].quantile(0.25) Q3 = df[col].quantile(0.75) IQR = Q3 - Q1 mask = (df[col] < Q1 - 1.5 * IQR) | (df[col] > Q3 + 1.5 * IQR) outlier_count = mask.sum() if outlier_count > 0: outliers[col] = { 'count': int(outlier_count), 'pct': round(outlier_count / len(df) * 100, 2), 'examples': df[mask][col].head(3).tolist(), } results['outliers'] = outliers return results # ===== ГЕНЕРАЦИЯ ОТЧЁТОВ ===== def generate_daily_report(df: pd.DataFrame, date: str = None) -> pd.DataFrame: """Сводная таблица по часам за день""" if date: df = df[df.index.date == pd.Timestamp(date).date()] # Агрегация по часам hourly = df.resample('1h').agg({ 'temperature': ['mean', 'min', 'max'], 'current': ['mean', 'max'], 'pressure': ['mean', 'min', 'max'], 'running': 'mean', # Доступность за час 'fault': 'any', # Были ли аварии }).round(2) # Плоские имена колонок hourly.columns = ['_'.join(col) for col in hourly.columns] hourly['availability_pct'] = (hourly['running_mean'] * 100).round(1) hourly['had_fault'] = hourly['fault_any'] return hourly # ===== EXCEL ОТЧЁТ ===== def export_to_excel(df: pd.DataFrame, hourly: pd.DataFrame, kpi: dict, filepath: str): """Красивый Excel-отчёт с несколькими листами""" with pd.ExcelWriter(filepath, engine='xlsxwriter') as writer: workbook = writer.book # Форматы header_fmt = workbook.add_format({ 'bold': True, 'bg_color': '#2C3E50', 'font_color': 'white', 'border': 1 }) number_fmt = workbook.add_format({'num_format': '0.0#', 'border': 1}) pct_fmt = workbook.add_format({'num_format': '0.0%', 'border': 1}) bad_fmt = workbook.add_format({'bg_color': '#FFB3B3', 'border': 1}) # ===== Лист 1: KPI ===== ws_kpi = workbook.add_worksheet('KPI') ws_kpi.write('A1', 'Показатель', header_fmt) ws_kpi.write('B1', 'Значение', header_fmt) avail = kpi.get('availability', {}) row = 1 for key, val in avail.items(): ws_kpi.write(row, 0, key) ws_kpi.write(row, 1, val) row += 1 ws_kpi.set_column('A:A', 25) ws_kpi.set_column('B:B', 15) # ===== Лист 2: Почасовой отчёт ===== hourly.to_excel(writer, sheet_name='Почасовой отчёт', startrow=1) ws = writer.sheets['Почасовой отчёт'] # Условное форматирование: красим аварийные часы ws.conditional_format('A2:Z1000', { 'type': 'formula', 'criteria': '=$G2=TRUE', # Если был fault 'format': bad_fmt }) # ===== Лист 3: Сырые данные (последние 1000 строк) ===== df.tail(1000).to_excel(writer, sheet_name='Данные') print(f"Отчёт сохранён: {filepath}") FastAPI: REST API для промышленных данных# pip install fastapi uvicorn from fastapi import FastAPI, HTTPException, Query from pydantic import BaseModel from typing import Optional from datetime import datetime import uvicorn app = FastAPI( title="Industrial Data API", description="API для доступа к данным производственного оборудования", version="1.0.0" ) # Модели данных class TelemetryPoint(BaseModel): device: str temperature: float current: float pressure: float running: bool timestamp: datetime class DeviceCommand(BaseModel): device: str command: str # "start", "stop", "set_setpoint" value: Optional[float] = None operator: str # Имитация БД (в реальности — запросы к InfluxDB/TimescaleDB) telemetry_db = [] @app.get("/api/v1/devices", summary="Список устройств") async def get_devices(): return { "devices": [ {"id": "pump1", "name": "Насос 1", "location": "Линия 1", "online": True}, {"id": "pump2", "name": "Насос 2", "location": "Линия 1", "online": True}, {"id": "valve1","name": "Клапан 1","location": "Линия 2","online": False}, ] } @app.get("/api/v1/telemetry/{device_id}", summary="Телеметрия устройства") async def get_telemetry( device_id: str, hours: int = Query(default=1, ge=1, le=720, description="Глубина истории"), resample: str = Query(default="1min", description="Гранулярность: 1s, 1min, 5min, 1h") ): # Проверка устройства valid_devices = ["pump1", "pump2", "valve1"] if device_id not in valid_devices: raise HTTPException(status_code=404, detail=f"Устройство '{device_id}' не найдено") # Запрос к historian (имитация) # В реальности: query_influxdb(device_id, hours, resample) return { "device": device_id, "from": (datetime.now().replace(hour=0, minute=0)).isoformat(), "to": datetime.now().isoformat(), "resample": resample, "points": [ {"time": datetime.now().isoformat(), "temperature": 85.3, "current": 15.2, "pressure": 5.8, "running": True} ] } @app.post("/api/v1/commands", summary="Отправить команду устройству", status_code=202) async def send_command(cmd: DeviceCommand): # Валидация команды valid_commands = ["start", "stop", "set_setpoint"] if cmd.command not in valid_commands: raise HTTPException(status_code=400, detail=f"Неизвестная команда: {cmd.command}") if cmd.command == "set_setpoint" and cmd.value is None: raise HTTPException(status_code=400, detail="set_setpoint требует параметр value") # Аудит-лог (обязательно для промышленных систем!) print(f"[AUDIT] {datetime.now()} | Operator: {cmd.operator} | " f"Device: {cmd.device} | Command: {cmd.command} | Value: {cmd.value}") # Отправить команду (через очередь, Modbus, OPC UA...) # command_queue.put(cmd) return {"status": "accepted", "command_id": "cmd_123456"} @app.get("/api/v1/health", summary="Healthcheck") async def health(): return {"status": "ok", "timestamp": datetime.now().isoformat()} # Запуск: uvicorn main:app --host 0.0.0.0 --port 8080 --reload asyncio: асинхронный опрос оборудованияimport asyncio import aiohttp import json from datetime import datetime async def poll_device_modbus(device_id: str, host: str, interval: float = 1.0): """Асинхронный опрос устройства через Modbus TCP""" from pymodbus.client import AsyncModbusTcpClient async with AsyncModbusTcpClient(host=host, port=502) as client: print(f"Подключён к {device_id} ({host})") while True: start = asyncio.get_event_loop().time() try: result = await client.read_input_registers(address=0, count=4, slave=1) if not result.isError(): data = { 'device': device_id, 'timestamp': datetime.now().isoformat(), 'temperature': result.registers[0] / 10.0, 'current': result.registers[1] / 10.0, 'pressure': result.registers[2] / 100.0, 'running': bool(result.registers[3] & 1), } # Публикуем данные (в очередь, БД, MQTT...) print(f"{device_id}: T={data['temperature']}°C") else: print(f"{device_id}: Ошибка Modbus") except Exception as e: print(f"{device_id}: {e}") await asyncio.sleep(5) # Пауза перед повтором continue # Точный интервал опроса elapsed = asyncio.get_event_loop().time() - start await asyncio.sleep(max(0, interval - elapsed)) async def main(): """Параллельный опрос нескольких устройств""" devices = [ ("pump1", "192.168.1.10"), ("pump2", "192.168.1.11"), ("valve1", "192.168.1.12"), ] # Запускаем все опросы параллельно tasks = [poll_device_modbus(dev_id, host, interval=1.0) for dev_id, host in devices] await asyncio.gather(*tasks) # Все работают одновременно! asyncio.run(main()) Полезные однострочники для инженераimport subprocess, json, struct, serial from pathlib import Path # Быстрый Modbus опрос из командной строки: # python -c "from pymodbus.client import ModbusTcpClient; c=ModbusTcpClient('192.168.1.10'); c.connect(); print(c.read_input_registers(0,4,slave=1).registers)" # Конвертация hex-дампа в float: def hex_to_float(hex_str: str) -> float: return struct.unpack('>f', bytes.fromhex(hex_str.replace(' ','')))[0] print(hex_to_float("42 48 00 00")) # → 50.0 # Поиск COM-портов: import serial.tools.list_ports for p in serial.tools.list_ports.comports(): print(f"{p.device}: {p.description}") # Быстрый парсинг CSV с временными метками: df = pd.read_csv('data.csv', parse_dates=['time'], index_col='time') print(df.resample('5min').mean()) # Сохранение данных в Parquet (быстрее CSV в 10-50 раз): df.to_parquet('data.parquet', compression='snappy') df2 = pd.read_parquet('data.parquet') ЗаключениеPython — это не замена C для микроконтроллеров и не замена SQL для баз данных. Это клей, который соединяет всё: читает данные из любого источника, анализирует, визуализирует, отправляет куда надо. Для инженера ключевые библиотеки: NumPy (быстрые вычисления), Pandas (анализ данных), SciPy (сигналы и системы), Matplotlib/Plotly (визуализация), pymodbus (Modbus), pyserial (UART), asyncua (OPC UA), FastAPI (REST API). Вложите неделю в изучение NumPy и Pandas — окупится сотнями часов сэкономленного времени на анализе данных, отчётах и автоматизации рутины.

Зачем RTOS на микроконтроллереПростой проект — один while(1) цикл. Всё хорошо: считали датчик, обновили дисплей, проверили кнопку. Но что если: Нужно принять UART-пакет точно за 10 мс, иначе потеряем байты Одновременно управлять тремя независимыми ПИД-контурами Обрабатывать CAN-сообщения с задержкой не более 5 мс И параллельно вести логирование на SD-карту Суперцикл (while(1)) ломается: длинная операция блокирует всё остальное. Прерывания помогают, но сложная логика в прерываниях — путь к хаосу. FreeRTOS решает это элегантно: каждая задача — отдельный "поток" со своим стеком и приоритетом. Планировщик переключает их так быстро (обычно каждые 1 мс), что кажется будто они работают одновременно. Задача с высоким приоритетом всегда получает процессор раньше. Ключевые концепции FreeRTOSTask (Задача)// Прототип задачи — бесконечный цикл! void vTaskFunction(void *pvParameters) { // Инициализация задачи int *param = (int *)pvParameters; for (;;) // Никогда не выходит! { // Работа задачи... // Уступить процессор (обязательно в каждом цикле!) vTaskDelay(pdMS_TO_TICKS(100)); // Пауза 100 мс } vTaskDelete(NULL); // Никогда не достигается, но хорошая практика } // Создание задачи: TaskHandle_t xTaskHandle = NULL; xTaskCreate( vTaskFunction, // Функция задачи "TaskName", // Имя (для отладки) configMINIMAL_STACK_SIZE * 4, // Размер стека в словах NULL, // Параметр (pvParameters) tskIDLE_PRIORITY + 2, // Приоритет (выше = важнее) &xTaskHandle // Хендл задачи ); ПриоритетыconfigMAX_PRIORITIES = 7 (типично) Приоритет 6: КРИТИЧЕСКИЙ (ISR-уровень, прерывания) Приоритет 5: Коммуникации реального времени (CAN, UART) Приоритет 4: Управление (ПИД-контроллеры) Приоритет 3: Мониторинг, аварийная логика Приоритет 2: UI, дисплей, кнопки Приоритет 1: Логирование, некритичные задачи Приоритет 0: Idle task (только когда все остальные ждут) Архитектура многозадачного приложенияРеальный пример: контроллер насосной станции на STM32F4. // ===== ЗАГОЛОВКИ ===== #include "FreeRTOS.h" #include "task.h" #include "queue.h" #include "semphr.h" #include "timers.h" // ===== ГЛОБАЛЬНЫЕ ОБЪЕКТЫ FreeRTOS ===== QueueHandle_t xSensorQueue; // Данные датчиков QueueHandle_t xCommandQueue; // Команды управления QueueHandle_t xLogQueue; // Сообщения лога SemaphoreHandle_t xI2CMutex; // Защита I2C шины SemaphoreHandle_t xUARTMutex; // Защита UART (printf) TimerHandle_t xHeartbeatTimer; // Мигание LED watchdog // ===== СТРУКТУРЫ ДАННЫХ ===== typedef struct { float temperature; float pressure; float flow; uint32_t timestamp_ms; uint8_t quality; // 0=BAD, 1=UNCERTAIN, 2=GOOD } SensorData_t; typedef enum { CMD_START, CMD_STOP, CMD_SET_SETPOINT, CMD_RESET_FAULT, } CommandType_t; typedef struct { CommandType_t type; float value; uint8_t source; // 0=HMI, 1=Modbus, 2=Auto } Command_t; typedef struct { char message[80]; uint8_t level; // 0=DEBUG, 1=INFO, 2=WARN, 3=ERROR uint32_t timestamp_ms; } LogMessage_t; // ===== ВСПОМОГАТЕЛЬНЫЙ МАКРОС ДЛЯ PRINTF ===== // Потокобезопасный printf через мьютекс #define LOG(level, fmt, ...) do { \ LogMessage_t msg; \ msg.level = (level); \ msg.timestamp_ms = xTaskGetTickCount(); \ snprintf(msg.message, sizeof(msg.message), fmt, ##__VA_ARGS__); \ xQueueSend(xLogQueue, &msg, 0); \ } while(0) #define LOG_INFO(fmt,...) LOG(1, fmt, ##__VA_ARGS__) #define LOG_WARN(fmt,...) LOG(2, "[WARN] " fmt, ##__VA_ARGS__) #define LOG_ERROR(fmt,...) LOG(3, "[ERR!] " fmt, ##__VA_ARGS__) // ===== ЗАДАЧА 1: ЧТЕНИЕ ДАТЧИКОВ (Приоритет 4) ===== static void vSensorTask(void *pvParam) { SensorData_t data; TickType_t xLastWakeTime = xTaskGetTickCount(); const TickType_t xPeriod = pdMS_TO_TICKS(100); // 10 Гц LOG_INFO("Sensor task started"); for (;;) { // Ждём ровно 100 мс от последнего пробуждения // vTaskDelayUntil гарантирует точный период! vTaskDelayUntil(&xLastWakeTime, xPeriod); data.timestamp_ms = xTaskGetTickCount(); // Захватываем I2C шину if (xSemaphoreTake(xI2CMutex, pdMS_TO_TICKS(50)) == pdTRUE) { data.temperature = BMP280_ReadTemperature(); data.pressure = BMP280_ReadPressure(); data.quality = 2; // GOOD xSemaphoreGive(xI2CMutex); } else { // I2C занята дольше 50 мс — что-то пошло не так data.quality = 0; // BAD LOG_WARN("I2C timeout in sensor task"); } // Читаем расходомер через 4-20мА data.flow = ADC_ReadFlow(); // Отправляем данные в очередь (не блокируем — если полная, пропускаем) if (xQueueSend(xSensorQueue, &data, 0) != pdTRUE) { LOG_WARN("Sensor queue full!"); } } } // ===== ЗАДАЧА 2: ПИД-УПРАВЛЕНИЕ (Приоритет 5) ===== static void vControlTask(void *pvParam) { SensorData_t sensorData; Command_t command; float setpoint = 5.0f; // Уставка давления, бар float output = 0.0f; bool running = false; bool fault = false; // ПИД параметры float kp = 2.0f, ki = 0.5f, kd = 0.1f; float integral = 0.0f, prevError = 0.0f; const float TS = 0.1f; // Совпадает с периодом датчиков LOG_INFO("Control task started"); for (;;) { // Ждём новые данные датчиков (блокирующий ждём до 200 мс) if (xQueueReceive(xSensorQueue, &sensorData, pdMS_TO_TICKS(200)) == pdTRUE) { // Проверяем входящие команды (неблокирующий) while (xQueueReceive(xCommandQueue, &command, 0) == pdTRUE) { switch (command.type) { case CMD_START: running = true; fault = false; integral = 0; LOG_INFO("Pump STARTED by source %d", command.source); break; case CMD_STOP: running = false; output = 0; LOG_INFO("Pump STOPPED by source %d", command.source); break; case CMD_SET_SETPOINT: setpoint = command.value; LOG_INFO("Setpoint changed to %.1f bar", setpoint); break; case CMD_RESET_FAULT: fault = false; LOG_INFO("Fault reset"); break; } } // Защиты if (sensorData.quality == 0) { running = false; fault = true; output = 0; LOG_ERROR("Sensor fault! Emergency stop."); } if (sensorData.pressure > 12.0f) { running = false; fault = true; output = 0; LOG_ERROR("High pressure! %.2f bar > 12.0 bar", sensorData.pressure); } if (sensorData.temperature > 90.0f) { running = false; fault = true; output = 0; LOG_ERROR("Motor overtemp! %.1f°C > 90°C", sensorData.temperature); } // ПИД вычисление if (running && !fault) { float error = setpoint - sensorData.pressure; integral += ki * error * TS; integral = fmaxf(-50.0f, fminf(50.0f, integral)); // Anti-windup float derivative = -(sensorData.pressure - prevError) / TS; prevError = sensorData.pressure; output = kp * error + integral + kd * derivative; output = fmaxf(0.0f, fminf(100.0f, output)); } else { output = 0.0f; integral = 0.0f; } // Применяем управляющий сигнал VFD_SetFrequency(output * 0.5f); // 0-100% → 0-50 Гц } else { // Таймаут ожидания данных датчика — авария LOG_ERROR("Sensor data timeout!"); running = false; output = 0; VFD_SetFrequency(0); } } } // ===== ЗАДАЧА 3: MODBUS SLAVE (Приоритет 3) ===== static void vModbusTask(void *pvParam) { uint8_t rxBuf[64]; uint8_t rxLen = 0; LOG_INFO("Modbus task started"); for (;;) { // Ждём байт из UART (через семафор от прерывания) if (UART_WaitForData(rxBuf, &rxLen, pdMS_TO_TICKS(100))) { // Обрабатываем Modbus запрос if (Modbus_ProcessRequest(rxBuf, rxLen)) { // Если команда — отправляем в очередь управления Command_t cmd; if (Modbus_ExtractCommand(&cmd)) { xQueueSend(xCommandQueue, &cmd, pdMS_TO_TICKS(10)); } } } } } // ===== ЗАДАЧА 4: ЛОГИРОВАНИЕ (Приоритет 1, самый низкий) ===== static void vLoggingTask(void *pvParam) { LogMessage_t msg; const char *levelNames[] = {"DBG", "INF", "WRN", "ERR"}; for (;;) { // Ждём сообщение из очереди if (xQueueReceive(xLogQueue, &msg, portMAX_DELAY) == pdTRUE) { // Пишем в UART (захватываем мьютекс) if (xSemaphoreTake(xUARTMutex, pdMS_TO_TICKS(50)) == pdTRUE) { printf("[%6lu][%s] %s\r\n", (unsigned long)msg.timestamp_ms, levelNames[msg.level % 4], msg.message); xSemaphoreGive(xUARTMutex); } // Пишем на SD-карту (низкий приоритет = не мешаем критичным задачам) // SD_AppendLog(&msg); } } } // ===== ТАЙМЕР HEARTBEAT ===== static void vHeartbeatCallback(TimerHandle_t xTimer) { // Мигаем светодиодом — система жива HAL_GPIO_TogglePin(GPIOC, GPIO_PIN_13); } // ===== ИНИЦИАЛИЗАЦИЯ ===== void App_Init(void) { // Создаём очереди xSensorQueue = xQueueCreate(5, sizeof(SensorData_t)); xCommandQueue = xQueueCreate(10, sizeof(Command_t)); xLogQueue = xQueueCreate(20, sizeof(LogMessage_t)); // Создаём мьютексы xI2CMutex = xSemaphoreCreateMutex(); xUARTMutex = xSemaphoreCreateMutex(); // Создаём задачи xTaskCreate(vSensorTask, "Sensors", 512, NULL, 4, NULL); xTaskCreate(vControlTask, "Control", 1024, NULL, 5, NULL); xTaskCreate(vModbusTask, "Modbus", 512, NULL, 3, NULL); xTaskCreate(vLoggingTask, "Logging", 256, NULL, 1, NULL); // Создаём таймер heartbeat (500 мс) xHeartbeatTimer = xTimerCreate("Heartbeat", pdMS_TO_TICKS(500), pdTRUE, NULL, vHeartbeatCallback); xTimerStart(xHeartbeatTimer, 0); // Запуск планировщика vTaskStartScheduler(); // Никогда не должно дойти сюда! for (;;); } Очереди: безопасная передача данных между задачамиОчередь — это основной механизм коммуникации в FreeRTOS. Thread-safe, FIFO, блокирующий. // Создание очереди на 10 элементов типа uint32_t QueueHandle_t xQueue = xQueueCreate(10, sizeof(uint32_t)); // Отправка (из задачи) uint32_t value = 42; xQueueSend(xQueue, &value, pdMS_TO_TICKS(100)); // Ждём 100мс если полная // Отправка с высоким приоритетом (в начало очереди) xQueueSendToFront(xQueue, &value, 0); // Отправка из прерывания (другая функция!) BaseType_t xHigherPriorityTaskWoken = pdFALSE; xQueueSendFromISR(xQueue, &value, &xHigherPriorityTaskWoken); portYIELD_FROM_ISR(xHigherPriorityTaskWoken); // Уступить если нужно // Приём (блокирует до данных или таймаута) uint32_t received; if (xQueueReceive(xQueue, &received, portMAX_DELAY) == pdTRUE) { // Данные получены } // "Подсмотреть" без извлечения xQueuePeek(xQueue, &received, 0); // Мониторинг UBaseType_t count = uxQueueMessagesWaiting(xQueue); // Сколько элементов UBaseType_t space = uxQueueSpacesAvailable(xQueue); // Сколько свободно Семафоры и мьютексы: защита разделяемых ресурсов// ===== МЬЮТЕКС (Mutual Exclusion) ===== // Для защиты ресурсов (I2C, SPI, UART, глобальные переменные) SemaphoreHandle_t xMutex = xSemaphoreCreateMutex(); // Правильный паттерн: void safe_i2c_read(uint8_t addr, uint8_t reg, uint8_t *buf, uint8_t len) { if (xSemaphoreTake(xMutex, pdMS_TO_TICKS(100)) == pdTRUE) { HAL_I2C_Mem_Read(&hi2c1, addr, reg, 1, buf, len, 100); xSemaphoreGive(xMutex); } else { // Таймаут — логируем, возвращаем ошибку } } // ===== ДВОИЧНЫЙ СЕМАФОР (уведомление о событии) ===== // Задача ждёт событие от ISR SemaphoreHandle_t xDataReadySemaphore = xSemaphoreCreateBinary(); // В прерывании (данные готовы): void HAL_UART_RxCpltCallback(UART_HandleTypeDef *huart) { BaseType_t xWoken = pdFALSE; xSemaphoreGiveFromISR(xDataReadySemaphore, &xWoken); portYIELD_FROM_ISR(xWoken); } // В задаче (ждём данные): void vProcessingTask(void *pvParam) { for (;;) { // Эффективное ожидание — задача не потребляет CPU! xSemaphoreTake(xDataReadySemaphore, portMAX_DELAY); // Данные готовы — обрабатываем process_uart_data(); } } // ===== СЧЁТНЫЙ СЕМАФОР (ограничение конкурентного доступа) ===== // Пример: максимум 3 одновременных подключения SemaphoreHandle_t xConnectionSlots = xSemaphoreCreateCounting(3, 3); void handle_new_connection() { if (xSemaphoreTake(xConnectionSlots, pdMS_TO_TICKS(5000)) == pdTRUE) { // Слот получен serve_client(); xSemaphoreGive(xConnectionSlots); // Освободить слот } else { // Нет свободных слотов send_busy_response(); } } Программные таймеры// Таймер однократный (one-shot) vs периодический TimerHandle_t xOneShotTimer; TimerHandle_t xPeriodicTimer; void vTimerCallback(TimerHandle_t xTimer) { // pvTimerGetTimerID позволяет использовать один callback для многих таймеров uint32_t timerID = (uint32_t)pvTimerGetTimerID(xTimer); switch (timerID) { case 1: // Однократный таймер — отключить нагреватель через 30 сек Heater_Off(); break; case 2: // Периодический — опрос watchdog External_WDT_Kick(); break; } } void setup_timers(void) { // One-shot таймер (не перезапускается автоматически) xOneShotTimer = xTimerCreate( "Heater", pdMS_TO_TICKS(30000), // 30 секунд pdFALSE, // pdFALSE = one-shot (void *)1, // ID таймера vTimerCallback ); // Периодический таймер xPeriodicTimer = xTimerCreate( "WDT", pdMS_TO_TICKS(500), // 500 мс pdTRUE, // pdTRUE = периодический (void *)2, vTimerCallback ); xTimerStart(xPeriodicTimer, 0); // Запустить один-шот когда нужно: // xTimerStart(xOneShotTimer, 0); // Сбросить периодический (перезапустить отсчёт): // xTimerReset(xPeriodicTimer, 0); // Изменить период на лету: // xTimerChangePeriod(xPeriodicTimer, pdMS_TO_TICKS(1000), 0); } Управление памятью и отладка// Мониторинг стека задачи (важно для нахождения переполнений!) void vCheckStackTask(void *pvParam) { for (;;) { vTaskDelay(pdMS_TO_TICKS(5000)); // Минимальный остаток стека (в словах) с начала работы UBaseType_t hwm = uxTaskGetStackHighWaterMark(NULL); if (hwm < 50) { // Меньше 50 слов — опасно! printf("WARNING: Task '%s' stack low! HWM=%lu words\r\n", pcTaskGetName(NULL), (unsigned long)hwm); } } } // Вывод информации о всех задачах (для отладки) void vPrintTaskStats(void) { char buffer[512]; vTaskList(buffer); // Требует configUSE_TRACE_FACILITY=1 printf("Task Name\t\tState\tPrio\tStack\tNum\r\n%s", buffer); // Загрузка CPU по задачам (требует configGENERATE_RUN_TIME_STATS=1) vTaskGetRunTimeStats(buffer); printf("\r\nTask\t\t\tTime\t\t%%\r\n%s", buffer); } // Обработчик нехватки памяти void vApplicationMallocFailedHook(void) { taskDISABLE_INTERRUPTS(); printf("FATAL: malloc failed! Heap exhausted.\r\n"); for (;;); } // Переполнение стека задачи void vApplicationStackOverflowHook(TaskHandle_t xTask, char *pcTaskName) { taskDISABLE_INTERRUPTS(); printf("FATAL: Stack overflow in task '%s'!\r\n", pcTaskName); for (;;); } FreeRTOS на ESP32ESP-IDF (официальный SDK ESP32) использует FreeRTOS как основу. На двухядерном ESP32 задачи можно привязывать к конкретному ядру: // ESP32-специфичное создание задачи с указанием ядра xTaskCreatePinnedToCore( vWiFiTask, // Функция "WiFi", // Имя 8192, // Стек (в байтах для ESP32!) NULL, // Параметры 5, // Приоритет &xWiFiHandle, // Хендл 0 // Ядро: 0 = Protocol CPU, 1 = Application CPU ); // WiFi и Bluetooth — всегда на ядре 0 (Protocol CPU) // Ваш код — лучше на ядре 1 (Application CPU) // Это разделяет сетевой стек и бизнес-логику // Встроенный мониторинг задач ESP-IDF: void print_esp_task_info(void) { printf("Free heap: %u bytes\r\n", esp_get_free_heap_size()); printf("Min free heap: %u bytes\r\n", esp_get_minimum_free_heap_size()); } Типичные ошибки FreeRTOS1. Вызов обычных функций из ISR // НЕПРАВИЛЬНО — заблокирует прерывание! void HAL_GPIO_EXTI_Callback(uint16_t pin) { xQueueSend(xQueue, &data, portMAX_DELAY); // ОШИБКА: блокирует ISR! } // ПРАВИЛЬНО — FromISR версии функций: void HAL_GPIO_EXTI_Callback(uint16_t pin) { BaseType_t xWoken = pdFALSE; xQueueSendFromISR(xQueue, &data, &xWoken); portYIELD_FROM_ISR(xWoken); } 2. Бесконечная задача без yield // НЕПРАВИЛЬНО — монополизирует процессор! void vBadTask(void *p) { for (;;) { do_something(); // Нет vTaskDelay или блокирующего ожидания! } } // ПРАВИЛЬНО: void vGoodTask(void *p) { for (;;) { do_something(); vTaskDelay(pdMS_TO_TICKS(10)); // Уступаем хотя бы 10 мс } } 3. Доступ к глобальным данным без защиты // НЕПРАВИЛЬНО — race condition! float g_temperature = 0; void vSensor(void *p) { g_temperature = read_sensor(); } void vControl(void *p) { if (g_temperature > 80) alarm(); } // ПРАВИЛЬНО — через очередь или мьютекс ЗаключениеFreeRTOS превращает микроконтроллер из последовательного автомата в полноценную многозадачную систему. Это не усложнение ради усложнения — это решение реальных проблем: независимость задач, чёткие интерфейсы через очереди, защита ресурсов через мьютексы. Начните с малого: замените суперцикл двумя задачами — одна читает датчик, другая управляет выходом, общаются через очередь. Это уже даст почувствовать преимущества. FreeRTOS поддерживается на STM32, ESP32, Arduino (с ограничениями), Raspberry Pi Pico и десятках других платформ. Документация на freertos.org — отличная, с примерами и объяснениями.

Зачем специализированная БД для временных рядовТехнологические данные — это всегда временной ряд: температура каждую секунду, давление каждые 100 мс, состояние оборудования каждые 10 мс. PostgreSQL или MySQL могут хранить такие данные. Но при миллионах записей в день начинаются проблемы. Почему реляционные БД плохо справляются: Индексы B-Tree неэффективны для временных запросов ("за последний час") Запись строк в таблицу с индексами — медленно при высоком темпе GROUP BY time_interval требует дорогих вычислений Партиционирование по времени нужно настраивать вручную Хранение тысяч тегов → тысячи колонок или плохая схема Что умеют Time-Series TSDB: Оптимизированная запись: 100 000+ точек/сек на скромном железе Встроенное сжатие (delta-delta, XOR float compression) Автоматические retention policies (TTL данных) Downsampling: автоматически агрегируем "горячие" данные в "холодные" Встроенные временны́е функции: moving average, rate, derivative InfluxDB 2.x: промышленный стандарт IoTОсновные концепцииMeasurement — аналог таблицы: measurement: "telemetry" Tags — индексированные метаданные (строки): tags: device="conveyor1", location="line1", area="factory" Fields — неиндексированные данные (числа, строки, bool): fields: temperature=87.3, current=15.5, running=true Timestamp — время с нано-точностью. Точка данных (Point): measurement,tags fields timestamp telemetry,device=conveyor1,location=line1 temperature=87.3,current=15.5 1710000000000000000 Почему Tags vs Fields важноTags: ИНДЕКСИРОВАНЫ → используйте для группировки/фильтрации device, location, sensor_type, unit_id Fields: НЕ индексированы → используйте для числовых данных temperature, pressure, current, voltage ОШИБКА: положить temperature в Tag — поиск по значению работает, но карданальность огромная → индекс разрастётся → InfluxDB замедлится. ОШИБКА: положить device_id в Field — нельзя эффективно фильтровать по устройству. Python клиент InfluxDB 2.x:from influxdb_client import InfluxDBClient, Point, WritePrecision from influxdb_client.client.write_api import SYNCHRONOUS, WriteOptions from datetime import datetime, timezone import time INFLUX_URL = "http://localhost:8086" INFLUX_TOKEN = "your-api-token-here" INFLUX_ORG = "factory" INFLUX_BUCKET = "process_data" # Клиент с батчевой записью client = InfluxDBClient(url=INFLUX_URL, token=INFLUX_TOKEN, org=INFLUX_ORG) write_api = client.write_api(write_options=WriteOptions( batch_size = 1000, # Накапливаем до 1000 точек flush_interval = 5_000, # Или сбрасываем каждые 5 секунд jitter_interval = 500, # ±500мс для сглаживания нагрузки retry_interval = 5_000, # Retry при ошибке через 5с max_retry_time = 180_000, # Максимум 3 минуты retry )) query_api = client.query_api() # ===== ЗАПИСЬ ===== def write_single_point(device: str, location: str, temperature: float, current: float, running: bool): """Запись одной точки""" point = ( Point("telemetry") .tag("device", device) .tag("location", location) .field("temperature", temperature) .field("current", current) .field("running", int(running)) # bool → int (InfluxDB лучше хранит) .time(datetime.now(timezone.utc)) ) write_api.write(bucket=INFLUX_BUCKET, record=point) def write_batch(measurements: list[dict]): """ Эффективная пакетная запись. measurements: [{'device': 'pump1', 'temp': 25.3, 'current': 12.1}, ...] """ points = [] for m in measurements: p = ( Point("telemetry") .tag("device", m['device']) .tag("location", m.get('location', 'unknown')) .field("temperature", float(m.get('temp', 0))) .field("current", float(m.get('current', 0))) .field("pressure", float(m.get('pressure', 0))) ) points.append(p) write_api.write(bucket=INFLUX_BUCKET, record=points) # Запись в нативном line protocol (максимальная производительность): def write_line_protocol(lines: list[str]): """ Прямая запись в line protocol — самый быстрый способ. Формат: measurement[,tag=value...] field=value[,field=value...] [timestamp] """ write_api.write(bucket=INFLUX_BUCKET, record='\n'.join(lines), write_precision=WritePrecision.NANOSECONDS) # Пример: lines = [ "telemetry,device=pump1,location=line1 temperature=87.3,current=15.5 1710000000000000000", "telemetry,device=pump2,location=line1 temperature=72.1,current=8.2 1710000000000000000", "telemetry,device=valve1,location=line2 position=75.0 1710000000000000000", ] write_line_protocol(lines) # ===== ЗАПРОСЫ (Flux) ===== def query_last_hour(device: str) -> list[dict]: """Последний час данных устройства""" flux = f''' from(bucket: "{INFLUX_BUCKET}") |> range(start: -1h) |> filter(fn: (r) => r._measurement == "telemetry") |> filter(fn: (r) => r.device == "{device}") |> filter(fn: (r) => r._field == "temperature" or r._field == "current") |> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value") |> sort(columns: ["_time"]) ''' tables = query_api.query(flux) results = [] for table in tables: for record in table.records: results.append({ 'time': record.get_time().isoformat(), 'temperature': record.values.get('temperature'), 'current': record.values.get('current'), }) return results def query_aggregated_stats(device: str, window: str = "5m", range_start: str = "-24h") -> list[dict]: """ Агрегированная статистика по временным окнам. window: "1m", "5m", "1h", "1d" """ flux = f''' from(bucket: "{INFLUX_BUCKET}") |> range(start: {range_start}) |> filter(fn: (r) => r._measurement == "telemetry" and r.device == "{device}") |> filter(fn: (r) => r._field == "temperature") |> aggregateWindow( every: {window}, fn: (tables=<-, column) => tables |> reduce( identity: {{mean: 0.0, min: 99999.0, max: -99999.0, count: 0}}, fn: (r, accumulator) => ({{ mean: accumulator.mean + r._value, min: if r._value < accumulator.min then r._value else accumulator.min, max: if r._value > accumulator.max then r._value else accumulator.max, count: accumulator.count + 1, }}) ), createEmpty: false ) ''' # Для простого avg/min/max лучше использовать встроенные функции: flux_simple = f''' from(bucket: "{INFLUX_BUCKET}") |> range(start: {range_start}) |> filter(fn: (r) => r._measurement == "telemetry" and r.device == "{device}" and r._field == "temperature") |> aggregateWindow(every: {window}, fn: mean, createEmpty: false) |> yield(name: "mean") ''' tables = query_api.query(flux_simple) return [{'time': r.get_time().isoformat(), 'mean_temp': r.get_value()} for table in tables for r in table.records] def query_anomalies(threshold_high: float = 85.0, range_start: str = "-7d") -> list[dict]: """Поиск аномалий — превышений порога""" flux = f''' from(bucket: "{INFLUX_BUCKET}") |> range(start: {range_start}) |> filter(fn: (r) => r._measurement == "telemetry" and r._field == "temperature") |> filter(fn: (r) => r._value > {threshold_high}) |> group(columns: ["device"]) |> sort(columns: ["_time"], desc: true) ''' tables = query_api.query(flux) return [{ 'device': r.values.get('device'), 'time': r.get_time().isoformat(), 'value': r.get_value(), 'excess': round(r.get_value() - threshold_high, 2), } for table in tables for r in table.records] def query_device_availability(range_start: str = "-30d") -> list[dict]: """Доступность (availability) по устройствам за период""" flux = f''' import "math" total = from(bucket: "{INFLUX_BUCKET}") |> range(start: {range_start}) |> filter(fn: (r) => r._measurement == "telemetry" and r._field == "running") |> group(columns: ["device"]) |> count() |> rename(columns: {{_value: "total_count"}}) running = from(bucket: "{INFLUX_BUCKET}") |> range(start: {range_start}) |> filter(fn: (r) => r._measurement == "telemetry" and r._field == "running") |> filter(fn: (r) => r._value == 1) |> group(columns: ["device"]) |> count() |> rename(columns: {{_value: "running_count"}}) join(tables: {{total, running}}, on: ["device"]) |> map(fn: (r) => ({{ r with availability_pct: math.round(x: r.running_count / r.total_count * 1000.0) / 10.0 }})) ''' tables = query_api.query(flux) return [{'device': r.values.get('device'), 'availability': r.values.get('availability_pct')} for table in tables for r in table.records] Retention Policies и DownsamplingХранить сырые данные с секундным разрешением 10 лет — безумно дорого. Правильная стратегия: "Горячие" данные: 1 секунда, 30 дней → быстрый SSD "Тёплые" данные: 1 минута, 1 год → обычный SSD "Холодные" данные: 1 час, 10 лет → HDD/объектное хранилище Конфигурация в InfluxDB 2.x:# Создание bucket с retention 30 дней (сырые данные) influx bucket create \ --name process_data_raw \ --retention 30d \ --org factory # Bucket для агрегированных данных (бессрочно) influx bucket create \ --name process_data_aggregated \ --retention 0 \ --org factory Задача downsampling (Flux):def setup_downsampling_task(): """ Создаём задачу InfluxDB для автоматического downsampling. Каждые 5 минут агрегируем сырые данные в минутные. """ flux_task = ''' option task = { name: "Downsampling: raw→1min", every: 5m, // Запускать каждые 5 минут offset: 1m, // Смещение (ждём пока данные придут) } // Читаем сырые данные за последние 5 минут data = from(bucket: "process_data_raw") |> range(start: -task.every) |> filter(fn: (r) => r._measurement == "telemetry") // Агрегируем каждую числовую метрику data |> filter(fn: (r) => r._field == "temperature" or r._field == "current" or r._field == "pressure") |> aggregateWindow(every: 1m, fn: mean, createEmpty: false) |> set(key: "_measurement", value: "telemetry_1m") |> to(bucket: "process_data_aggregated") // Для бинарных данных (running) — используем last data |> filter(fn: (r) => r._field == "running") |> aggregateWindow(every: 1m, fn: last, createEmpty: false) |> set(key: "_measurement", value: "telemetry_1m") |> to(bucket: "process_data_aggregated") ''' # Создание задачи через API tasks_api = client.tasks_api() task = tasks_api.create_task_every( name="Downsampling: raw→1min", flux=flux_task, every="5m", organization=INFLUX_ORG ) print(f"Задача создана: {task.id}") TimescaleDB: PostgreSQL для временных рядовTimescaleDB — расширение PostgreSQL. Если вы уже используете PostgreSQL и знаете SQL — это лучший выбор. Вы получаете TSDB-оптимизации при сохранении полного SQL. -- Установка расширения CREATE EXTENSION IF NOT EXISTS timescaledb; -- Обычная таблица PostgreSQL CREATE TABLE telemetry ( time TIMESTAMPTZ NOT NULL, device TEXT NOT NULL, location TEXT NOT NULL, temperature FLOAT, current FLOAT, pressure FLOAT, running BOOLEAN, quality TEXT DEFAULT 'GOOD' ); -- Превращаем в hypertable (TimescaleDB магия!) SELECT create_hypertable('telemetry', 'time', chunk_time_interval => INTERVAL '1 day' -- Партиция = 1 день ); -- Индекс на часто используемые теги CREATE INDEX ON telemetry (device, time DESC); CREATE INDEX ON telemetry (location, time DESC); -- Compression (сжатие старых данных) ALTER TABLE telemetry SET ( timescaledb.compress, timescaledb.compress_segmentby = 'device', timescaledb.compress_orderby = 'time DESC' ); -- Автоматическое сжатие данных старше 7 дней SELECT add_compression_policy('telemetry', INTERVAL '7 days'); -- Автоматическое удаление старых данных (30 дней) SELECT add_retention_policy('telemetry', INTERVAL '30 days'); Запросы (обычный SQL!):-- Последний час данных с устройства SELECT time, temperature, current, running FROM telemetry WHERE device = 'pump1' AND time > NOW() - INTERVAL '1 hour' ORDER BY time DESC; -- Среднее по 5-минутным окнам SELECT time_bucket('5 minutes', time) AS bucket, device, ROUND(AVG(temperature)::numeric, 2) AS avg_temp, ROUND(MIN(temperature)::numeric, 2) AS min_temp, ROUND(MAX(temperature)::numeric, 2) AS max_temp, COUNT(*) AS samples FROM telemetry WHERE device = 'pump1' AND time > NOW() - INTERVAL '24 hours' GROUP BY bucket, device ORDER BY bucket DESC; -- Обнаружение аномалий (значение > avg + 2*stddev) WITH stats AS ( SELECT device, AVG(temperature) AS avg_temp, STDDEV(temperature) AS std_temp FROM telemetry WHERE time > NOW() - INTERVAL '7 days' GROUP BY device ) SELECT t.time, t.device, t.temperature, s.avg_temp, s.std_temp, (t.temperature - s.avg_temp) / NULLIF(s.std_temp, 0) AS z_score FROM telemetry t JOIN stats s ON t.device = s.device WHERE t.time > NOW() - INTERVAL '24 hours' AND ABS((t.temperature - s.avg_temp) / NULLIF(s.std_temp, 0)) > 2.0 ORDER BY ABS((t.temperature - s.avg_temp) / NULLIF(s.std_temp, 0)) DESC LIMIT 50; -- Доступность оборудования за месяц SELECT device, COUNT(*) FILTER (WHERE running = true) AS running_count, COUNT(*) AS total_count, ROUND( COUNT(*) FILTER (WHERE running = true)::numeric / COUNT(*) * 100, 1 ) AS availability_pct, SUM(CASE WHEN running THEN 1 ELSE 0 END) * EXTRACT(EPOCH FROM INTERVAL '1 second') / 3600.0 AS running_hours FROM telemetry WHERE time > NOW() - INTERVAL '30 days' GROUP BY device ORDER BY availability_pct DESC; Непрерывные агрегации (Continuous Aggregates):-- Создаём материализованное представление с автообновлением CREATE MATERIALIZED VIEW telemetry_5min WITH (timescaledb.continuous) AS SELECT time_bucket('5 minutes', time) AS bucket, device, location, AVG(temperature) AS avg_temp, MIN(temperature) AS min_temp, MAX(temperature) AS max_temp, AVG(current) AS avg_current, MAX(current) AS max_current, BOOL_OR(running) AS any_running, COUNT(*) AS sample_count FROM telemetry GROUP BY bucket, device, location WITH NO DATA; -- Автоматическое обновление каждые 5 минут SELECT add_continuous_aggregate_policy('telemetry_5min', start_offset => INTERVAL '15 minutes', end_offset => INTERVAL '5 minutes', schedule_interval => INTERVAL '5 minutes' ); -- Запрос к агрегированным данным (мгновенно!) SELECT * FROM telemetry_5min WHERE device = 'pump1' AND bucket > NOW() - INTERVAL '24 hours' ORDER BY bucket DESC; Python + SQLAlchemy + TimescaleDB:from sqlalchemy import create_engine, text from sqlalchemy.orm import Session import pandas as pd from datetime import datetime, timedelta, timezone DATABASE_URL = "postgresql://user:password@localhost:5432/factory_db" engine = create_engine(DATABASE_URL, pool_size=10, max_overflow=20) class TelemetryRepository: def write_batch(self, records: list[dict]) -> int: """Пакетная запись телеметрии""" if not records: return 0 with engine.begin() as conn: result = conn.execute( text(""" INSERT INTO telemetry (time, device, location, temperature, current, pressure, running) VALUES (:time, :device, :location, :temperature, :current, :pressure, :running) ON CONFLICT DO NOTHING """), records ) return result.rowcount def get_latest(self, device: str, fields: list[str] = None) -> dict | None: """Последнее значение устройства""" field_list = ', '.join(fields or ['temperature', 'current', 'pressure', 'running']) with engine.connect() as conn: row = conn.execute( text(f""" SELECT time, {field_list} FROM telemetry WHERE device = :device ORDER BY time DESC LIMIT 1 """), {'device': device} ).fetchone() return dict(row._mapping) if row else None def get_as_dataframe(self, device: str, hours: int = 24) -> pd.DataFrame: """Загрузка данных в Pandas DataFrame для анализа""" query = text(""" SELECT time, temperature, current, pressure, running FROM telemetry WHERE device = :device AND time > :since ORDER BY time """) with engine.connect() as conn: df = pd.read_sql( query, conn, params={'device': device, 'since': datetime.now(timezone.utc) - timedelta(hours=hours)}, parse_dates=['time'], index_col='time' ) return df def detect_anomalies_zscore(self, device: str, field: str = 'temperature', threshold: float = 2.5) -> pd.DataFrame: """Обнаружение аномалий методом z-score""" df = self.get_as_dataframe(device, hours=24) if df.empty or field not in df.columns: return pd.DataFrame() mean = df[field].mean() std = df[field].std() if std == 0: return pd.DataFrame() df['z_score'] = (df[field] - mean) / std anomalies = df[df['z_score'].abs() > threshold].copy() anomalies['is_high'] = anomalies['z_score'] > 0 return anomalies[['z_score', field, 'is_high']] def get_equipment_report(self, days: int = 30) -> pd.DataFrame: """Отчёт по оборудованию за период""" query = text(""" SELECT device, COUNT(*) as total_records, COUNT(*) FILTER (WHERE running) as running_records, ROUND((COUNT(*) FILTER (WHERE running)::numeric / COUNT(*) * 100)::numeric, 1) as availability_pct, ROUND(AVG(temperature)::numeric, 1) as avg_temp, ROUND(MAX(temperature)::numeric, 1) as max_temp, ROUND(AVG(current)::numeric, 2) as avg_current FROM telemetry WHERE time > NOW() - MAKE_INTERVAL(days => :days) GROUP BY device ORDER BY device """) with engine.connect() as conn: return pd.read_sql(query, conn, params={'days': days}) Выбор TSDB: сравнительная таблицаКритерий InfluxDB 2.x TimescaleDB Prometheus Основа Собственный движок PostgreSQL Собственный Язык запросов Flux (мощный, непривычный) SQL PromQL Производительность записи ★★★★★ ★★★★ ★★★ SQL-совместимость ❌ ✅ (полная) ❌ Сжатие ★★★★★ ★★★★ ★★★ Масштабирование InfluxDB Enterprise TimescaleDB Thanos/Cortex Лицензия BSL (OSS ограничен) Apache 2 Apache 2 Интеграция с Grafana ★★★★★ ★★★★★ ★★★★★ Лучше для IoT, большой объём тегов Существующий PostgreSQL-стек DevOps мониторинг ЗаключениеВыбор TSDB зависит от контекста. InfluxDB — лучший выбор для чистых IoT/телеметрия проектов: максимальная производительность, мощный Flux для временны́х вычислений, отличная экосистема. TimescaleDB — если уже есть PostgreSQL инфраструктура, нужны JOINs с другими данными или разработчики лучше знают SQL. Ключевые принципы для production: всегда настраивайте retention policies (данные должны автоматически удаляться), используйте downsampling для долгосрочного хранения агрегатов, настройте сжатие (экономия 90%+ дискового пространства), мониторьте производительность самой TSDB. Deadband-фильтрация на уровне edge-узла (не писать если значение не изменилось существенно) снижает нагрузку на БД в 5–50 раз для медленно меняющихся процессов. Это первое что нужно сделать перед любой оптимизацией TSDB.

Git: это не просто "сохранить файл"Git изобрёл Линус Торвальдс в 2005 году за две недели — потому что существующие системы контроля версий его раздражали. Результат стал стандартом де-факто для всей современной разработки. Но большинство разработчиков используют только 10% возможностей Git: git add, git commit, git push. И потом удивляются, почему в команде хаос, история проекта нечитаема, а деплой — это страшный ритуал. Правильное использование Git — это не набор команд, это культура разработки. Сегодня разберём, как устроена эта культура в реальных командах. Анатомия правильного коммитаКоммит — это единица изменений. Плохой коммит: "исправил баги и добавил фичи". Хороший коммит: одно логическое изменение, понятное описание. Conventional Commits — стандарт сообщений<type>(<scope>): <description> [optional body] [optional footer(s)] Типы: feat — новая функциональность fix — исправление бага docs — только документация style — форматирование, точки с запятой (нет изменений логики) refactor — рефакторинг (нет новой функциональности, нет фикса) perf — оптимизация производительности test — добавление тестов chore — обслуживание: обновление зависимостей, конфигурации CI ci — изменения CI/CD конфигурации revert — откат предыдущего коммита Примеры: # Плохо: git commit -m "fix" git commit -m "wip" git commit -m "changes" git commit -m "поправил немного" # Хорошо: git commit -m "fix(auth): исправлена утечка токена при logout" git commit -m "feat(modbus): добавлена поддержка FC15 (write multiple coils)" git commit -m "perf(historian): оптимизирован batch-insert, +340% throughput" git commit -m "docs(api): добавлены примеры для /api/v1/devices endpoint" # С телом для сложных изменений: git commit -m "fix(plc): исправлено переполнение счётчика при rollover Счётчик типа UINT использовался для значений >65535. Изменён на DINT (32-бит). Затронутые устройства: все узлы с FC03. Closes #247" Почему это важно?Автоматический CHANGELOG — инструменты как conventional-changelog генерируют его автоматически Семантическое версионирование — feat → minor, fix → patch, feat! или BREAKING CHANGE → major Читаемая история — через год понятно что и зачем было сделано Быстрый поиск — git log --grep="fix(modbus)" найдёт все фиксы Modbus Стратегии ветвленияGit Flow — классика для релизного циклаmain ────────────────────────────────────── (production-ready, теги версий) \ / release/1.2.0 ─────────────────────────── (только bagfixes перед релизом) \ / develop ──────────────────────────────── (интеграция фич) \ \ / feat/A feat/B feat/C Ветки: main — всегда стабильный, деплоится в прод, только через merge из release/* develop — основная ветка разработки, всегда должна собираться feature/* — новые фичи, создаются из develop, мержатся в develop release/* — подготовка релиза (версия, changelog), только bugfix hotfix/* — срочные фиксы прод, мержатся в main И develop # Создать фичу: git checkout develop git checkout -b feature/modbus-fc15-support # Завершить фичу: git checkout develop git merge --no-ff feature/modbus-fc15-support # --no-ff сохраняет историю git branch -d feature/modbus-fc15-support # Подготовить релиз: git checkout develop git checkout -b release/1.2.0 # Обновить версию, CHANGELOG... git commit -m "chore(release): version 1.2.0" git checkout main git merge --no-ff release/1.2.0 git tag -a v1.2.0 -m "Release 1.2.0" git checkout develop git merge --no-ff release/1.2.0 Trunk-Based Development — для быстрых командВсе разработчики работают в одной ветке (main), фичи прячутся за feature-флагами. Деплой несколько раз в день. Подходит для опытных команд с хорошим покрытием тестами. # Только короткоживущие ветки (1-2 дня максимум) git checkout -b task/PLC-247-fix-counter-overflow # ... работа ... git push origin task/PLC-247-fix-counter-overflow # Pull Request → Review → Merge в main # Деплой автоматически GitHub Flow — для непрерывного деплояУпрощённый Git Flow без develop: main = то, что в проде Feature branches — от main, в main через PR Деплой = merge в main Code Review: как делать правильноCode review — не поиск ошибок, это обмен знаниями и повышение качества. Хороший review делает команду сильнее. Для автора PR:## Описание Добавлена поддержка записи нескольких coils (FC15) в Modbus slave. ## Мотивация Клиент запросил управление 16 выходными реле через один Modbus-запрос вместо 16 отдельных FC05. Уменьшает нагрузку на шину в 16 раз. ## Изменения - `ModbusSlave::handle_fc15()` — новый обработчик функционального кода 15 - Обновлён маппинг coils на GPIO пины - Добавлены unit-тесты: 8 тест-кейсов ## Тестирование - [x] Unit-тесты: все зелёные - [x] Интеграционный тест с реальным Modbus-мастером (Python pymodbus) - [x] Проверен на железе: Raspberry Pi + MCP2551 ## Breaking Changes Нет. FC05 продолжает работать. ## Связанные Issues Closes #247 Для ревьюера:Проверяйте: Логику — правильно ли реализовано то, что задумано? Граничные случаи — что при пустых данных? При переполнении? При сетевой ошибке? Безопасность — нет ли SQL-инъекций, XSS, незащищённых данных? Производительность — нет ли N+1 запросов, бесконечных циклов? Тесты — покрывают ли они описанную функциональность? Документацию — понятно ли из кода и комментариев что происходит? Не проверяйте: Стиль форматирования (для этого есть линтеры и форматтеры) Личные предпочтения (если оба подхода корректны) Тон комментариев: ❌ Это неправильно, так делать нельзя ✅ Здесь возможно переполнение при dlc > 8, как насчёт проверки? ❌ Почему ты использовал цикл вместо map()? ✅ Можно ли тут использовать list comprehension для читаемости? ❌ Нет, переделай. ✅ Мне кажется, паттерн Strategy тут подошёл бы лучше — как думаешь? GitHub Actions: автоматизация всего# .github/workflows/ci.yml name: CI/CD Pipeline on: push: branches: [ main, develop ] pull_request: branches: [ main, develop ] jobs: # ===== ЛИНТИНГ И СТАТИЧЕСКИЙ АНАЛИЗ ===== lint: name: Lint & Static Analysis runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - name: Set up Python uses: actions/setup-python@v4 with: python-version: '3.11' cache: 'pip' - name: Install dependencies run: | pip install flake8 pylint mypy black isort pip install -r requirements.txt - name: Check formatting (black) run: black --check --diff . - name: Check imports (isort) run: isort --check-only --diff . - name: Lint (flake8) run: flake8 . --max-line-length=100 --exclude=.venv,migrations - name: Type check (mypy) run: mypy src/ --strict --ignore-missing-imports # ===== ТЕСТИРОВАНИЕ ===== test: name: Unit & Integration Tests runs-on: ubuntu-latest needs: lint services: # Поднимаем сервисы для интеграционных тестов redis: image: redis:7 ports: ['6379:6379'] mosquitto: image: eclipse-mosquitto:2 ports: ['1883:1883'] strategy: matrix: python-version: ['3.10', '3.11', '3.12'] # Тестируем на всех версиях steps: - uses: actions/checkout@v4 - name: Set up Python ${{ matrix.python-version }} uses: actions/setup-python@v4 with: python-version: ${{ matrix.python-version }} cache: 'pip' - name: Install dependencies run: pip install -r requirements.txt -r requirements-dev.txt - name: Run tests with coverage run: | pytest tests/ \ --cov=src \ --cov-report=xml \ --cov-report=term-missing \ --cov-fail-under=80 \ -v env: REDIS_URL: redis://localhost:6379 MQTT_HOST: localhost - name: Upload coverage to Codecov uses: codecov/codecov-action@v3 with: file: ./coverage.xml # ===== СБОРКА DOCKER ОБРАЗА ===== build: name: Build & Push Docker Image runs-on: ubuntu-latest needs: test if: github.ref == 'refs/heads/main' || startsWith(github.ref, 'refs/tags/') steps: - uses: actions/checkout@v4 - name: Set up Docker Buildx uses: docker/setup-buildx-action@v3 - name: Login to Registry uses: docker/login-action@v3 with: registry: ghcr.io username: ${{ github.actor }} password: ${{ secrets.GITHUB_TOKEN }} - name: Extract metadata id: meta uses: docker/metadata-action@v5 with: images: ghcr.io/${{ github.repository }} tags: | type=ref,event=branch type=semver,pattern={{version}} type=semver,pattern={{major}}.{{minor}} type=sha,prefix=sha- - name: Build and push uses: docker/build-push-action@v5 with: context: . platforms: linux/amd64,linux/arm64 # Для x86 серверов И Raspberry Pi push: true tags: ${{ steps.meta.outputs.tags }} cache-from: type=gha cache-to: type=gha,mode=max # ===== ДЕПЛОЙ ===== deploy-staging: name: Deploy to Staging runs-on: ubuntu-latest needs: build if: github.ref == 'refs/heads/develop' environment: staging steps: - name: Deploy via SSH uses: appleboy/ssh-action@v1 with: host: ${{ secrets.STAGING_HOST }} username: deploy key: ${{ secrets.STAGING_SSH_KEY }} script: | cd /opt/gateway docker compose pull docker compose up -d --remove-orphans docker compose ps deploy-production: name: Deploy to Production runs-on: ubuntu-latest needs: build if: startsWith(github.ref, 'refs/tags/v') environment: production # Требует одобрения в GitHub steps: - name: Deploy to production uses: appleboy/ssh-action@v1 with: host: ${{ secrets.PROD_HOST }} username: deploy key: ${{ secrets.PROD_SSH_KEY }} script: | cd /opt/gateway export IMAGE_TAG=${{ github.ref_name }} docker compose pull docker compose up -d --remove-orphans # Smoke test sleep 10 curl -f http://localhost:8080/health || (docker compose logs && exit 1) Git Hooks: автоматизация на уровне репозитория# .git/hooks/pre-commit (запускается перед каждым коммитом) #!/bin/bash set -e echo "🔍 Pre-commit проверки..." # Форматирование Python if command -v black &> /dev/null; then black --check . --quiet if [ $? -ne 0 ]; then echo "❌ Форматирование не соответствует black. Запустите: black ." exit 1 fi fi # Быстрые тесты (только изменённые файлы) CHANGED_PY=$(git diff --cached --name-only --diff-filter=ACM | grep '\.py$') if [ -n "$CHANGED_PY" ]; then pytest tests/unit/ -x -q --tb=short fi echo "✅ Все проверки прошли" Лучше использовать pre-commit framework:# .pre-commit-config.yaml repos: - repo: https://github.com/pre-commit/pre-commit-hooks rev: v4.5.0 hooks: - id: trailing-whitespace - id: end-of-file-fixer - id: check-yaml - id: check-json - id: check-merge-conflict - id: detect-private-key # Не допускаем секреты в коде! - repo: https://github.com/psf/black rev: 23.10.1 hooks: - id: black - repo: https://github.com/pycqa/isort rev: 5.12.0 hooks: - id: isort - repo: https://github.com/commitizen-tools/commitizen rev: v3.12.0 hooks: - id: commitizen # Проверяет формат сообщения коммита # Установка: pip install pre-commit pre-commit install # Устанавливает хуки в .git/hooks/ pre-commit run --all-files # Запуск вручную Семантическое версионирование и автоматический релизSemantic Versioning: MAJOR.MINOR.PATCHPATCH (1.2.3 → 1.2.4): багфиксы, обратно совместимые изменения MINOR (1.2.4 → 1.3.0): новая функциональность, обратно совместимая MAJOR (1.3.0 → 2.0.0): несовместимые изменения API Автоматический выпуск с commitizen:# pyproject.toml: [tool.commitizen] name = "cz_conventional_commits" tag_format = "v$version" version_scheme = "semver" version_files = [ "src/__init__.py:__version__", "pyproject.toml:version" ] update_changelog_on_bump = true # Команды: cz bump # Автоматически определяет тип bumpa из коммитов cz bump --major # Принудительно major cz changelog # Генерирует CHANGELOG.md Автоматический CHANGELOG.md из conventional commits: ## v1.3.0 (2024-03-15) ### Features - **modbus**: добавлена поддержка FC15 (write multiple coils) (#247) - **historian**: реализован deadband-алгоритм сжатия, экономия 78% места ### Bug Fixes - **uart**: исправлена потеря байт при высоких нагрузках (#251) - **pid**: устранено интегральное насыщение при длительной работе ### Performance - **batch-write**: оптимизирован bulk insert в InfluxDB, +340% throughput Практические советы.gitignore — не игнорируйте важное# Python __pycache__/ *.pyc *.pyo .venv/ .env venv/ # IDE .idea/ .vscode/ *.swp *.swo # Сборка dist/ build/ *.egg-info/ # Тесты .coverage htmlcov/ .pytest_cache/ # Секреты (НИКОГДА не коммитить!) *.key *.pem .env.local config.secret.yaml # OS .DS_Store Thumbs.db Работа с секретами — никогда в репозиторий!# Плохо: секреты в коде MQTT_PASSWORD = "supersecret123" # Хорошо: из переменных окружения MQTT_PASSWORD = os.getenv("MQTT_PASSWORD") # В .env локально, в CI — secrets # Для локальной разработки: .env файл (в .gitignore!) # pip install python-dotenv from dotenv import load_dotenv load_dotenv() # Для проверки что секрет не утёк: git log --all --full-history -- "*.env" # Поиск в истории git grep "supersecret" # Поиск в текущем состоянии Интерактивный rebase для чистой истории# Перед мержем PR привести историю в порядок git rebase -i origin/main # Редактор покажет: # pick a1b2c3 WIP fix something # pick d4e5f6 another fix # pick g7h8i9 добавил логирование # Меняем на: # reword a1b2c3 fix(modbus): исправлен CRC при DLC=0 # squash d4e5f6 # Объединить с предыдущим # pick g7h8i9 feat(logging): добавлено структурированное логирование # Результат: чистая, осмысленная история ЗаключениеGit — это не инструмент, это язык коммуникации в команде. Правильные коммиты рассказывают историю проекта. Грамотное ветвление изолирует работу. CI/CD устраняет ручной труд и человеческие ошибки при деплое. Начните с малого: установите .pre-commit-config.yaml с black и detect-private-key. Перейдите на conventional commits. Добавьте один GitHub Actions workflow с тестами. Каждый из этих шагов принесёт немедленную пользу. Инвестиция в культуру работы с кодом возвращается многократно: меньше времени на дебаггинг, меньше страха перед деплоем, больше времени на реальную разработку.

Когда Linux вместо микроконтроллераВыбор между голым МК (Arduino/STM32) и Linux-системой (Raspberry Pi/промышленный ПК) — одно из ключевых архитектурных решений. Linux выигрывает когда нужно: Сложные сетевые протоколы (TCP/IP стек, TLS, MQTT, OPC UA) Работа с файлами: логирование, конфигурация, обновления ПО Высокоуровневые вычисления: Python/NumPy, ML-инференс Несколько параллельных задач с разной логикой Веб-интерфейс или REST API Большой объём RAM/Flash (база данных, historian) Микроконтроллер выигрывает когда нужно: Детерминированное реальное время (< 1 мс) Мгновенный старт (Linux загружается 10–30 секунд) Минимальное энергопотребление Дешёвое серийное производство Золотое правило: Linux для "мозга" и коммуникаций, микроконтроллер для "мышц" и реального времени. Оба — в одной системе, связанные UART или SPI. Raspberry Pi в промышленности: что учестьRaspberry Pi не проектировался для промышленного применения, но активно используется. Ключевые ограничения и решения: Ограничение 1: SD-карта умирает SD-карты не рассчитаны на постоянную запись. В промышленном применении — выход из строя за 3–12 месяцев. Решения: # 1. Read-only файловая система (overlayfs) # В /boot/cmdline.txt добавить: overlayroot=tmpfs # Данные писать только на специальный раздел с journaling # 2. Переместить tmpfs для логов в RAM # /etc/fstab: tmpfs /tmp tmpfs defaults,noatime,size=100m 0 0 tmpfs /var/log tmpfs defaults,noatime,size=50m 0 0 tmpfs /var/tmp tmpfs defaults,noatime,size=20m 0 0 # 3. Использовать SSD через USB3 или eMMC-модуль (CM4) Ограничение 2: Нет RTC (часов реального времени) При потере питания время сбивается. # Установить модуль DS3231 через I2C # /boot/config.txt: dtoverlay=i2c-rtc,ds3231 # Синхронизация при загрузке: sudo hwclock --hctosys # Hardware clock → System clock Ограничение 3: Нет аппаратного watchdog "из коробки" # Включить встроенный watchdog BCM2835 # /boot/config.txt: dtparam=watchdog=on # /etc/systemd/system.conf: RuntimeWatchdogSec=10 # Сброс если systemd не пингует 10 секунд RebootWatchdogSec=60 # Проверка: ls /dev/watchdog # Должен существовать GPIO: управление пинами из userspacesysfs (устаревший, но всё ещё работает):# Экспортируем GPIO 17 echo "17" > /sys/class/gpio/export # Устанавливаем направление echo "out" > /sys/class/gpio/gpio17/direction # Устанавливаем значение echo "1" > /sys/class/gpio/gpio17/value # Читаем состояние входа cat /sys/class/gpio/gpio18/value libgpiod (современный стандарт):# Установка sudo apt install gpiod libgpiod-dev # Командная строка gpioget gpiochip0 17 # Прочитать GPIO 17 gpioset gpiochip0 17=1 # Установить в HIGH gpioset gpiochip0 17=0 18=1 # Установить несколько gpioinfo gpiochip0 # Информация о всех пинах # Python + gpiod # pip install gpiod import gpiod import time # Открываем чип chip = gpiod.Chip('gpiochip0') # Настраиваем пины led_line = chip.get_line(17) button_line = chip.get_line(18) led_config = gpiod.LineRequest() led_config.consumer = "myapp" led_config.request_type = gpiod.LineRequest.DIRECTION_OUTPUT led_line.request(led_config) btn_config = gpiod.LineRequest() btn_config.consumer = "myapp" btn_config.request_type = gpiod.LineRequest.EVENT_BOTH_EDGES # Прерывания! button_line.request(btn_config) try: while True: # Ожидание события с таймаутом 100 мс event_happened = button_line.event_wait(nsec=100_000_000) if event_happened: event = button_line.event_read() if event.type == gpiod.LineEvent.RISING_EDGE: print("Кнопка нажата") led_line.set_value(1) else: print("Кнопка отпущена") led_line.set_value(0) finally: led_line.release() button_line.release() SPI и I2C из userspaceI2C (smbus2):# pip install smbus2 import smbus2 import time class BME280_Linux: """Работа с датчиком BME280 через Linux I2C""" ADDR = 0x76 REG_ID = 0xD0 REG_CTRL = 0xF4 REG_DATA = 0xF7 def __init__(self, bus_num: int = 1): self.bus = smbus2.SMBus(bus_num) def read_reg(self, reg: int) -> int: return self.bus.read_byte_data(self.ADDR, reg) def write_reg(self, reg: int, value: int): self.bus.write_byte_data(self.ADDR, reg, value) def read_burst(self, reg: int, length: int) -> bytes: return bytes(self.bus.read_i2c_block_data(self.ADDR, reg, length)) def init(self): chip_id = self.read_reg(self.REG_ID) if chip_id != 0x60: raise RuntimeError(f"BME280 не найден, ID={chip_id:#x}") # Нормальный режим, oversampling ×4 self.write_reg(0xF4, 0x97) # ctrl_meas self.write_reg(0xF5, 0xA0) # config: IIR filter 16 time.sleep(0.1) def read(self) -> dict: data = self.read_burst(self.REG_DATA, 6) raw_press = (data[0] << 12) | (data[1] << 4) | (data[2] >> 4) raw_temp = (data[3] << 12) | (data[4] << 4) | (data[5] >> 4) # Упрощённое преобразование (без компенсации) # В реальности нужно читать калибровочные коэффициенты! temp = raw_temp / 5120.0 pressure = raw_press / 25600.0 / 100.0 # гПа return {'temperature': round(temp, 1), 'pressure': round(pressure, 1)} # Использование: sensor = BME280_Linux(bus_num=1) # /dev/i2c-1 sensor.init() while True: data = sensor.read() print(f"T={data['temperature']}°C, P={data['pressure']}гПа") time.sleep(1) SPI (spidev):import spidev import time class MCP3208_ADC: """12-битный АЦП MCP3208 через SPI""" CHANNELS = 8 def __init__(self, bus: int = 0, device: int = 0, speed_hz: int = 1_000_000): self.spi = spidev.SpiDev() self.spi.open(bus, device) # /dev/spidev0.0 self.spi.max_speed_hz = speed_hz self.spi.mode = 0 def read_channel(self, channel: int) -> int: """Чтение канала 0-7, возвращает 0-4095""" if not 0 <= channel < self.CHANNELS: raise ValueError(f"Канал {channel} вне диапазона 0-7") # MCP3208: 3 байта транзакции # Байт 1: старт-бит + single/diff + D2 # Байт 2: D1, D0, X, X, X, X, X, X # Байт 3: X, X, X, X, X, X, X, X cmd = [0x06 | (channel >> 2), (channel & 0x03) << 6, 0x00] response = self.spi.xfer2(cmd) # Из ответа: байт 1 биты 1-0 + байт 2 все 8 бит = 12 бит result = ((response[1] & 0x0F) << 8) | response[2] return result def read_voltage(self, channel: int, vref: float = 3.3) -> float: """Чтение в вольтах""" raw = self.read_channel(channel) return raw * vref / 4095.0 def read_all(self, vref: float = 3.3) -> list: """Чтение всех 8 каналов""" return [self.read_voltage(ch, vref) for ch in range(self.CHANNELS)] def close(self): self.spi.close() # Использование: adc = MCP3208_ADC() while True: voltages = adc.read_all() for ch, v in enumerate(voltages): print(f"CH{ch}: {v:.3f}В", end=" ") print() time.sleep(0.5) UART и последовательный портimport serial import serial.tools.list_ports # Найти все доступные порты def list_serial_ports(): ports = serial.tools.list_ports.comports() for port in ports: print(f"{port.device}: {port.description} ({port.hwid})") # На Raspberry Pi: # /dev/ttyAMA0 или /dev/serial0 — встроенный UART (GPIO 14/15) # /dev/ttyUSB0 — USB-UART адаптер # /dev/ttyACM0 — USB CDC (Arduino) # Важно для RPi: отключить console на /dev/serial0 # sudo raspi-config → Interface Options → Serial Port # "Would you like a login shell to be accessible over serial?" → No # "Would you like the serial port hardware to be enabled?" → Yes class SerialDevice: """Надёжная работа с последовательным портом""" def __init__(self, port: str, baudrate: int = 9600, timeout: float = 1.0): self.port_name = port self.baudrate = baudrate self.timeout = timeout self.ser = None def connect(self): try: self.ser = serial.Serial( port = self.port_name, baudrate = self.baudrate, bytesize = serial.EIGHTBITS, parity = serial.PARITY_NONE, stopbits = serial.STOPBITS_ONE, timeout = self.timeout ) print(f"Подключено: {self.port_name}") return True except serial.SerialException as e: print(f"Ошибка подключения: {e}") return False def send(self, data: bytes) -> bool: try: self.ser.write(data) return True except serial.SerialException: return False def send_line(self, text: str) -> bool: return self.send((text + '\n').encode('utf-8')) def read_line(self) -> str | None: try: line = self.ser.readline() if line: return line.decode('utf-8', errors='replace').strip() return None except serial.SerialException: return None def close(self): if self.ser and self.ser.is_open: self.ser.close() systemd: управление сервисами приложенияПравильное промышленное приложение на Linux должно запускаться как системный сервис — автостарт при загрузке, перезапуск при сбое, логирование. Создание systemd unit:# /etc/systemd/system/industrial-gateway.service [Unit] Description=Industrial IoT Gateway After=network.target Wants=network-online.target After=network-online.target # Зависимость от другого сервиса (например, MQTT-брокера) # Requires=mosquitto.service # After=mosquitto.service [Service] Type=simple User=pi Group=pi WorkingDirectory=/opt/gateway # Переменные окружения из файла EnvironmentFile=/etc/gateway/config.env # Команда запуска ExecStart=/opt/gateway/venv/bin/python /opt/gateway/main.py # Перезапуск при сбое Restart=on-failure RestartSec=10s StartLimitIntervalSec=60s StartLimitBurst=3 # Максимум 3 попытки за 60 секунд # Watchdog интеграция с systemd # Приложение должно вызывать sd_notify WATCHDOG=1 каждые N секунд WatchdogSec=30s # Логирование StandardOutput=journal StandardError=journal SyslogIdentifier=gateway # Безопасность (опционально, но рекомендуется) NoNewPrivileges=true PrivateTmp=true [Install] WantedBy=multi-user.target # Установка и управление: sudo systemctl daemon-reload sudo systemctl enable industrial-gateway sudo systemctl start industrial-gateway sudo systemctl status industrial-gateway # Логи: journalctl -u industrial-gateway -f # В реальном времени journalctl -u industrial-gateway --since today journalctl -u industrial-gateway -n 100 # Последние 100 строк Watchdog из Python (sd-notify):# pip install sdnotify import sdnotify import time import threading notifier = sdnotify.SystemdNotifier() def watchdog_thread(): """Пингуем systemd watchdog каждые 10 секунд""" while True: notifier.notify("WATCHDOG=1") time.sleep(10) def main(): # Сообщаем systemd что мы готовы notifier.notify("READY=1") notifier.notify("STATUS=Инициализация...") # Запускаем watchdog в фоне wdg = threading.Thread(target=watchdog_thread, daemon=True) wdg.start() try: # Основной цикл приложения while True: notifier.notify("STATUS=Работает нормально") # ... бизнес-логика time.sleep(1) except Exception as e: notifier.notify(f"STATUS=ОШИБКА: {e}") raise Сетевая конфигурация для промышленного шлюза# /etc/dhcpcd.conf — статический IP для промышленной сети interface eth0 static ip_address=192.168.1.200/24 static routers=192.168.1.1 static domain_name_servers=192.168.1.1 8.8.8.8 # Или через NetworkManager (современный способ): sudo nmcli con add type ethernet ifname eth0 con-name industrial \ ipv4.method manual ipv4.addresses 192.168.1.200/24 \ ipv4.gateway 192.168.1.1 ipv4.dns "192.168.1.1 8.8.8.8" sudo nmcli con up industrial Bonding (резервирование сети):# Два сетевых интерфейса — один основной, второй резервный # /etc/network/interfaces: auto bond0 iface bond0 inet static address 192.168.1.200 netmask 255.255.255.0 gateway 192.168.1.1 bond-slaves eth0 eth1 bond-mode active-backup # При отказе eth0 — переключаемся на eth1 bond-miimon 100 # Проверка связи каждые 100 мс bond-primary eth0 Buildroot: минимальный Linux-образДля серийного производства не нужен полный Raspberry Pi OS с Python IDE и LibreOffice. Нужен минимальный образ с только нужными компонентами. Buildroot — система сборки кастомных Linux-образов: # Клонируем Buildroot git clone https://git.buildroot.net/buildroot cd buildroot # Начинаем с дефолтной конфигурации для Raspberry Pi make raspberrypi4_64_defconfig # Настраиваем через menuconfig make menuconfig # Target packages → Networking applications → mosquitto (MQTT-брокер) # Target packages → Libraries → python3 # Target packages → Libraries → python-paho-mqtt # System configuration → Root password # Собираем (первый раз ~1-2 часа) make -j4 # Результат: output/images/sdcard.img # Записываем на SD: sudo dd if=output/images/sdcard.img of=/dev/sdX bs=4M status=progress Преимущества кастомного образа: Размер: 50–200 МБ вместо 4–8 ГБ Быстрый старт: 5–8 секунд вместо 30+ Безопасность: минимальная поверхность атаки Reproducible builds: одинаковый образ на всех устройствах Python-приложение как надёжный шлюз#!/opt/gateway/venv/bin/python3 """ Промышленный IoT-шлюз: Modbus RTU → MQTT """ import logging import signal import sys import time import json import threading from pathlib import Path from typing import Optional import sdnotify import paho.mqtt.client as mqtt from pymodbus.client import ModbusSerialClient # Настройка логирования (в journald через stderr) logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(levelname)s] %(name)s: %(message)s', stream=sys.stderr ) log = logging.getLogger('gateway') class IndustrialGateway: def __init__(self): self.running = False self.notifier = sdnotify.SystemdNotifier() self.mqtt_client: Optional[mqtt.Client] = None self.modbus_client: Optional[ModbusSerialClient] = None # Конфигурация из переменных окружения import os self.mqtt_host = os.getenv('MQTT_HOST', 'localhost') self.mqtt_port = int(os.getenv('MQTT_PORT', '1883')) self.modbus_port = os.getenv('MODBUS_PORT', '/dev/serial0') self.modbus_baud = int(os.getenv('MODBUS_BAUD', '9600')) self.poll_interval = float(os.getenv('POLL_INTERVAL', '1.0')) def setup(self): """Инициализация подключений""" log.info("Инициализация шлюза...") # MQTT self.mqtt_client = mqtt.Client(client_id="industrial-gateway") self.mqtt_client.on_connect = self._on_mqtt_connect self.mqtt_client.on_disconnect = self._on_mqtt_disconnect self.mqtt_client.will_set("gateway/status", '{"online": false}', retain=True) self.mqtt_client.connect_async(self.mqtt_host, self.mqtt_port) self.mqtt_client.loop_start() # Modbus self.modbus_client = ModbusSerialClient( port=self.modbus_port, baudrate=self.modbus_baud, timeout=1.0 ) if not self.modbus_client.connect(): log.error(f"Не удалось подключиться к Modbus: {self.modbus_port}") def _on_mqtt_connect(self, client, userdata, flags, rc): if rc == 0: log.info(f"MQTT подключён: {self.mqtt_host}") client.publish("gateway/status", '{"online": true}', retain=True) else: log.error(f"MQTT ошибка подключения: {rc}") def _on_mqtt_disconnect(self, client, userdata, rc): log.warning(f"MQTT отключён (rc={rc}), переподключение...") def poll_device(self, device_addr: int, reg_start: int, reg_count: int) -> Optional[list]: """Опрос устройства Modbus""" try: result = self.modbus_client.read_input_registers( address=reg_start, count=reg_count, slave=device_addr ) if result.isError(): log.warning(f"Modbus ошибка: устройство {device_addr}") return None return result.registers except Exception as e: log.error(f"Исключение при опросе {device_addr}: {e}") return None def publish(self, topic: str, data: dict): """Публикация в MQTT""" try: payload = json.dumps(data, ensure_ascii=False) self.mqtt_client.publish(topic, payload) except Exception as e: log.error(f"Ошибка публикации: {e}") def run(self): """Главный цикл""" self.running = True # Сигналы завершения signal.signal(signal.SIGTERM, self._shutdown) signal.signal(signal.SIGINT, self._shutdown) self.setup() # Сообщаем systemd о готовности self.notifier.notify("READY=1") log.info("Шлюз запущен и готов к работе") poll_count = 0 while self.running: loop_start = time.monotonic() # Опрашиваем частотник (адрес 1, регистры 0-5) vfd_data = self.poll_device(1, 0, 6) if vfd_data: self.publish("factory/line1/vfd1/telemetry", { 'status_word': vfd_data[0], 'freq_hz': vfd_data[1] / 100.0, 'current_a': vfd_data[2] / 10.0, 'voltage_v': vfd_data[3], 'power_kw': vfd_data[4] / 10.0, 'fault_code': vfd_data[5], 'running': bool(vfd_data[0] & 0x0001), 'fault': bool(vfd_data[0] & 0x0008), }) # Опрашиваем датчик давления (адрес 5, регистры 0-1) pressure_data = self.poll_device(5, 0, 2) if pressure_data: import struct raw = struct.pack('>HH', pressure_data[0], pressure_data[1]) pressure = struct.unpack('>f', raw)[0] self.publish("factory/line1/pressure1/telemetry", { 'pressure_bar': round(pressure, 2) }) poll_count += 1 # Watchdog if poll_count % 5 == 0: self.notifier.notify("WATCHDOG=1") self.notifier.notify(f"STATUS=Опросов: {poll_count}") # Точный таймер интервала опроса elapsed = time.monotonic() - loop_start sleep_time = self.poll_interval - elapsed if sleep_time > 0: time.sleep(sleep_time) elif sleep_time < -0.1: log.warning(f"Опрос занял {elapsed:.3f}с (лимит {self.poll_interval}с)") def _shutdown(self, signum, frame): log.info(f"Получен сигнал {signum}, завершаем работу...") self.running = False if self.mqtt_client: self.mqtt_client.publish("gateway/status", '{"online": false}', retain=True) self.mqtt_client.loop_stop() if self.modbus_client: self.modbus_client.close() sys.exit(0) if __name__ == "__main__": gateway = IndustrialGateway() gateway.run() Безопасность Linux-устройства в сети# Минимальный hardening для промышленного устройства # 1. Обновление системы (автоматически) sudo apt install unattended-upgrades sudo dpkg-reconfigure unattended-upgrades # 2. UFW файрвол sudo apt install ufw sudo ufw default deny incoming sudo ufw default allow outgoing sudo ufw allow ssh sudo ufw allow 1883 # MQTT (только если нужен снаружи) sudo ufw enable # 3. Отключить неиспользуемые сервисы sudo systemctl disable bluetooth sudo systemctl disable avahi-daemon # 4. Изменить стандартный пароль! passwd pi # ОБЯЗАТЕЛЬНО # 5. SSH ключи вместо паролей # Скопировать ключ: ssh-copy-id pi@192.168.1.200 # /etc/ssh/sshd_config: # PasswordAuthentication no # PermitRootLogin no # 6. Логирование попыток взлома sudo apt install fail2ban ЗаключениеLinux на встраиваемых системах — мощнейший инструмент. Raspberry Pi, как платформа, имеет ограничения для промышленного применения, но при правильной настройке (read-only FS, watchdog, systemd-сервисы, резервирование) служит надёжно годами. Для новых промышленных проектов рассмотрите специализированные платформы: Raspberry Pi CM4 (eMMC, нет SD-карты), BeagleBone Black (PRU для реального времени), Toradex Colibri/Apalis (промышленный температурный диапазон, многолетняя поддержка), IEI, Advantech (сертифицированные промышленные платформы). Главный принцип для промышленного Linux: устройство должно работать без участия человека годами. Watchdog, автоперезапуск сервисов, защита файловой системы от записи, автоматические обновления безопасности — это не опции, это базовые требования.