Source code for cgl.plugins.otio.tools.exrheader

import sys
import os
import re
import subprocess
from concurrent.futures import ThreadPoolExecutor




def find_exrheader_exec():
    if sys.platform.startswith('win32'):
        path = os.path.abspath(os.path.join(os.path.dirname(__file__), 'bin', 'exrheader.exe'))
        if os.path.exists(path):
            return path
    return 'exrheader'




def read_exr_header(path):
    cmd = [find_exrheader_exec(), path]
    p = subprocess.Popen(cmd, stderr=subprocess.PIPE, stdout=subprocess.PIPE)
    stdout, stderr = p.communicate()
    if p.returncode != 0:
        raise ValueError(stderr.decode("utf-8"))

    data = stdout.decode("utf-8")

    layers = {}
    rgba = []
    metadata = {}
    file_version = None

    for line in data.splitlines():
        if not line:
            continue

        if line.startswith('file format version:'):
            version = line.split(':')
            if len(version) == 2:
                v = [l.strip() for l in version[1].split(",")]
                file_version = int(v[0])

            continue

        if line.startswith("   "):
            channel_data = [l.strip() for l in line.split(',')]
            if len(channel_data) == 3:
                channel_name = channel_data[0].split('.')
                if len(channel_name) == 1:
                    rgba.append(channel_name[0].upper())

                elif len(channel_name) == 2:
                    layer_name = channel_name[0]
                    channel_name = channel_name[1]
                    if layer_name not in layers:
                        layers[layer_name] = []
                    layers[layer_name].append(channel_name.upper())
                else:
                    print(f"unknown channel: {line}")

        else:
            META_FORMAT=r'(?P<key>\S+)\s+[(]\s*type\s*(?P<type>\w+)\s*[)]\s*:\s*(?P<value>.*)'

            m = re.match(META_FORMAT, line)
            if not m:
                continue
            d = m.groupdict()
            key = d['key']

            if key in ('channels', ):
                continue

            value = d['value']
            data_type = d['type']
            if data_type == 'string':
                value = value.strip('"')

            metadata[key] = {'type': data_type, 'value' : value}

    if not rgba:
        raise ValueError("Unable to fine RGBA channels")

    return {"rgba": rgba,
            "path": path,
            "layers" : layers,
            'file_format_version' : file_version,
            'metadata': metadata}




def batch_iter_meta(file_list):
    with ThreadPoolExecutor() as exec:
        futures = []

        for path in file_list:
            f = exec.submit(read_exr_header, path)
            futures.append(f)

        for f in futures:
            yield f.result()



if __name__ == "__main__":

    import glob
    test_file = "Z:\\prod\\premise\\render\\AlchemyTestB\\shots\\47\\0400\\lay\\default\\mark\\000.000\\high\\47_0400.*.exr"
    files = glob.glob(test_file)
    # for path in files:
    #     meta = read_exr_header(path)
    #     print(meta['path'])
    for meta in batch_iter_meta(files):
        print(meta['path'])

    # test_file = "47_1900.0001.exr"
    # header = read_exr_header(sys.argv[1])

    # pprint(header)