<commit_before>"""Mixins and core functionality."""
import json
import os
from contextlib import closing
from io import BytesIO
from itertools import repeat
from math import ceil
from multiprocessing.pool import ThreadPool

import numpy as np
from PIL import Image

from helios.core.request_manager import RequestManager
from helios.core.session import Session
from helios.utilities import logging_utils


class SDKCore(object):
    """
    Core class for Python interface to Helios Core APIs.

    This class must be inherited by any additional Core API classes.
    """
    max_threads = 32

    def __init__(self, session=None):
        """
        Initialize core capabilities.

        If a session has been started prior to initialization it can be used
        via the session input parameter. If this is not used a session will
        be started automatically.

        Args:
            session (Session object, optional): An instance of the
                Session. Defaults to None. If unused a session will be
                created for you.

        """
        # Start session or use custom session.
        if session is None:
            self.session = Session()
        else:
            self.session = session

        # If the session hasn't been started, start it.
        if not self.session.token:
            self.session.start_session()

        # Create request manager to handle all API requests.
        self.request_manager = RequestManager(self.session.token,
                                              pool_maxsize=self.max_threads)

    @staticmethod
    def parse_query_inputs(input_dict):
        # Check for unique case: sensors
        if 'sensors' in input_dict.keys():
            query_str = input_dict.pop('sensors') + '&'
        else:
            query_str = ''

        # Parse input_dict into a str
        for key in input_dict.keys():
            if isinstance(input_dict[key], (list, tuple)):
                query_str += (str(key) +
                              '=' +
                              ','.join([str(x) for x in input_dict[key]]) +
                              '&')
            else:
                query_str += (str(key) + '=' + str(input_dict[key]) + '&')
        query_str = query_str[:-1]

        return query_str

    @staticmethod
    def check_headers_for_dud(header_data):
        if 'x-amz-meta-helios' in header_data:
            meta_block = json.loads(header_data['x-amz-meta-helios'])
            if meta_block['isOutcast'] or meta_block['isDud'] or meta_block['isFrozen']:
                return True
        return False

    @property
    def base_api_url(self):
        return self.session.api_url
<commit_msg>Better dictionary iteration in parse_query_inputs. Removes None values from dict.<commit_after>"""Mixins and core functionality."""
import json
import os
from contextlib import closing
from io import BytesIO
from itertools import repeat
from math import ceil
from multiprocessing.pool import ThreadPool

import numpy as np
from PIL import Image

from helios.core.request_manager import RequestManager
from helios.core.session import Session
from helios.utilities import logging_utils


class SDKCore(object):
    """
    Core class for Python interface to Helios Core APIs.

    This class must be inherited by any additional Core API classes.
    """
    max_threads = 32

    def __init__(self, session=None):
        """
        Initialize core capabilities.

        If a session has been started prior to initialization it can be used
        via the session input parameter. If this is not used a session will
        be started automatically.

        Args:
            session (Session object, optional): An instance of the
                Session. Defaults to None. If unused a session will be
                created for you.

        """
        # Start session or use custom session.
        if session is None:
            self.session = Session()
        else:
            self.session = session

        # If the session hasn't been started, start it.
        if not self.session.token:
            self.session.start_session()

        # Create request manager to handle all API requests.
        self.request_manager = RequestManager(self.session.token,
                                              pool_maxsize=self.max_threads)

    @staticmethod
    def parse_query_inputs(input_dict):
        # Check for unique case: sensors
        if 'sensors' in input_dict:
            query_str = input_dict.pop('sensors') + '&'
        else:
            query_str = ''

        # Parse input_dict into a str
        for key, val in input_dict.items():
            if val is None:
                continue

            if isinstance(val, (list, tuple)):
                query_str += (str(key) +
                              '=' +
                              ','.join([str(x) for x in val]) +
                              '&')
            else:
                query_str += (str(key) + '=' + str(val) + '&')

        # Remove final ampersand
        query_str = query_str[:-1]

        return query_str

    @staticmethod
    def check_headers_for_dud(header_data):
        if 'x-amz-meta-helios' in header_data:
            meta_block = json.loads(header_data['x-amz-meta-helios'])
            if meta_block['isOutcast'] or meta_block['isDud'] or meta_block['isFrozen']:
                return True
        return False

    @property
    def base_api_url(self):
        return self.session.api_url

{'fraction_non_alphanumeric': 0.07696496641858777, 'fraction_numerical': 0.0010891268832819024, 'mean_word_length': 1.9607737775389575, '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': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '76045', 'n_tokens_mistral': 1472, 'n_tokens_neox': 1420, 'n_words': 488}