Help optimizing python code for Video QC

[groucho86]

Hi everyone,
I'm working on a script that takes a Rec2020 PQ 16bit TIFF image sequence and confirms two things: that it's properly constrained to P3 and that it does not exceed 1000nit.

I understand that a python for loop is not the way to get good performance. How can I optimize this code?

Here's what I have so far.


source_filepath = '/tmp/test.%04d.tiff'
clip = core.imwri.Read(filename=source_filepath, firstnum=1001,float_output=False)
clip = core.resize.Bicubic(clip, format=vs.RGBS)

## convert to X'Y'Z to check Y value
xyz = core.fmtc.primaries(clip=clip, prims='2020', primd='ciexyz')
xyz = core.std.PlaneStats(xyz, plane=1, prop='Nit')

## convert to P3D65/Linear to check for negative values (indicating out of gamut)
lin = core.fmtc.transfer(xyz, transs='2084', transd='linear' )
lin = core.fmtc.primaries(clip=clip, prims='2020', primd ='p3d65')
lin_r = core.std.PlaneStats(lin, plane=0, prop='Gamut')
lin_g = core.std.PlaneStats(lin, plane=1, prop='Gamut')
lin_b = core.std.PlaneStats(lin, plane=2, prop='Gamut')

qc_d = {}

for i in range(xyz.num_frames):

qc_d[i] = {}

if xyz.get_frame(i).props['NitMax'] > 0.751816896:
qc_d[i]['IllegalNit'] = True
else:
qc_d[i]['IllegalNit'] = False


if any(i < 0 for i in [lin_r.get_frame(i).props['GamutMin'],
lin_g.get_frame(i).props['GamutMin'],
lin_b.get_frame(i).props['GamutMin']]):
qc_d[i]['IllegalGamut'] = True

else:
qc_d[i]['IllegalGamut'] = False



Ultimately I just need a list of frames that have gamut and/or exceeding nit values.

I first tried ModifyFrame to tag "bad" frames but was struggling with that. Which is why I went for a "QC dictionary" instead.

[_Al_]

Modify frame would need some sort of request for frames anyway to analyze it .

Maybe you can use one more thread in Python , I get 30% boost, if using ThreadPoolExecutor for those analysis:


## convert to X'Y'Z to check Y value
xyz = core.fmtc.primaries(clip=clip, prims='2020', primd='ciexyz')
xyz = core.std.PlaneStats(xyz, plane=1, prop='Nit')

## convert to P3D65/Linear to check for negative values (indicating out of gamut)
lin = core.fmtc.transfer(xyz, transs='2084', transd='linear' )
lin = core.fmtc.primaries(clip=clip, prims='2020', primd ='p3d65')
lin = core.std.PlaneStats(lin, plane=0, prop='GamutR')\
.std.PlaneStats(lin, plane=1, prop='GamutG')\
.std.PlaneStats(lin, plane=2, prop='GamutB')

def analyze_nit(clip):
return [ 1 if xyz.get_frame(i).props['NitMax'] > 0.751816896 else 0 for i in range(clip.num_frames)]

def analyze_gamut(clip):
qc_gamut = []
for i in range(clip.num_frames):
f = clip.get_frame(i)
qc_gamut.append( 1 if f.props['GamutRMin']<0 or f.props['GamutGMin']<0 or f.props['GamutBMin']<0 else 0 )
return qc_gamut

from concurrent import futures

with futures.ThreadPoolExecutor(max_workers=2) as exe:
job1 = exe.submit(analyze_nit, xyz)
job2 = exe.submit(analyze_gamut, lin)
qc_nit = job1.result()
qc_gamut = job2.result()

values are bints (1 or 0, just thinking ahead about that plot) could be reached by index from those lists (gc_nit, gc_gamut), but you can change them to booleans. To have a list is kind of useful if you want to have immediate visual where those illegals values are located in video simply adding matplotlib graph capabilities, just add at the end of your script:

import matplotlib.pyplot as plt #pip install matplotlib
x_axis = list(range(0, len(xyz)))
plt.title("illegal nit(yellow) and gamut(red)")
plt.xlabel("Frames")
plt.ylabel("Illegal values for frame")
plt.scatter(x_axis, qc_nit, s=10, c='yellow')
plt.scatter(x_axis, qc_gamut, s=10, c='red')
plt.show()

[_Al_]

or maybe those plane stats could be created as well in those threaded jobs:

## convert to X'Y'Z to check Y value
xyz = core.fmtc.primaries(clip=clip, prims='2020', primd='ciexyz')

## convert to P3D65/Linear to check for negative values (indicating out of gamut)
lin = core.fmtc.transfer(xyz, transs='2084', transd='linear' )
lin = core.fmtc.primaries(clip=clip, prims='2020', primd ='p3d65')

from concurrent import futures

def analyze_nit(clip):
clip = clip.std.PlaneStats(plane=1, prop='Nit')
return [ 1 if clip.get_frame(i).props['NitMax'] > 0.751816896 else 0 for i in range(clip.num_frames)]

def analyze_gamut(clip):
clip = core.std.PlaneStats(clip, plane=0, prop='GamutR')\
.std.PlaneStats(clip, plane=1, prop='GamutG')\
.std.PlaneStats(clip, plane=2, prop='GamutB')
qc_gamut = []
for i in range(clip.num_frames):
f = clip.get_frame(i)
qc_gamut.append( 1 if f.props['GamutRMin']<0 or f.props['GamutGMin']<0 or f.props['GamutBMin']<0 else 0 )
return qc_gamut

with futures.ThreadPoolExecutor(max_workers=2) as exe:
job1 = exe.submit(analyze_nit, xyz)
job2 = exe.submit(analyze_gamut, lin)
qc_nit = job1.result()
qc_gamut = job2.result()

[WolframRhodium]

source_filepath = '/tmp/test.%04d.tiff'
clip = core.imwri.Read(filename=source_filepath, firstnum=1001,float_output=False)
clip = core.resize.Bicubic(clip, format=vs.RGBS)

## convert to X'Y'Z to check Y value
xyz = core.fmtc.primaries(clip=clip, prims='2020', primd='ciexyz')
xyz = core.std.PlaneStats(xyz, plane=1, prop='Nit')

## convert to P3D65/Linear to check for negative values (indicating out of gamut)
lin = core.fmtc.transfer(xyz, transs='2084', transd='linear' )
lin = core.fmtc.primaries(clip=clip, prims='2020', primd ='p3d65')
lin_r = core.std.PlaneStats(lin, plane=0, prop='Gamut')
lin_g = core.std.PlaneStats(lin, plane=1, prop='Gamut')
lin_b = core.std.PlaneStats(lin, plane=2, prop='Gamut')
##############################################

num_frames = xyz.num_frames
IllegalNit_frames = []
IllegalGamutMin_frames = []
def process(n, f):
xyz_frame, *lin_rgb_frame = f

fout = xyz_frame.copy()

if xyz_frame.props['NitMax'] > 0.751816896:
IllegalNit_frames.append(n)
fout.props['IllegalNit'] = True
else:
fout.props['IllegalNit'] = False

if any(lf.props['GamutMin'] < 0 for lf in lin_rgb_frame):
IllegalGamutMin_frames.append(n)
fout.props['IllegalGamutMin'] = True
else:
fout.props['IllegalGamutMin'] = False

global num_frames
num_frames -= 1
if num_frames == 0:
with open("out.txt", "w") as file:
file.write('IllegalNit: ' + ' '.join(str(n) for n in IllegalNit_frames) + '\n')
file.write('IllegalGamutMin: ' + ' '.join(str(n) for n in IllegalGamutMin_frames) + '\n')

return fout

res = core.std.ModifyFrame(xyz, [xyz, lin_r, lin_g, lin_b], process)
res = core.text.FrameProps(res, ['IllegalNit', 'IllegalGamutMin'])
res.set_output()


The script can be used in two ways: either view per-frame properties in vsedit, or you may run the script (using vspipe or Benchmark feature from vsedit) and a list of results will be written to disk.