def _process(self, submission_id, on_done):
        log.info("Processing submission from queue_name: {0}, submission_id: {1}".format(self.queue_name, submission_id))
        try:
            submission = self._get_submission(submission_id)

            if submission is None:
                statsd.increment('xqueue.consumer.consumer_callback.submission_does_not_exist',
                                 tags=['queue:{0}'.format(self.queue_name)])
                log.error("Queued pointer refers to nonexistent entry in Submission DB: queue_name: {0}, submission_id: {1}".format(
                    self.queue_name,
                    submission_id
                ))

            # if item has been retired, skip grading
            if submission and not submission.retired:
                self._deliver_submission(submission)

            # close transaction
            transaction.commit()
        except Exception as e:
            # We need a wide catch here to correctly rollback the
            # transaction and acknowledge the message if something
            # goes wrong
            statsd.increment('xqueue.consumer.consumer_callback.unknown_error',
                             tags=['queue:{0}'.format(self.queue_name)])
            log.error("Error processing submission_id: {0} on queue_name: {1}, {2}" .format(
                submission_id,
                self.queue_name,
                e,
            ))
            transaction.rollback()
        finally:
            # acknowledge that the message was processed
            on_done()

def post_grade_to_lms(header, body):
    '''
    Send grading results back to LMS
        header:  JSON-serialized xqueue_header (string)
        body:    grader reply (string)

    Returns:
        success: Flag indicating successful exchange (Boolean)
    '''
    header_dict = json.loads(header)
    lms_callback_url = header_dict['lms_callback_url']

    payload = {'xqueue_header': header, 'xqueue_body': body}

    # Quick kludge retries to fix prod problem with 6.00x push graders. We're
    # seeing abrupt disconnects when servers are taken out of the ELB, causing
    # in flight lms_ack requests to fail. This just tries five times before
    # giving up.
    attempts = 0
    success = False
    while (not success) and attempts < 5:
        (success, lms_reply) = _http_post(lms_callback_url,
                                          payload,
                                          settings.REQUESTS_TIMEOUT)
        attempts += 1

    if success:
        statsd.increment('xqueue.consumer.post_grade_to_lms.success')
    else:
        log.error("Unable to return to LMS: lms_callback_url: {0}, payload: {1}, lms_reply: {2}".format(lms_callback_url, payload, lms_reply))
        statsd.increment('xqueue.consumer.post_grade_to_lms.failure')

    return success

def get_single_qitem(queue_name):
    '''
    Retrieve a single queued item, if one exists, from the named queue

    Returns (success, qitem):
        success: Flag whether retrieval is successful (Boolean)
                 If no items in the queue, then return False
        qitem:   Retrieved item
    '''
    queue_name = str(queue_name)

    # Pull a single submission (if one exists) from the named queue
    credentials = pika.PlainCredentials(settings.RABBITMQ_USER,
                                        settings.RABBITMQ_PASS)

    connection = pika.BlockingConnection(pika.ConnectionParameters(
        heartbeat_interval=5,
        credentials=credentials, host=settings.RABBIT_HOST,
        virtual_host=settings.RABBIT_VHOST))
    channel = connection.channel()
    channel.queue_declare(queue=queue_name, durable=True)

    # qitem is the item from the queue
    method, header, qitem = channel.basic_get(queue=queue_name)

    if method is None or method.NAME == 'Basic.GetEmpty':  # Got nothing
        connection.close()
        return (False, '')
    else:
        channel.basic_ack(method.delivery_tag)
        connection.close()
        statsd.increment('xqueue.consumer.get_single_qitem',
                         tags=['queue:{0}'.format(queue_name)])
        return (True, qitem)

    def purchased_callback(self):
        """
        When purchased, this should enroll the user in the course.  We are assuming that
        course settings for enrollment date are configured such that only if the (user.email, course_id) pair is found
        in CourseEnrollmentAllowed will the user be allowed to enroll.  Otherwise requiring payment
        would in fact be quite silly since there's a clear back door.
        """
        try:
            course_loc = CourseDescriptor.id_to_location(self.course_id)
            course_exists = modulestore().has_item(self.course_id, course_loc)
        except ValueError:
            raise PurchasedCallbackException(
                "The customer purchased Course {0}, but that course doesn't exist!".format(self.course_id))

        if not course_exists:
            raise PurchasedCallbackException(
                "The customer purchased Course {0}, but that course doesn't exist!".format(self.course_id))

        CourseEnrollment.enroll(user=self.user, course_id=self.course_id, mode=self.mode)

        log.info("Enrolled {0} in paid course {1}, paid ${2}".format(self.user.email, self.course_id, self.line_cost))
        org, course_num, run = self.course_id.split("/")
        statsd.increment("shoppingcart.PaidCourseRegistration.purchased_callback.enrollment",
                         tags=["org:{0}".format(org),
                               "course:{0}".format(course_num),
                               "run:{0}".format(run)])

def send_nearby_carts(lat_long, category="Anything"):
    if category not in TAGS_BY_TRUCK:
        return jsonify(data="Valid categories: %s" % str(TAGS_BY_TRUCK.keys()))
    lat_long = lat_long.split(",")
    latitude = float(lat_long[0])
    longitude = float(lat_long[1])
    unsorted_result = []
    unsorted_result_lat_long = []
    result_feet = []
    result_miles = []
    for index in find_nearby_carts(longitude, latitude, IDX):
        if index in TAGS_BY_TRUCK[category]:
            unsorted_result.append(CARTS[index])
            unsorted_result_lat_long.append(CARTS[index]["latitude"] + "," + CARTS[index]["longitude"])
    distances, addresses = get_distances_and_addresses(lat_long[0] + "," + lat_long[1], unsorted_result_lat_long)
    if len(distances) == len(unsorted_result):
        for i in range(len(unsorted_result)):
            unsorted_result[i]["distance"] = distances[i]
            unsorted_result[i]["address"] = addresses[i]
            if "ft" in unsorted_result[i]["distance"]:
                result_feet.append(unsorted_result[i])
            else:
                result_miles.append(unsorted_result[i])
        result_feet.sort(key=operator.itemgetter("distance"))
        result_miles.sort(key=operator.itemgetter("distance"))
        result = result_feet + result_miles
    else:
        result = unsorted_result
    statsd.increment("cart_api.requests", tags=["support", "page:nearby_carts"])
    return jsonify(data=result)

def get_service(service_id):
    """ Get a single service's data
    """

    result = g.firebase.get('/services', service_id)
    statsd.increment('firebase.services.get')
    return jsonify(dict(service=result))

def get_services():
    """ Get the collection of all services
    """
    
    result = g.firebase.get('/services', None)
    statsd.increment('firebase.services.get')
    return jsonify(dict(services=result))

    def publish(event):
        if event.get('event_name') != 'grade':
            return

        student_module, created = StudentModule.objects.get_or_create(
            course_id=course_id,
            student=user,
            module_type=descriptor.location.category,
            module_state_key=descriptor.location.url(),
            defaults={'state': '{}'},
        )
        student_module.grade = event.get('value')
        student_module.max_grade = event.get('max_value')
        student_module.save()

        # Bin score into range and increment stats
        score_bucket = get_score_bucket(student_module.grade, student_module.max_grade)
        org, course_num, run = course_id.split("/")

        tags = ["org:{0}".format(org),
                "course:{0}".format(course_num),
                "run:{0}".format(run),
                "score_bucket:{0}".format(score_bucket)]

        if grade_bucket_type is not None:
            tags.append('type:%s' % grade_bucket_type)

        statsd.increment("lms.courseware.question_answered", tags=tags)

def register():
    statsd.increment('api_calls.register')
    form = SignupForm(request.form)
    logger.info(request.form)
    if not form.validate():
        msg = {
            'success': False,
            'msg': form.errors}
        return jsonify(msg)
    user = session.query(Person).\
        filter(Person.email == form.email.data
               ).first()
    if user:
        msg = {
            'success': False,
            'msg': user.email + ' is already registered!',
            'parameter': 'email', }
        return jsonify(msg)
    u = Person(form)
    session.add(u)
    session.commit()
    try:
        pm = PostMonkey(apikey=MC_APIKEY, timeout=10)
        pm.listSubscribe(
            id=MC_LISTID, email_address=form.email.data)
    except MailChimpException, e:
        app.logger.error(str(e))

def score():
    statsd.increment('api_calls.score.%s' % request.method)
    if request.method == 'GET':
        row = session.query(Score).filter(
            Score.person_id == g.user._id)
        res = []
        for pt in row:
            res.append(pt.json_data())
        return jsonify({
            'msg': res,
            'success': False})

    '''post function'''
    logger.info(request.form)
    form = ScoreForm(request.form)
    if not form.validate():
        msg = {
            'success': False,
            'msg': form.errors}
        return jsonify(msg)
    session.add(Score(form, g.user._id))
    session.commit()
    msg = {
        'success': True,
        'msg': 'Added',
        'id': form.id.data, }
    return jsonify(msg)

def finalize_expired_submission(sub):
    """
    Expire submissions by posting back to LMS with error message.
    Input:
        timed_out_list from check_if_expired method
    Output:
        Success code.
    """

    grader_dict = {
        'score': 0,
        'feedback': error_template.format(errors="Error scoring submission."),
        'status': GraderStatus.failure,
        'grader_id': "0",
        'grader_type': sub.next_grader_type,
        'confidence': 1,
        'submission_id' : sub.id,
        }

    sub.state = SubmissionState.finished
    sub.save()

    grade = create_grader(grader_dict,sub)

    statsd.increment("open_ended_assessment.grading_controller.expire_submissions.finalize_expired_submission",
                     tags=[
                         "course:{0}".format(sub.course_id),
                         "location:{0}".format(sub.location),
                         'grader_type:{0}'.format(sub.next_grader_type)
                     ])

    return True

def reset_skipped_subs():
    """
    Reset submissions marked skipped to return them to the queue.
    """
    
    # Mihara: There's no reason not to do that which I can see.    
    counter = Submission.objects.filter(
        state=SubmissionState.skipped,
        posted_results_back_to_queue=False
        ).update(state=SubmissionState.waiting_to_be_graded)
    
    # Mihara: Seriously, why did they write it like that?
    #counter=0
    #unique_locations=[x['location'] for x in Submission.objects.all().values('location').distinct()]
    #for location in unique_locations:
    #    subs_pending_total= Submission.objects.filter(
    #        location=location,
    #        state=SubmissionState.skipped
    #    ).order_by('-date_created')
    #    for sub in subs_pending_total:
    #        sub.state=SubmissionState.waiting_to_be_graded
    #        counter+=1
    if counter>0:
        statsd.increment("open_ended_assessment.grading_controller.expire_submissions.reset_skipped_subs",
            tags=["counter:{0}".format(counter)])
        log.debug("Reset {0} submission from skipped state".format(counter))

def post_services(args):
    """
    Sample Request data:

    {
      "name": NAME,
      "data": {
        "terms": [
          {
            "policy_name" : POLICY_NAME,
            "policy_desc": POLICY_DESC,
            "policy_values": POLICY_OPTIONS
          },
        ],
        "full_terms_url": URL
      }
    }
    """
    
    data = request.get_json()
    
    try:
        result = g.firebase.put('/services',
                                args.get('name', None),
                                args.get('data', {}),
                                params={'print': 'silent'},
                                headers={'X_FANCY_HEADER': 'VERY FANCY'})
        statsd.increment('firebase.services.put')
    except:
        pass
    
    return jsonify({'success': True})

def reset_ml_subs_to_in():
    """
    Reset submissions marked ML to instructor if there are not enough instructor submissions to grade
    This happens if the instructor skips too many submissions
    """
    counter=0
    unique_locations=[x['location'] for x in list(Submission.objects.values('location').distinct())]
    for location in unique_locations:
        subs_graded, subs_pending = staff_grading_util.count_submissions_graded_and_pending_instructor(location)
        subs_pending_total= Submission.objects.filter(
            location=location,
            state=SubmissionState.waiting_to_be_graded,
            preferred_grader_type="ML"
        ).order_by('-date_created')[:settings.MIN_TO_USE_ML]
        if ((subs_graded+subs_pending) < settings.MIN_TO_USE_ML and subs_pending_total.count() > subs_pending):
            for sub in subs_pending_total:
                if sub.next_grader_type=="ML" and sub.get_unsuccessful_graders().count()==0:
                    staff_grading_util.set_ml_grading_item_back_to_instructor(sub)
                    counter+=1
                if (counter+subs_graded + subs_pending)> settings.MIN_TO_USE_ML:
                    break
    if counter>0:
        statsd.increment("open_ended_assessment.grading_controller.expire_submissions.reset_ml_subs_to_in",
            tags=["counter:{0}".format(counter)])
        log.debug("Reset {0} submission from ML to IN".format(counter))

    def publish(event):
        """A function that allows XModules to publish events. This only supports grade changes right now."""
        if event.get("event_name") != "grade":
            return

        # Construct the key for the module
        key = KeyValueStore.Key(
            scope=Scope.user_state, student_id=user.id, block_scope_id=descriptor.location, field_name="grade"
        )

        student_module = field_data_cache.find_or_create(key)
        # Update the grades
        student_module.grade = event.get("value")
        student_module.max_grade = event.get("max_value")
        # Save all changes to the underlying KeyValueStore
        student_module.save()

        # Bin score into range and increment stats
        score_bucket = get_score_bucket(student_module.grade, student_module.max_grade)
        org, course_num, run = course_id.split("/")

        tags = [
            "org:{0}".format(org),
            "course:{0}".format(course_num),
            "run:{0}".format(run),
            "score_bucket:{0}".format(score_bucket),
        ]

        if grade_bucket_type is not None:
            tags.append("type:%s" % grade_bucket_type)

        statsd.increment("lms.courseware.question_answered", tags=tags)

def check_if_grading_finished_for_duplicates():
    duplicate_submissions = Submission.objects.filter(
        preferred_grader_type = "PE",
        is_duplicate= True,
        posted_results_back_to_queue=False,
    )
    log.info(duplicate_submissions)
    counter=0
    for sub in duplicate_submissions:
        if sub.duplicate_submission_id is not None:
            try:
                original_sub=Submission.objects.get(id=sub.duplicate_submission_id)
                if original_sub.state == SubmissionState.finished:
                    finalize_grade_for_duplicate_peer_grader_submissions(sub, original_sub)
                    counter+=1
                    log.debug("Finalized one duplicate submission: Original: {0} Duplicate: {1}".format(original_sub,sub))
            except:
                log.error("Could not finalize grade for submission with id {0}".format(sub.duplicate_submission_id))

    statsd.increment("open_ended_assessment.grading_controller.expire_submissions.check_if_duplicate_grading_finished",
        tags=[
            "counter:{0}".format(counter),
        ])
    log.info("Finalized {0} duplicate submissions".format(counter))
    return True

def reset_timed_out_submissions(subs):
    """
    Check if submissions have timed out, and reset them to waiting to grade state if they have
    Input:
        subs - A QuerySet of submissions
    Output:
        status code indicating success
    """
    now = timezone.now()
    min_time = datetime.timedelta(seconds=settings.RESET_SUBMISSIONS_AFTER)
    timed_out_subs=subs.filter(date_modified__lt=now-min_time)
    timed_out_sub_count=timed_out_subs.count()
    count = 0

    for i in xrange(0, timed_out_sub_count):
        sub = subs[i]
        if sub.state == SubmissionState.being_graded:
            sub.state = SubmissionState.waiting_to_be_graded
            sub.save()
            count += 1

    if count>0:
        statsd.increment("open_ended_assessment.grading_controller.expire_submissions.reset_timed_out_submissions",
            tags=["counter:{0}".format(count)])
        log.debug("Reset {0} submissions that had timed out in their current grader.".format(count))

    return True

def reset_timed_out_submissions():
    """
    Check if submissions have timed out, and reset them to waiting to grade state if they have
    Input:
        subs - A QuerySet of submissions
    Output:
        status code indicating success
    """
    now = timezone.now()
    min_time = datetime.timedelta(seconds=settings.RESET_SUBMISSIONS_AFTER)

    # have to split into 2 queries now, because we are giving some finished submissions to peer graders when
    # there's nothing to grade

    reset_waiting_count = (Submission.objects
                           .filter(date_modified__lt=now-min_time,
                                   state=SubmissionState.being_graded,
                                   posted_results_back_to_queue=False)
                           .update(state=SubmissionState.waiting_to_be_graded))

    reset_finished_count = (Submission.objects
                            .filter(date_modified__lt=now-min_time,
                                    state=SubmissionState.being_graded,
                                    posted_results_back_to_queue=True)
                            .update(state=SubmissionState.finished))

    reset_count = reset_waiting_count + reset_finished_count
    if reset_count>0:
        statsd.increment("open_ended_assessment.grading_controller.expire_submissions.reset_timed_out_submissions",
            tags=["counter:{0}".format(reset_count)])
        log.debug("Reset {0} submissions that had timed out in their current grader.  {1}->W {2}->F"
                  .format(reset_count, reset_waiting_count, reset_finished_count))

    return True

		    def onChannelHangup(self, ev):
		        statsd.increment('freeswitch.channels.finished')
		        if (ev.Hangup_Cause in config.freeSwitch.normalHangupCauses):
		            statsd.increment('freeswitch.channels.finished.normally')
		            statsd.increment('freeswitch.channels.finished.normally.'+ev.Hangup_Cause.lower())
		        else:
		            statsd.increment('freeswitch.channels.finished.abnormally')
		            statsd.increment('freeswitch.channels.finished.abnormally.'+ev.Hangup_Cause.lower())

def showIndex():
    options = {
        "/location/<lat_long>": 'Pass comma separated latitude,longitude value to get info for nearby carts.Values returned are in the format:{"data": [list of carts containing address, applicant, distance, facilitytype, fooditems, latitude, and longitude]}',
        "/location/<lat_long>/<category>": "Returns all carts near comma separated latitude longitude matching a particular category. List of available categories can be found using the /categories option. Results formatted the same as /location/<lat_long>",
        "/categories": 'Returns categories in format "data"=[list of categories]',
    }
    statsd.increment("cart-api.requests", tags=["support", "page:options"])
    return jsonify(options)