def get_segment_state_for_optimization()

in source/dataplaneapi/runtime/app.py [0:0]

• 73 lines of code
• 17 McCabe index (conditional complexity)

def get_segment_state_for_optimization():
    """
    Retrieve one or more non-optimized segments identified in the current/prior chunks and all the dependent detectors output 
    around the segments for optimization. Use "SearchWindowSeconds" to control the time window around the segments within which 
    the dependent detectors output are queried for.

    Body:

    .. code-block:: python

        {
            "Program": string,
            "Event": string,
            "ChunkNumber": integer,
            "Classifier": string,
            "Detectors": list,
            "AudioTrack": integer,
            "SearchWindowSeconds": integer
        }

    Returns:

        List containing one or more non-optimized segments identified in the current/prior chunks along with all the dependent 
        detectors output around the segments
    
    Raises:
        400 - BadRequestError
        500 - ChaliceViewError
    """
    try:
        request = json.loads(app.current_request.raw_body.decode())

        validate(instance=request, schema=API_SCHEMA["get_segment_state_for_optimization"])

        print("Got a valid schema")

        program = request["Program"]
        event = request["Event"]
        chunk_number = request["ChunkNumber"]
        classifier = request["Classifier"]
        detectors = request["Detectors"] if "Detectors" in request else []
        audio_track = str(request["AudioTrack"]) if "AudioTrack" in request else None
        opto_audio_track = audio_track if audio_track is not None else "1"
        search_win_sec = request["SearchWindowSeconds"]

        output = []

        plugin_result_table = ddb_resource.Table(PLUGIN_RESULT_TABLE_NAME)

        print(
            f"Getting all the non-optimized segments identified in the current/prior chunks for program '{program}', event '{event}', classifier '{classifier}' and chunk number '{chunk_number}'")

        response = plugin_result_table.query(
            KeyConditionExpression=Key("PK").eq(f"{program}#{event}#{classifier}"),
            FilterExpression=Attr("ChunkNumber").lte(chunk_number) & (
                    Attr("OptoStart").not_exists() | Attr(f"OptoStart.{opto_audio_track}").not_exists() | Attr(
                "OptoEnd").not_exists() | Attr(f"OptoEnd.{opto_audio_track}").not_exists()),
            ConsistentRead=True
        )

        if "Items" not in response or len(response["Items"]) < 1:
            print(
                f"No non-optimized segment was identified in the current/prior chunks for program '{program}', event '{event}', classifier '{classifier}' and chunk number '{chunk_number}'")

        else:
            print(
                f"Got one or more non-optimized segments identified in the current/prior chunks for program '{program}', event '{event}', classifier '{classifier}' and chunk number '{chunk_number}'")

            segments = response["Items"]

            while "LastEvaluatedKey" in response:
                response = plugin_result_table.query(
                    ExclusiveStartKey=response["LastEvaluatedKey"],
                    KeyConditionExpression=Key("PK").eq(f"{program}#{event}#{classifier}"),
                    FilterExpression=Attr("ChunkNumber").lte(chunk_number) & (Attr("OptoStart").not_exists() | Attr(
                        f"OptoStart.{opto_audio_track}").not_exists() | Attr("OptoEnd").not_exists() | Attr(
                        f"OptoEnd.{opto_audio_track}").not_exists()),
                    ConsistentRead=True
                )

                segments.extend(response["Items"])

            for segment in segments:
                segment_start = segment["Start"]
                segment_end = segment["End"] if "End" in segment else None

                print(
                    f"Getting all the dependent detectors output around segment Start '{segment_start}' within a search window of '{search_win_sec}' seconds")

                # Get dependent detectors output for optimizing the segment Start
                detectors_output = get_detectors_output_for_segment(program, event, detectors, search_win_sec,
                                                                    audio_track, start=segment_start)

                # Get dependent detectors output for optimizing the segment End if segment End is present
                if segment_end is not None and segment_start != segment_end:
                    print(
                        f"Getting all the dependent detectors output around segment End '{segment_end}' within a search window of '{search_win_sec}' seconds")

                    detectors_output.extend(
                        get_detectors_output_for_segment(program, event, detectors, search_win_sec, audio_track,
                                                         end=segment_end))

                output.append(
                    {
                        "Segment": segment,
                        "DependentDetectorsOutput": detectors_output
                    }
                )

    except BadRequestError as e:
        print(f"Got chalice BadRequestError: {str(e)}")
        raise

    except ValidationError as e:
        print(f"Got jsonschema ValidationError: {str(e)}")
        raise BadRequestError(e.message)

    except Exception as e:
        print(
            f"Unable to get the non-optimized segments and dependent detectors output for program '{program}', event '{event}' and chunk number '{chunk_number}': {str(e)}")
        raise ChaliceViewError(
            f"Unable to get the non-optimized segments and dependent detectors output for program '{program}', event '{event}' and chunk number '{chunk_number}': {str(e)}")

    else:
        return replace_decimals(output)