Raspberry Pi и Linux — отправка события

Программа на Raspberry Pi обнаруживает событие и вызывает отправитель. Пример ниже принимает готовый тип события, объект и значение температуры; чтение датчика, пороги и восстановление остаются в вашей программе.

Подготовьте канал

Создайте защищённый входящий канал и текстовый шаблон по IoT-инструкции. Для различения аварии и восстановления подключите температурные правила.

Задайте переменные DEPESHER_URL и DEPESHER_TOKEN в окружении процесса. Для службы используйте файл окружения с доступом только владельцу службы; не храните токен в самом скрипте. Python 3 использует системное хранилище сертификатов через стандартный TLS-контекст.

Сохраните отправитель

Файл send_event.py, Python 3.9 или новее, дополнительных библиотек не требует:

import argparse
import json
import math
import os
import ssl
import sys
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("event", choices=["temperature_high", "restored"])
    parser.add_argument("value", type=float)
    parser.add_argument("--site", default="Склад №1")
    parser.add_argument("--device", default="sensor-03")
    args = parser.parse_args()

    url = os.environ.get("DEPESHER_URL", "")
    token = os.environ.get("DEPESHER_TOKEN", "")
    if not url.startswith("https://") or not token:
        print("Нужны HTTPS URL и токен Депешера", file=sys.stderr)
        return 1
    if not math.isfinite(args.value):
        print("Значение должно быть конечным числом", file=sys.stderr)
        return 1

    body = json.dumps({
        "site": {"name": args.site},
        "device": {"id": args.device},
        "event": {"type": args.event},
        "measurement": {"value": args.value, "unit": "°C"},
    }, ensure_ascii=False, allow_nan=False).encode("utf-8")
    request = Request(url, data=body, method="POST", headers={
        "Content-Type": "application/json; charset=utf-8",
        "Authorization": "Bearer " + token,
    })

    try:
        with urlopen(request, timeout=15, context=ssl.create_default_context()) as response:
            if response.status != 202:
                print("Неожиданный HTTP-ответ", file=sys.stderr)
                return 1
    except HTTPError as error:
        print("HTTP " + str(error.code), file=sys.stderr)
        if error.code == 429:
            print("Retry-After: " + error.headers.get("Retry-After", ""), file=sys.stderr)
        return 1
    except (URLError, TimeoutError):
        print("Не удалось выполнить запрос", file=sys.stderr)
        return 1

    print("Событие принято Депешером")
    return 0


if __name__ == "__main__":
    sys.exit(main())

Запрос и TLS используют стандартные urllib.request и ssl. Код не выводит URL или токен в журнал и не повторяет запрос автоматически.

Проверьте два события

python3 send_event.py temperature_high 12.5
python3 send_event.py restored 4.0

Ожидайте «Событие принято Депешером», два сообщения в истории и получение в выбранных каналах. С температурными правилами первое будет «Критично», второе — «Информация». При базовом IoT-шаблоне первая строка события содержит temperature_high, значение — 12.5 °C.

После проверки вызывайте скрипт только при установленном программой переходе состояния. Повторный запуск создаст новое событие; лимит частоты, гистерезис и очередь при потере сети реализуйте в источнике. Ответ 202 не подтверждает доставку, а тайм-аут не доказывает отсутствие принятого сообщения.

Для обычного текстового результата вместо JSON можно использовать CLI Депешера. Рабочие уведомления нескольких объектов показаны в сервисном сценарии.