MRCall+SMS3.py

def key(time, base, adjustment):
    return (time+adjustment) - (time+adjustment)%base
    
def map_entries(index, entry):
    if entry.type == "call":
        yield (entry.number+entry.number_to+key(entry.time, 5*60, 0), entry) 
    elif entry.type == "SMS":
        yield (entry.number+entry.number_to+key(entry.time, 5*60, 0), entry)
        yield (entry.number+entry.number_to+key(entry.time, 5*60, -5*60), entry)
    else:
        yield (None, None)


def reduce_entries(key, entries):
    for i in range(0, len(entries)) :
        entry = entries[i]
        if entry.type == "call":
            sms = find_sms(entries,
                           entry.number_to,
                           i+1,
                           entry.time,
                           entry.time + 5*60)
            if sms != None:
                keyx = sms.number+sms.time
                yield (keyx, entry,sms)

def reduce_deduplicate(key, entries):
    for i in range(0, len(entries)) :
        entry = entries[i]
        if entry.type == "sms":
            sms = find_sms(entries,
                           entry.number_to,
                           i+1,
                           entry.time,
                           entry.time + 5*60)
            if sms != None:
                entries.remove(entry)
                entries.remove(sms)
                yield (entry,sms)
        

def find_sms(entries, number_to, start, time_start, time_end):
    for i in range(start, len(entries)) :
        entry = source[i]
        if (entry.type == "sms" and
            entry.number_to == number_to and
            entry.time > time_start and
            entry.time < time_end) :
            return entry
        elif entry.time > time_end:
            return None
        

# Run Map
for i in range(0, len(source)) :
    for key, entry in map_entries(i, source[i]):
        if key != None:
            sorted_map[key].append(entry)

# Sort the map(keys and values)
sorted_map.sort()

# Run Reduce
for k in sorted_map.keys():
    for keyx, call, sms in reduce_entries(k, sorted_map[k]):
        sorted_map2[keyx].append(call)
        sorted_map2[keyx].append(sms)

# Sort the map(keys and values)
sorted_map2.sort()

# Run Reduce
for k in sorted_map2.keys():
    for (call, sms) in reduce_deduplicate(k, sorted_map2[k]):
        found_items.append(entry+sms)