# 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

async def process_death_waiter(process):
    return_code = await process.wait()
    return process, return_code


def find_caches(pipeline_steps, pipeline_cwd):
    if not any(step.get('cache') for step in pipeline_steps):
        # If no step requires caching then bail
        return pipeline_steps

    for i, step in reversed(list(enumerate(pipeline_steps))):
        cache_filename = os.path.join(pipeline_cwd,
                                      '.cache',
                                      step['_cache_hash'])
        if os.path.exists(cache_filename):
            try:
                canary = gzip.open(cache_filename, "rt")
                canary.seek(1)
                canary.close()
            except Exception:  #noqa
                continue
            logging.info('Found cache for step %d: %s', i, step['run'])
            pipeline_steps = pipeline_steps[i+1:]
            step = {
                'run': 'cache_loader',
                'parameters': {
                    'load-from': os.path.join('.cache', step['_cache_hash'])
                }
            }

# 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


async def process_death_waiter(process):
    return_code = await process.wait()
    return process, return_code


def find_caches(pipeline_steps, pipeline_cwd):
    if not any(step.get('cache') for step in pipeline_steps):
        # If no step requires caching then bail
        return pipeline_steps

    for i, step in reversed(list(enumerate(pipeline_steps))):
        cache_filename = os.path.join(pipeline_cwd,
                                      '.cache',
                                      step['_cache_hash'])
        if os.path.exists(cache_filename):
            try:
                canary = gzip.open(cache_filename, "rt")
                canary.seek(1)
                canary.close()
            except Exception:  #noqa
                continue
            logging.info('Found cache for step %d: %s', i, step['run'])
            pipeline_steps = pipeline_steps[i+1:]
            step = {
                'run': 'cache_loader',
                'parameters': {
                    'load-from': os.path.join('.cache', step['_cache_hash'])
                }
            }

            step['executor'] = resolve_executor(step, '.', [])