# Path: datapackage_pipelines/specs/specs.py
# SPEC_PARSERS = [
#     BasicPipelineParser(),
#     SourceSpecPipelineParser()
# ]
# YAML_LOADER = yaml.CLoader if 'CLoader' in yaml.__dict__ else yaml.Loader
# def resolve_processors(spec: PipelineSpec):
# def process_schedules(spec: PipelineSpec):
# def process_environment(spec: PipelineSpec):
# def find_specs(root_dir='.') -> PipelineSpec:
# def pipelines(prefixes=None, ignore_missing_deps=False, root_dir='.', status_manager=None):
# def register_all_pipelines(root_dir='.'):
# 
# Path: datapackage_pipelines/status/status_manager.py
# def status_mgr(root_dir='.') -> StatusManager:
#     global _status
#     global _root_dir
# 
#     if _status is not None and _root_dir == root_dir:
#         return _status
#     _root_dir = root_dir
#     _status = StatusManager(host=os.environ.get('DPP_REDIS_HOST'), root_dir=root_dir)
#     return _status
# 
# Path: datapackage_pipelines/utilities/stat_utils.py
# STATS_DPP_KEY = '.dpp'
# 
# STATS_OUT_DP_URL_KEY = 'out-datapackage-url'
# 
# Path: datapackage_pipelines/utilities/extended_json.py
# class json(object):
#     dumps = _dumps
#     loads = _loads
#     dumpl = _dumpl
#     loadl = _loadl
#     dump = _dump
#     load = _load
#     JSONDecodeError = _json.JSONDecodeError
# 
# Path: datapackage_pipelines/manager/runners/runner_config.py
# class RunnerConfiguration(object):
#     ENV_VAR = 'DPP_RUNNER_CONFIG'
#     DEFAULT_RUNNER_CONFIG = 'dpp-runners.yaml'
#     def __init__(self):
#     def get_runner_class(self, kind):
#     def get_runner(self, name):


import asyncio
import gzip
import logging
import os
from concurrent.futures import CancelledError
from json.decoder import JSONDecodeError
from ..specs.specs import resolve_executor
from ..status import status_mgr
from ..utilities.stat_utils import STATS_DPP_KEY, STATS_OUT_DP_URL_KEY
from ..utilities.extended_json import json
from .runners import runner_config
                }
            }
            step['executor'] = resolve_executor(step, '.', [])
            pipeline_steps.insert(0, step)
            break

    return pipeline_steps


async def construct_process_pipeline(pipeline_steps, pipeline_cwd, errors, debug=False):
    error_collectors = []
    processes = []
    error_queue = asyncio.Queue()
    rfd = None

    error_aggregator = \
        asyncio.ensure_future(dequeue_errors(error_queue, errors))

    pipeline_steps.append({
        'run': '(sink)',
        'executor': SINK,
        '_cache_hash': pipeline_steps[-1]['_cache_hash']
    })

    for i, step in enumerate(pipeline_steps):

        new_rfd, wfd = os.pipe()

        if debug:
            logging.info("- %s", step['run'])

# Path: datapackage_pipelines/specs/specs.py
# SPEC_PARSERS = [
#     BasicPipelineParser(),
#     SourceSpecPipelineParser()
# ]
# YAML_LOADER = yaml.CLoader if 'CLoader' in yaml.__dict__ else yaml.Loader
# def resolve_processors(spec: PipelineSpec):
# def process_schedules(spec: PipelineSpec):
# def process_environment(spec: PipelineSpec):
# def find_specs(root_dir='.') -> PipelineSpec:
# def pipelines(prefixes=None, ignore_missing_deps=False, root_dir='.', status_manager=None):
# def register_all_pipelines(root_dir='.'):
# 
# Path: datapackage_pipelines/status/status_manager.py
# def status_mgr(root_dir='.') -> StatusManager:
#     global _status
#     global _root_dir
# 
#     if _status is not None and _root_dir == root_dir:
#         return _status
#     _root_dir = root_dir
#     _status = StatusManager(host=os.environ.get('DPP_REDIS_HOST'), root_dir=root_dir)
#     return _status
# 
# Path: datapackage_pipelines/utilities/stat_utils.py
# STATS_DPP_KEY = '.dpp'
# 
# STATS_OUT_DP_URL_KEY = 'out-datapackage-url'
# 
# Path: datapackage_pipelines/utilities/extended_json.py
# class json(object):
#     dumps = _dumps
#     loads = _loads
#     dumpl = _dumpl
#     loadl = _loadl
#     dump = _dump
#     load = _load
#     JSONDecodeError = _json.JSONDecodeError
# 
# Path: datapackage_pipelines/manager/runners/runner_config.py
# class RunnerConfiguration(object):
#     ENV_VAR = 'DPP_RUNNER_CONFIG'
#     DEFAULT_RUNNER_CONFIG = 'dpp-runners.yaml'
#     def __init__(self):
#     def get_runner_class(self, kind):
#     def get_runner(self, name):


# Path: datapackage_pipelines/manager/tasks.py
import asyncio
import gzip
import logging
import os
from concurrent.futures import CancelledError
from json.decoder import JSONDecodeError
from ..specs.specs import resolve_executor
from ..status import status_mgr
from ..utilities.stat_utils import STATS_DPP_KEY, STATS_OUT_DP_URL_KEY
from ..utilities.extended_json import json
from .runners import runner_config

                }
            }
            step['executor'] = resolve_executor(step, '.', [])
            pipeline_steps.insert(0, step)
            break

    return pipeline_steps


async def construct_process_pipeline(pipeline_steps, pipeline_cwd, errors, debug=False):
    error_collectors = []
    processes = []
    error_queue = asyncio.Queue()
    rfd = None

    error_aggregator = \
        asyncio.ensure_future(dequeue_errors(error_queue, errors))

    pipeline_steps.append({
        'run': '(sink)',
        'executor': SINK,
        '_cache_hash': pipeline_steps[-1]['_cache_hash']
    })

    for i, step in enumerate(pipeline_steps):

        new_rfd, wfd = os.pipe()

        if debug:
            logging.info("- %s", step['run'])

        runner = runner_config.get_runner(step.get('runner'))