def generate_collapsing_window_segments()

in distill/segmentation/segment.py [0:0]

• 33 lines of code
• 12 McCabe index (conditional complexity)

def generate_collapsing_window_segments(target_dict, field_name, field_values_of_interest, label=""):
    """
    Generates segments based on a window to time in which the given field name has a value matching one of the values
    indicated by the field_values_of_interest list.

    :param target_dict: A dictionary of User Ale logs assumed to be ordered by clientTime (Date/Time Objects or integers).
    :param field_name: A string indicating the field name meant to be matched by the field values.
    :param field_values_of_interest: A list of field values to be matched in order to start/end a segment.
    :param label: An optional string argument that provides a prefix for the returned dictionary keys.
    """
    key_list = list(target_dict.keys())

    start_end_val_lists = []
    start_end_vals = []
    segment_names = []
    index = 0
    segment_started = False

    for i in range(len(key_list)):
        if field_name in target_dict[key_list[i]] and any(item in target_dict[key_list[i]][field_name] for item in field_values_of_interest):
            if not segment_started:
                # Start a new Segment
                start_end = [target_dict[key_list[i]]['clientTime'], None]
                start_end_val_lists.append(start_end)
                segment_names.append(label + str(index))
                segment_started = True

            # End Segment if end of dictionary
            elif segment_started and i == (len(key_list) - 1):
                # End the Segment
                start_end_tuple = (start_end_val_lists[index][0], target_dict[key_list[i]]['clientTime'])
                start_end_vals.append(start_end_tuple)
                index += 1
                segment_started = False
        else:
            if segment_started:
                # End the Segment
                start_end_tuple = (start_end_val_lists[index][0], target_dict[key_list[i - 1]]['clientTime'])
                start_end_vals.append(start_end_tuple)
                index += 1
                segment_started = False

    if 0 < len(start_end_vals) < len(segment_names):
        start_end_vals.append((start_end_val_lists[index][0], target_dict[key_list[len(key_list) - 1]]['clientTime']))

    # Create Segments object with create_segment
    segments = create_segment(target_dict, segment_names, start_end_vals)
    for segment in segments:
        segment.segment_type = Segment_Type.GENERATE_COLLAPSING_WINDOWS
        segment.generate_field_name = field_name
        segment.generate_matched_values = field_values_of_interest

    return segments