Як налаштувати єСвітло. Приклади від розробника застосунку

💡 Усі статті, обговорення, новини про Python — в одному місці. Приєднуйтесь до Python спільноти!

Привіт, я Віталій, Software engineer в Chimplie. Останні 5 років працював з Python, розробляв мікросервісну архітектуру на AWS і маю досвід роботи з багатьма AWS-сервісами. За допомогою цих інструментів я вирішив побудувати систему телеграм-каналів, які будуть сповіщати мене про включення або відключення електроенергії у мене вдома, а також вдома у моєї мами та бабусі.

Щодня у мене вимикають світло, за графіком та без графіка, або ж за графіком світла не має бути, але воно чомусь є. І коли немає світла, мені доводиться шукати, де є світло та інтернет, щоб можна було попрацювати. Особливо це актуально, якщо весь день заповнений зустрічами, і потрібно постійно бути онлайн з можливістю прийняти відеодзвінок. І, сидячи онлайн десь у кафе по сусідству, хочеться знати, коли вдома увімкнулось світло, бо, як то кажуть, «в гостях добре, а вдома краще».

Особливість моєї розробки в тому, що вона дозволяє швидко підключати нові адреси. У цій статті я покроково розповім, як це робити.

Крок 1. Raspberry Pi + Lambda

Ще рік тому купив собі Raspberry Pi, і весь час шукав йому застосування, але ніяк не міг придумати. Але тепер, здається, настав «зоряний час» для Raspberry Pi: він вмикається автоматично, коли вмикають світло, і вимикається разом з відключенням світла. На Raspberry OS, як і на будь-яких UNIX-based операційних системах є CRON де можна програмувати, щоб з певною періодичністю виконувався будь-який програмний код.

На AWS створюємо Lambda. Лямбда є ефективним інструментом виконання будь-яких обчислень на AWS. AWS дозволяє зробити один мільйон викликів lambda протягом місяця безкоштовно (free tier). Тож пишемо Lambda, яка буде приймати id мого будинку, та буде записувати інформацію про виклик в базу DynamoDB, та час такого виклику. DynamoDB — нереляційна база даних від AWS, що ідеально працює з Lambda. AWS надає 25 ГБ DynamoDB безкоштовно, тож і тут не доведеться платити.

На Raspberry створюємо python-скрипт, який буде викликати нашу AWS lambda щохвилини, таким чином повідомляючи, що Raspberry Pi жива, тобто вдома є світло.

payload = {"id": os.environ["ADDRESS_ID"]}

if __name__ == "__main__":
    lambda_.invoke(
        FunctionName="eSvitloPong",
        InvocationType='RequestResponse',
        Payload=json.dumps(payload).encode(),
    )

AWS Lambda ж буде записувати час виклику в DynamoDB, таким чином в базі даних буде зберігатись час, коли востаннє Rasberry Pi подавала ознаки життя:

def lambda_handler(event, context):
    id = event['id']

    dynamodb.update_item(
        TableName='eSvitlo',
        Key={'id': {"S": id}},
        UpdateExpression="set pong=:pong",
        ExpressionAttributeValues={":pong": {"S": datetime.now().isoformat()}}
    )

    return {
        'statusCode': 200,
        'body': json.dumps('Ping')
    }

Крок 2. Telegram

Для наступного кроку створимо телеграм-канал. Це можна зробити, використовуючи клієнтський застосунок Telegram. Там же знаходимо бота, який створює телеграм-ботів @BotFather. Створюємо телеграм-бота та зберігаємо його токен.

Він нам буде потрібен, щоб надсилати повідомлення від телеграм-бота в телеграм-канал. Додаємо нашого телеграм-бота в телеграм-канал, і робимо його адміністратором. Тепер він зможе надсилати повідомлення в каналі.

Також нам потрібно буде знати id каналу, в який бот буде надсилати повідомлення, це можна зробити за допомогою хаку, описаного тут. Id каналу запишемо в тій же таблиці DynamoDB.

Крок 3. Ще одна Lambda

Тепер залишилось створити ще одну AWS Lambda, яка буде перевіряти коли за адресою востаннє було світло. Будемо вважати, що якщо Raspberry Pi востаннє була жива більш як 7 хвилин тому, то світла нема. А якщо вона після цього ожила, то світло зʼявилось. Програмуємо нашу лямбду на таку перевірку, і якщо світло вимкнулось або увімкнулось, будемо надсилати повідомлення в канал, використовуючи нашого телеграм-бота, створеного на минулому кроці.

def lambda_handler(event, context):
    items = dynamodb.scan(
        TableName='eSvitlo',
    )['Items']

    for item in items:
        id = item['id']['S']
        ts = datetime.fromisoformat(item['pong']['S'])
        status = item['electricity_status']['S']
        channel_id = int(item['channel_id']['S'])

        now = datetime.now()

        print(f"State: {status}")
        print(f"TS: {ts}")

        if ((now - ts).total_seconds() > 60 * 7) and status == "on":
            send_telegram_message(channel_id, "Немає світла 🕯")
            status = "off"

        if ((now - ts).total_seconds() <= 60 * 7) and status == "off":
            send_telegram_message(channel_id, "Є світло 💡")
            status = "on"

        dynamodb.update_item(
            TableName='eSvitlo',
            Key={'id': {"S": id}},
            UpdateExpression="set electricity_status=:electricity_status",
            ExpressionAttributeValues={":electricity_status": {"S": status}}
        )

    return {
        'statusCode': 200,
        'body': json.dumps('Ok')
    }

І все, готово, залишилось скинути цей телеграм канал у вайбер-чат ОСББ, і насолоджуватись тим як він працює.

Крок 4. Роутер зі статичною IP-адресою

Наступною моєю ціллю було підключити такі ж канали в моєї мами та бабусі. Що ж, у них немає Rasbperry PI, а звичайний ноутбук не залишиш постійно увімкненим.

На допомогу прийшов інтернет-провайдер. Майже кожен провайдер надає послугу статичної IP-адреси. Це коли ти можеш набрати в браузері щось типу 10.11.12.1 і зайти на адмінку роутера. Отже, через контакт-центр оператора (в моєму випадку — «Київстар» та «Воля») замовляємо цю послугу і налаштовуємо роутер, дозволяючи віддалений доступ до нього з IP-адреси нашої Lambda. Про налаштування роутера написано ось тут. Про налаштування IP-адреси для Lambda написано ось тут.

Оскільки я не сильно знайомий з AWS VPC, то ж просто мовчки виконав те, що написано в статті, створив VPC, Internet Gateway, і NAT Gateway з Elastic IP-адресою. На жаль, IP-адреса, видана AWS, буде вже коштувати певних грошей, орієнтовно 4$ в місяць. Тепер наш роутер можна пінгувати.

Крок 5. Ще одна AWS Lambda

Наостанок, залишилось створити AWS, що буде пінгувати роутер, виконуючи простий HEAD-запит (HEAD для мінімізації кількості трафіку): якщо роутер відповів, значить є світло, і ми оновлюємо DynamoDB таблицю, якщо роутер не відповідає — немає світла.

Для мінімізації кількості часу, який працює наша Lambda, будемо використовувати asyncio + aiohttp. Це дозволяє нам пінгувати багато роутерів одночасно:

async def ping_router(ip: str, port: str):
    async with aiohttp.ClientSession() as session:
        try:
            await session.head(url=f"http://{ip}:{port}", timeout=10)
            return True
        except (asyncio.exceptions.TimeoutError, aiohttp.ClientConnectorError):
            return False


async def item_handler(item):
    id = item['id']['S']

    if 'ip' not in item:
        print(f"skip {id}")
        return

    ip = item['ip']['S']
    port = item['port']['S']

    alive = await ping_router(ip, port)

    if alive:
        print(f"{id} is alive")
        pong = datetime.now().isoformat()

        dynamodb.update_item(
            TableName='eSvitlo',
            Key={'id': {"S": id}},
            UpdateExpression="set pong=:pong",
            ExpressionAttributeValues={":pong": {"S": pong}}
        )
    else:
        print(f"{id} is not alive")

Крок 6. Усе готово

Тепер залишилось створити телеграм-канали для кожної адреси і додати їх до нашої бази даних. Готово:

Архітектура

Код

Вирішив, що в моєму коді немає нічого секретного, тому користуйтесь.

А ще, якщо хтось має статичний IP та хоче підключити собі такий канал, пишіть.