data-generation-pipeline.md

Data Generation Pipeline

In this file we will be going a bit in depth on the code for the data generation pipeline and how the containers communicate with eachother. This file will also explicitly say when code or even the architecture was changed from what was made in previous iterations or from what we though we would do.

Sensor Data Simulation (sensors.py animals.py)

Data Generation Setup (sensors.py)

The Data Generation part of the project starts by getting all existing animals from the PostgreSQL database to start simulating their behaviour from where they last where. This requires doing a query with a JOIN to get the polygon that defines that animal's farm and all their other data. We also need to use redis to retrive the animal's last known BPM, Temp and position.

def get_existing_animals(socket: socket):
    conn = get_pg_con()
    cursor = conn.cursor()

    r = redis.Redis(host="redis", port=6379, db=0, decode_responses=True)
    if r.ping():
        print("CONNECTED TO REDIS")

    query = "SELECT * FROM animals LEFT OUTER JOIN farms ON animals.farm_id = farms.id;"
    cursor.execute(query)

    for row in cursor:
        print(f"ANIMAL ROW {row}")
        animal_data = r.get(f"animal_lts:{row[0]}")
        
        farm_poly: list = []
        if str(row[-2]).__contains__("),("):
           farm_poly = farm_polygon_str_to_tuple_list(str(row[-2]))

        x = 0
        y = 0
        UID = row[0]

        if(animal_data != "" and r.exists(f"animal_lts:{row[0]}")):
            animal_json = json.loads(animal_data)
            x = animal_json["x"]
            y = animal_json["y"]
            UID = animal_json["UID"]

        create_animal_set_entry(row[7], UID, x, y, row[6], farm_poly, socket)

Then after retrieving all the already existing animals, our app must be able to handle animal registration and removal at runtime, i.e. when an animal is added or removed via the REST API the "sensors" must start simulating data for that animal immediatly. For that we use a WebSocket that receives messages from the API and then creates a thread to simulate the data for that animal or joins the dead animal's thread to stop that specific data simulation.

async def ws_handler(ws):
    """Handles animal addition and death registration in runtime."""
    r = redis.Redis(host="redis", port=6379, db=0, decode_responses=True)
    if r.ping():
        print("CONNECTED TO REDIS")

    async for message in ws:
        json_msg = json.loads(message)
        if json_msg["op_code"] == "birth":
            farm_id     = json_msg["farm_id"]
            UID         = json_msg["UID"]
            species     = json_msg["species"]
            birthdate   = json_msg["birthdate"]
            outline_str = json_msg["outline"]

            print(outline_str)
            outline = farm_polygon_str_to_tuple_list(outline_str)

            x = 0
            y = 0

            if r.exists(f"animal_lts:{UID}"):
                r_json = json.loads(r.get(f"animal_lts:{UID}"))
                x      = r_json["x"] 
                y      = r_json["y"]

            create_animal_set_entry(farm_id, UID, x, y, species, outline, animal_socket)

            print(f"\033[92mSIMULATING FOR {json_msg} \033[0m")

        elif json_msg["op_code"] == "death":
            UID = json_msg["UID"]
            key = f"animal_lts:{UID}"
            
            animal_to_remove = None
            for animal in animal_set:
                if animal.UID == UID:
                    animal.die()
                    animal_to_remove = animal
                    break
            
            if animal_to_remove:
                animal_set.remove(animal_to_remove)
                if r.exists(key):
                    r.delete(key)

                print(f"\033[91mREMOVED {json_msg} \033[0m")

Animal Data Generation (animals.py)

To simulate the data we use an abstract animal class for the methods not specific to any animal species and then have some methods that needs every specific animal class to implement, such as the methods that return the animal's base movement rate and speed and base BPM and Temp. It is also here that we parse the Farm Polygon string and have the animal "decide" if it will try running away this second. The chance for an animal to run away every seconds is 1% although this should probably be changed as it makes the animals very prone to escape.

I will not be putting any code snippets here as it is too big of a class and I don't want to clutter this doc too much.

Animal Data Send Off

To send off the data to the NGINX server we create a struct to neatly pack all the data together and then send it as a UDP packet through the network.

NGINX

There is indeed a NGINX server in between the sensors.py script and the ingestion worker to work as a reverse proxy and it'd be fun if we could have Ip Whitelisting as some sort of unnecessary security measure.

Sensor Data Processing (ingestion worker)

The C code. We make use of a Thread Pool to process the data coming from the sensors as fast as possible. The amount of threads is hard-coded for now although it'd be cool if it could be passed as a command line argument when starting the container in the dockerfile.

Then for each one of the workers we create connections to all the services we will need and then follow this same loop:

• Wait for a UDP packet.
• Unpack the data to a struct.

ssize_t bytes_received = recvfrom(t_pool->socket_fd, &buffer, sizeof(animal_packet_t), 0, 
    (struct sockaddr*) &sensor_addr, &addr_len);
        
if(bytes_received != sizeof(animal_packet_t)){
    printf("INVALID DATA FORMAT FOR PACKET\n");
    continue;
}

animal_t animal = AnimalUnpack(buffer);

• Update the most recent data on Redis.
• Publish the most recent data on a Redis PUB/SUB.

redisReply* update_reply = redisCommand(redis, "SET animal_lts:%ld %s", animal.UID, payload);

if(update_reply == NULL){
    printf("ERROR ON REDIS INSERT");
}
else freeReplyObject(update_reply);

redisReply* publish_reply = redisCommand(redis, "PUBLISH animal_updates %s", payload);
if(publish_reply == NULL){
    printf("ERROR ON REDIS PUBLISH");
}

• Send the data to RabbitMQ.

int status = amqp_basic_publish(conn, 1, amqp_cstring_bytes(""), amqp_cstring_bytes(RMQ_QUEUE), 0, 0, &properties, amqp_cstring_bytes(payload));
if(status == AMQP_STATUS_OK){
    printf("[✔] MESSAGE QUEUED: ");
}

• Verify animal's state is ok.
• If not hit the api/notif/internal_notification endpoint and send a message to the workers.

int8_t animal_in_farm = CheckAnimalInFarm(redis, &animal) == 0;

Data Insertion

To finish off the data generation pipeline all we need is to insert the data just generated into our TimescaleDB instance. To do this we used another python script.

First we get a connection to our RabbitMQ instance and start consuming the messages.

def get_RMQ_con():
    conn = pika.BlockingConnection(pika.ConnectionParameters("rabbitMQ", 5672))
    channel = conn.channel()
    channel.queue_declare(queue="sensor_queue", durable=False)

    channel.basic_consume(queue="sensor_queue", auto_ack=True, on_message_callback=callback)

    return channel

def main():
    RMQ_channel = get_RMQ_con()
    RMQ_channel.start_consuming()

Then we create a connection to our database and consume the message inserting it in the DB.

def callback(ch, method, properties, body):
    data = json.loads(body.decode())
    print(f"[x] RECEIVED: {data}")

    conn = get_db_con()
    cursor = conn.cursor()

    query = """ 
        INSERT INTO SENSOR_DATA (time, uuid, uid, x_pos, y_pos, bpm, temp)
        VALUES(NOW(), 1, %s, %s, %s, %s, %s)
    """

    try:
        cursor.execute(query, (data["UID"], data["x"], data["y"], data["bpm"], data["temp"]))
        conn.commit()
    except Exception as e:
        print(f"FAILED TO INSERT: {e}")
        exit(-1)

    cursor.close()
    conn.close()