<commit_before>import logging
import os
import subprocess

import boto3
import click
from retrying import retry

from ..utils.io import IngestStatus
from ..models import Scene
from ..ingest.models import Ingest
from ..ingest import (
    create_landsat8_ingest,
    create_sentinel2_ingest,
    create_ingest_definition
)
from ..uploads.landsat8.settings import datasource_id as landsat_id
from ..uploads.sentinel2.settings import datasource_id as sentinel2_id
from ..utils.exception_reporting import wrap_rollbar
from ..utils.emr import get_cluster_id, wait_for_emr_success

logger = logging.getLogger(__name__)

BATCH_JAR_PATH = os.getenv('BATCH_JAR_PATH', 'rf-batch-761c316.jar')


@click.command(name='ingest-scene')
@click.argument('scene_id')
@click.option('--ignore-previous', is_flag=True,
              help='Boolean to ignore scene status for ingestion')
@wrap_rollbar
def ingest_scene(scene_id, ignore_previous):
    """Ingest a scene into Raster Foundry

    Args:
        scene_id (str): ID of scene to ingest
        ignore_previous (bool): whether to ignore ingest_status when processing ingest
    """
    logger.info("Ingesting Scene: %s", scene_id)
    s3_uri, ingest_definition_id = save_ingest_def_to_s3(scene_id, ignore_previous)
    launch_spark_ingest_job(s3_uri, ingest_definition_id, scene_id)
    wait_for_status(ingest_definition_id)


def save_ingest_def_to_s3(scene_id, ignore_previous=False):
    """Create ingest definition and upload to S3

    Args:
        ignore_previous (bool): boolean to ignore previous ingest status
        scene_id (str): ID of scene to create ingest definition for
    """
    scene = Scene.from_id(scene_id)

    logger.info('Beginning to create ingest definition for scene %s for user %s...',
                scene_id, scene.owner)
    if scene.ingestStatus not in [IngestStatus.TOBEINGESTED, IngestStatus.FAILED] and not ignore_previous:
        raise Exception('Scene is no longer waiting to be ingested, error error')

    scene.ingestStatus = IngestStatus.INGESTING
    logger.info('Updating scene (%s) status to ingesting', scene_id)
    scene.update()
    logger.info('Successfully updated scene (%s) status', scene_id)

    logger.info('Creating ingest definition')
    if scene.datasource == landsat_id:
        ingest_definition = create_landsat8_ingest(scene)
    elif scene.datasource == sentinel2_id:
        ingest_definition = create_sentinel2_ingest(scene)
    else:
        ingest_definition = create_ingest_definition(scene)
    ingest_definition.put_in_s3()
    logger.info('Successfully created and pushed ingest definition for scene %s', scene)

    Ingest.delete_status_from_s3(ingest_definition.id)
    return ingest_definition.s3_uri, ingest_definition.id


def launch_spark_ingest_job(ingest_def_uri, ingest_def_id, scene_id):
    """Launch ingest job and wait for success/failure

    Args:
        ingest_def_uri (str): S3 URI for location of ingest definition
        ingest_def_id (str): ID for ingest definition
        scene_id (str): ID for scene to be ingested

    Returns:
        bool
    """

    logger.info('Launching Spark ingest job with ingest definition %s for scene %s',
                ingest_def_uri, scene_id)
    cluster_id = get_cluster_id()
    emr_response = execute_ingest_emr_job(scene_id, ingest_def_uri, ingest_def_id, cluster_id)
    logger.info('Launched Spark ingest job for %s with ingest ID %s. Waiting for status changes.',
                scene_id, ingest_def_id)
    step_id = emr_response['StepIds'][0]
    is_success = wait_for_emr_success(step_id, cluster_id)
    return is_success

@retry(wait_exponential_multiplier=2000, wait_exponential_max=30000, stop_max_attempt_number=5)
def execute_ingest_emr_job(scene_id, ingest_s3_uri, ingest_def_id, cluster_id):
    """Kick off ingest in AWS EMR

    Args:
        scene_id (str): id of the scene to ingest
        ingest_s3_uri (str): URI for ingest definition
        ingest_def_id (str): ID to namespace ingest job
        cluster_id (str): ID of the cluster to submit work to

    Returns:
        dict
    """
    logger.info('Constructing ingest step request for %s for cluster %s for ingest id %s',
                scene_id, cluster_id, ingest_def_id)
    step = {
        'ActionOnFailure': 'CONTINUE',
        'Name': 'ingest-{}'.format(ingest_def_id),
        'HadoopJarStep': {
            'Args': ['/usr/bin/spark-submit',
                     '--master',
                     'yarn',
                     '--deploy-mode',
                     'cluster',
                     '--conf',<commit_msg>Add local option to ingest command to run ingests locally<commit_after>import logging
import os
import subprocess

import boto3
import click
from retrying import retry

from ..utils.io import IngestStatus
from ..models import Scene
from ..ingest.models import Ingest
from ..ingest import (
    create_landsat8_ingest,
    create_sentinel2_ingest,
    create_ingest_definition
)
from ..uploads.landsat8.settings import datasource_id as landsat_id
from ..uploads.sentinel2.settings import datasource_id as sentinel2_id
from ..utils.exception_reporting import wrap_rollbar
from ..utils.emr import get_cluster_id, wait_for_emr_success

logger = logging.getLogger(__name__)

BATCH_JAR_PATH = os.getenv('BATCH_JAR_PATH', 'rf-batch-761c316.jar')


@click.command(name='ingest-scene')
@click.argument('scene_id')
@click.option('--ignore-previous', is_flag=True,
              help='Boolean to ignore scene status for ingestion')
@click.option('--local', is_flag=True,
              help='Boolean to run using a spark local cluster rather than EMR job')
@wrap_rollbar
def ingest_scene(scene_id, ignore_previous, local):
    """Ingest a scene into Raster Foundry

    Args:
        scene_id (str): ID of scene to ingest
        ignore_previous (bool): whether to ignore ingest_status when processing ingest
    """
    logger.info("Ingesting Scene: %s", scene_id)
    s3_uri, ingest_definition_id = save_ingest_def_to_s3(scene_id, ignore_previous)
    launch_spark_ingest_job(s3_uri, ingest_definition_id, scene_id, local)
    wait_for_status(ingest_definition_id)


def save_ingest_def_to_s3(scene_id, ignore_previous=False):
    """Create ingest definition and upload to S3

    Args:
        ignore_previous (bool): boolean to ignore previous ingest status
        scene_id (str): ID of scene to create ingest definition for
    """
    scene = Scene.from_id(scene_id)

    logger.info('Beginning to create ingest definition for scene %s for user %s...',
                scene_id, scene.owner)
    if scene.ingestStatus not in [IngestStatus.TOBEINGESTED, IngestStatus.FAILED] and not ignore_previous:
        raise Exception('Scene is no longer waiting to be ingested, error error')

    scene.ingestStatus = IngestStatus.INGESTING
    logger.info('Updating scene (%s) status to ingesting', scene_id)
    scene.update()
    logger.info('Successfully updated scene (%s) status', scene_id)

    logger.info('Creating ingest definition')
    if scene.datasource == landsat_id:
        ingest_definition = create_landsat8_ingest(scene)
    elif scene.datasource == sentinel2_id:
        ingest_definition = create_sentinel2_ingest(scene)
    else:
        ingest_definition = create_ingest_definition(scene)
    ingest_definition.put_in_s3()
    logger.info('Successfully created and pushed ingest definition for scene %s', scene)

    Ingest.delete_status_from_s3(ingest_definition.id)
    return ingest_definition.s3_uri, ingest_definition.id


def launch_spark_ingest_job(ingest_def_uri, ingest_def_id, scene_id, local=False):
    """Launch ingest job and wait for success/failure

    Args:
        ingest_def_uri (str): S3 URI for location of ingest definition
        ingest_def_id (str): ID for ingest definition
        scene_id (str): ID for scene to be ingested
        local (bool): whether to execute ingest as a spark local job

    Returns:
        bool
    """

    logger.info('Launching Spark ingest job with ingest definition %s for scene %s',
                ingest_def_uri, scene_id)
    if local:
        return execute_local_ingest_job(scene_id, ingest_def_uri)
    else:
        cluster_id = get_cluster_id()
        emr_response = execute_ingest_emr_job(scene_id, ingest_def_uri, ingest_def_id, cluster_id)
        logger.info('Launched Spark ingest job for %s with ingest ID %s. Waiting for status changes.',
                    scene_id, ingest_def_id)
        step_id = emr_response['StepIds'][0]
        is_success = wait_for_emr_success(step_id, cluster_id)
        return is_success


def execute_local_ingest_job(scene_id, ingest_s3_uri):
    """Execute ingest job locally with a spark local cluster"""

    command = ['spark-submit',
               '--master', 'local[32]', '--driver-memory', '48g',
               '--class', 'com.azavea.rf.batch.ingest.spark.Ingest',
               '/opt/raster-foundry/jars/rf-batch.jar',
               '-t', '--overwrite', '-s', scene_id, '-j', ingest_s3_uri
    ]
    subprocess.check_call(command)
    logger.info('Finished Ingesting %s in spark local', scene_id)
    return True


@retry(wait_exponential_multiplier=2000, wait_exponential_max=30000, stop_max_attempt_number=5)
def execute_ingest_emr_job(scene_id, ingest_s3_uri, ingest_def_id, cluster_id):
    """Kick off ingest in AWS EMR

    Args:
        scene_id (str): id of the scene to ingest
        ingest_s3_uri (str): URI for ingest definition
        ingest_def_id (str): ID to namespace ingest job
        cluster_id (str): ID of the cluster to submit work to

    Returns:
        dict
    """
    logger.info('Constructing ingest step request for %s for cluster %s for ingest id %s',
                scene_id, cluster_id, ingest_def_id)
    step = {
        'ActionOnFailure': 'CONTINUE',
        'Name': 'ingest-{}'.format(ingest_def_id),
        'HadoopJarStep': {
            'Args': ['/usr/bin/spark-submit',
                     '--master',
                     'yarn',
                     '--deploy-mode',
                     'cluster',
                     '--conf',
{'fraction_non_alphanumeric': 0.08863771564544913, 'fraction_numerical': 0.00793178663493952, 'mean_word_length': 3.3180650684931505, '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': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2225007', 'n_tokens_mistral': 3250, 'n_tokens_neox': 3130, 'n_words': 831}