Target Quality to use implemented percentile for weighted vmaf reading

TargetQuality/per_shot.py

@@ -5,9 +5,9 @@ from math import log as ln
 
 import subprocess
 from subprocess import STDOUT, PIPE
-from .target_quality import make_pipes, vmaf_probe, transform_vmaf, weighted_search, get_target_q, adapt_probing_rate
+from .target_quality import vmaf_probe, transform_vmaf, weighted_search, get_target_q, \
+    adapt_probing_rate, read_weighted_vmaf
 from Projects import Project
-from VMAF import read_weighted_vmaf
 
 import matplotlib
 from matplotlib import pyplot as plt

TargetQuality/target_quality.py

@@ -1,12 +1,11 @@
 import subprocess
 from subprocess import STDOUT, PIPE
 from Av1an.commandtypes import CommandPair, Command
-
 from Projects import Project
-from VMAF import call_vmaf, read_weighted_vmaf
+from VMAF import call_vmaf, read_json
 from Chunks.chunk import Chunk
-from math import isnan
 from math import log as ln
+from math import ceil, floor
 from Av1an.bar import process_pipe
 import numpy as np
 from scipy import interpolate
@@ -16,10 +15,47 @@ def transform_vmaf(vmaf):
     if vmaf<99.99:
         return -ln(1-vmaf/100)
     else:
-        #return -ln(1-99.99/100)
+        # return -ln(1-99.99/100)
         return 9.210340371976184
 
 
+def read_weighted_vmaf(file, percentile=0):
+    """Reads vmaf file with vmaf scores in it and return N percentile score from it.
+
+    :return: N percentile score
+    :rtype: float
+    """
+
+    jsn = read_json(file)
+
+    vmafs = sorted([x['metrics']['vmaf'] for x in jsn['frames']])
+
+    percentile = percentile if percentile != 0 else 0.25
+    score = get_percentile(vmafs, percentile)
+
+    return round(score, 2)
+
+
+def get_percentile(scores, percent):
+    """
+    Find the percentile of a list of values.
+    :param scores: - is a list of values. Note N MUST BE already sorted.
+    :param percent: - a float value from 0.0 to 1.0.
+    :return: - the percentile of the values
+    """
+    scores = sorted(scores)
+    key = lambda x: x
+
+    k = (len(scores)-1) * percent
+    f = floor(k)
+    c = ceil(k)
+    if f == c:
+        return key(scores[int(k)])
+    d0 = key(scores[int(f)]) * (c-k)
+    d1 = key(scores[int(c)]) * (k-f)
+    return d0+d1
+
+
 def adapt_probing_rate(rate, frames):
     """
     Change probing rate depending on amount of frames in scene.
@@ -64,7 +100,6 @@ def get_target_q(scores, target_quality):
     return int(q[0]), round(q[1], 3)
 
 
-
 def weighted_search(num1, vmaf1, num2, vmaf2, target):
     """
     Returns weighted value closest to searched
@@ -86,7 +121,6 @@ def weighted_search(num1, vmaf1, num2, vmaf2, target):
     return new_point
 
 
-
 def probe_cmd(chunk: Chunk, q, ffmpeg_pipe, encoder, probing_rate) -> CommandPair:
     """
     Generate and return commands for probes at set Q values
@@ -95,8 +129,8 @@ def probe_cmd(chunk: Chunk, q, ffmpeg_pipe, encoder, probing_rate) -> CommandPai
     should be faster than the actual encoding commands.
     These should not be moved into encoder classes at this point.
     """
-    pipe = ['ffmpeg', '-y', '-hide_banner', '-loglevel', 'error', '-i', '-', '-vf', f'select=not(mod(n\\,{probing_rate}))',
+    pipe = ['ffmpeg', '-y', '-hide_banner', '-loglevel', 'error', '-i', '-', '-vf',
-            *ffmpeg_pipe]
+            f'select=not(mod(n\\,{probing_rate}))', *ffmpeg_pipe]
 
     probe_name = gen_probes_names(chunk, q).with_suffix('.ivf').as_posix()
 

VMAF/vmaf.py

@@ -30,29 +30,6 @@ def read_json(file):
         return fl
 
 
-def read_weighted_vmaf(file, percentile=0):
-    """Reads vmaf file with vmaf scores in it and return N percentile score from it.
-
-    :return: N percentile score
-    :rtype: float
-    """
-
-    jsn = read_json(file)
-
-    vmafs = [x['metrics']['vmaf'] for x in jsn['frames']]
-
-    if percentile == 0:
-        # Using 2 standart deviations to weight for bad frames
-        mean = np.mean(vmafs)
-        minimum = np.min(vmafs)
-        score = (mean * 1/3) + (minimum * 2/3)
-
-    else:
-        score = round(np.percentile(vmafs, percentile), 2)
-
-    return round(score, 2)
-
-
 def call_vmaf(chunk: Chunk, encoded: Path, n_threads, model, res,
               fl_path: Path = None, vmaf_filter=None, vmaf_rate=0):
     cmd = ''