Using AWS Lambda to track the electricity status in the apartments

Intro

From October 2022 to February 2023, Russia inflicted massive blows on Ukraine’s energy infrastructure. As a result, power outages were implemented throughout the country, affecting the entire population. These outages lasted 8–12 hours per day, and in case of emergencies or directly after attacks, power could be unavailable for several days. These strikes greatly impacted the IT industry, which was forced to seek alternative sources of electricity. Some IT professionals, including myself, decided to utilize our knowledge to help alleviate the difficulties faced in this challenging situation.

Over the past five years, I have been working extensively with Python, focusing on developing microservice architectures on AWS. I have gained experience with numerous AWS services, which has enabled me to create a system of Telegram channels that notify me when power is turned on or off at my house, as well as at my mother’s and grandmother’s homes.

Over the last few months, my lights had been switched off according to a pre-determined schedule or even at random times. Sometimes, the lights might have been on when they shouldn’t have been. When there was a power outage, I had to find a location with both electricity and internet access to continue working. This was particularly important on days filled with meetings, as I needed to remain online and available for video calls. While working from a nearby cafe, it was useful to know when power was restored at home because, as the saying goes, “there’s no place like home.”

One distinctive feature of my system is its ability to rapidly connect new addresses. In this article, I will provide a step-by-step guide on how to implement this process.

Step 1: Raspberry Pi + Lambda

A year ago, I purchased a Raspberry Pi but struggled to find a practical use for it. However, the time has come for it to shine: the Raspberry Pi automatically powers on when the electricity is on and shuts down when the power is off. Raspberry OS, like other UNIX-based operating systems, features CRON, which allows you to schedule the execution of program code at specific intervals.

We proceed to create an AWS Lambda function. Lambda is a powerful tool for performing computations on AWS. The platform allows you to make one million Lambda calls per month for free (within the free tier). We then write a Lambda function that accepts my house’s ID and logs the call information and timestamp to the DynamoDB database. DynamoDB is a serverless, non-relational database from AWS that works seamlessly with Lambda. AWS also provides 25GB of free DynamoDB storage, so there are no additional costs.

On the Raspberry Pi, we develop a Python script that calls our AWS Lambda function every minute, effectively reporting that the Raspberry Pi is operational, which in turn indicates that there is power at home.

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

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

The AWS Lambda function will log the timestamp of each call in the DynamoDB database, effectively storing the most recent moment when the Raspberry Pi was active and indicating the presence of power at home:

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')
    }

Step 2: Telegram

In the next step, we will set up a Telegram channel. This can be accomplished using the Telegram client application. Within the application, locate the @BotFather bot, which facilitates the creation of Telegram bots. Proceed to create a Telegram bot and make sure to save its access token for future use.

We will need this token to send messages from the Telegram bot to the Telegram channel. After creating the bot, add it to the Telegram channel and grant it administrator privileges. With these permissions, the bot will be able to send messages within the channel.

We will also need to know the ID of the channel to which the bot will send messages. This can be accomplished using a workaround, as described here. Record the channel ID in the same DynamoDB table for reference.

Step 3: One more lambda

Now, the remaining task is to create another AWS Lambda function that checks when the power was last on at the given address. We will consider the power to be off if the Raspberry Pi was last active more than 7 minutes ago. Conversely, if the Raspberry Pi comes back online after this period, we will assume that the power has been restored. We will program our Lambda function to perform this check and send a message to the Telegram channel using the previously created Telegram bot whenever the power status changes (either goes off or comes back on).

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')
    }

That’s it! The system is now complete. All that’s left to do is to share the Telegram channel with the appropriate Viber chat group (e.g., a building management group chat) and enjoy the convenience of the automated power status updates.

Step 4. Wi-Fi router with the static IP address

My next goal was to set up similar channels for my mother's and grandmother's homes. However, they don't have Raspberry Pis, and leaving an ordinary laptop constantly powered on is not feasible.

That's where the internet service provider (ISP) comes in handy. Almost every ISP offers the option of a static IP address. This means you can type something like 10.11.12.1 into your browser and access the router's admin panel. So, through the ISP's contact center (in my case, Kyivstar and Volia), we order the static IP address service and configure the router to allow remote access from our Lambda function's IP address. Information on configuring the router can be found here. Information on configuring the IP address for Lambda can be found here.

As I am not very familiar with AWS VPC, I simply followed the instructions in the article, creating a VPC, Internet Gateway, and NAT Gateway with an Elastic IP address. Unfortunately, the IP address provided by AWS will incur a small cost, roughly $4 per month. With these configurations in place, our router is now accessible and can be pinged remotely.

Step 5. One more lambda

Finally, we need to create an AWS Lambda function that will ping the router by performing a simple HEAD request (using HEAD to minimize data traffic). If the router responds, it indicates that there is power, and we update the DynamoDB table accordingly. If the router doesn’t respond, it means there is no power.

To minimize the runtime of our Lambda function, we will use asyncio and aiohttp. This enables us to ping multiple routers simultaneously:

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")

Step 6. Done

Now, all that’s left to do is to create Telegram channels for each address and add them to our database. Once that’s completed, the system is ready to go:

Code

I’ve decided that there’s nothing confidential in my code, so feel free to use it: https://github.com/p1v2/eSvitlo

Conclusion

Throughout February 2023, Ukrainian energy specialists successfully restored the full electricity supply to all consumers in Ukraine, and power outages became a thing of the past. I am glad that my Telegram bots are no longer needed by anyone, but I would like to share my experience in building an energy monitoring system, as it may be useful in the future, perhaps in a different context.