def lambda_handler()

in broadcast-monitoring/src/sports_detect/app/main.py [0:0]
30 lines of code
5 McCabe index (conditional complexity)

def lambda_handler(event, context):
    """

    :param event: e.g.
    {
      "parsed": {...
      },
      "config": {
        "audio_check_enabled": true,
        "station_logo_check_enabled": true,
        "language_detect_check_enabled": false,
        "team_detect_check_enabled": true,
        "sports_detect_check_enabled": true
      },
      "frame": {
        ...
        "S3_Bucket": "aws-rnd-broadcast-maas-video-processing-dev",
        "S3_Key": "frames/test_video_single_pipeline/test_1/original/2020/02/10/23/42:43:646000.jpg",
        "Resized_S3_Key": "frames/test_video_single_pipeline/test_1/resized/2020/02/10/23/42:43:646000.jpg"
      }
    }
    :param context:
    :return:
    """

    frame_info = event['frame']
    bucket = frame_info['S3_Bucket']
    key = frame_info['S3_Key']
    min_confidence = int(os.getenv('SPORTS_MIN_CONFIDENCE', 60))
    model_arn = os.getenv('SPORTS_MODEL_ARN')

    logger.info('Sports Detection for image: %s', os.path.join(bucket, key))

    img_data = {'S3Object': {'Bucket': bucket, 'Name': key}}

    with DDBUpdateBuilder(key={'Stream_ID': frame_info['Stream_ID'], 'DateTime': frame_info['DateTime']},
                          table_name=DDB_FRAME_TABLE) as update_builder:
        try:
            response = rekognition.detect_custom_labels(
                Image=img_data, MinConfidence=min_confidence, ProjectVersionArn=model_arn
            )
        except ClientError as e:
            logger.error('Error calling detect)sports: %s', e)
            update_builder.update_attr('Sports_Detect_Error', e.response['Error']['Code'])
            raise e
        else:
            result = response.get('CustomLabels', [])

            if not result:
                logger.info('No sports detected')
            else:
                res_out = [f'{r["Name"]}: {r["Confidence"]}' for r in result]
                logger.info('Sports detected: %s', json.dumps(res_out, indent=4))

            # extract expected program
            expected_program = event['parsed']['expectedProgram']
            for name, value in SportsCheck().execute(expected_program, result):
                logger.info("Writing to %s [%s]: %s", DDB_FRAME_TABLE, name, value)
                update_builder.update_attr(name, value, convert_to_ddb)

        return result