Skip to content

Instantly share code, notes, and snippets.

@4re
Last active July 24, 2022 12:24
Show Gist options
  • Save 4re/e04ebfb300de4e4b7501622afb1764a7 to your computer and use it in GitHub Desktop.
Save 4re/e04ebfb300de4e4b7501622afb1764a7 to your computer and use it in GitHub Desktop.
Ugly hack to load mpls in vapoursynth
"""Parse blu-ray .mpls
Code from https://gist.github.com/dk00/0a0634c5666cf1b8ab9f
"""
import os
import collections
import io
import struct
import json
import argparse
from vapoursynth import core
__version__ = "1.0.3"
def _int_be(data):
return {1: ord, 2: lambda b: struct.unpack(">H", b)[0], 4: lambda b: struct.unpack(">I", b)[0]}[
len(data)
](data)
def load(file, fix_overlap=True):
# https://github.com/lerks/BluRay/wiki/MPLS
file.seek(8)
addr_items, addr_marks = _int_be(file.read(4)), _int_be(file.read(4))
file.seek(addr_items + 6)
item_count = _int_be(file.read(2))
file.seek(2, io.SEEK_CUR)
def read_item():
block_size = _int_be(file.read(2))
item = collections.OrderedDict()
item["name"] = file.read(5).decode()
file.seek(7, io.SEEK_CUR)
item["times"] = [_int_be(file.read(4)), _int_be(file.read(4))]
file.seek(block_size - 20, io.SEEK_CUR)
return item
items = [read_item() for _ in range(item_count)]
file.seek(addr_marks + 4)
mark_count = _int_be(file.read(2))
def read_mark():
file.seek(2, io.SEEK_CUR)
index = _int_be(file.read(2))
time = _int_be(file.read(4))
file.seek(6, io.SEEK_CUR)
return index, time
for _ in range(mark_count):
index, time = read_mark()
if time > items[index]["times"][-2]:
items[index]["times"].insert(-1, time)
if fix_overlap:
b = None
for item in items:
a, b = b, item["times"]
if a and b[0] < a[-1] < b[-1]:
a[-1] = b[0]
if len(b) > 1 and b[-1] - b[-2] < 90090:
b.pop()
return items
def source(mpls_file, format=None, cache=None, decoder="lsmas"):
if decoder == "ffms2":
decoder = core.ffms2.Source
elif decoder == "lsmas":
decoder = core.lsmas.LWLibavSource
else:
raise ValueError('decoder must be "ffms2" or "lsmas".')
stream_dir = os.path.join(os.path.split(os.path.split(mpls_file)[0])[0], "STREAM")
clip_list = []
ts_list = []
for i in load(open(mpls_file, "rb")):
ts_list.append(i["name"])
for i in ts_list:
clip_list.append(decoder(os.path.join(stream_dir, i + ".m2ts"), format=format, cache=cache))
return core.std.Splice(clip_list)
def main():
ap = argparse.ArgumentParser()
ap.add_argument("-f", "--format", help="Sets output format (raw, json, index).", action="store")
ap.add_argument("filename")
args = ap.parse_args()
data = load(open(ap.parse_args().filename, "rb"))
if not args.format:
args.format = "raw"
if args.format == "json":
print(json.dumps(data, indent=2))
elif args.format == "index":
print(len(data))
elif args.format == "raw":
print(data)
else:
raise ValueError('Wrong value for "--format". Allowed values are: raw, json, index.')
if __name__ == "__main__":
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment