MetropolisControlSystem.py

import json
from datetime import datetime
from threading import Thread

import requests
import sys
from kafka import KafkaConsumer

from MetropolisStorage.Storage import Storage


class MetropolisControlSystem(Thread):
    def __init__(self, name, kafka_server, storage):
        """
        This class will be the handler of the control requests 
        coming from the data processing system
        
        :param name: string, name of the control unit 
        :param kafka_server: string, kafka server address 
        :param storage: MetropolisStorage.Storage 
        """
        Thread.__init__(self)
        self._name =     name
        self._server =   kafka_server
        self._storage =  storage
        self._consumer = None

    def initialize(self):
        try:
            self._consumer = KafkaConsumer(self._name, bootstrap_servers=self._server)
        except:
            self._consumer = None
        return self._consumer

    def run(self):
        if self._consumer is not None:
            for msg in self._consumer:
                # convert the message as a json object to get the id attribute
                jsonlamp = json.loads(str(msg.value, 'utf-8'), encoding='utf-8')
                # for debug purposes ... TODO remove in production
                # print(json.loads(str(msg.value)))
                # get the ip address linked to the given id
                ip_addr = self._storage.control().get_object(int(jsonlamp["id"]))
                if int(jsonlamp["id"]) == 100:
                    print("Returned:", datetime.now().timestamp())
                # # send the message to the rightful lamp
                requests.get("http://" + str(ip_addr), msg)