def fetch_resource(
        resource_name, resource_id=None, method='GET', data=None,
        server_url=None, token=None):
    f = getattr(requests, method.lower())
    server_url = server_url if server_url else get_server_url()
    url = get_resource_url(server_url, resource_name, resource_id)
    token = token if token else get_token()
    kw = {} if data is None else {'json': data}
    try:
        response = f(url, headers={'Authorization': 'Bearer ' + token}, **kw)
    except requests.ConnectionError:
        raise CrossComputeConnectionError({
            'url': 'could not connect to server ' + url})
    status_code = response.status_code
    d = {'url': url, 'token': token, 'statusCode': status_code}
    if status_code in [401, 403]:
        d['statusHelp'] = 'please check your server url and token'
        raise CrossComputeConnectionError(d)
    try:
        response_json = response.json()
    except ValueError:
        d['statusHelp'] = 'could not parse response as json'
        d['responseContent'] = response.content.decode('utf-8')
        raise CrossComputeConnectionError(d)
    if status_code != 200:
        if response_json:
            d.update(response_json)
        raise (
            CrossComputeExecutionError if status_code == 400 else
            CrossComputeImplementationError)(d)
    return response_json def get_bash_configuration_text():
    return BASH_CONFIGURATION_TEXT.format(
        client_url=get_client_url(),
        server_url=get_server_url(),
        token=get_token('YOUR-TOKEN')) def get_token(default=None):
    return get_environment_value('CROSSCOMPUTE_TOKEN', default) def yield_echo(statistics_dictionary, is_quiet=False, as_json=False):
    statistics_dictionary['ping count'] = 0
    try:
        for echo_message in get_echoes_client():
            event_name = echo_message.event
            if event_name == 'message':
                if not is_quiet and not as_json:
                    print('.', end='', flush=True)
                statistics_dictionary['ping count'] += 1
                continue
            elif not is_quiet:
                print('$', end='', flush=True)
            event_dictionary = json.loads(echo_message.data)
            yield event_name, event_dictionary
    except KeyboardInterrupt:
        raise CrossComputeKeyboardInterrupt def normalize_result_definition(raw_result_definition, folder=None):
def normalize_data(raw_data, view_name, variable_id=None, folder=None):
def normalize_data_dictionary(
        raw_data_dictionary, view_name, variable_id=None, folder=None):
def normalize_report_variable_dictionaries(
        raw_variable_dictionaries, folder=None):
def normalize_result_variable_dictionaries(
        raw_variable_dictionaries, variable_definitions, folder=None):
def normalize_report_template_dictionaries(
        raw_template_dictionaries, folder=None):
def normalize_tool_template_dictionaries(
        raw_template_dictionaries, variable_dictionaries, folder=None):
def normalize_test_dictionaries(raw_test_dictionaries):
def normalize_environment_dictionary(raw_environment_dictionary):
def normalize_block_dictionaries(raw_block_dictionaries, with_data=True):
def get_print_dictionary(dictionary, folder):
def get_template_block_dictionaries(
        dictionary, variable_dictionaries, folder=None): def save_table_csv(target_path, value, variable_id, value_by_id_by_path): VIEW_NAMES = [
    'text',
    'number',
    'markdown',
    'table',
    'image',
    'map',
    'electricity-network',
    'file',
]
DEFAULT_VIEW_NAME = 'text'
PRINT_FORMAT_NAMES = [
    'pdf'
]
DEBUG_VARIABLE_DEFINITIONS = [{
    'id': 'stdout',
    'name': 'Standard Output',
    'view': 'text',
    'path': 'stdout.txt',
}, {
    'id': 'stderr',
    'name': 'Standard Error',
    'view': 'text',
    'path': 'stderr.txt',
}] class CrossComputeConnectionError(HTTPInternalServerError, CrossComputeError):
    pass class CrossComputeDefinitionError(HTTPBadRequest, CrossComputeError):
    pass class CrossComputeError(Exception):
    pass class CrossComputeExecutionError(HTTPBadRequest, CrossComputeError):
    pass class CrossComputeImplementationError(
        HTTPInternalServerError, CrossComputeError):
    pass class CrossComputeKeyboardInterrupt(KeyboardInterrupt, CrossComputeError):
    pass def sanitize_name(name):
    return compact_whitespace(''.join(
        _ if is_valid_name_character(_) else ' ' for _ in name)) def sanitize_json_value(value):
    if isinstance(value, dict):
        return {
            sanitize_json_value(k): sanitize_json_value(v)
            for k, v in value.items()}
    if isinstance(value, list):
        return [sanitize_json_value(_) for _ in value]
    if isinstance(value, str):
        return value
    return None if value is None or isnan(value) else value
import json
import re
import requests
import subprocess
import time
from collections import defaultdict
from concurrent.futures import (
    ThreadPoolExecutor)
from copy import deepcopy
from functools import partial
from invisibleroads_macros_disk import (
    TemporaryStorage, make_folder, make_random_folder)
from io import StringIO
from itertools import product, repeat
from mimetypes import guess_type
from os import environ, getcwd
from os.path import (
    abspath, basename, dirname, getsize, exists, isdir, join, splitext)
from sys import exc_info
from traceback import print_exception
from .connection import (
    fetch_resource,
    get_bash_configuration_text,
    get_token,
    yield_echo)
from .definition import (
    get_nested_value,
    get_template_dictionary,
    get_variable_dictionary_by_id,
    load_definition)
from .serialization import (
    define_load,
    define_save,
    load_value_json,
    render_object,
    save_json,
    SAVE_BY_EXTENSION_BY_VIEW)
from ..constants import DEBUG_VARIABLE_DEFINITIONS, S
from ..exceptions import (
    CrossComputeConnectionError,
    CrossComputeDefinitionError,
    CrossComputeError,
    CrossComputeExecutionError,
    CrossComputeImplementationError,
    CrossComputeKeyboardInterrupt)
from ..macros import (
    sanitize_name,
    sanitize_json_value)
from ..symmetries import download
# TODO: Consider sanitize_json_value for result input, output, log, debug




# https://stackoverflow.com/a/14693789/192092
ANSI_ESCAPE_PATTERN = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])')


class SafeDict(dict):
    # https://stackoverflow.com/a/17215533/192092

    def __missing__(self, key):
        return '{' + key + '}'


def run_report_automation(report_definition, is_mock=True, log=None):

    style_dictionary = get_nested_value(