EvaluationOAD.py

import os
from typing import List, Tuple, Any, Dict, Type

import numpy as np

import Metrics
from Metrics.BoundsBasedMetrics.LatencyAwareScore import LatencyAwareScore
from Metrics.ExistingResult import ExistingResult, CURRENT_APPROACH_NAME

from Classes.Label import Label
from Classes.Sequence import Sequence
from Graphics.ExportVisualResult import drawSequence
from Metrics.BoundsBasedMetrics.ActionBasedScore import ActionBasedScore
from Metrics.BoundsBasedMetrics.BoundedOnlineDetection import BoundedOnlineDetectionOverlap
from Metrics.BoundsBasedMetrics.BoundsBasedMetric import BoundsBasedMetric
from Metrics.BoundsBasedMetrics.BoundedOfflineOverlap import BoundedOfflineOverlap
from Metrics.BoundsBasedMetrics.DetectionToActionPoint import DetectionToActionPoint
from Metrics.FrameBasedMetrics.DetectionToActionPoint_simplified import DetectionToActionPoint_simplified
from Metrics.FrameBasedMetrics.FrameBasedMetric import FrameBasedMetric
from Metrics.FrameBasedMetrics.PerFrame_All import PerFrame_All
from Metrics.FrameBasedMetrics.SmoothRatioPrediction_Liu import SmoothRatioObservation_Liu

alphas_BoundedOverlap = [0.0, 0.2, 0.4, 0.6, 0.8, 0.95] # for BoffD and BOD
tolerance_T = 4  # 4 frames, for Action based F1 (Bloom G3D)

# Latency aware
delta_frames_latency = 10
useStartFrameInsteadOfNegativeDelta = True

exportPerSequenceDetailResult = False # for all metric
canCorrect = False  # for BOD
summary = True  # to recap some results, specific to some metrics and score
CURRENT_APPROACH_NAME.NAME = "OURS" #the name that will appear in generated curves for your approach
boundsPredictionFolder = "Bounds/" # input folder for bounds prediction
framesPredictionFolder = "Frames/" # input folder for frame prediction (can be ignored if you only eval using bounds metrics)
outputResultFolder = "Results/" # output folder for results
outputMFResultFolder = "ResultsMultiFold/" # output folder for results of multi folds (aggregration of results of each fold)


def defineMetricsBounded(nbClass):
    """
    Define the metrics to be used for the evaluation of the bounds prediction
    :param nbClass: the number of classes
    :return:
    """
    metricsBounded = []
    metricsBounded += [BoundedOfflineOverlap(a, nbClass, exportPerSequenceDetailResult) for a in
                       alphas_BoundedOverlap]
    metricsBounded += [BoundedOnlineDetectionOverlap(a, nbClass, canCorrect, exportPerSequenceDetailResult) for a in
                       alphas_BoundedOverlap]
    metricsBounded += [BoundedOnlineDetectionOverlap(a, nbClass, True, exportPerSequenceDetailResult) for a in
                       alphas_BoundedOverlap]
    metricsBounded += [ActionBasedScore(tolerance_T, nbClass, exportPerSequenceDetailResult)]
    metricsBounded += [DetectionToActionPoint(nbClass, exportPerSequenceDetailResult)]
    metricsBounded += [LatencyAwareScore(nbClass, exportPerSequenceDetailResult, delta_frames_latency,
                                         useStartFrameInsteadOfNegativeDelta)]
    return metricsBounded


def defineMetricsPerFrame(nbClass):
    """
    Define the metrics to be used for the evaluation of the frame prediction
    :param nbClass:  the number of classes
    :return:
    """
    metricsBounded = []
    metricsBounded += [PerFrame_All(nbClass, exportPerSequenceDetailResult)]
    metricsBounded += [DetectionToActionPoint_simplified(nbClass, exportPerSequenceDetailResult)]
    metricsBounded += [SmoothRatioObservation_Liu(nbClass, exportPerSequenceDetailResult)]
    return metricsBounded


def EvaluationGivenOneProtocoleFile(pathToEval,pathLabel,nbClass,doExportVisual,pathExistingResults):
    """
    Evaluate the results of one protocole file
    :param pathToEval: the path to the folder containing the bounds and/or frame prediction
    :param pathLabel: the path to the folder containing the ground truth (classic format see https://www-intuidoc.irisa.fr/oad-datasets/)
    :param nbClass: the number of classes
    :param doExportVisual: if True, export the visual result of the evaluation -one image per sequence, 3 lines per image: ground truth, bounds pred, and frame pred (Along time)
    :param pathExistingResults: the path to the file containing the existing results (if any)
    :return:
    """

    if pathExistingResults is not None:
        print("Existing results will be loaded from: ", pathExistingResults)
        try:
            with open(pathExistingResults, "r") as f:
                lines = f.readlines()
                try:
                    ExistingResult.existingResults = eval("\n".join(lines))
                    print("set existing results", ExistingResult.existingResults)
                except SyntaxError as e:
                    print(
                        "Syntaxe error while reading existing results file. Please check the file, should be python syntax.",
                        "File not used.",
                        e.msg)
                    ExistingResult.existingResults = None
        except FileNotFoundError as e:
            print("Existing results file not found. File not used.", e.strerror)
            ExistingResult.existingResults = None

    metricsBounded = defineMetricsBounded(nbClass)
    metricsFramebased = defineMetricsPerFrame(nbClass)

    files = os.listdir(pathToEval)
    assert len(files) != 0, "No files/folder in the pathEval. Need Bounds/Frames folders, or subfolders for multi folds"

    if doExportVisual:
        ExportVisual(pathToEval, pathLabel, nbClass)

    # if doBoundsEval:
    EvaluateBoundedPrediction(pathToEval, pathLabel, metricsBounded)
    # if doPerFrameEval:
    EvaluatePerFramePrediction(pathToEval, pathLabel, metricsFramebased)


def ExportVisual(pathToEval, pathLabel, nbClass):
    labelsPredPerFrame = None
    filesInFold = os.listdir(pathToEval)
    # filter to avoid evntual Results folder
    filesInFold = [f for f in filesInFold if not f.startswith(outputMFResultFolder[:-1]) and not f.startswith(outputResultFolder[:-1])]
    foldsPredFolders = [""]
    if boundsPredictionFolder[:-1] not in filesInFold:  # multi folds
        print("Multi folds detected")
        foldsPredFolders = [f + "/" for f in filesInFold]

    for ifoldFolder in range(len(foldsPredFolders)):
        pathBoundForThisFold = pathToEval + foldsPredFolders[ifoldFolder] + "/" + boundsPredictionFolder
        filesInFold = os.listdir(pathBoundForThisFold)
        if os.path.isdir(pathBoundForThisFold + filesInFold[0]):
            raise Exception("Sub Folders founds in pathBoundsPrediction, No sub-fold allowed",
                            pathBoundForThisFold + filesInFold[0], "is a directory")
        pathFramesPrediction = pathToEval + foldsPredFolders[ifoldFolder] + "/" + framesPredictionFolder
        for file in filesInFold:
            labelsGT = Label.LabelsFromFile(pathLabel + file)
            labelsPred = Label.LabelsFromFile(pathBoundForThisFold + file)
            if os.path.exists(pathFramesPrediction + file):
                labelsPredPerFrame = ReadPerFramePrediction(pathFramesPrediction + file)
            drawSequence(pathToEval+foldsPredFolders[ifoldFolder] + outputResultFolder + "Visual/", file, labelsGT, labelsPred,
                         labelsPredPerFrame, nbClass)


def EvaluateBoundedPrediction(pathToEval, pathLabels: str, metricsBounded: List[BoundsBasedMetric]):
    print("Bounded Evaluations")
    folderBoundBasedEval = "BoundBasedEval/"
    filesInFold = os.listdir(pathToEval)
    # filter to avoid evntual Results folder
    filesInFold = [f for f in filesInFold if not f.startswith(outputMFResultFolder[:-1]) and not f.startswith(outputResultFolder[:-1])]

    multiFold = False
    foldsPredFolders = [pathToEval+boundsPredictionFolder]
    pathOutputResults = [pathToEval+outputResultFolder + folderBoundBasedEval]
    if boundsPredictionFolder[:-1] not in filesInFold:
        print("\t Mutli-Fold evaluation")
        multiFold = True
        foldsPredFolders = [pathToEval+f + "/" + boundsPredictionFolder for f in filesInFold]
        pathOutputResults = [pathToEval+f + "/" + outputResultFolder + folderBoundBasedEval for f in filesInFold]

    fold: List[Tuple[List[Sequence], List[List[Label]]]] = []
    for ifoldFolder in range(len(foldsPredFolders)):
        filesFold = os.listdir(foldsPredFolders[ifoldFolder])
        if os.path.isdir(foldsPredFolders[ifoldFolder] + filesFold[0]):
            raise Exception("Sub Folders founds in bounds folder, No sub-fold allowed",
                            foldsPredFolders[ifoldFolder] + filesFold[0], "is a directory")

        sequences: List[Sequence] = []
        preds: List[List[Label]] = []
        for file in filesFold:
            # print("Evaluating file: ", file)
            labelsGT = Label.LabelsFromFile(pathLabels + file)
            sequence = Sequence(file, labelsGT)
            labelsPred = Label.LabelsFromFile(foldsPredFolders[ifoldFolder] + file)
            sequences.append(sequence)
            preds.append(labelsPred)
        fold.append((sequences, preds))

    """
    Dict[type(Metric), List[Metric]]
    """
    metricPerType: List[Dict] = [{} for _ in range(len(fold))]
    metricB: BoundsBasedMetric
    for metricB in metricsBounded:
        outputResultMetric: List = []
        for iFold, (sequences, predictions) in enumerate(fold):
            if type(metricB) not in metricPerType[iFold]:
                metricPerType[iFold][type(metricB)] = []
            metricPerType[iFold][type(metricB)].append(metricB)
            result = metricB.Evaluate(sequences, predictions, pathOutputResults[iFold])
            outputResultMetric.append(result)
        if multiFold:
            pathMFB = pathToEval + outputMFResultFolder + folderBoundBasedEval
            if not os.path.exists(pathMFB):
                os.makedirs(pathMFB)
            metricB.AggregateMultiFold(outputResultMetric, pathMFB, filesInFold)

    if summary:
        for iFold in range(len(fold)):
            for metricType, metrics in metricPerType[iFold].items():
                metricType.Summary(metrics, pathOutputResults[iFold])
    print("DONE - Bound Based Evaluations")


def ReadPerFramePrediction(file) -> List[int]:
    f = open(file, "r")
    line = f.readlines()[0]
    f.close()
    return list(map(int, line.split(" ")))


def EvaluatePerFramePrediction(pathToEval,pathLabels,  metricsFramebased):
    print("Evaluation of frames Prediction")

    print("Per frame Evaluations")
    folderFrameBasedEval = "FrameBasedEval/"
    filesInFold = os.listdir(pathToEval)
    # filter to avoid evntual Results folder
    filesInFold = [f for f in filesInFold if
                   not f.startswith(outputMFResultFolder[:-1]) and not f.startswith(outputResultFolder[:-1])]

    multiFold = False
    foldsPredFolders = [pathToEval + framesPredictionFolder]
    pathOutputResults = [pathToEval + outputResultFolder + folderFrameBasedEval]
    if framesPredictionFolder[:-1] not in filesInFold:
        print("\t Mutli-Fold evaluation")
        multiFold = True
        foldsPredFolders = [pathToEval + f + "/" + framesPredictionFolder for f in filesInFold]
        pathOutputResults = [pathToEval + f + "/" + outputResultFolder + folderFrameBasedEval for f in filesInFold]

    fold: List[Tuple[List[Sequence], List[List[int]]]] = []
    for ifoldFolder in range(len(foldsPredFolders)):
        filesFold = os.listdir(foldsPredFolders[ifoldFolder])
        if os.path.isdir(foldsPredFolders[ifoldFolder] + filesFold[0]):
            raise Exception("Sub Folders founds in path frames Prediction, No sub-fold allowed")

        sequences: List[Sequence] = []
        preds: List[List[int]] = []
        for file in filesFold:
            # print("Evaluating file: ", file)
            labelsGT = Label.LabelsFromFile(pathLabels + file)
            sequence = Sequence(file, labelsGT)
            if foldsPredFolders[ifoldFolder] is not None:
                perFramePred = ReadPerFramePrediction(foldsPredFolders[ifoldFolder] + file)
                preds.append(perFramePred)
            sequences.append(sequence)
        fold.append((sequences, preds))

    metricPerType: List[Dict] = [{} for _ in range(len(fold))]
    metricB: FrameBasedMetric
    for metricB in metricsFramebased:
        outputResultMetric: List = []
        for iFold, (sequences, predictions) in enumerate(fold):
            if type(metricB) not in metricPerType[iFold]:
                metricPerType[iFold][type(metricB)] = []
            metricPerType[iFold][type(metricB)].append(metricB)
            result = metricB.Evaluate(sequences, predictions, pathOutputResults[iFold])
            outputResultMetric.append(result)
        if multiFold:
            pathMFB = pathToEval + outputMFResultFolder + folderFrameBasedEval
            if not os.path.exists(pathMFB):
                os.makedirs(pathMFB)
            metricB.AggregateMultiFold(outputResultMetric, pathMFB, filesInFold)

    if summary:
        for iFold in range(len(fold)):
            for metricType, metrics in metricPerType[iFold].items():
                metricType.Summary(metrics, pathOutputResults[iFold])
    print("DONE - Per frame Evaluations")


def readProtocole(pathProtocole,doForSubFolder=False):
    """

    :param pathProtocole: path to the protocole file, contains:
        - pathToEval: path to the folder containing the predictions
        - pathLabel: path to the folder containing the labels
        - nbClass: number of classes
        - doExportVisual: boolean, if true export visual results
        - pathExistingResults: path to the folder containing the existing results (can be None)
    :param doForSubFolder: if true, the evaluation is done for each subfolder of pathToEval
    :return:
    """
    f = open(pathProtocole, "r")
    lines = f.readlines()
    f.close()
    args = eval("".join(lines))
    # dbName = args["dbName"]
    pathToEval = args["pathToEval"]
    pathLabel = args["pathLabel"]
    nbClass = args["nbClass"]
    doExportVisual = args["doExportVisual"]
    pathExistingResults = args["pathExistingResults"]

    if doForSubFolder:
        subfolds = os.listdir(pathToEval)
        for subfold in subfolds:
            if os.path.isdir(pathToEval+subfold):
                EvaluationGivenOneProtocoleFile(pathToEval+subfold+"/",pathLabel,nbClass,doExportVisual,pathExistingResults)
    else:
        EvaluationGivenOneProtocoleFile(pathToEval,pathLabel,nbClass,doExportVisual,pathExistingResults)



# pathProtocole = "C:\workspace2\Datasets\OAD\log\protocole.txt"
# pathProtocole = "C:\workspace2\Datasets\OAD\log\protocoleVariants.txt"
# pathProtocole = "C:\workspace2\Datasets\G3D\protocole.txt"
# pathProtocole = "C:\workspace2\Datasets\MAD\protocole.txt"
# pathProtocole = "C:\workspace2\Datasets\G3D\protocoleOLT.txt"
# pathProtocole = "C:\workspace2\Datasets\MSRC12\protocoleOLT.txt"
# pathProtocole = "C:\workspace2\Datasets\MSRC12\protocoleJCRNN.txt"
# pathProtocole = "C:\workspace2\Datasets\OAD\log\protocoleOLT.txt"
# pathProtocole = "C:\workspace2\Datasets\OAD\log\protocoleSSNet.txt"
# pathProtocole = "C:\workspace2\Datasets\G3D\protocoleSSNet.txt"
# pathProtocole = "C:\workspace2\Datasets\MSRC12\protocoleSSNet.txt"
# pathProtocole = "C:\workspace2\Datasets\Chalearn\protocoleSSNet.txt"
pathProtocole = "C:\workspace2\Datasets\Chalearn\protocol.txt"
# pathProtocole = "C:\workspace2\Datasets\Chalearn\protocole.txt"
# pathProtocole = "C:\workspace2\Datasets\Chalearn\protocoleVariants.txt"

readProtocole(pathProtocole,True)

# structure of folders required :
# - pathToEval (expOut)
# - - setOfEval1 [optional level, if not set, put 2nd arg to false: readProtocol(path,False)]
# - - - model1
# - - - - bounds
# - - - - frames
# - - - - Results [output of the evaluation, result of this script execution]
# - - - model2
# - - - ...
# - - ResultsMultiFold [output of the evaluation, result of this script execution, aggregated result of all models]

# - - setOfEval2
# - - - ....
#

#OUTPUT in expOut (fil