astrobase/lcproc/awsrun.py
def runcp_producer_loop(
        lightcurve_list,
        input_queue,
        input_bucket,
        result_queue,
        result_bucket,
        pfresult_list=None,
        runcp_kwargs=None,
        process_list_slice=None,
        purge_queues_when_done=False,
        delete_queues_when_done=False,
        download_when_done=True,
        save_state_when_done=True,
        s3_client=None,
        sqs_client=None
):
    """This sends checkplot making tasks to the input queue and monitors the
    result queue for task completion.

    Parameters
    ----------

    lightcurve_list : str or list of str
        This is either a string pointing to a file containing a list of light
        curves filenames to process or the list itself. The names must
        correspond to the full filenames of files stored on S3, including all
        prefixes, but not include the 's3://<bucket name>/' bit (these will be
        added automatically).

    input_queue : str
        This is the name of the SQS queue which will receive processing tasks
        generated by this function. The queue URL will automatically be obtained
        from AWS.

    input_bucket : str
        The name of the S3 bucket containing the light curve files to process.

    result_queue : str
        This is the name of the SQS queue that this function will listen to for
        messages from the workers as they complete processing on their input
        elements. This function will attempt to match input sent to the
        `input_queue` with results coming into the `result_queue` so it knows
        how many objects have been successfully processed. If this function
        receives task results that aren't in its own input queue, it will
        acknowledge them so they complete successfully, but not download them
        automatically. This handles leftover tasks completing from a previous
        run of this function.

    result_bucket : str
        The name of the S3 bucket which will receive the results from the
        workers.

    pfresult_list : list of str or None
        This is a list of periodfinder result pickle S3 URLs associated with
        each light curve. If provided, this will be used to add in phased light
        curve plots to each checkplot pickle. If this is None, the worker loop
        will produce checkplot pickles that only contain object information,
        neighbor information, and unphased light curves.

    runcp_kwargs : dict
        This is a dict used to pass any extra keyword arguments to the
        `lcproc.checkplotgen.runcp` function that will be run by the worker
        loop.

    process_list_slice : list
        This is used to index into the input light curve list so a subset of the
        full list can be processed in this specific run of this function.

        Use None for a slice index elem to emulate single slice spec behavior:

        process_list_slice = [10, None]  -> lightcurve_list[10:]
        process_list_slice = [None, 500] -> lightcurve_list[:500]

    purge_queues_when_done : bool
        If this is True, and this function exits (either when all done, or when
        it is interrupted with a Ctrl+C), all outstanding elements in the
        input/output queues that have not yet been acknowledged by workers or by
        this function will be purged. This effectively cancels all outstanding
        work.

    delete_queues_when_done : bool
        If this is True, and this function exits (either when all done, or when
        it is interrupted with a Ctrl+C'), all outstanding work items will be
        purged from the input/queues and the queues themselves will be deleted.

    download_when_done : bool
        If this is True, the generated checkplot pickle for each input work item
        will be downloaded immediately to the current working directory when the
        worker functions report they're done with it.

    save_state_when_done : bool
        If this is True, will save the current state of the work item queue and
        the work items acknowledged as completed to a pickle in the current
        working directory. Call the `runcp_producer_loop_savedstate` function
        below to resume processing from this saved state later.

    s3_client : boto3.Client or None
        If None, this function will instantiate a new `boto3.Client` object to
        use in its S3 download operations. Alternatively, pass in an existing
        `boto3.Client` instance to re-use it here.

    sqs_client : boto3.Client or None
        If None, this function will instantiate a new `boto3.Client` object to
        use in its SQS operations. Alternatively, pass in an existing
        `boto3.Client` instance to re-use it here.

    Returns
    -------

    dict or str
        Returns the current work state as a dict or str path to the generated
        work state pickle depending on if `save_state_when_done` is True.

    """

    if not sqs_client:
        sqs_client = boto3.client('sqs')
    if not s3_client:
        s3_client = boto3.client('s3')

    if isinstance(lightcurve_list, str) and os.path.exists(lightcurve_list):

        # get the LC list
        with open(lightcurve_list, 'r') as infd:
            lclist = infd.readlines()

        lclist = [x.replace('\n','') for x in lclist if len(x) > 0]

        if process_list_slice is not None:
            lclist = lclist[process_list_slice[0]:process_list_slice[1]]

        lclist = [x[1:] for x in lclist if x.startswith('/')]
        lclist = ['s3://%s/%s' % (input_bucket, x) for x in lclist]

    # this handles direct invocation using lists of s3:// urls of light curves
    elif isinstance(lightcurve_list, list):
        lclist = lightcurve_list

    # set up the input and output queues

    # check if the queues by the input and output names given exist already
    # if they do, go ahead and use them
    # if they don't, make new ones.
    try:
        inq = sqs_client.get_queue_url(QueueName=input_queue)
        inq_url = inq['QueueUrl']
        LOGINFO('input queue already exists, skipping creation...')
    except ClientError as e:
        inq = awsutils.sqs_create_queue(input_queue, client=sqs_client)
        inq_url = inq['url']

    try:
        outq = sqs_client.get_queue_url(QueueName=result_queue)
        outq_url = outq['QueueUrl']
        LOGINFO('result queue already exists, skipping creation...')
    except ClientError as e:
        outq = awsutils.sqs_create_queue(result_queue, client=sqs_client)
        outq_url = outq['url']

    LOGINFO('input queue: %s' % inq_url)
    LOGINFO('output queue: %s' % outq_url)

    # wait until queues are up
    LOGINFO('waiting for queues to become ready...')
    time.sleep(10.0)

    # for each item in the lightcurve_list, send it to the input queue and wait
    # until it's done to send another one

    if pfresult_list is None:
        pfresult_list = [None for x in lclist]

    for lc, pf in zip(lclist, pfresult_list):

        this_item = {
            'target': lc,
            'action': 'runcp',
            'args': (pf,),
            'kwargs':runcp_kwargs if runcp_kwargs is not None else {},
            'outbucket': result_bucket,
            'outqueue': outq_url
        }

        resp = awsutils.sqs_put_item(inq_url, this_item, client=sqs_client)
        if resp:
            LOGINFO('sent %s to queue: %s' % (lc,inq_url))

    # now block until all objects are done
    done_objects = {}

    LOGINFO('all items queued, waiting for results...')

    # listen to the kill and term signals and raise KeyboardInterrupt when
    # called
    signal.signal(signal.SIGINT, kill_handler)
    signal.signal(signal.SIGTERM, kill_handler)

    while len(list(done_objects.keys())) < len(lclist):

        try:

            result = awsutils.sqs_get_item(outq_url, client=sqs_client)

            if result is not None and len(result) > 0:

                recv = result[0]
                try:
                    processed_object = recv['item']['target']
                except KeyError:
                    LOGWARNING('unknown target in received item: %s' % recv)
                    processed_object = 'unknown-lc'

                cpf = recv['item']['cpf']
                receipt = recv['receipt_handle']

                if processed_object in lclist:

                    if processed_object not in done_objects:
                        done_objects[processed_object] = [cpf]
                    else:
                        done_objects[processed_object].append(cpf)

                    LOGINFO('done with %s -> %s' % (processed_object, cpf))

                    if download_when_done:

                        getobj = awsutils.awsutils.s3_get_url(
                            cpf,
                            client=s3_client
                        )
                        LOGINFO('downloaded %s -> %s' % (cpf, getobj))

                else:
                    LOGWARNING('processed object returned is not in '
                               'queued target list, probably from an '
                               'earlier run. accepting but not downloading.')

                awsutils.sqs_delete_item(outq_url, receipt)

        except KeyboardInterrupt as e:

            LOGWARNING('breaking out of producer wait-loop')
            break


    # delete the input and output queues when we're done
    LOGINFO('done with processing.')
    time.sleep(1.0)

    if purge_queues_when_done:
        LOGWARNING('purging queues at exit, please wait 10 seconds...')
        sqs_client.purge_queue(QueueUrl=inq_url)
        sqs_client.purge_queue(QueueUrl=outq_url)
        time.sleep(10.0)

    if delete_queues_when_done:
        LOGWARNING('deleting queues at exit')
        awsutils.sqs_delete_queue(inq_url)
        awsutils.sqs_delete_queue(outq_url)

    work_state = {
        'done': done_objects,
        'in_progress': list(set(lclist) - set(done_objects.keys())),
        'args':((os.path.abspath(lightcurve_list) if
                 isinstance(lightcurve_list, str) else lightcurve_list),
                input_queue,
                input_bucket,
                result_queue,
                result_bucket),
        'kwargs':{'pfresult_list':pfresult_list,
                  'runcp_kwargs':runcp_kwargs,
                  'process_list_slice':process_list_slice,
                  'download_when_done':download_when_done,
                  'purge_queues_when_done':purge_queues_when_done,
                  'save_state_when_done':save_state_when_done,
                  'delete_queues_when_done':delete_queues_when_done}
    }

    if save_state_when_done:
        with open('runcp-queue-producer-loop-state.pkl','wb') as outfd:
            pickle.dump(work_state, outfd, pickle.HIGHEST_PROTOCOL)

    # at the end, return the done_objects dict
    # also return the list of unprocessed items if any
    return work_state
def runcp_producer_loop(
        lightcurve_list,
        input_queue,
        input_bucket,
        result_queue,
        result_bucket,
        pfresult_list=None,
        runcp_kwargs=None,
        process_list_slice=None,
        purge_queues_when_done=False,
        delete_queues_when_done=False,
        download_when_done=True,
        save_state_when_done=True,
        s3_client=None,
        sqs_client=None
):
    """This sends checkplot making tasks to the input queue and monitors the
    result queue for task completion.

    Parameters
    ----------

    lightcurve_list : str or list of str
        This is either a string pointing to a file containing a list of light
        curves filenames to process or the list itself. The names must
        correspond to the full filenames of files stored on S3, including all
        prefixes, but not include the 's3://<bucket name>/' bit (these will be
        added automatically).

    input_queue : str
        This is the name of the SQS queue which will receive processing tasks
        generated by this function. The queue URL will automatically be obtained
        from AWS.

    input_bucket : str
        The name of the S3 bucket containing the light curve files to process.

    result_queue : str
        This is the name of the SQS queue that this function will listen to for
        messages from the workers as they complete processing on their input
        elements. This function will attempt to match input sent to the
        `input_queue` with results coming into the `result_queue` so it knows
        how many objects have been successfully processed. If this function
        receives task results that aren't in its own input queue, it will
        acknowledge them so they complete successfully, but not download them
        automatically. This handles leftover tasks completing from a previous
        run of this function.

    result_bucket : str
        The name of the S3 bucket which will receive the results from the
        workers.

    pfresult_list : list of str or None
        This is a list of periodfinder result pickle S3 URLs associated with
        each light curve. If provided, this will be used to add in phased light
        curve plots to each checkplot pickle. If this is None, the worker loop
        will produce checkplot pickles that only contain object information,
        neighbor information, and unphased light curves.

    runcp_kwargs : dict
        This is a dict used to pass any extra keyword arguments to the
        `lcproc.checkplotgen.runcp` function that will be run by the worker
        loop.

    process_list_slice : list
        This is used to index into the input light curve list so a subset of the
        full list can be processed in this specific run of this function.

        Use None for a slice index elem to emulate single slice spec behavior:

        process_list_slice = [10, None]  -> lightcurve_list[10:]
        process_list_slice = [None, 500] -> lightcurve_list[:500]

    purge_queues_when_done : bool
        If this is True, and this function exits (either when all done, or when
        it is interrupted with a Ctrl+C), all outstanding elements in the
        input/output queues that have not yet been acknowledged by workers or by
        this function will be purged. This effectively cancels all outstanding
        work.

    delete_queues_when_done : bool
        If this is True, and this function exits (either when all done, or when
        it is interrupted with a Ctrl+C'), all outstanding work items will be
        purged from the input/queues and the queues themselves will be deleted.

    download_when_done : bool
        If this is True, the generated checkplot pickle for each input work item
        will be downloaded immediately to the current working directory when the
        worker functions report they're done with it.

    save_state_when_done : bool
        If this is True, will save the current state of the work item queue and
        the work items acknowledged as completed to a pickle in the current
        working directory. Call the `runcp_producer_loop_savedstate` function
        below to resume processing from this saved state later.

    s3_client : boto3.Client or None
        If None, this function will instantiate a new `boto3.Client` object to
        use in its S3 download operations. Alternatively, pass in an existing
        `boto3.Client` instance to re-use it here.

    sqs_client : boto3.Client or None
        If None, this function will instantiate a new `boto3.Client` object to
        use in its SQS operations. Alternatively, pass in an existing
        `boto3.Client` instance to re-use it here.

    Returns
    -------

    dict or str
        Returns the current work state as a dict or str path to the generated
        work state pickle depending on if `save_state_when_done` is True.

    """

    if not sqs_client:
        sqs_client = boto3.client('sqs')
    if not s3_client:
        s3_client = boto3.client('s3')

    if isinstance(lightcurve_list, str) and os.path.exists(lightcurve_list):

        # get the LC list
        with open(lightcurve_list, 'r') as infd:
            lclist = infd.readlines()

        lclist = [x.replace('\n','') for x in lclist if len(x) > 0]

        if process_list_slice is not None:
            lclist = lclist[process_list_slice[0]:process_list_slice[1]]

        lclist = [x[1:] for x in lclist if x.startswith('/')]
        lclist = ['s3://%s/%s' % (input_bucket, x) for x in lclist]

    # this handles direct invocation using lists of s3:// urls of light curves
    elif isinstance(lightcurve_list, list):
        lclist = lightcurve_list

    # set up the input and output queues

    # check if the queues by the input and output names given exist already
    # if they do, go ahead and use them
    # if they don't, make new ones.
    try:
        inq = sqs_client.get_queue_url(QueueName=input_queue)
        inq_url = inq['QueueUrl']
        LOGINFO('input queue already exists, skipping creation...')
    except ClientError as e:
        inq = awsutils.sqs_create_queue(input_queue, client=sqs_client)
        inq_url = inq['url']

    try:
        outq = sqs_client.get_queue_url(QueueName=result_queue)
        outq_url = outq['QueueUrl']
        LOGINFO('result queue already exists, skipping creation...')
    except ClientError as e:
        outq = awsutils.sqs_create_queue(result_queue, client=sqs_client)
        outq_url = outq['url']

    LOGINFO('input queue: %s' % inq_url)
    LOGINFO('output queue: %s' % outq_url)

    # wait until queues are up
    LOGINFO('waiting for queues to become ready...')
    time.sleep(10.0)

    # for each item in the lightcurve_list, send it to the input queue and wait
    # until it's done to send another one

    if pfresult_list is None:
        pfresult_list = [None for x in lclist]

    for lc, pf in zip(lclist, pfresult_list):

        this_item = {
            'target': lc,
            'action': 'runcp',
            'args': (pf,),
            'kwargs':runcp_kwargs if runcp_kwargs is not None else {},
            'outbucket': result_bucket,
            'outqueue': outq_url
        }

        resp = awsutils.sqs_put_item(inq_url, this_item, client=sqs_client)
        if resp:
            LOGINFO('sent %s to queue: %s' % (lc,inq_url))

    # now block until all objects are done
    done_objects = {}

    LOGINFO('all items queued, waiting for results...')

    # listen to the kill and term signals and raise KeyboardInterrupt when
    # called
    signal.signal(signal.SIGINT, kill_handler)
    signal.signal(signal.SIGTERM, kill_handler)

    while len(list(done_objects.keys())) < len(lclist):

        try:

            result = awsutils.sqs_get_item(outq_url, client=sqs_client)

            if result is not None and len(result) > 0:

                recv = result[0]
                try:
                    processed_object = recv['item']['target']
                except KeyError:
                    LOGWARNING('unknown target in received item: %s' % recv)
                    processed_object = 'unknown-lc'

                cpf = recv['item']['cpf']
                receipt = recv['receipt_handle']

                if processed_object in lclist:

                    if processed_object not in done_objects:
                        done_objects[processed_object] = [cpf]
                    else:
                        done_objects[processed_object].append(cpf)

                    LOGINFO('done with %s -> %s' % (processed_object, cpf))

                    if download_when_done:

                        getobj = awsutils.awsutils.s3_get_url(
                            cpf,
                            client=s3_client
                        )
                        LOGINFO('downloaded %s -> %s' % (cpf, getobj))

                else:
                    LOGWARNING('processed object returned is not in '
                               'queued target list, probably from an '
                               'earlier run. accepting but not downloading.')

                awsutils.sqs_delete_item(outq_url, receipt)

        except KeyboardInterrupt as e:

            LOGWARNING('breaking out of producer wait-loop')
            break


    # delete the input and output queues when we're done
    LOGINFO('done with processing.')
    time.sleep(1.0)

    if purge_queues_when_done:
        LOGWARNING('purging queues at exit, please wait 10 seconds...')
        sqs_client.purge_queue(QueueUrl=inq_url)
        sqs_client.purge_queue(QueueUrl=outq_url)
        time.sleep(10.0)

    if delete_queues_when_done:
        LOGWARNING('deleting queues at exit')
        awsutils.sqs_delete_queue(inq_url)
        awsutils.sqs_delete_queue(outq_url)

    work_state = {
        'done': done_objects,
        'in_progress': list(set(lclist) - set(done_objects.keys())),
        'args':((os.path.abspath(lightcurve_list) if
                 isinstance(lightcurve_list, str) else lightcurve_list),
                input_queue,
                input_bucket,
                result_queue,
                result_bucket),
        'kwargs':{'pfresult_list':pfresult_list,
                  'runcp_kwargs':runcp_kwargs,
                  'process_list_slice':process_list_slice,
                  'download_when_done':download_when_done,
                  'purge_queues_when_done':purge_queues_when_done,
                  'save_state_when_done':save_state_when_done,
                  'delete_queues_when_done':delete_queues_when_done}
    }

    if save_state_when_done:
        with open('runcp-queue-producer-loop-state.pkl','wb') as outfd:
            pickle.dump(work_state, outfd, pickle.HIGHEST_PROTOCOL)

    # at the end, return the done_objects dict
    # also return the list of unprocessed items if any
    return work_state
This sends checkplot making tasks to the input queue and monitors the
    result queue for task completion.

    Parameters
    ----------

    lightcurve_list : str or list of str
        This is either a string pointing to a file containing a list of light
        curves filenames to process or the list itself. The names must
        correspond to the full filenames of files stored on S3, including all
        prefixes, but not include the 's3://<bucket name>/' bit (these will be
        added automatically).

    input_queue : str
        This is the name of the SQS queue which will receive processing tasks
        generated by this function. The queue URL will automatically be obtained
        from AWS.

    input_bucket : str
        The name of the S3 bucket containing the light curve files to process.

    result_queue : str
        This is the name of the SQS queue that this function will listen to for
        messages from the workers as they complete processing on their input
        elements. This function will attempt to match input sent to the
        `input_queue` with results coming into the `result_queue` so it knows
        how many objects have been successfully processed. If this function
        receives task results that aren't in its own input queue, it will
        acknowledge them so they complete successfully, but not download them
        automatically. This handles leftover tasks completing from a previous
        run of this function.

    result_bucket : str
        The name of the S3 bucket which will receive the results from the
        workers.

    pfresult_list : list of str or None
        This is a list of periodfinder result pickle S3 URLs associated with
        each light curve. If provided, this will be used to add in phased light
        curve plots to each checkplot pickle. If this is None, the worker loop
        will produce checkplot pickles that only contain object information,
        neighbor information, and unphased light curves.

    runcp_kwargs : dict
        This is a dict used to pass any extra keyword arguments to the
        `lcproc.checkplotgen.runcp` function that will be run by the worker
        loop.

    process_list_slice : list
        This is used to index into the input light curve list so a subset of the
        full list can be processed in this specific run of this function.

        Use None for a slice index elem to emulate single slice spec behavior:

        process_list_slice = [10, None]  -> lightcurve_list[10:]
        process_list_slice = [None, 500] -> lightcurve_list[:500]

    purge_queues_when_done : bool
        If this is True, and this function exits (either when all done, or when
        it is interrupted with a Ctrl+C), all outstanding elements in the
        input/output queues that have not yet been acknowledged by workers or by
        this function will be purged. This effectively cancels all outstanding
        work.

    delete_queues_when_done : bool
        If this is True, and this function exits (either when all done, or when
        it is interrupted with a Ctrl+C'), all outstanding work items will be
        purged from the input/queues and the queues themselves will be deleted.

    download_when_done : bool
        If this is True, the generated checkplot pickle for each input work item
        will be downloaded immediately to the current working directory when the
        worker functions report they're done with it.

    save_state_when_done : bool
        If this is True, will save the current state of the work item queue and
        the work items acknowledged as completed to a pickle in the current
        working directory. Call the `runcp_producer_loop_savedstate` function
        below to resume processing from this saved state later.

    s3_client : boto3.Client or None
        If None, this function will instantiate a new `boto3.Client` object to
        use in its S3 download operations. Alternatively, pass in an existing
        `boto3.Client` instance to re-use it here.

    sqs_client : boto3.Client or None
        If None, this function will instantiate a new `boto3.Client` object to
        use in its SQS operations. Alternatively, pass in an existing
        `boto3.Client` instance to re-use it here.

    Returns
    -------

    dict or str
        Returns the current work state as a dict or str path to the generated
        work state pickle depending on if `save_state_when_done` is True.