<commit_before># Import Built-Ins
import logging
import time

# Import Homebrew
from btfxwss.connection import WebSocketConnection
from btfxwss.queue_processor import QueueProcessor


# Init Logging Facilities
log = logging.getLogger(__name__)


def is_connected(func):
    def wrapped(self, *args, **kwargs):
        if self.conn and self.conn.connected.is_set():
            return func(self, *args, **kwargs)
        else:
            log.error("Cannot call %s() on unestablished connection!",
                      func.__name__)
            return None
    return wrapped


class BtfxWss:
    """Websocket Client Interface to Bitfinex WSS API

    It features separate threads for the connection and data handling.
    Data can be accessed using the provided methods.
    """

    def __init__(self, key=None, secret=None, log_level=None, **wss_kwargs):
        """
        Initializes BtfxWss Instance.
        :param key: Api Key as string
        :param secret: Api secret as string
        :param addr: Websocket API Address
        """
        self.key = key if key else ''
        self.secret = secret if secret else ''

        self.conn = WebSocketConnection(log_level=log_level,
                                        **wss_kwargs)
        self.queue_processor = QueueProcessor(self.conn.q,
                                              log_level=log_level)

    ##############
    # Properties #
    ##############
    @property
    def channel_configs(self):
        return self.conn.channel_configs

    @property
    def account(self):
        return self.queue_processor.account

    ##############################################
    # Client Initialization and Shutdown Methods #
    ##############################################

    def start(self):
        """Start the client.

        :return:
        """
        self.conn.start()
        self.queue_processor.start()

    def stop(self):
        """Stop the client.

        :return:
        """
        self.conn.disconnect()
        self.queue_processor.join()

    def reset(self):
        """Reset the client.

        :return:
        """
        self.conn.reconnect()
        while not self.conn.connected.is_set():
            log.info("reset(): Waiting for connection to be set up..")
            time.sleep(1)

        for key in self.channel_configs:
            self.conn.send(**self.channel_configs[key])

    ##########################
    # Data Retrieval Methods #
    ##########################

    def tickers(self, pair):
        """Return a queue containing all received ticker data.

        :param pair:
        :return: Queue()
        """
        key = ('ticker', pair)
        if key in self.queue_processor.tickers:
            return self.queue_processor.tickers[key]
        else:
            raise KeyError(pair)

    def books(self, pair):
        """Return a queue containing all received book data.

        :param pair:
        :return: Queue()
        """
        key = ('book', pair)
        if key in self.queue_processor.books:
            return self.queue_processor.books[key]
        else:
            raise KeyError(pair)

    def raw_books(self, pair):
        """Return a queue containing all received raw book data.

        :param pair:
        :return: Queue()
        """
        key = ('raw_book', pair)
        if key in self.queue_processor.raw_books:
            return self.queue_processor.raw_books[key]
        else:
            raise KeyError(pair)

    def trades(self, pair):
        """Return a queue containing all received trades data.

        :param pair:
        :return: Queue()
        """
        key = ('trades', pair)
        if key in self.queue_processor.trades:
            return self.queue_processor.trades[key]
        else:
            raise KeyError(pair)

    def candles(self, pair, timeframe=None):
        """Return a queue containing all received candles data.

        :param pair: str, Symbol pair to request data for
        :param timeframe: str
        :return: Queue()
        """
        timeframe = '1m' if not timeframe else timeframe
        key = ('candles', pair, timeframe)
        if key in self.queue_processor.candles:
            return self.queue_processor.candles[key]
        else:
            raise KeyError(pair)

    ##########################################
    # Subscription and Configuration Methods #
    ##########################################

    def _subscribe(self, channel_name, identifier, **kwargs):
        q = {'event': 'subscribe', 'channel': channel_name}
        q.update(**kwargs)
        log.debug("_subscribe: %s", q)
        self.conn.send(**q)
        self.channel_configs[identifier] = q

    def _unsubscribe(self, channel_name, identifier, **kwargs):

        channel_id = self.channel_configs[identifier]
        q = {'event': 'unsubscribe', 'chanId': channel_id}
        q.update(kwargs)
        self.conn.send(**q)
        self.channel_configs.pop(identifier)

    def config(self, decimals_as_strings=True, ts_as_dates=False,
               sequencing=False, **kwargs):
        """Send configuration to websocket server

        :param decimals_as_strings: bool, turn on/off decimals as strings
        :param ts_as_dates: bool, decide to request timestamps as dates instead
        :param sequencing: bool, turn on sequencing
        :param kwargs:
        :return:<commit_msg>[jrhee17-111] Don't throw KeyError to allow blocking until events can be received<commit_after># Import Built-Ins
import logging
import time

# Import Homebrew
from btfxwss.connection import WebSocketConnection
from btfxwss.queue_processor import QueueProcessor


# Init Logging Facilities
log = logging.getLogger(__name__)


def is_connected(func):
    def wrapped(self, *args, **kwargs):
        if self.conn and self.conn.connected.is_set():
            return func(self, *args, **kwargs)
        else:
            log.error("Cannot call %s() on unestablished connection!",
                      func.__name__)
            return None
    return wrapped


class BtfxWss:
    """Websocket Client Interface to Bitfinex WSS API

    It features separate threads for the connection and data handling.
    Data can be accessed using the provided methods.
    """

    def __init__(self, key=None, secret=None, log_level=None, **wss_kwargs):
        """
        Initializes BtfxWss Instance.
        :param key: Api Key as string
        :param secret: Api secret as string
        :param addr: Websocket API Address
        """
        self.key = key if key else ''
        self.secret = secret if secret else ''

        self.conn = WebSocketConnection(log_level=log_level,
                                        **wss_kwargs)
        self.queue_processor = QueueProcessor(self.conn.q,
                                              log_level=log_level)

    ##############
    # Properties #
    ##############
    @property
    def channel_configs(self):
        return self.conn.channel_configs

    @property
    def account(self):
        return self.queue_processor.account

    ##############################################
    # Client Initialization and Shutdown Methods #
    ##############################################

    def start(self):
        """Start the client.

        :return:
        """
        self.conn.start()
        self.queue_processor.start()

    def stop(self):
        """Stop the client.

        :return:
        """
        self.conn.disconnect()
        self.queue_processor.join()

    def reset(self):
        """Reset the client.

        :return:
        """
        self.conn.reconnect()
        while not self.conn.connected.is_set():
            log.info("reset(): Waiting for connection to be set up..")
            time.sleep(1)

        for key in self.channel_configs:
            self.conn.send(**self.channel_configs[key])

    ##########################
    # Data Retrieval Methods #
    ##########################

    def tickers(self, pair):
        """Return a queue containing all received ticker data.

        :param pair:
        :return: Queue()
        """
        key = ('ticker', pair)
        return self.queue_processor.tickers[key]

    def books(self, pair):
        """Return a queue containing all received book data.

        :param pair:
        :return: Queue()
        """
        key = ('book', pair)
        return self.queue_processor.books[key]

    def raw_books(self, pair):
        """Return a queue containing all received raw book data.

        :param pair:
        :return: Queue()
        """
        key = ('raw_book', pair)
        return self.queue_processor.raw_books[key]

    def trades(self, pair):
        """Return a queue containing all received trades data.

        :param pair:
        :return: Queue()
        """
        key = ('trades', pair)
        return self.queue_processor.trades[key]

    def candles(self, pair, timeframe=None):
        """Return a queue containing all received candles data.

        :param pair: str, Symbol pair to request data for
        :param timeframe: str
        :return: Queue()
        """
        timeframe = '1m' if not timeframe else timeframe
        key = ('candles', pair, timeframe)
        return self.queue_processor.candles[key]

    ##########################################
    # Subscription and Configuration Methods #
    ##########################################

    def _subscribe(self, channel_name, identifier, **kwargs):
        q = {'event': 'subscribe', 'channel': channel_name}
        q.update(**kwargs)
        log.debug("_subscribe: %s", q)
        self.conn.send(**q)
        self.channel_configs[identifier] = q

    def _unsubscribe(self, channel_name, identifier, **kwargs):

        channel_id = self.channel_configs[identifier]
        q = {'event': 'unsubscribe', 'chanId': channel_id}
        q.update(kwargs)
        self.conn.send(**q)
        self.channel_configs.pop(identifier)

    def config(self, decimals_as_strings=True, ts_as_dates=False,
               sequencing=False, **kwargs):
        """Send configuration to websocket server

        :param decimals_as_strings: bool, turn on/off decimals as strings
        :param ts_as_dates: bool, decide to request timestamps as dates instead
        :param sequencing: bool, turn on sequencing
        :param kwargs:
        :return:
{'fraction_non_alphanumeric': 0.1427744549488713, 'fraction_numerical': 0.0008682230368512445, 'mean_word_length': 2.260062893081761, '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': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1499637', 'n_tokens_mistral': 2851, 'n_tokens_neox': 2726, 'n_words': 785}