object-publisher.py

import time
import random
import pickle
import configparser
import json
import paho.mqtt.client as mqtt
import threading
import uuid
import ast
import logging
import numpy as np
import hashlib
import zmq
import sys

logging.basicConfig(level=logging.INFO)

# Create a ConfigParser object
config = configparser.ConfigParser()

# Read the configuration file
config.read("config.ini")

# Access values from the CAMERA section
posteID = config.getint("CAMERA", "id")
width = config.getint("CAMERA", "width")
height = config.getint("CAMERA", "height")
lat = config.get("CAMERA", "latitude")
long = config.get("CAMERA", "longitude")
heading = config.get("CAMERA", "heading")
try:
    camera_id = config.getint("CAMERA", "camera")
except:
    camera_id = "0"


# Get the current timestamp
timestamp = int(time.time())
random.seed(timestamp)
poste_id = int(posteID)
detection_dic_id = {}
detection_dic_time = {}


# Access values from the DETECTION section
detection_server_ip = config.get("DETECTION", "frameIngestorIp")
broker_ip = config.get("DETECTION", "objectPublisherIp")

# Print the values to verify
print(f"poste ID: {posteID}")
print(f"Resolution: {width}x{height}")
print(f"Detection Server IP: {detection_server_ip}")
print(f"Broker IP: {broker_ip}")


with open("names.pkl", "rb") as fp:  # load id of avaliable detections
    vis = pickle.load(fp)
    vis = dict(map(reversed, vis.items()))

reversed_vis = dict(map(reversed, vis.items()))  # index to identification

etsi_correspondence = {
    0: "unknown",
    1: "person",
    2: "bicycle",
    4: "motorcycle",
    5: "car",
    6: "bus",
    7: "truck",
}

etsi_correspondence = {value: key for key,
                       value in etsi_correspondence.items()}


client = mqtt.Client()
try:
    client.connect(str(broker_ip), 1883, 60)
    path = "/jetson/camera/tracking/objects"
except:
    client.connect(str(broker_ip), 1884, 60)

    if camera_id is not "0":
        path = "p" + str(poste_id) + "/jetson/camera/" + \
            str(camera_id)+"/tracking/objects"
    else:
        path = "p" + str(poste_id) + "/jetson/camera/tracking/objects"

print(path)
mqtt_th = threading.Thread(target=client.loop_forever)
mqtt_th.start()
logging.info("Started MQTT")


# Set up ZeroMQ context and socket
context = zmq.Context()

#  Socket to talk to server
socket = context.socket(zmq.SUB)
socket.connect("tcp://" + detection_server_ip + ":11001")
# Subscribe to all messages (empty string subscribes to everything)
socket.setsockopt_string(zmq.SUBSCRIBE, "P" + str(poste_id))


while True:
    start = time.time()
    message = socket.recv_string()

    try:
        full_dic = ast.literal_eval("".join(message.split(" ")[1:]))
    except:
        raise Exception("Error in the message received")
        continue

    tracking_dic = full_dic.copy()

    boxes = tracking_dic["boxes"][0]
    confs = tracking_dic["confs"][0]
    clss = tracking_dic["clss"][0]
    timestamp_start = tracking_dic["timestamp_start_yolo"]
    timestamp_end = tracking_dic["model_processing_time"]
    global_ids = tracking_dic["global_ids"][0]

    number_detected = len(tracking_dic["clss"][0])
    avg_position_per_class = {}

    names = []
    others = []

    for i in detection_dic_time.copy().keys():
        if time.time() - detection_dic_time[i] > 10:
            detection_dic_time.pop(i, None)
            detection_dic_id.pop(i, None)

    try:
        for ind in range(0, len(global_ids)):
            # Transform class indexes into names
            name = reversed_vis[int(clss[ind])]

            names.append(name)

            if global_ids[ind] not in detection_dic_time.keys():
                hex_time = time.time()
                # Concatenate the constant with the data
                data_with_constant = (
                    str(posteID) + str(global_ids[ind]) + str(timestamp)
                )

                # Create a hash object using SHA-256 algorithm
                hash_object = hashlib.sha256()

                # Update the hash object with the bytes-like object (UTF-8 encoded string in this case)
                hash_object.update(data_with_constant.encode("utf-8"))

                hexa = hash_object.hexdigest()
                # Truncate the hash to 32 bits (8 characters)
                truncated_hash = hexa[:7]

                detection_dic_id[global_ids[ind]] = int(truncated_hash, 16)
                detection_dic_time[global_ids[ind]] = time.time()
            dic = {
                "objectID": detection_dic_id[global_ids[ind]],
                "globalID": global_ids[ind],
                "classification": etsi_correspondence[reversed_vis[vis[names[ind]]]],
                "confidence": int(str(confs[ind])),
                "bbox": {
                    "top_left_x": int(boxes[ind][0]),
                    "top_left_y": int(boxes[ind][1]),
                    "width": int(boxes[ind][2] - boxes[ind][0]),
                    "height": int(boxes[ind][3] - boxes[ind][1]),
                },
                "latitude": tracking_dic["gps"][global_ids[ind]][0],
                "longitude": tracking_dic["gps"][global_ids[ind]][1],
                "heading": None,
                "speed": None,
                "event": "",
            }
            bbox = [
                [int(boxes[ind][0]), int(boxes[ind][1])],
                [
                    int(boxes[ind][2] - boxes[ind][0]),
                    int(boxes[ind][3] - boxes[ind][1]),
                ],
            ]

            # Foot Aprox
            detection_position = [
                bbox[0][0] + (bbox[1][0] / 2),
                bbox[0][1] + bbox[1][1] - 1,
            ]

            others.append(dic)
    except Exception as e:
        trace = []
        tb = e.__traceback__
        while tb is not None:
            trace.append(
                {
                    "filename": tb.tb_frame.f_code.co_filename,
                    "name": tb.tb_frame.f_code.co_name,
                    "lineno": tb.tb_lineno,
                }
            )
            tb = tb.tb_next
        print(str({"type": type(e).__name__, "message": str(e), "trace": trace}))
        sys.exit(0)

    logging.info(
        "Classes={} | names={} | confs={} | # = {}".format(
            clss, names, confs, len(confs)
        )
    )

    final_json_p = {
        "numberOf": len(others),
        "listOfObjects": others,
        "timestamp": time.time(),
        "receiverID": int(poste_id),
        "cameraID": str(camera_id),
        "test": {
            "timestamp_start_yolo": timestamp_start[0],
            "timestamp_end_yolo": timestamp_end,
        },
    }
    client.publish(path,
                   payload=json.dumps(final_json_p),
                   qos=0,
                   retain=False,
                   )
    time.sleep(0.01)
    restart_time = time.time()