<commit_before>import asyncio
import mqtt_interface.mqttinterface as mqtt
import mqtt_interface.pipeline as pipeline
import concurrent.futures as futures
import functools as ft
import json
import queue
from utils.utils import *

### For logging ###
import logging
import logging.handlers as handlers

logging.basicConfig(format='%(levelname)s %(asctime)s %(message)s', datefmt='%m/%d/%Y %I:%M:%S %p')

class VizierNode:

    def __init__(self, broker_host, broker_port, node_descriptor, logging_config = None):
        self.mqtt_client = mqtt.MQTTInterface(port=broker_port, host=broker_host)
        self.node_descriptor = node_descriptor
        self.end_point = node_descriptor["end_point"]
        self.expanded_links = generate_links_from_descriptor(node_descriptor)
        self.links = {}
        self.host = broker_host
        self.port = broker_port

        self.publishable_mapping = {}
        self.offerable_mapping = {}
        self.offerable_data = {}
        self.gettable_mapping = {}

        self.executor = futures.ThreadPoolExecutor(max_workers=100)

        if(logging_config):
            logging.configDict(logging_config)
            self.logger = logging.getLogger(__name__)
        else:
            self.logger = logging.getLogger(__name__)
            self.logger.setLevel(logging.DEBUG)

    # QUICK CURRY FUNCTION FOR HANDLING GET REQUESTS ON CHANNELS
    def _create_get_handler(self, info):
        """ Returns a message handler for use as a callback """

        def f(network_message):

            try:
                network_message = json.loads(network_message.payload.decode(encoding='UTF-8'))
            except Exception as e:
                # TODO: (PAUL) Change the logger to be passed into function
                logger = logging.getLogger(__name__)
                logger.error("Got malformed network message")

            # Make sure that we actually got a valid get request
            if('response' in network_message and 'link' in network_message['response']):
                response_channel = network_message['response']['link']
                json_message = create_vizier_get_response(info, message_type="data")
                self.mqtt_client.send_message(response_channel, json.dumps(json_message).encode(encoding='UTF-8'))

        return f

    def _get_request(self, link, retries=10, timeout=5):

        response_link = self.end_point + '/' + link + '/' + 'response'
        _, q = self.mqtt_client.subscribe(response_link)
        decoded_message = None
        get_request = json.dumps(create_vizier_get_message(link, response_link)).encode(encoding='UTF-8')

        for _ in range(retries):

            self.mqtt_client.send_message(link, get_request)

            # We expect this to potentially fail with a timeout
            try:
                mqtt_message = q.get(timeout=timeout)
            except Exception as e:
                pass

            # Try to decode packet.  Could potentially fail
            try:
                decoded_message = json.loads(mqtt_message.payload.decode(encoding='UTF-8'))
                break
            except Exception as e:
                self.logger.error(repr(e))
                self.logger.error('Malformed network packet')

        if(decoded_message is None):
            self.logger.error('Get request on topic (%s) failed', topic)

        # Make sure that we unsubscribe from the response channel
        self.mqtt_client.unsubscribe(response_link)
        return decoded_message<commit_msg>Don't print out expected except.<commit_after>import asyncio
import mqtt_interface.mqttinterface as mqtt
import mqtt_interface.pipeline as pipeline
import concurrent.futures as futures
import functools as ft
import json
import queue
from utils.utils import *

### For logging ###
import logging
import logging.handlers as handlers

logging.basicConfig(format='%(levelname)s %(asctime)s %(message)s', datefmt='%m/%d/%Y %I:%M:%S %p')

class VizierNode:

    def __init__(self, broker_host, broker_port, node_descriptor, logging_config = None):
        self.mqtt_client = mqtt.MQTTInterface(port=broker_port, host=broker_host)
        self.node_descriptor = node_descriptor
        self.end_point = node_descriptor["end_point"]
        self.expanded_links = generate_links_from_descriptor(node_descriptor)
        self.links = {}
        self.host = broker_host
        self.port = broker_port

        self.publishable_mapping = {}
        self.offerable_mapping = {}
        self.offerable_data = {}
        self.gettable_mapping = {}

        self.executor = futures.ThreadPoolExecutor(max_workers=100)

        if(logging_config):
            logging.configDict(logging_config)
            self.logger = logging.getLogger(__name__)
        else:
            self.logger = logging.getLogger(__name__)
            self.logger.setLevel(logging.DEBUG)

    # QUICK CURRY FUNCTION FOR HANDLING GET REQUESTS ON CHANNELS
    def _create_get_handler(self, info):
        """ Returns a message handler for use as a callback """

        def f(network_message):

            try:
                network_message = json.loads(network_message.payload.decode(encoding='UTF-8'))
            except Exception as e:
                # TODO: (PAUL) Change the logger to be passed into function
                logger = logging.getLogger(__name__)
                logger.error("Got malformed network message")

            # Make sure that we actually got a valid get request
            if('response' in network_message and 'link' in network_message['response']):
                response_channel = network_message['response']['link']
                json_message = create_vizier_get_response(info, message_type="data")
                self.mqtt_client.send_message(response_channel, json.dumps(json_message).encode(encoding='UTF-8'))

        return f

    def _get_request(self, link, retries=10, timeout=5):

        response_link = self.end_point + '/' + link + '/' + 'response'
        _, q = self.mqtt_client.subscribe(response_link)
        decoded_message = None
        get_request = json.dumps(create_vizier_get_message(link, response_link)).encode(encoding='UTF-8')

        for _ in range(retries):

            self.mqtt_client.send_message(link, get_request)

            # We expect this to potentially fail with a timeout
            try:
                mqtt_message = q.get(timeout=timeout)
            except Exception as e:
                # Don't try to decode message if we didn't get anything
                continue

            # Try to decode packet.  Could potentially fail
            try:
                decoded_message = json.loads(mqtt_message.payload.decode(encoding='UTF-8'))
                break
            except Exception as e:
                # Just pass here because we expect to fail.  In the future,
                # split the exceptions up into reasonable cases.
                pass

        if(decoded_message is None):
            self.logger.error('Get request on topic (%s) failed', topic)

        # Make sure that we unsubscribe from the response channel
        self.mqtt_client.unsubscribe(response_link)
        return decoded_message
{'fraction_non_alphanumeric': 0.08729496802891298, 'fraction_numerical': 0.0027800945232137894, 'mean_word_length': 2.5356265356265357, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4792923', 'n_tokens_mistral': 2027, 'n_tokens_neox': 1931, 'n_words': 482}