VRChat の Panorama Camera で撮影したステレオ 360° 画像を左目と右目に分離して保存、それぞれに 360 photo の XMP メタデータを付与する

convert_pano.py

import argparse
import datetime
import os
import re
import xml.etree.ElementTree as ET
from pathlib import Path
from multiprocessing import Queue, Process, freeze_support

from PIL import Image       # pip install Pillow
from PIL.PngImagePlugin import PngInfo


SOURCE_GLOB = 'VRChat *.png'
RE_DATETIME = r'(?P<yyyy>\d{4}) (?P<mm>\d{1,2}) (?P<dd>\d{1,2}) (?P<HH>\d{1,2}) (?P<MM>\d{1,2}) (?P<SS>\d{1,2})'
RE_PARENT   = re.compile(r'VRChat ' + RE_DATETIME + r' (?P<serial>\d{3})\.png$')
RE_CHILD    = re.compile(r'VRChat ' + RE_DATETIME + r' (?P<serial>\d{3})\.360\.png$')


PI_PACKET_HEADER = "<?xpacket begin=\"\ufeff\" id=\"W5M0MpCehiHzreSzNTczkc9d\"?>"
PI_PACKET_TRAILER = "<?xpacket end=\"r\"?>"
XMP_PNG_iTXt_KEY = "XML:com.adobe.xmp"

def do_utime(path: str):
    m = RE_CHILD.match(path)
    dt = datetime.datetime(
        int(m.group('yyyy')),
        int(m.group('mm')),
        int(m.group('dd')),
        int(m.group('HH')),
        int(m.group('MM')),
        int(m.group('SS'))
    )
    tm = dt.timestamp()
    print(f"{path}: utime({dt.strftime('%Y-%m-%d %H:%M:%S')})")
    os.utime(path, times=(tm, tm))

# reference: https://github.com/yasirkula/Unity360ScreenshotCapture
def gen_photosphere_metadata(image: Image):
    root = ET.Element('x:xmpmeta', attrib={'xmlns:x': 'adobe:ns:meta/', 'x:xmptk': 'Adobe XMP Core 5.1.0-jc003' })
    rdf = ET.SubElement(root, 'rdf:RDF', attrib={ 'xmlns:rdf': 'http://www.w3.org/1999/02/22-rdf-syntax-ns#' })
    description = ET.SubElement(rdf, 'rdf:Description', attrib={ 
        'rdf:about': '', 
        'xmlns:GPano': 'http://ns.google.com/photos/1.0/panorama/'
    })
    PHOTOSPHERE_TAGS = {
        'GPano:UsePanoramaViewer': 'True',
        'GPano:CaptureSoftware': 'VRChat',
        'GPano:StitchingSoftware': 'VRChat',
        'GPano:ProjectionType': 'equirectangular',
        'GPano:PoseHeadingDegrees': '0.0',
        'GPano:InitialViewHeadingDegrees': '0.0',
        'GPano:InitialViewPitchDegrees': '0.0',
        'GPano:InitialViewRollDegrees': '0.0',
        'GPano:InitialHorizontalFOVDegrees': '60.0',
        'GPano:CroppedAreaLeftPixels': '0',
        'GPano:CroppedAreaTopPixels': '0',
        'GPano:CroppedAreaImageWidthPixels': str(image.width),
        'GPano:CroppedAreaImageHeightPixels': str(image.height),
        'GPano:FullPanoWidthPixels': str(image.width),
        'GPano:FullPanoHeightPixels': str(image.height),
    }
    for k, v in PHOTOSPHERE_TAGS.items():
        ET.SubElement(description, k).text = v

    b = ET.tostring(root, encoding='UTF-8', xml_declaration=False)
    return b"".join([PI_PACKET_HEADER.encode('UTF-8'), b, PI_PACKET_TRAILER.encode('UTF-8')])


def crop_worker(queue):
    for filename, stem in iter(queue.get, None):
        out_filename = f'{stem}.360.png'
        print(f"{filename}: Generating")
        with Image.open(filename) as image:
            pi = PngInfo()
            buf = gen_photosphere_metadata(image)
            pi.add_itxt(XMP_PNG_iTXt_KEY, buf, lang='', tkey='', zip=False)

            image.save(out_filename, 'PNG', pnginfo=pi)
            do_utime(out_filename)


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('path', type=Path, help='path to find images')
    parser.add_argument('--jobs', '-j', type=int, default=8, help='number of concurrency')
    parser.add_argument('--force', '-f', action='store_true', help='overwrite files')
    args = parser.parse_args()

    procs = []
    queue = Queue()

    for i in range(args.jobs):
        p = Process(target=crop_worker, args=(queue, ))
        procs.append(p)
        p.start()

    files = [x for x in args.path.glob(SOURCE_GLOB) if x.is_file() and RE_PARENT.match(x.name)]
    for f in files:
        new_name = f'{f.stem}.360.png'
        new_path = [x for x in Path('.').glob(new_name) if x.is_file()]
        if len(new_path) == 0 or args.force:
            queue.put((str(f), f.stem))
        else:
            for path in new_path:
                print(f"{f}: {path} exists, skipped")

    for i in range(args.jobs):
        queue.put(None)
    for p in procs:
        p.join()

if __name__ == '__main__':
    freeze_support()
    main()