from kafka import KafkaConsumer
import json
import logging
import time


class Consumer():

    consumer = None
    name_server = 'kafka:9093'

    def start(self):
        """ This method initialises a KafkaConsumer reading messages from the beginning """
        global consumer

        self.consumer = KafkaConsumer(bootstrap_servers='kafka:9093', group_id=None, auto_offset_reset='earliest')
        #self.consumer = KafkaConsumer(topic, bootstrap_servers=self.name_servername_server, group_id=None,
                                      #enable_auto_commit=False, auto_offset_reset='earliest')

        return self.consumer

    def subscribe(self, topics):
        """" This method subscribes the 5G-TRMF to a set of interesting topics. The topics parameter must be a list """
        global consumer

        try:
            self.consumer.subscribe(topics)
            return 1
        except Exception as e:
            return 0

    def stop(self):
        """ This method finishes a KafkaConsumer connection as well as unsubscribing the topics registered """
        global consumer

        self.consumer.unsubscribe()
        self.consumer.close()

    def start_reading(self, data_lock, historical):
        """ This method begins to retrieve messages from a KafkaTopic.
        IT MUST BE LAUNCHED AS A THREAD TO AVOID BLOCKING THE APP """
        logging.basicConfig(level=logging.INFO)
        global consumer

        for message in self.consumer:
            trust_information = json.loads(message.value.decode())
            data_lock.acquire()

            if trust_information["trustor"]["trustorDID"] in historical:
                historical[trust_information["trustor"]["trustorDID"]].append(trust_information)
            else:
                historical[trust_information["trustor"]["trustorDID"]] = [trust_information]

            data_lock.release()
            logging.info("New message: %s", trust_information)


    def readSLANotification(self, historical, trustor, trustee, offerDID):
        """ This function retrieves all notifications of potential SLA violations generated by the SLA Breach Predictor.
        Currently, we are simulating that the TMF is subscribed to the real SLA Breach Predicto Kafka topic.
        TODO -> Verify which trustor (5GZORRO Participant) the notification is associated with as the Trust Framework may be
         managing the trust of more than one at the same time. if message.key.decode('utf-8') =="""
        global consumer

        notifications = []
        for message in consumer:
            sla_information = json.loads(message.value.decode())
            notifications.append(sla_information)

        return notifications

    def readLastTrustValues(self, historical, trustor, trustee, last_interaction, current_interation_number):
        """ This method is utilised to retrieve all new trust information generated by a particular trustee on which we want
        to update our previous trust score. This method only retrieves new inputs """
        values = []

        """ Starting from the end to discover new trust information faster """
        for interactions in reversed(historical):

            interation_number = interactions["currentInteractionNumber"]

            """ Looking for all new interactions not previously contemplated"""
            if interactions["trustor"]["trustorDID"] == trustor and \
                        interactions["trustor"]["trusteeDID"] == trustee and \
                        int(interation_number) > int(last_interaction) and \
                        int(interation_number) == int(current_interation_number):
                data = {"trustorDID": interactions["trustor"]["trustorDID"],
                            "trusteeDID": interactions["trustor"]["trusteeDID"],
                            "offerDID": interactions["trustor"]["offerDID"],
                            "trusteeSatisfaction": interactions["trustee"]["trusteeSatisfaction"],
                            "credibility": interactions["trustor"]["credibility"],
                            "transactionFactor": interactions["trustor"]["transactionFactor"],
                            "communityFactor": interactions["trustor"]["communityFactor"],
                            "interaction_number": interactions["trustor"]["direct_parameters"]["interactionNumber"],
                            "totalInteractionNumber": interactions["trustor"]["direct_parameters"]["totalInteractionNumber"],
                            "userSatisfaction": interactions["trustor"]["direct_parameters"]["userSatisfaction"],
                            "trust_value": interactions["trust_value"],
                            "initEvaluationPeriod": interactions["initEvaluationPeriod"],
                            "endEvaluationPeriod": interactions["endEvaluationPeriod"]
                        }
                values.append(data)

        return values

    def readLastTrustInterationValues(self, historical, trustor, trustee, offer, current_interation_number):
        """ This method is utilised to retrieve all new trust information generated by a particular trustee on the current
         interaction number X """

        data = {}

        for interactions in reversed(historical):
            if interactions["trustor"]["trustorDID"] == trustor and \
                    interactions["trustor"]["trusteeDID"] == trustee and \
                    interactions["trustor"]["offerDID"] == offer and \
                    current_interation_number > 0:

                interation_number = interactions["trustor"]["direct_parameters"]["interactionNumber"]

                """ Checking whether the current interaction is the one we are looking for"""
                if interation_number == current_interation_number-1:
                    data = {"trustorDID": interactions["trustor"]["trustorDID"],
                                "trusteeDID": interactions["trustor"]["trusteeDID"],
                                "offerDID": interactions["trustor"]["offerDID"],
                                "trusteeSatisfaction": interactions["trustee"]["trusteeSatisfaction"],
                                "credibility": interactions["trustor"]["credibility"],
                                "transactionFactor": interactions["trustor"]["transactionFactor"],
                                "communityFactor": interactions["trustor"]["communityFactor"],
                                "interaction_number": interactions["trustor"]["direct_parameters"]["interactionNumber"],
                                "totalInteractionNumber": interactions["trustor"]["direct_parameters"]["totalInteractionNumber"],
                                "userSatisfaction": interactions["trustor"]["direct_parameters"]["userSatisfaction"],
                                "trust_value": interactions["trust_value"],
                                "initEvaluationPeriod": interactions["initEvaluationPeriod"],
                                "endEvaluationPeriod": interactions["endEvaluationPeriod"]
                             }
                    return data

        return data

    def readLastTrustValue(self, historical, trustor, trustee):
        """ This method obtains the last trust value recorded in the historical for a specific a trustor, and trustee.
        Only specific information is returned """

        data = {}

        for interactions in reversed(historical):
            if interactions["trustor"]["trustorDID"] == trustor and \
                        interactions["trustor"]["trusteeDID"] == trustee:
                data = {"trustorDID": interactions["trustor"]["trustorDID"],
                            "trusteeDID": interactions["trustor"]["trusteeDID"],
                            "offerDID": interactions["trustor"]["offerDID"],
                            "trusteeSatisfaction": interactions["trustee"]["trusteeSatisfaction"],
                            "credibility": interactions["trustor"]["credibility"],
                            "transactionFactor": interactions["trustor"]["transactionFactor"],
                            "communityFactor": interactions["trustor"]["communityFactor"],
                            "interaction_number": interactions["trustor"]["direct_parameters"]["interactionNumber"],
                            "totalInteractionNumber": interactions["trustor"]["direct_parameters"]["totalInteractionNumber"],
                            "userSatisfaction": interactions["trustor"]["direct_parameters"]["userSatisfaction"],
                            "trust_value": interactions["trust_value"],
                            "initEvaluationPeriod": interactions["initEvaluationPeriod"],
                            "endEvaluationPeriod": interactions["endEvaluationPeriod"]
                        }
                return data

        return data


    def readLastTrustValueOffer(self, historical, trustor, trustee, offer):
        """ This method obtains the last trust value recorded in the historical for a specific a trustor, trustee and offer.
        Only specific information is returned """

        data = {}

        for interactions in reversed(historical):
            if interactions["trustor"]["trustorDID"] == trustor and \
                        interactions["trustor"]["trusteeDID"] == trustee and \
                        interactions["trustor"]["offerDID"] == offer:
                data = {"trustorDID": interactions["trustor"]["trustorDID"],
                            "trusteeDID": interactions["trustor"]["trusteeDID"],
                            "offerDID": interactions["trustor"]["offerDID"],
                            "trusteeSatisfaction": interactions["trustee"]["trusteeSatisfaction"],
                            "credibility": interactions["trustor"]["credibility"],
                            "transactionFactor": interactions["trustor"]["transactionFactor"],
                            "communityFactor": interactions["trustor"]["communityFactor"],
                            "interaction_number": interactions["trustor"]["direct_parameters"]["interactionNumber"],
                            "totalInteractionNumber": interactions["trustor"]["direct_parameters"]["totalInteractionNumber"],
                            "userSatisfaction": interactions["trustor"]["direct_parameters"]["userSatisfaction"],
                            "trust_value": interactions["trust_value"],
                            "initEvaluationPeriod": interactions["initEvaluationPeriod"],
                            "endEvaluationPeriod": interactions["endEvaluationPeriod"]
                         }
                return data

        return data


    def readAllInformationTrustValue(self, historical, trustor, trustee, offer):
        """ This method obtains the last trust value recorded in Kafka for a specific a trustor, trustee and offer. All
         previously recorded trust information is returned """

        data = {}

        for interactions in reversed(historical):
            if interactions["trustor"]["trustorDID"] == trustor and \
                        interactions["trustor"]["trusteeDID"] == trustee and \
                        interactions["trustor"]["offerDID"] == offer:
                data = {"trustorDID": interactions["trustor"]["trustorDID"],
                            "trusteeDID": interactions["trustor"]["trusteeDID"],
                            "offerDID": interactions["trustor"]["offerDID"],
                            "trusteeSatisfaction": interactions["trustee"]["trusteeSatisfaction"],
                            "credibility": interactions["trustor"]["credibility"],
                            "transactionFactor": interactions["trustor"]["transactionFactor"],
                            "communityFactor": interactions["trustor"]["communityFactor"],
                            "interaction_number": interactions["trustor"]["direct_parameters"]["interactionNumber"],
                            "totalInteractionNumber": interactions["trustor"]["direct_parameters"]["totalInteractionNumber"],
                            "userSatisfaction": interactions["trustor"]["direct_parameters"]["userSatisfaction"],
                            "trust_value": interactions["trust_value"],
                            "initEvaluationPeriod": interactions["initEvaluationPeriod"],
                            "endEvaluationPeriod": interactions["endEvaluationPeriod"]
                        }
                return data

        return data

    def readTrusteeInteractions(self, historical, trustee):
        """ This function counts all interactions with a particular trustee in the historical"""

        counter = 0

        for interactions in reversed(historical):
            if interactions["trustor"]["trusteeDID"] == trustee:
                counter += 1

        return counter

    def readOfferTrusteeInteractions(self, historical, trustee, offerTrusteDIDs):
        """ This function counts all interactions with a particular offer in the historical """

        counter = 0

        for interactions in reversed(historical):
            if interactions["trustor"]["trusteeDID"] == trustee and \
                    interactions["trustor"]["offerDID"] == offerTrusteDIDs:
                counter += 1

        return counter

    def readSatisfactionSummation(self, historical, trustor, trustee):
        """ This method returns the satisfaction average rate between a trustor and a trustee  """

        counter = 0
        satisfactionsummation = 0.0
        iterations = 0

        for interactions in reversed(historical):
            iterations+=1
            if interactions["trustor"]["trustorDID"] == trustor and \
                        interactions["trustor"]["trusteeDID"] == trustee:
                counter += 1
                satisfactionsummation = satisfactionsummation + interactions["trustor"]["direct_parameters"]["userSatisfaction"]

        return round(satisfactionsummation/counter, 3)
from kafka import KafkaConsumer
import json
import logging
import time


class Consumer():

    consumer = None
    name_server = 'kafka:9093'

    def start(self):
        """ This method initialises a KafkaConsumer reading messages from the beginning """
        global consumer

        self.consumer = KafkaConsumer(bootstrap_servers='kafka:9093', group_id=None, auto_offset_reset='earliest')
        #self.consumer = KafkaConsumer(topic, bootstrap_servers=self.name_servername_server, group_id=None,
                                      #enable_auto_commit=False, auto_offset_reset='earliest')

        return self.consumer

    def subscribe(self, topics):
        """" This method subscribes the 5G-TRMF to a set of interesting topics. The topics parameter must be a list """
        global consumer

        try:
            self.consumer.subscribe(topics)
            return 1
        except Exception as e:
            return 0

    def stop(self):
        """ This method finishes a KafkaConsumer connection as well as unsubscribing the topics registered """
        global consumer

        self.consumer.unsubscribe()
        self.consumer.close()

    def start_reading(self, data_lock, historical):
        """ This method begins to retrieve messages from a KafkaTopic.
        IT MUST BE LAUNCHED AS A THREAD TO AVOID BLOCKING THE APP """
        logging.basicConfig(level=logging.INFO)
        global consumer

        for message in self.consumer:
            trust_information = json.loads(message.value.decode())
            data_lock.acquire()

            if trust_information["trustor"]["trustorDID"] in historical:
                historical[trust_information["trustor"]["trustorDID"]].append(trust_information)
            else:
                historical[trust_information["trustor"]["trustorDID"]] = [trust_information]

            data_lock.release()
            logging.info("New message: %s", trust_information)


    def readSLANotification(self, historical, trustor, trustee, offerDID):
        """ This function retrieves all notifications of potential SLA violations generated by the SLA Breach Predictor.
        Currently, we are simulating that the TMF is subscribed to the real SLA Breach Predicto Kafka topic.
        TODO -> Verify which trustor (5GZORRO Participant) the notification is associated with as the Trust Framework may be
         managing the trust of more than one at the same time. if message.key.decode('utf-8') =="""
        global consumer

        notifications = []
        for message in consumer:
            sla_information = json.loads(message.value.decode())
            notifications.append(sla_information)

        return notifications

    def readLastTrustValues(self, historical, trustor, trustee, last_interaction, current_interation_number):
        """ This method is utilised to retrieve all new trust information generated by a particular trustee on which we want
        to update our previous trust score. This method only retrieves new inputs """
        values = []

        """ Starting from the end to discover new trust information faster """
        for interactions in reversed(historical):

            interation_number = interactions["currentInteractionNumber"]

            """ Looking for all new interactions not previously contemplated"""
            if interactions["trustor"]["trustorDID"] == trustor and \
                        interactions["trustor"]["trusteeDID"] == trustee and \
                        int(interation_number) > int(last_interaction) and \
                        int(interation_number) == int(current_interation_number):
                data = {"trustorDID": interactions["trustor"]["trustorDID"],
                            "trusteeDID": interactions["trustor"]["trusteeDID"],
                            "offerDID": interactions["trustor"]["offerDID"],
                            "trusteeSatisfaction": interactions["trustee"]["trusteeSatisfaction"],
                            "credibility": interactions["trustor"]["credibility"],
                            "transactionFactor": interactions["trustor"]["transactionFactor"],
                            "communityFactor": interactions["trustor"]["communityFactor"],
                            "interaction_number": interactions["trustor"]["direct_parameters"]["interactionNumber"],
                            "totalInteractionNumber": interactions["trustor"]["direct_parameters"]["totalInteractionNumber"],
                            "userSatisfaction": interactions["trustor"]["direct_parameters"]["userSatisfaction"],
                            "trust_value": interactions["trust_value"],
                            "initEvaluationPeriod": interactions["initEvaluationPeriod"],
                            "endEvaluationPeriod": interactions["endEvaluationPeriod"]
                        }
                values.append(data)

        return values

    def readLastTrustInterationValues(self, historical, trustor, trustee, offer, current_interation_number):
        """ This method is utilised to retrieve all new trust information generated by a particular trustee on the current
         interaction number X """

        data = {}

        for interactions in reversed(historical):
            if interactions["trustor"]["trustorDID"] == trustor and \
                    interactions["trustor"]["trusteeDID"] == trustee and \
                    interactions["trustor"]["offerDID"] == offer and \
                    current_interation_number > 0:

                interation_number = interactions["trustor"]["direct_parameters"]["interactionNumber"]

                """ Checking whether the current interaction is the one we are looking for"""
                if interation_number == current_interation_number-1:
                    data = {"trustorDID": interactions["trustor"]["trustorDID"],
                                "trusteeDID": interactions["trustor"]["trusteeDID"],
                                "offerDID": interactions["trustor"]["offerDID"],
                                "trusteeSatisfaction": interactions["trustee"]["trusteeSatisfaction"],
                                "credibility": interactions["trustor"]["credibility"],
                                "transactionFactor": interactions["trustor"]["transactionFactor"],
                                "communityFactor": interactions["trustor"]["communityFactor"],
                                "interaction_number": interactions["trustor"]["direct_parameters"]["interactionNumber"],
                                "totalInteractionNumber": interactions["trustor"]["direct_parameters"]["totalInteractionNumber"],
                                "userSatisfaction": interactions["trustor"]["direct_parameters"]["userSatisfaction"],
                                "trust_value": interactions["trust_value"],
                                "initEvaluationPeriod": interactions["initEvaluationPeriod"],
                                "endEvaluationPeriod": interactions["endEvaluationPeriod"]
                             }
                    return data

        return data

    def readLastTrustValue(self, historical, trustor, trustee):
        """ This method obtains the last trust value recorded in the historical for a specific a trustor, and trustee.
        Only specific information is returned """

        data = {}

        for interactions in reversed(historical):
            if interactions["trustor"]["trustorDID"] == trustor and \
                        interactions["trustor"]["trusteeDID"] == trustee:
                data = {"trustorDID": interactions["trustor"]["trustorDID"],
                            "trusteeDID": interactions["trustor"]["trusteeDID"],
                            "offerDID": interactions["trustor"]["offerDID"],
                            "trusteeSatisfaction": interactions["trustee"]["trusteeSatisfaction"],
                            "credibility": interactions["trustor"]["credibility"],
                            "transactionFactor": interactions["trustor"]["transactionFactor"],
                            "communityFactor": interactions["trustor"]["communityFactor"],
                            "interaction_number": interactions["trustor"]["direct_parameters"]["interactionNumber"],
                            "totalInteractionNumber": interactions["trustor"]["direct_parameters"]["totalInteractionNumber"],
                            "userSatisfaction": interactions["trustor"]["direct_parameters"]["userSatisfaction"],
                            "trust_value": interactions["trust_value"],
                            "initEvaluationPeriod": interactions["initEvaluationPeriod"],
                            "endEvaluationPeriod": interactions["endEvaluationPeriod"]
                        }
                return data

        return data


    def readLastTrustValueOffer(self, historical, trustor, trustee, offer):
        """ This method obtains the last trust value recorded in the historical for a specific a trustor, trustee and offer.
        Only specific information is returned """

        data = {}

        for interactions in reversed(historical):
            if interactions["trustor"]["trustorDID"] == trustor and \
                        interactions["trustor"]["trusteeDID"] == trustee and \
                        interactions["trustor"]["offerDID"] == offer:
                data = {"trustorDID": interactions["trustor"]["trustorDID"],
                            "trusteeDID": interactions["trustor"]["trusteeDID"],
                            "offerDID": interactions["trustor"]["offerDID"],
                            "trusteeSatisfaction": interactions["trustee"]["trusteeSatisfaction"],
                            "credibility": interactions["trustor"]["credibility"],
                            "transactionFactor": interactions["trustor"]["transactionFactor"],
                            "communityFactor": interactions["trustor"]["communityFactor"],
                            "interaction_number": interactions["trustor"]["direct_parameters"]["interactionNumber"],
                            "totalInteractionNumber": interactions["trustor"]["direct_parameters"]["totalInteractionNumber"],
                            "userSatisfaction": interactions["trustor"]["direct_parameters"]["userSatisfaction"],
                            "trust_value": interactions["trust_value"],
                            "initEvaluationPeriod": interactions["initEvaluationPeriod"],
                            "endEvaluationPeriod": interactions["endEvaluationPeriod"]
                         }
                return data

        return data


    def readAllInformationTrustValue(self, historical, trustor, trustee, offer):
        """ This method obtains the last trust value recorded in Kafka for a specific a trustor, trustee and offer. All
         previously recorded trust information is returned """

        data = {}

        for interactions in reversed(historical):
            if interactions["trustor"]["trustorDID"] == trustor and \
                        interactions["trustor"]["trusteeDID"] == trustee and \
                        interactions["trustor"]["offerDID"] == offer:
                data = {"trustorDID": interactions["trustor"]["trustorDID"],
                            "trusteeDID": interactions["trustor"]["trusteeDID"],
                            "offerDID": interactions["trustor"]["offerDID"],
                            "trusteeSatisfaction": interactions["trustee"]["trusteeSatisfaction"],
                            "credibility": interactions["trustor"]["credibility"],
                            "transactionFactor": interactions["trustor"]["transactionFactor"],
                            "communityFactor": interactions["trustor"]["communityFactor"],
                            "interaction_number": interactions["trustor"]["direct_parameters"]["interactionNumber"],
                            "totalInteractionNumber": interactions["trustor"]["direct_parameters"]["totalInteractionNumber"],
                            "userSatisfaction": interactions["trustor"]["direct_parameters"]["userSatisfaction"],
                            "trust_value": interactions["trust_value"],
                            "initEvaluationPeriod": interactions["initEvaluationPeriod"],
                            "endEvaluationPeriod": interactions["endEvaluationPeriod"]
                        }
                return data

        return data

    def readTrusteeInteractions(self, historical, trustee):
        """ This function counts all interactions with a particular trustee in the historical"""

        counter = 0

        for interactions in reversed(historical):
            if interactions["trustor"]["trusteeDID"] == trustee:
                counter += 1

        return counter

    def readOfferTrusteeInteractions(self, historical, trustee, offerTrusteDIDs):
        """ This function counts all interactions with a particular offer in the historical """

        counter = 0

        for interactions in reversed(historical):
            if interactions["trustor"]["trusteeDID"] == trustee and \
                    interactions["trustor"]["offerDID"] == offerTrusteDIDs:
                counter += 1

        return counter

    def readSatisfactionSummation(self, historical, trustor, trustee):
        """ This method returns the satisfaction average rate between a trustor and a trustee  """

        counter = 0
        satisfactionsummation = 0.0
        iterations = 0

        for interactions in reversed(historical):
            iterations+=1
            if interactions["trustor"]["trustorDID"] == trustor and \
                        interactions["trustor"]["trusteeDID"] == trustee:
                counter += 1
                satisfactionsummation = satisfactionsummation + interactions["trustor"]["direct_parameters"]["userSatisfaction"]

        return round(satisfactionsummation/counter, 3)
This method initialises a KafkaConsumer reading messages from the beginning #self.consumer = KafkaConsumer(topic, bootstrap_servers=self.name_servername_server, group_id=None, #enable_auto_commit=False, auto_offset_reset='earliest') " This method subscribes the 5G-TRMF to a set of interesting topics. The topics parameter must be a list This method finishes a KafkaConsumer connection as well as unsubscribing the topics registered This method begins to retrieve messages from a KafkaTopic.         IT MUST BE LAUNCHED AS A THREAD TO AVOID BLOCKING THE APP This function retrieves all notifications of potential SLA violations generated by the SLA Breach Predictor.         Currently, we are simulating that the TMF is subscribed to the real SLA Breach Predicto Kafka topic.         TODO -> Verify which trustor (5GZORRO Participant) the notification is associated with as the Trust Framework may be          managing the trust of more than one at the same time. if message.key.decode('utf-8') == This method is utilised to retrieve all new trust information generated by a particular trustee on which we want         to update our previous trust score. This method only retrieves new inputs Starting from the end to discover new trust information faster Looking for all new interactions not previously contemplated This method is utilised to retrieve all new trust information generated by a particular trustee on the current          interaction number X Checking whether the current interaction is the one we are looking for This method obtains the last trust value recorded in the historical for a specific a trustor, and trustee.         Only specific information is returned This method obtains the last trust value recorded in the historical for a specific a trustor, trustee and offer.         Only specific information is returned This method obtains the last trust value recorded in Kafka for a specific a trustor, trustee and offer. All          previously recorded trust information is returned This function counts all interactions with a particular trustee in the historical This function counts all interactions with a particular offer in the historical This method returns the satisfaction average rate between a trustor and a trustee