def detect_deadspace()

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

• 35 lines of code
• 11 McCabe index (conditional complexity)

def detect_deadspace(target_dict, deadspace_limit, start_time_limit, end_time_limit, label=""):
    """
    Detects deadspace in a dictionary of User Ale logs.  Detected instances of deadspace are captured in Segment
    objects based on the start and end time limits indicated by the function parameters.

    :param target_dict: A dictionary of User Ale logs assumed to be ordered by clientTime (Date/Time Objects or integers).
    :param deadspace_limit: An integer representing the amount of time (in seconds) considered to be 'deadspace'.
    :param start_time_limit: Amount of time (in seconds) prior to a detected deadspace event that should be included in the deadspace segment.
    :param end_time_limit: Amount of time (in seconds) to keep the segment window open after a detected deadspace event.
    :param label: An optional string argument that provides a prefix for the returned dictionary keys.

    :return: A Segments object containing newly created Segment objects.
    """

    # Iterate through the target dictionary using key list
    start_end_vals = []
    segment_names = []
    key_list = list(target_dict.keys())
    index = 0
    for i in range(len(key_list)):
        # Check for deadspace
        if i < len(key_list) - 1:
            curr_time = target_dict[key_list[i]]['clientTime']
            next_time = target_dict[key_list[i + 1]]['clientTime']
            if isinstance(curr_time, int) and isinstance(next_time, int):
                time_delta = next_time - curr_time
                if time_delta > deadspace_limit * 1000:
                    # Deadspace detected
                    start_time = curr_time - (start_time_limit * 1000)
                    end_time = next_time + (end_time_limit * 1000)
                    start_end_tuple = (start_time, end_time)
                    start_end_vals.append(start_end_tuple)
                    segment_names.append(label + str(index))
                    index += 1
            elif isinstance(curr_time, datetime.datetime) and isinstance(next_time, datetime.datetime):
                time_delta = next_time - curr_time
                if time_delta > datetime.timedelta(seconds=deadspace_limit):
                    # Deadspace detected
                    start_time = curr_time - datetime.timedelta(seconds=start_time_limit)
                    end_time = next_time + datetime.timedelta(seconds=end_time_limit)
                    start_end_tuple = (start_time, end_time)
                    start_end_vals.append(start_end_tuple)
                    segment_names.append(label + str(index))
                    index += 1
            else:
                raise TypeError('clientTime field is not consistently represented as an integer or datetime object')

    # Create segment dictionary with create_segment
    segments = create_segment(target_dict, segment_names, start_end_vals)
    for segment in segments:
        segment.segment_type = Segment_Type.DEADSPACE
        segment.generate_field_name = None
        segment.generate_matched_values = None

    return segments