Implementando Soluciones IoT con AWS GreenGrass

En este artículo, continuaremos nuestra serie sobre cómo aprender a usar AWS GreenGrass, implementando la lógica de negocio como se mencionó en el artículo anterior.

Vamos directo al grano, en un archivo llamado main.py vas a escribir lo siguiente:

from random import randrange

# Generar un entero aleatorio entre 0 y 9
def generar_entero_aleatorio() -> int:
    return randrange(10)

if __name__ == "__main__":
    x = generar_entero_aleatorio()
    print(f"El entero al azar es: {x}")

Al ejecutar este ejemplo usando python3 main.py, observamos que efectivamente se genera un entero al azar y se muestra en la terminal. Sin embargo, ya que estamos implementando una solución IoT, de nada nos sirve tener una salida en una terminal.

¿Cómo hacemos entonces para que este dato sea enviado a la nube de AWS? Recuerda que la función del núcleo de GreenGrass es extender las capacidades de AWS a dispositivos locales, permitiendo ejecutar tareas directamente en los dispositivos IoT, incluso sin conexión a Internet. Necesitamos comunicarnos con ese núcleo de alguna manera.

Imaginemos un dispositivo IoT genérico con un sistema operativo que use el kernel de Linux y ejecute el núcleo de AWS GreenGrass. Entre las múltiples funciones del núcleo, una es gestionar la comunicación con la nube de AWS de manera segura y robusta. Para que la función de Python pueda enviar el mensaje a la nube, usaremos el núcleo como intermediario.

Instalamos el AWS GreenGrass SDK para Python usando pip install awsiotsdk, y luego modificamos el main.py :

import time
from random import randrange

from awsiot.greengrasscoreipc.clientv2 import GreengrassCoreIPCClientV2
from awsiot.greengrasscoreipc.model import JsonMessage, PublishMessage


# Generar un entero aleatorio entre 0 y 9
def generar_entero_aleatorio() -> int:
    return randrange(10)

# Enviar el dato al núcleo GreenGrass
def enviar_dato_nucleo(dato: int) -> bool:
    mensaje = {"dato": dato}
    publish_message = PublishMessage(json_message=JsonMessage(message=mensaje))

    ipc_client = GreengrassCoreIPCClientV2()
    try:
        ipc_client.publish_to_topic(topic="datos", publish_message=publish_message)
    except Exception as exc:
        print("Error al enviar dato")   
        return False
    else:
        return True
    finally:
        ipc_client.close()     

if __name__ == "__main__":
    while True:
        x = generar_entero_aleatorio()
        enviar_dato_nucleo(x)
        time.sleep(60)

En el siguiente artículo, empaquetaremos la solución como un componente de GreenGrass basado en Docker y luego la desplegaremos en el núcleo.