def _cluster_to_usage_data(cluster, basic_summary=None, now=None):
    bcs = basic_summary or _cluster_to_basic_summary(cluster)

    if now is None:
        now = datetime.utcnow()

    if not bcs['created']:
        return []

    # Figure out billing rate per second for the job, given that
    # normalizedinstancehours is how much we're charged up until
    # the next full hour.
    full_hours = math.ceil(_to_secs(bcs['ran']) / 60.0 / 60.0)
    nih_per_sec = bcs['nih'] / (full_hours * 3600.0)

    # Don't actually count a step as billed for the full hour until
    # the cluster finishes. This means that our total "nih_billed"
    # will be less than normalizedinstancehours in the cluster, but it
    # also keeps stats stable for steps that have already finished.
    if bcs['end']:
        cluster_end_billing = bcs['created'] + timedelta(hours=full_hours)
    else:
        cluster_end_billing = now

    intervals = []

    # make a fake step for cluster startup and bootstrapping, so we don't
    # consider that wasted.
    intervals.append({
        'label': bcs['label'],
        'owner': bcs['owner'],
        'start': bcs['created'],
        'end': bcs['ready'] or bcs['end'] or now,
        'step_num': None,
    })

    for step in getattr(cluster, 'steps', ()):
        step_status = getattr(step, 'status', None)
        step_timeline = getattr(step_status, 'timeline', None)

        # we've reached the last step that's actually run
        if not hasattr(step_timeline, 'startdatetime'):
            break

        step_start = _to_datetime(step_timeline.startdatetime)

        step_end = _to_datetime(getattr(step_timeline, 'enddatetime', None))
        if step_end is None:
            # step started running and was cancelled. credit it for 0 usage
            if bcs['end']:
                step_end = step_start
            # step is still running
            else:
                step_end = now

        m = _STEP_NAME_RE.match(getattr(step, 'name', ''))
        if m:
            step_label = m.group(1)
            step_owner = m.group(2)
            step_num = int(m.group(6))
        else:
            step_label, step_owner, step_num = None, None, None

        intervals.append({
            'label': step_label,
            'owner': step_owner,
            'start': step_start,
            'end': step_end,
            'step_num': step_num,
        })

    # fill in end_billing
    for i in range(len(intervals) - 1):
        intervals[i]['end_billing'] = intervals[i + 1]['start']

    intervals[-1]['end_billing'] = cluster_end_billing

    # fill normalized usage information
    for interval in intervals:

        interval['nih_used'] = (
            nih_per_sec *
            _to_secs(interval['end'] - interval['start']))

        interval['date_to_nih_used'] = dict(
            (d, nih_per_sec * secs)
            for d, secs
            in _subdivide_interval_by_date(interval['start'],
                                           interval['end']).items())

        interval['hour_to_nih_used'] = dict(
            (d, nih_per_sec * secs)
            for d, secs
            in _subdivide_interval_by_hour(interval['start'],
                                           interval['end']).items())

        interval['nih_billed'] = (
            nih_per_sec *
            _to_secs(interval['end_billing'] - interval['start']))

        interval['date_to_nih_billed'] = dict(
            (d, nih_per_sec * secs)
            for d, secs
            in _subdivide_interval_by_date(interval['start'],
                                           interval['end_billing']).items())

        interval['hour_to_nih_billed'] = dict(
            (d, nih_per_sec * secs)
            for d, secs
            in _subdivide_interval_by_hour(interval['start'],
                                           interval['end_billing']).items())

        # time billed but not used
        interval['nih_bbnu'] = interval['nih_billed'] - interval['nih_used']

        interval['date_to_nih_bbnu'] = {}
        for d, nih_billed in interval['date_to_nih_billed'].items():
            nih_bbnu = nih_billed - interval['date_to_nih_used'].get(d, 0.0)
            if nih_bbnu:
                interval['date_to_nih_bbnu'][d] = nih_bbnu

        interval['hour_to_nih_bbnu'] = {}
        for d, nih_billed in interval['hour_to_nih_billed'].items():
            nih_bbnu = nih_billed - interval['hour_to_nih_used'].get(d, 0.0)
            if nih_bbnu:
                interval['hour_to_nih_bbnu'][d] = nih_bbnu

    return intervals
Break billing/usage information for a cluster down by job.
dataset/ETHPy150Open Yelp/mrjob/mrjob/tools/emr/audit_usage.py/_cluster_to_usage_data