Ruffification and fixed bug

This commit is contained in:
2026-09-17 10:19:11 -07:00
parent 13d9cd4914
commit e6320e84c6
25 changed files with 787 additions and 659 deletions
+28 -28
View File
@@ -15,15 +15,16 @@
# import importlib
import os
import sys
sys.path.insert(0, os.path.abspath('../../src'))
sys.path.insert(0, os.path.abspath("../../src"))
sys.path.insert(0, os.path.abspath("../../../src"))
print(sys.path)
# -- Project information -----------------------------------------------------
project = u'wavinfo'
copyright = u'2018-2025, Jamie Hardt'
author = u'Jamie Hardt'
project = "wavinfo"
copyright = "2018-2025, Jamie Hardt"
author = "Jamie Hardt"
# The short X.Y version
version = "4.0"
@@ -42,34 +43,34 @@ release = "4.0.0"
# extensions coming with Sphinx (named 'sphinx.ext.*') or your custom
# ones.
extensions = [
'sphinx.ext.autodoc',
'sphinx.ext.todo',
'sphinx.ext.coverage',
"sphinx.ext.autodoc",
"sphinx.ext.todo",
"sphinx.ext.coverage",
]
# Add any paths that contain templates here, relative to this directory.
templates_path = ['_templates']
templates_path = ["_templates"]
# The suffix(es) of source filenames.
# You can specify multiple suffix as a list of string:
#
# source_suffix = ['.rst', '.md']
source_suffix = '.rst'
source_suffix = ".rst"
# The master toctree document.
master_doc = 'index'
master_doc = "index"
# The language for content autogenerated by Sphinx. Refer to documentation
# for a list of supported languages.
#
# This is also used if you do content translation via gettext catalogs.
# Usually you set "language" from the command line for these cases.
language = 'en'
language = "en"
# List of patterns, relative to source directory, that match files and
# directories to ignore when looking for source files.
# This pattern also affects html_static_path and html_extra_path.
exclude_patterns = [u'_build', 'Thumbs.db', '.DS_Store']
exclude_patterns = ["_build", "Thumbs.db", ".DS_Store"]
# The name of the Pygments (syntax highlighting) style to use.
pygments_style = None
@@ -80,7 +81,7 @@ pygments_style = None
# The theme to use for HTML and HTML Help pages. See the documentation for
# a list of builtin themes.
#
html_theme = 'sphinx_rtd_theme'
html_theme = "sphinx_rtd_theme"
# Theme options are theme-specific and customize the look and feel of a theme
# further. For a list of options available for each theme, see the
@@ -91,7 +92,7 @@ html_theme = 'sphinx_rtd_theme'
# Add any paths that contain custom static files (such as style sheets) here,
# relative to this directory. They are copied after the builtin static files,
# so a file named "default.css" will overwrite the builtin "default.css".
html_static_path = ['_static']
html_static_path = ["_static"]
# Custom sidebar templates, must be a dictionary that maps document names
# to template names.
@@ -107,7 +108,7 @@ html_static_path = ['_static']
# -- Options for HTMLHelp output ---------------------------------------------
# Output file base name for HTML help builder.
htmlhelp_basename = 'wavinfodoc'
htmlhelp_basename = "wavinfodoc"
# -- Options for LaTeX output ------------------------------------------------
@@ -116,15 +117,12 @@ latex_elements = {
# The paper size ('letterpaper' or 'a4paper').
#
# 'papersize': 'letterpaper',
# The font size ('10pt', '11pt' or '12pt').
#
# 'pointsize': '10pt',
# Additional stuff for the LaTeX preamble.
#
# 'preamble': '',
# Latex figure (float) alignment
#
# 'figure_align': 'htbp',
@@ -134,8 +132,7 @@ latex_elements = {
# (source start file, target name, title,
# author, documentclass [howto, manual, or own class]).
latex_documents = [
(master_doc, 'wavinfo.tex', u'wavinfo Documentation',
u'Jamie Hardt', 'manual'),
(master_doc, "wavinfo.tex", "wavinfo Documentation", "Jamie Hardt", "manual"),
]
@@ -143,10 +140,7 @@ latex_documents = [
# One entry per manual page. List of tuples
# (source start file, name, description, authors, manual section).
man_pages = [
(master_doc, 'wavinfo', u'wavinfo Documentation',
[author], 1)
]
man_pages = [(master_doc, "wavinfo", "wavinfo Documentation", [author], 1)]
# -- Options for Texinfo output ----------------------------------------------
@@ -155,9 +149,15 @@ man_pages = [
# (source start file, target name, title, author,
# dir menu entry, description, category)
texinfo_documents = [
(master_doc, 'wavinfo', u'wavinfo Documentation',
author, 'wavinfo', 'One line description of project.',
'Miscellaneous'),
(
master_doc,
"wavinfo",
"wavinfo Documentation",
author,
"wavinfo",
"One line description of project.",
"Miscellaneous",
),
]
@@ -176,7 +176,7 @@ epub_title = project
# epub_uid = ''
# A list of files that should not be packed into the epub file.
epub_exclude_files = ['search.html']
epub_exclude_files = ["search.html"]
# -- Extension configuration -------------------------------------------------
+10 -3
View File
@@ -26,7 +26,7 @@
"source": [
"from wavinfo import WavInfoReader\n",
"\n",
"path = '../tests/test_files/sounddevices/A101_1.WAV'\n",
"path = \"../tests/test_files/sounddevices/A101_1.WAV\"\n",
"\n",
"info = WavInfoReader(path)"
]
@@ -113,7 +113,12 @@
}
],
"source": [
"(info.fmt.sample_rate, info.fmt.channel_count, info.fmt.block_align, info.fmt.bits_per_sample)"
"(\n",
" info.fmt.sample_rate,\n",
" info.fmt.channel_count,\n",
" info.fmt.block_align,\n",
" info.fmt.bits_per_sample,\n",
")"
]
},
{
@@ -271,7 +276,9 @@
],
"source": [
"path = \"../tests/test_files/cue_chunks/izotoperx_cues_test.wav\"\n",
"info = WavInfoReader(path, info_encoding=\"utf-8\") # iZotope RX seems to encode marker text as UTF-8\n",
"info = WavInfoReader(\n",
" path, info_encoding=\"utf-8\"\n",
") # iZotope RX seems to encode marker text as UTF-8\n",
"\n",
"for cue in info.cues.each_cue():\n",
" print(f\"Cue ID: {cue[0]}\")\n",
+1 -2
View File
@@ -2,8 +2,7 @@
Probe WAVE Files for iXML, Broadcast-WAVE and other metadata.
"""
__all__ = ['WavInfoReader', 'WavInfoEOFError']
__all__ = ["WavInfoReader", "WavInfoEOFError"]
from .wave_reader import WavInfoReader
from .riff_parser import WavInfoEOFError
+35 -34
View File
@@ -18,7 +18,7 @@ class MyJSONEncoder(json.JSONEncoder):
if isinstance(o, Enum):
return o._name_
elif isinstance(o, bytes):
return 'base64:' + b64encode(o).decode('ascii')
return "base64:" + b64encode(o).decode("ascii")
else:
return super().default(o)
@@ -50,7 +50,7 @@ class MetaBrowser(Cmd):
if isinstance(val, int):
print(f" - {key}: {val}")
elif isinstance(val, str):
print(f" - {key}: \"{val}\"")
print(f' - {key}: "{val}"')
elif isinstance(val, dict):
print(f" - {key}: Dict ({len(val)} keys)")
elif isinstance(val, list):
@@ -63,7 +63,7 @@ class MetaBrowser(Cmd):
print(f" - {key}: Unknown")
def do_ls(self, _):
'List items at the current node: LS'
"List items at the current node: LS"
root = self.cwd
if isinstance(root, list):
@@ -94,7 +94,7 @@ class MetaBrowser(Cmd):
if argv[0] in self.cwd.keys():
self.path = self.path + [argv[0]]
else:
print(f"Key \"{argv[0]}\" does not exist")
print(f'Key "{argv[0]}" does not exist')
if len(self.path) > 0:
self.prompt = "(" + "/".join(self.path) + ") "
@@ -102,41 +102,40 @@ class MetaBrowser(Cmd):
self.prompt = "(wavinfo) "
def do_bye(self, _):
'Exit the interactive browser: BYE'
"Exit the interactive browser: BYE"
return True
def main():
version = importlib.metadata.version('wavinfo')
version = importlib.metadata.version("wavinfo")
manpath = os.path.dirname(__file__) + "/man"
parser = OptionParser()
parser.usage = 'wavinfo (--adm | --ixml) <FILE> +'
parser.usage = "wavinfo (--adm | --ixml) <FILE> +"
# parser.add_option('--install-manpages',
# help="Install manual pages for wavinfo",
# default=False,
# action='store_true')
parser.add_option('--man',
help="Read the manual and exit.",
default=False,
action='store_true')
parser.add_option(
"--man", help="Read the manual and exit.", default=False, action="store_true"
)
parser.add_option('--adm', dest='adm',
help='Output ADM XML',
default=False,
action='store_true')
parser.add_option(
"--adm", dest="adm", help="Output ADM XML", default=False, action="store_true"
)
parser.add_option('--ixml', dest='ixml',
help='Output iXML',
default=False,
action='store_true')
parser.add_option(
"--ixml", dest="ixml", help="Output iXML", default=False, action="store_true"
)
parser.add_option('-i',
help='Read metadata with an interactive prompt',
default=False,
action='store_true')
parser.add_option(
"-i",
help="Read metadata with an interactive prompt",
default=False,
action="store_true",
)
(options, args) = parser.parse_args(sys.argv)
@@ -149,6 +148,7 @@ def main():
if options.man:
import shlex
print("Which man page?")
print("1) wavinfo usage")
print("7) General info on Wave file metadata")
@@ -176,26 +176,27 @@ def main():
raise MissingDataError("ixml")
else:
ret_dict = {
'filename': arg,
'run_date': datetime.datetime.now().isoformat(),
'application': f"wavinfo {version}",
'scopes': {}
"filename": arg,
"run_date": datetime.datetime.now().isoformat(),
"application": f"wavinfo {version}",
"scopes": {},
}
for scope, name, value in this_file.walk():
if scope not in ret_dict['scopes'].keys():
ret_dict['scopes'][scope] = {}
if scope not in ret_dict["scopes"].keys():
ret_dict["scopes"][scope] = {}
ret_dict['scopes'][scope][name] = value
ret_dict["scopes"][scope][name] = value
if options.i:
interactive_dict.append(ret_dict)
else:
json.dump(ret_dict, cls=MyJSONEncoder, fp=sys.stdout,
indent=2)
json.dump(ret_dict, cls=MyJSONEncoder, fp=sys.stdout, indent=2)
except MissingDataError as e:
print("MissingDataError: Missing metadata (%s) in file %s" %
(e, arg), file=sys.stderr)
print(
"MissingDataError: Missing metadata (%s) in file %s" % (e, arg),
file=sys.stderr,
)
continue
except Exception as e:
raise e
+12 -11
View File
@@ -1,4 +1,5 @@
import struct
# from collections import namedtuple
from typing import NamedTuple, Dict
@@ -10,38 +11,38 @@ class RF64Context(NamedTuple):
bigchunk_table: Dict[str, int]
def parse_rf64(stream, signature=b'RF64') -> RF64Context:
def parse_rf64(stream, signature=b"RF64") -> RF64Context:
start = stream.tell()
assert stream.read(4) == b'WAVE'
assert stream.read(4) == b"WAVE"
ds64_chunk = riff_parser.parse_chunk(stream)
assert type(ds64_chunk) is riff_parser.ChunkDescriptor, \
assert type(ds64_chunk) is riff_parser.ChunkDescriptor, (
f"Expected ds64 chunk here, found {type(ds64_chunk)}"
)
ds64_field_spec = "<QQQI"
ds64_fields_size = struct.calcsize(ds64_field_spec)
assert ds64_chunk.ident == b'ds64'
assert ds64_chunk.ident == b"ds64"
ds64_data = ds64_chunk.read_data(stream)
assert len(ds64_data) >= ds64_fields_size
riff_size, data_size, sample_count, length_lookup_table = struct.unpack(
ds64_field_spec, ds64_data[0:ds64_fields_size]
)
)
bigchunk_table = {}
chunksize64format = "<4sL"
# chunksize64size = struct.calcsize(chunksize64format)
for _ in range(length_lookup_table):
bigname, bigsize = struct.unpack_from(chunksize64format,
ds64_data,
offset=ds64_fields_size)
bigname, bigsize = struct.unpack_from(
chunksize64format, ds64_data, offset=ds64_fields_size
)
bigchunk_table[bigname] = bigsize
bigchunk_table[b'data'] = data_size
bigchunk_table[b"data"] = data_size
bigchunk_table[signature] = riff_size
stream.seek(start, 0)
return RF64Context(sample_count=sample_count,
bigchunk_table=bigchunk_table)
return RF64Context(sample_count=sample_count, bigchunk_table=bigchunk_table)
+12 -11
View File
@@ -12,7 +12,7 @@ class WavInfoEOFError(EOFError):
class ListChunkDescriptor(NamedTuple):
signature: bytes
children: List[Union['ChunkDescriptor', 'ListChunkDescriptor']]
children: List[Union["ChunkDescriptor", "ListChunkDescriptor"]]
class ChunkDescriptor(NamedTuple):
@@ -48,14 +48,15 @@ def parse_chunk(stream, rf64_context=None):
if len(ident) != 4 or len(size_bytes) != 4:
raise WavInfoEOFError(identifier=ident, chunk_start=header_start)
data_size = struct.unpack('<I', size_bytes)[0]
data_size = struct.unpack("<I", size_bytes)[0]
if data_size == 0xFFFFFFFF:
if rf64_context is None and ident in {b'RF64', b'BW64'}:
if rf64_context is None and ident in {b"RF64", b"BW64"}:
rf64_context = parse_rf64(stream=stream, signature=ident)
assert rf64_context is not None, \
assert rf64_context is not None, (
"Sentinel data size 0xFFFFFFFF found outside of RF64 context"
)
data_size = rf64_context.bigchunk_table[ident]
@@ -63,14 +64,14 @@ def parse_chunk(stream, rf64_context=None):
if displacement % 2:
displacement += 1
if ident in {b'RIFF', b'LIST', b'RF64', b'BW64', b'list'}:
return parse_list_chunk(stream=stream, length=data_size,
rf64_context=rf64_context)
if ident in {b"RIFF", b"LIST", b"RF64", b"BW64", b"list"}:
return parse_list_chunk(
stream=stream, length=data_size, rf64_context=rf64_context
)
else:
data_start = stream.tell()
stream.seek(displacement, 1)
return ChunkDescriptor(ident=ident,
start=data_start,
length=data_size,
rf64_context=rf64_context)
return ChunkDescriptor(
ident=ident, start=data_start, length=data_size, rf64_context=rf64_context
)
+97 -78
View File
@@ -2,15 +2,14 @@
ADM Reader
"""
from struct import unpack, unpack_from, calcsize
from io import BytesIO
from collections import namedtuple
from io import BytesIO
from struct import calcsize, unpack, unpack_from
from typing import Optional
from lxml import etree as ET
ChannelEntry = namedtuple('ChannelEntry', "track_index uid track_ref pack_ref")
ChannelEntry = namedtuple("ChannelEntry", "track_index uid track_ref pack_ref")
class WavADMReader:
@@ -32,19 +31,19 @@ class WavADMReader:
offset = calcsize(header_fmt)
for _ in range(uid_count):
track_index, uid, track_ref, pack_ref = unpack_from(uid_fmt,
chna_data,
offset)
track_index, uid, track_ref, pack_ref = unpack_from(
uid_fmt, chna_data, offset
)
# these values are either ascii or all null
self.channel_uids.append(
ChannelEntry(track_index - 1,
uid.decode('ascii'),
track_ref.decode('ascii'),
pack_ref.decode('ascii')
)
ChannelEntry(
track_index - 1,
uid.decode("ascii"),
track_ref.decode("ascii"),
pack_ref.decode("ascii"),
)
)
offset += calcsize(uid_fmt)
@@ -65,40 +64,40 @@ class WavADMReader:
afext = self.axml.find(".//audioFormatExtended", namespaces=nsmap)
program = afext.find("audioProgramme", namespaces=nsmap)
ret_dict['programme_id'] = program.get("audioProgrammeID")
ret_dict['programme_name'] = program.get("audioProgrammeName")
ret_dict['programme_start'] = program.get("start")
ret_dict['programme_end'] = program.get("end")
ret_dict['contents'] = []
ret_dict["programme_id"] = program.get("audioProgrammeID")
ret_dict["programme_name"] = program.get("audioProgrammeName")
ret_dict["programme_start"] = program.get("start")
ret_dict["programme_end"] = program.get("end")
ret_dict["contents"] = []
for content_ref in program.findall("audioContentIDRef",
namespaces=nsmap):
for content_ref in program.findall("audioContentIDRef", namespaces=nsmap):
content_dict = dict()
content_dict['content_id'] = cid = content_ref.text
content = afext.find("audioContent[@audioContentID='%s']" % cid,
namespaces=nsmap)
content_dict['content_name'] = content.get("audioContentName")
content_dict['objects'] = []
content_dict["content_id"] = cid = content_ref.text
content = afext.find(
"audioContent[@audioContentID='%s']" % cid, namespaces=nsmap
)
content_dict["content_name"] = content.get("audioContentName")
content_dict["objects"] = []
for object_ref in content.findall("audioObjectIDRef",
namespaces=nsmap):
for object_ref in content.findall("audioObjectIDRef", namespaces=nsmap):
object_dict = dict()
object_dict['object_id'] = oid = object_ref.text
object = afext.find("audioObject[@audioObjectID='%s']" % oid,
namespaces=nsmap)
object_dict["object_id"] = oid = object_ref.text
object = afext.find(
"audioObject[@audioObjectID='%s']" % oid, namespaces=nsmap
)
pack = object.find("audioPackFormatIDRef", namespaces=nsmap)
object_dict['object_name'] = object.get("audioObjectName")
object_dict['object_start'] = object.get("start")
object_dict['object_duration'] = object.get("duration")
object_dict['pack_id'] = pack.text
object_dict["object_name"] = object.get("audioObjectName")
object_dict["object_start"] = object.get("start")
object_dict["object_duration"] = object.get("duration")
object_dict["pack_id"] = pack.text
track_uid_list = []
for t in object.findall("audioTrackUIDRef", namespaces=nsmap):
track_uid_list.append(t.text)
object_dict['track_uids'] = track_uid_list
content_dict['objects'].append(object_dict)
object_dict["track_uids"] = track_uid_list
content_dict["objects"].append(object_dict)
ret_dict['contents'].append(content_dict)
ret_dict["contents"].append(content_dict)
return ret_dict
@@ -111,8 +110,9 @@ class WavADMReader:
*object_name*, *object_id*,
*pack_format_name*, *pack_type*, *channel_format_name*
"""
channel_info = next((x for x in self.channel_uids
if x.track_index == index), None)
channel_info = next(
(x for x in self.channel_uids if x.track_index == index), None
)
if channel_info is None:
return None
@@ -121,60 +121,78 @@ class WavADMReader:
nsmap = self.axml.getroot().nsmap
afext = self.axml.find(".//audioFormatExtended",
namespaces=nsmap)
afext = self.axml.find(".//audioFormatExtended", namespaces=nsmap)
trackformat_elem = afext.find(
"audioTrackFormat[@audioTrackFormatID='%s']"
% channel_info.track_ref, namespaces=nsmap)
"audioTrackFormat[@audioTrackFormatID='%s']" % channel_info.track_ref,
namespaces=nsmap,
)
stream_id = trackformat_elem[0].text
channelformatref_elem = afext.find(
("audioStreamFormat[@audioStreamFormatID='%s']"
"/audioChannelFormatIDRef") % stream_id,
namespaces=nsmap)
if channelformatref_elem:
("audioStreamFormat[@audioStreamFormatID='%s']/audioChannelFormatIDRef")
% stream_id,
namespaces=nsmap,
)
if channelformatref_elem is not None:
channelformat_id = channelformatref_elem.text
else:
channelformat_id = None
packformatref_elem = afext.find(
("audioStreamFormat[@audioStreamFormatID='%s']"
"/audioPackFormatIDRef") % stream_id,
namespaces=nsmap)
("audioStreamFormat[@audioStreamFormatID='%s']/audioPackFormatIDRef")
% stream_id,
namespaces=nsmap,
)
if packformatref_elem:
if packformatref_elem is not None:
packformat_id = packformatref_elem.text
else:
packformat_id = None
channelformat_elem = afext\
.find("audioChannelFormat[@audioChannelFormatID='%s']"
% channelformat_id,
namespaces=nsmap)
ret_dict['channel_format_name'] = channelformat_elem.get(
"audioChannelFormatName")
if channelformat_id:
channelformat_elem = afext.find(
"audioChannelFormat[@audioChannelFormatID='%s']" % channelformat_id,
namespaces=nsmap,
)
ret_dict["channel_format_name"] = channelformat_elem.get(
"audioChannelFormatName"
)
else:
ret_dict["channel_format_name"] = None
packformat_elem = afext.find(
"audioPackFormat[@audioPackFormatID='%s']" % packformat_id,
namespaces=nsmap)
ret_dict['pack_type'] = packformat_elem.get(
"typeDefinition")
ret_dict['pack_format_name'] = packformat_elem.get(
"audioPackFormatName")
"audioPackFormat[@audioPackFormatID='%s']" % packformat_id, namespaces=nsmap
)
if packformat_elem is not None:
ret_dict["pack_type"] = packformat_elem.get("typeDefinition")
ret_dict["pack_format_name"] = packformat_elem.get("audioPackFormatName")
else:
ret_dict["pack_type"] = None
ret_dict["pack_format_name"] = None
object_elem = afext.find("audioObject[audioPackFormatIDRef = '%s']"
% packformat_id,
namespaces=nsmap)
object_elem = afext.find(
"audioObject[audioPackFormatIDRef = '%s']" % packformat_id, namespaces=nsmap
)
ret_dict['audio_object_name'] = object_elem.get("audioObjectName")
object_id = object_elem.get("audioObjectID")
ret_dict['object_id'] = object_id
if object_elem is not None:
ret_dict["audio_object_name"] = object_elem.get("audioObjectName")
object_id = object_elem.get("audioObjectID")
ret_dict["object_id"] = object_id
content_elem = afext.find("audioContent/[audioObjectIDRef = '%s']"
% object_id,
namespaces=nsmap)
content_elem = afext.find(
"audioContent/[audioObjectIDRef = '%s']" % object_id, namespaces=nsmap
)
ret_dict['content_name'] = content_elem.get("audioContentName")
ret_dict['content_id'] = content_elem.get("audioContentID")
ret_dict["content_name"] = content_elem.get("audioContentName")
ret_dict["content_id"] = content_elem.get("audioContentID")
else:
ret_dict["audio_object_name"] = None
ret_dict["object_id"] = None
ret_dict["content_name"] = None
ret_dict["content_id"] = None
return ret_dict
@@ -188,6 +206,7 @@ class WavADMReader:
rd.update(self.track_info(channel_uid_rec.track_index))
return rd
return dict(channel_entries=list(map(lambda z: make_entry(z),
self.channel_uids)),
programme=self.programme())
return dict(
channel_entries=list(map(lambda z: make_entry(z), self.channel_uids)),
programme=self.programme(),
)
+22 -19
View File
@@ -13,16 +13,18 @@ class WavBextReader:
the BEXT metadata scope. According to EBU Rec 3285 this shall be
ASCII.
"""
packstring = "<256s" + "32s" + "32s" + "10s" + "8s" + "QH" + "64s" + \
"hhhhh" + "180s"
packstring = (
"<256s" + "32s" + "32s" + "10s" + "8s" + "QH" + "64s" + "hhhhh" + "180s"
)
rest_starts = struct.calcsize(packstring)
unpacked = struct.unpack(packstring, bext_data[:rest_starts])
def sanitize_bytes(b: bytes) -> str:
# honestly can't remember why I'm stripping nulls this way
first_null = next((index for index, byte in enumerate(b)
if byte == 0), None)
first_null = next(
(index for index, byte in enumerate(b) if byte == 0), None
)
trimmed = b if first_null is None else b[:first_null]
decoded = trimmed.decode(encoding)
return decoded
@@ -91,18 +93,19 @@ class WavBextReader:
# umid_str = None
return {'description': self.description,
'originator': self.originator,
'originator_ref': self.originator_ref,
'originator_date': self.originator_date,
'originator_time': self.originator_time,
'time_reference': self.time_reference,
'version': self.version,
'umid': self.umid,
'coding_history': self.coding_history,
'loudness_value': self.loudness_value,
'loudness_range': self.loudness_range,
'max_true_peak': self.max_true_peak,
'max_momentary_loudness': self.max_momentary_loudness,
'max_shortterm_loudness': self.max_shortterm_loudness
}
return {
"description": self.description,
"originator": self.originator,
"originator_ref": self.originator_ref,
"originator_date": self.originator_date,
"originator_time": self.originator_time,
"time_reference": self.time_reference,
"version": self.version,
"umid": self.umid,
"coding_history": self.coding_history,
"loudness_value": self.loudness_value,
"loudness_range": self.loudness_range,
"max_true_peak": self.max_true_peak,
"max_momentary_loudness": self.max_momentary_loudness,
"max_shortterm_loudness": self.max_shortterm_loudness,
}
+59 -45
View File
@@ -7,11 +7,12 @@ IBM Corporation and Microsoft Corporation
https://www.aelius.com/njh/wavemetatools/doc/riffmci.pdf
"""
from dataclasses import dataclass
from .riff_parser import ChunkDescriptor
from struct import unpack, calcsize
from typing import Optional, Tuple, NamedTuple, List, Dict, Any, Generator
from typing import Optional, Tuple, NamedTuple, List, Dict, Any, Generator
#: Country Codes used in the RIFF standard to resolve locale. These codes
#: appear in CSET and LTXT metadata.
@@ -100,6 +101,7 @@ class CueEntry(NamedTuple):
"""
A ``cue`` element structure.
"""
#: Cue "name" or id number
name: int
#: Cue position, as a frame count in the play order of the WAVE file. In
@@ -118,29 +120,37 @@ class CueEntry(NamedTuple):
return calcsize(cls.Format)
@classmethod
def read(cls, data: bytes) -> 'CueEntry':
assert len(data) == cls.format_size(), \
(f"cue data size incorrect, expected {calcsize(cls.Format)} "
"found {len(data)}")
def read(cls, data: bytes) -> "CueEntry":
assert len(data) == cls.format_size(), (
f"cue data size incorrect, expected {calcsize(cls.Format)} "
"found {len(data)}"
)
parsed = unpack(cls.Format, data)
return cls(name=parsed[0], position=parsed[1], chunk_id=parsed[2],
chunk_start=parsed[3], block_start=parsed[4],
sample_offset=parsed[5])
return cls(
name=parsed[0],
position=parsed[1],
chunk_id=parsed[2],
chunk_start=parsed[3],
block_start=parsed[4],
sample_offset=parsed[5],
)
class LabelEntry(NamedTuple):
"""
A ``labl`` structure.
"""
name: int
text: str
@classmethod
def read(cls, data: bytes, encoding: str):
return cls(name=unpack("<I", data[0:4])[0],
text=data[4:].decode(encoding).rstrip("\0"))
return cls(
name=unpack("<I", data[0:4])[0], text=data[4:].decode(encoding).rstrip("\0")
)
NoteEntry = LabelEntry
@@ -150,6 +160,7 @@ class RangeLabel(NamedTuple):
"""
A ``ltxt`` structure.
"""
name: int
length: int
purpose: str
@@ -162,21 +173,26 @@ class RangeLabel(NamedTuple):
@classmethod
def read(cls, data: bytes, fallback_encoding: str):
leader_struct_fmt = "<II4sHHHH"
parsed = unpack(leader_struct_fmt, data[0:calcsize(leader_struct_fmt)])
text_data = data[calcsize(leader_struct_fmt):]
parsed = unpack(leader_struct_fmt, data[0 : calcsize(leader_struct_fmt)])
text_data = data[calcsize(leader_struct_fmt) :]
if data[6] != 0:
fallback_encoding = f"cp{data[6]}"
return cls(name=parsed[0], length=parsed[1], purpose=parsed[2],
country=parsed[3], language=parsed[4],
dialect=parsed[5], codepage=parsed[6],
text=text_data.decode(fallback_encoding))
return cls(
name=parsed[0],
length=parsed[1],
purpose=parsed[2],
country=parsed[3],
language=parsed[4],
dialect=parsed[5],
codepage=parsed[6],
text=text_data.decode(fallback_encoding),
)
@dataclass
class WavCuesReader:
#: Every ``cue`` entry in the file
cues: List[CueEntry]
@@ -190,13 +206,15 @@ class WavCuesReader:
notes: List[NoteEntry]
@classmethod
def read_all(cls, f,
cues: Optional[ChunkDescriptor],
labls: List[ChunkDescriptor],
ltxts: List[ChunkDescriptor],
notes: List[ChunkDescriptor],
fallback_encoding: str) -> 'WavCuesReader':
def read_all(
cls,
f,
cues: Optional[ChunkDescriptor],
labls: List[ChunkDescriptor],
ltxts: List[ChunkDescriptor],
notes: List[ChunkDescriptor],
fallback_encoding: str,
) -> "WavCuesReader":
cue_list = []
if cues is not None:
cues_data = cues.read_data(f)
@@ -205,33 +223,31 @@ class WavCuesReader:
cues_count = unpack("<I", cues_data[0:offset])
for _ in range(cues_count[0]):
cue_bytes = cues_data[offset: offset + CueEntry.format_size()]
cue_bytes = cues_data[offset : offset + CueEntry.format_size()]
cue_list.append(CueEntry.read(cue_bytes))
offset += CueEntry.format_size()
label_list = []
for labl in labls:
label_list.append(
LabelEntry.read(labl.read_data(f),
encoding=fallback_encoding)
LabelEntry.read(labl.read_data(f), encoding=fallback_encoding)
)
range_list = []
for r in ltxts:
range_list.append(
RangeLabel.read(r.read_data(f),
fallback_encoding=fallback_encoding)
RangeLabel.read(r.read_data(f), fallback_encoding=fallback_encoding)
)
note_list = []
for note in notes:
note_list.append(
NoteEntry.read(note.read_data(f),
encoding=fallback_encoding)
NoteEntry.read(note.read_data(f), encoding=fallback_encoding)
)
return WavCuesReader(cues=cue_list, labels=label_list,
ranges=range_list, notes=note_list)
return WavCuesReader(
cues=cue_list, labels=label_list, ranges=range_list, notes=note_list
)
def each_cue(self) -> Generator[Tuple[int, int], None, None]:
"""
@@ -242,8 +258,7 @@ class WavCuesReader:
for cue in self.cues:
yield (cue.name, cue.sample_offset)
def label_and_note(self, cue_ident: int) -> Tuple[Optional[str],
Optional[str]]:
def label_and_note(self, cue_ident: int) -> Tuple[Optional[str], Optional[str]]:
"""
Get the label and note (extended comment) for a cue.
@@ -251,10 +266,10 @@ class WavCuesReader:
:returns: a tuple of the the cue's label (if present) and note (if
present)
"""
label = next((label.text for label in self.labels
if label.name == cue_ident), None)
note = next((n.text for n in self.notes
if n.name == cue_ident), None)
label = next(
(label.text for label in self.labels if label.name == cue_ident), None
)
note = next((n.text for n in self.notes if n.name == cue_ident), None)
return (label, note)
def range(self, cue_ident: int) -> Optional[int]:
@@ -264,23 +279,22 @@ class WavCuesReader:
:param cue_ident: the cue's name, its unique identifying number
:returns: the length of the marker's range, or `None`
"""
return next((r.length for r in self.ranges
if r.name == cue_ident), None)
return next((r.length for r in self.ranges if r.name == cue_ident), None)
def to_dict(self) -> Dict[str, Any]:
retval = dict()
for n, t in self.each_cue():
retval[n] = dict()
retval[n]['frame'] = t
retval[n]["frame"] = t
label, note = self.label_and_note(n)
r = self.range(n)
if label is not None:
retval[n]['label'] = label
retval[n]["label"] = label
if note is not None:
retval[n]['note'] = note
retval[n]["note"] = note
if r is not None:
retval[n]['length'] = r
retval[n]["length"] = r
return retval
+122 -85
View File
@@ -19,6 +19,7 @@ class SegmentType(IntEnum):
"""
Metadata segment type.
"""
EndMarker = 0x0
DolbyE = 0x1
# Reserved2 = 0x2
@@ -29,7 +30,7 @@ class SegmentType(IntEnum):
DolbyDigitalPlus = 0x7
AudioInfo = 0x8
DolbyAtmos = 0x9
DolbyAtmosSupplemental = 0xa
DolbyAtmosSupplemental = 0xA
@classmethod
def _missing_(cls, val):
@@ -82,6 +83,7 @@ class DolbyDigitalPlusMetadata:
"""
Dolby surround endcoding mode.
"""
RESERVED = 0b11
IN_USE = 0b10
NOT_IN_USE = 0b01
@@ -126,6 +128,7 @@ class DolbyDigitalPlusMetadata:
Dolby Digital Plus `acmod` field
§ 4.3.2.3
"""
RESERVED = 0b000
CH_ORD_1_0 = 0b001
"Mono"
@@ -163,6 +166,7 @@ class DolbyDigitalPlusMetadata:
Dolby Digital Plus `surmixlev` field
§ 4.3.3.2
"""
DOWN_3DB = 0b00
DOWN_6DB = 0b01
MUTE = 0b10
@@ -174,24 +178,28 @@ class DolbyDigitalPlusMetadata:
Per ATSC/A52 § 5.4.2.12, this is not in use and always 0xFF.
"""
pass
class MixLevel(int):
"""
§ 4.3.6.2
"""
pass
class DialnormLevel(int):
"""
§ 4.3.4.4
"""
pass
class RoomType(Enum):
"""
`roomtyp` 4.3.6.3
"""
NOT_INDICATED = 0b00
LARGE_ROOM_X_CURVE = 0b01
SMALL_ROOM_FLAT_CURVE = 0b10
@@ -203,6 +211,7 @@ class DolbyDigitalPlusMetadata:
should downmix.
§ 4.3.8.1
"""
NOT_INDICATED = 0b00
PRO_LOGIC = 0b01
STEREO = 0b10
@@ -213,6 +222,7 @@ class DolbyDigitalPlusMetadata:
Dolby Surround-EX mode.
`dsurexmod` § 4.3.9.1
"""
NOT_INDICATED = 0b00
NOT_SEX = 0b01
SEX = 0b10
@@ -222,6 +232,7 @@ class DolbyDigitalPlusMetadata:
"""
`dheadphonmod` § 4.3.9.2
"""
NOT_INDICATED = 0b00
NOT_DOLBY_HEADPHONE = 0b01
DOLBY_HEADPHONE = 0b10
@@ -246,6 +257,7 @@ class DolbyDigitalPlusMetadata:
`compr1` RF compression profile
§ 4.3.10 (fig 42)
"""
NONE = 0
FILM_STANDARD = 1
FILM_LIGHT = 2
@@ -341,9 +353,11 @@ class DolbyDigitalPlusMetadata:
return b
def program_info(b):
return (b & 0x40) > 0, \
DolbyDigitalPlusMetadata.BitStreamMode(b & 0x38 >> 3), \
DolbyDigitalPlusMetadata.AudioCodingMode(b & 0x7)
return (
(b & 0x40) > 0,
DolbyDigitalPlusMetadata.BitStreamMode(b & 0x38 >> 3),
DolbyDigitalPlusMetadata.AudioCodingMode(b & 0x7),
)
def ddplus_reserved1(_):
pass
@@ -351,39 +365,49 @@ class DolbyDigitalPlusMetadata:
def surround_config(b):
return (
DolbyDigitalPlusMetadata.CenterDownMixLevel(b & 0x30 >> 4),
DolbyDigitalPlusMetadata.SurroundDownMixLevel(b & 0xc >> 2),
DolbyDigitalPlusMetadata.DolbySurroundEncodingMode(b & 0x3)
)
DolbyDigitalPlusMetadata.SurroundDownMixLevel(b & 0xC >> 2),
DolbyDigitalPlusMetadata.DolbySurroundEncodingMode(b & 0x3),
)
def dialnorm_info(b):
return (b & 0x80) > 0, b & 0x40 > 0, b & 0x20 > 0, \
DolbyDigitalPlusMetadata.DialnormLevel(b & 0x1f)
return (
(b & 0x80) > 0,
b & 0x40 > 0,
b & 0x20 > 0,
DolbyDigitalPlusMetadata.DialnormLevel(b & 0x1F),
)
def langcod(b) -> int:
return b
def audio_prod_info(b):
return (b & 0x80) > 0, \
DolbyDigitalPlusMetadata.MixLevel(b & 0x7c >> 2), \
DolbyDigitalPlusMetadata.RoomType(b & 0x3)
return (
(b & 0x80) > 0,
DolbyDigitalPlusMetadata.MixLevel(b & 0x7C >> 2),
DolbyDigitalPlusMetadata.RoomType(b & 0x3),
)
# loro_center_downmix_level, loro_surround_downmix_level
def ext_bsi1_word1(b):
return DolbyDigitalPlusMetadata.DownMixLevelToken(b & 0x38 >> 3), \
DolbyDigitalPlusMetadata.DownMixLevelToken(b & 0x7)
return DolbyDigitalPlusMetadata.DownMixLevelToken(
b & 0x38 >> 3
), DolbyDigitalPlusMetadata.DownMixLevelToken(b & 0x7)
# downmix_mode, ltrt_center_downmix_level, ltrt_surround_downmix_level
def ext_bsi1_word2(b):
return DolbyDigitalPlusMetadata\
.PreferredDownMixMode(b & 0xC0 >> 6), \
DolbyDigitalPlusMetadata.DownMixLevelToken(b & 0x38 >> 3), \
DolbyDigitalPlusMetadata.DownMixLevelToken(b & 0x7)
return (
DolbyDigitalPlusMetadata.PreferredDownMixMode(b & 0xC0 >> 6),
DolbyDigitalPlusMetadata.DownMixLevelToken(b & 0x38 >> 3),
DolbyDigitalPlusMetadata.DownMixLevelToken(b & 0x7),
)
# surround_ex_mode, dolby_headphone_encoded, ad_converter_type
def ext_bsi2_word1(b):
return DolbyDigitalPlusMetadata.SurroundEXMode(b & 0x60 >> 5), \
DolbyDigitalPlusMetadata.HeadphoneMode(b & 0x18 >> 3), \
DolbyDigitalPlusMetadata.ADConverterType(b & 0x4 >> 2)
return (
DolbyDigitalPlusMetadata.SurroundEXMode(b & 0x60 >> 5),
DolbyDigitalPlusMetadata.HeadphoneMode(b & 0x18 >> 3),
DolbyDigitalPlusMetadata.ADConverterType(b & 0x4 >> 2),
)
def ddplus_reserved2(_):
pass
@@ -398,7 +422,7 @@ class DolbyDigitalPlusMetadata:
pass
def ddplus_info1(b):
return DolbyDigitalPlusMetadata.StreamDependency(b & 0xc >> 2)
return DolbyDigitalPlusMetadata.StreamDependency(b & 0xC >> 2)
def ddplus_reserved4(_):
pass
@@ -412,19 +436,24 @@ class DolbyDigitalPlusMetadata:
pid = program_id(buffer[0])
lfe_on, bitstream_mode, audio_coding_mode = program_info(buffer[1])
ddplus_reserved1(buffer[2:2])
center_downmix_level, surround_downmix_level, \
dolby_surround_encoded = surround_config(buffer[4])
langcode_present, copyright_bitstream, original_bitstream, \
dialnorm = dialnorm_info(buffer[5])
center_downmix_level, surround_downmix_level, dolby_surround_encoded = (
surround_config(buffer[4])
)
langcode_present, copyright_bitstream, original_bitstream, dialnorm = (
dialnorm_info(buffer[5])
)
langcode = langcod(buffer[6])
prod_info_exists, mixlevel, roomtype = audio_prod_info(buffer[7])
loro_center_downmix_level, \
loro_surround_downmix_level = ext_bsi1_word1(buffer[8])
downmix_mode, ltrt_center_downmix_level, \
ltrt_surround_downmix_level = ext_bsi1_word2(buffer[9])
surround_ex_mode, dolby_headphone_encoded, \
ad_converter_type = ext_bsi2_word1(buffer[10])
loro_center_downmix_level, loro_surround_downmix_level = ext_bsi1_word1(
buffer[8]
)
downmix_mode, ltrt_center_downmix_level, ltrt_surround_downmix_level = (
ext_bsi1_word2(buffer[9])
)
surround_ex_mode, dolby_headphone_encoded, ad_converter_type = ext_bsi2_word1(
buffer[10]
)
ddplus_reserved2(buffer[11:14])
compression = compr1(buffer[14])
@@ -436,32 +465,34 @@ class DolbyDigitalPlusMetadata:
reserved(buffer[27:69])
return DolbyDigitalPlusMetadata(
program_id=pid, lfe_on=lfe_on,
bitstream_mode=bitstream_mode,
audio_coding_mode=audio_coding_mode,
center_downmix_level=center_downmix_level,
surround_downmix_level=surround_downmix_level,
dolby_surround_encoded=dolby_surround_encoded,
langcode_present=langcode_present,
copyright_bitstream=copyright_bitstream,
original_bitstream=original_bitstream,
dialnorm=dialnorm,
langcode=langcode,
prod_info_exists=prod_info_exists,
mixlevel=mixlevel,
roomtype=roomtype,
loro_center_downmix_level=loro_center_downmix_level,
loro_surround_downmix_level=loro_surround_downmix_level,
downmix_mode=downmix_mode,
ltrt_center_downmix_level=ltrt_center_downmix_level,
ltrt_surround_downmix_level=ltrt_surround_downmix_level,
surround_ex_mode=surround_ex_mode,
dolby_headphone_encoded=dolby_headphone_encoded,
ad_converter_type=ad_converter_type,
compression_profile=compression,
dynamic_range=dynamic_range,
stream_dependency=stream_info,
datarate_kbps=data_rate)
program_id=pid,
lfe_on=lfe_on,
bitstream_mode=bitstream_mode,
audio_coding_mode=audio_coding_mode,
center_downmix_level=center_downmix_level,
surround_downmix_level=surround_downmix_level,
dolby_surround_encoded=dolby_surround_encoded,
langcode_present=langcode_present,
copyright_bitstream=copyright_bitstream,
original_bitstream=original_bitstream,
dialnorm=dialnorm,
langcode=langcode,
prod_info_exists=prod_info_exists,
mixlevel=mixlevel,
roomtype=roomtype,
loro_center_downmix_level=loro_center_downmix_level,
loro_surround_downmix_level=loro_surround_downmix_level,
downmix_mode=downmix_mode,
ltrt_center_downmix_level=ltrt_center_downmix_level,
ltrt_surround_downmix_level=ltrt_surround_downmix_level,
surround_ex_mode=surround_ex_mode,
dolby_headphone_encoded=dolby_headphone_encoded,
ad_converter_type=ad_converter_type,
compression_profile=compression,
dynamic_range=dynamic_range,
stream_dependency=stream_info,
datarate_kbps=data_rate,
)
@dataclass
@@ -488,7 +519,6 @@ class DolbyAtmosMetadata:
@classmethod
def load(cls, data: bytes):
assert len(data) == cls.SEGMENT_LENGTH
# (f"DolbyAtmosMetadata segment is incorrect length, "
# f"expected {cls.SEGMENT_LENGTH} actual was {len(data)}")
@@ -498,7 +528,7 @@ class DolbyAtmosMetadata:
h.seek(32, 1)
toolname = h.read(cls.TOOL_NAME_LENGTH)
toolname = unpack("%is" % cls.TOOL_NAME_LENGTH, toolname)[0]
toolname = toolname.decode('utf-8').strip('\0')
toolname = toolname.decode("utf-8").strip("\0")
vers = h.read(3)
major, minor, fix = unpack("BBB", vers)
@@ -508,10 +538,11 @@ class DolbyAtmosMetadata:
a_val = unpack("B", h.read(1))[0]
warp_mode = a_val & 0x7
return DolbyAtmosMetadata(tool_name=toolname,
tool_version=(major, minor, fix),
warp_mode=DolbyAtmosMetadata
.WarpMode(warp_mode))
return DolbyAtmosMetadata(
tool_name=toolname,
tool_version=(major, minor, fix),
warp_mode=DolbyAtmosMetadata.WarpMode(warp_mode),
)
@dataclass
@@ -531,15 +562,14 @@ class DolbyAtmosSupplementalMetadata:
NOT_INDICATED = 0x04
object_count: int
render_modes: List['DolbyAtmosSupplementalMetadata.BinauralRenderMode']
render_modes: List["DolbyAtmosSupplementalMetadata.BinauralRenderMode"]
trim_modes: List[int]
MAGIC = 0xf8726fbd
MAGIC = 0xF8726FBD
TRIM_CONFIG_COUNT = 9
@classmethod
def load(cls, data: bytes):
trim_modes = []
render_modes = []
@@ -564,9 +594,9 @@ class DolbyAtmosSupplementalMetadata:
binaural_mode &= 0x7
render_modes.append(binaural_mode)
return DolbyAtmosSupplementalMetadata(object_count=object_count,
render_modes=render_modes,
trim_modes=trim_modes)
return DolbyAtmosSupplementalMetadata(
object_count=object_count, render_modes=render_modes, trim_modes=trim_modes
)
class WavDolbyMetadataReader:
@@ -589,9 +619,9 @@ class WavDolbyMetadataReader:
retval = size
for b in bs:
retval += int(b)
retval &= 0xff
retval &= 0xFF
retval = ((~retval) + 1) & 0xff
retval = ((~retval) + 1) & 0xFF
return retval
@@ -614,8 +644,9 @@ class WavDolbyMetadataReader:
else:
seg_size = unpack("<H", h.read(2))[0]
seg_payload = h.read(seg_size)
expected_checksum = WavDolbyMetadataReader\
.segment_checksum(seg_payload, seg_size)
expected_checksum = WavDolbyMetadataReader.segment_checksum(
seg_payload, seg_size
)
checksum = unpack("B", h.read(1))[0]
segment = seg_payload
@@ -626,35 +657,41 @@ class WavDolbyMetadataReader:
elif stype == SegmentType.DolbyAtmosSupplemental:
segment = DolbyAtmosSupplementalMetadata.load(segment)
self.segment_list\
.append((stype, checksum == expected_checksum, segment))
self.segment_list.append(
(stype, checksum == expected_checksum, segment)
)
def dolby_digital_plus(self) -> List[DolbyDigitalPlusMetadata]:
"""
Every valid Dolby Digital Plus metadata segment in the file.
"""
return [x[2] for x in self.segment_list
if x[0] == SegmentType.DolbyDigitalPlus and x[1]]
return [
x[2]
for x in self.segment_list
if x[0] == SegmentType.DolbyDigitalPlus and x[1]
]
def dolby_atmos(self) -> List[DolbyAtmosMetadata]:
"""
Every valid Dolby Atmos metadata segment in the file.
"""
return [x[2] for x in self.segment_list
if x[0] == SegmentType.DolbyAtmos and x[1]]
return [
x[2] for x in self.segment_list if x[0] == SegmentType.DolbyAtmos and x[1]
]
def dolby_atmos_supplemental(self) -> List[DolbyAtmosSupplementalMetadata]:
"""
Every valid Dolby Atmos Supplemental metadata segment in the file.
"""
return [x[2] for x in self.segment_list
if x[0] == SegmentType.DolbyAtmosSupplemental and x[1]]
return [
x[2]
for x in self.segment_list
if x[0] == SegmentType.DolbyAtmosSupplemental and x[1]
]
def to_dict(self) -> dict:
ddp = map(lambda x: asdict(x), self.dolby_digital_plus())
atmos = map(lambda x: asdict(x), self.dolby_atmos())
# atmos_sup = map(lambda x: asdict(x), self.dolby_atmos_supplemental())
return dict(dolby_digital_plus=list(ddp),
dolby_atmos=list(atmos))
return dict(dolby_digital_plus=list(ddp), dolby_atmos=list(atmos))
+53 -45
View File
@@ -4,7 +4,6 @@ from typing import Optional
class WavInfoChunkReader:
def __init__(self, f, encoding):
self.encoding = encoding
@@ -12,56 +11,64 @@ class WavInfoChunkReader:
parsed_chunks = parse_chunk(f)
assert type(parsed_chunks) is ListChunkDescriptor
list_chunks = [chunk for chunk in parsed_chunks.children
if type(chunk) is ListChunkDescriptor]
list_chunks = [
chunk
for chunk in parsed_chunks.children
if type(chunk) is ListChunkDescriptor
]
self.info_chunk = next((chunk for chunk in list_chunks
if chunk.signature == b'INFO'), None)
self.info_chunk = next(
(chunk for chunk in list_chunks if chunk.signature == b"INFO"), None
)
#: 'ICOP' Copyright
self.copyright: Optional[str] = self._get_field(f, b'ICOP')
self.copyright: Optional[str] = self._get_field(f, b"ICOP")
#: 'IPRD' Product
self.product: Optional[str] = self._get_field(f, b'IPRD')
self.product: Optional[str] = self._get_field(f, b"IPRD")
self.album: Optional[str] = self.product
#: 'IGNR' Genre
self.genre: Optional[str] = self._get_field(f, b'IGNR')
self.genre: Optional[str] = self._get_field(f, b"IGNR")
#: 'ISBJ' Subject
self.subject: Optional[str] = self._get_field(f, b'ISBJ')
self.subject: Optional[str] = self._get_field(f, b"ISBJ")
#: 'IART' Artist, composer, author
self.artist: Optional[str] = self._get_field(f, b'IART')
self.artist: Optional[str] = self._get_field(f, b"IART")
#: 'ICMT' Comment
self.comment: Optional[str] = self._get_field(f, b'ICMT')
self.comment: Optional[str] = self._get_field(f, b"ICMT")
#: 'ISFT' Software, encoding application
self.software: Optional[str] = self._get_field(f, b'ISFT')
self.software: Optional[str] = self._get_field(f, b"ISFT")
#: 'ICRD' Created date
self.created_date: Optional[str] = self._get_field(f, b'ICRD')
self.created_date: Optional[str] = self._get_field(f, b"ICRD")
#: 'IENG' Engineer
self.engineer: Optional[str] = self._get_field(f, b'IENG')
self.engineer: Optional[str] = self._get_field(f, b"IENG")
#: 'ITCH' Technician
self.technician: Optional[str] = self._get_field(f, b'ITCH')
self.technician: Optional[str] = self._get_field(f, b"ITCH")
#: 'IKEY' Keywords, keyword list
self.keywords: Optional[str] = self._get_field(f, b'IKEY')
self.keywords: Optional[str] = self._get_field(f, b"IKEY")
#: 'INAM' Name, title
self.title: Optional[str] = self._get_field(f, b'INAM')
self.title: Optional[str] = self._get_field(f, b"INAM")
#: 'ISRC' Source
self.source: Optional[str] = self._get_field(f, b'ISRC')
self.source: Optional[str] = self._get_field(f, b"ISRC")
#: 'TAPE' Tape
self.tape: Optional[str] = self._get_field(f, b'TAPE')
self.tape: Optional[str] = self._get_field(f, b"TAPE")
#: 'IARL' Archival Location
self.archival_location: Optional[str] = self._get_field(f, b'IARL')
self.archival_location: Optional[str] = self._get_field(f, b"IARL")
#: 'ICSM' Commissioned
self.commissioned: Optional[str] = self._get_field(f, b'ICMS')
self.commissioned: Optional[str] = self._get_field(f, b"ICMS")
def _get_field(self, f, field_ident) -> Optional[str]:
search = next(((chunk.start, chunk.length)
for chunk in self.info_chunk.children
if chunk.ident == field_ident),
None)
search = next(
(
(chunk.start, chunk.length)
for chunk in self.info_chunk.children
if chunk.ident == field_ident
),
None,
)
if search is not None:
f.seek(search[0])
data = f.read(search[1])
return data.decode(self.encoding).rstrip('\0')
return data.decode(self.encoding).rstrip("\0")
else:
return None
@@ -69,26 +76,27 @@ class WavInfoChunkReader:
"""
A dictionary with all of the key/values read from the INFO scope.
"""
return {'copyright': self.copyright,
'product': self.product,
'album': self.album,
'genre': self.genre,
'artist': self.artist,
'comment': self.comment,
'software': self.software,
'created_date': self.created_date,
'engineer': self.engineer,
'keywords': self.keywords,
'title': self.title,
'source': self.source,
'tape': self.tape,
'commissioned': self.commissioned,
'archival_location': self.archival_location,
'subject': self.subject,
'technician': self.technician
}
return {
"copyright": self.copyright,
"product": self.product,
"album": self.album,
"genre": self.genre,
"artist": self.artist,
"comment": self.comment,
"software": self.software,
"created_date": self.created_date,
"engineer": self.engineer,
"keywords": self.keywords,
"title": self.title,
"source": self.source,
"tape": self.tape,
"commissioned": self.commissioned,
"archival_location": self.archival_location,
"subject": self.subject,
"technician": self.technician,
}
def __repr__(self):
return_val = self.to_dict()
return_val.update({'encoding': self.encoding})
return_val.update({"encoding": self.encoding})
return str(return_val)
+20 -18
View File
@@ -1,5 +1,6 @@
from lxml import etree as ET
import io
# from collections import namedtuple
from typing import Optional
from enum import IntEnum
@@ -22,6 +23,7 @@ class SteinbergMetadata:
"""
Steinberg speaker format enumeration.
"""
MONO = 0
STEREO = 1
LRC = 10
@@ -79,7 +81,8 @@ class SteinbergMetadata:
`AudioSpeakerArrangement` property
"""
val = self.parsed.find(
"./ATTR_LIST/ATTR[NAME = 'AudioSpeakerArrangement']/VALUE")
"./ATTR_LIST/ATTR[NAME = 'AudioSpeakerArrangement']/VALUE"
)
if val is not None:
return type(self).AudioSpeakerArrangement(int(val.text))
@@ -88,8 +91,7 @@ class SteinbergMetadata:
"""
AudioSampleFormatSize
"""
val = self.parsed.find(
"./ATTR_LIST/ATTR[NAME = 'AudioSampleFormatSize']/VALUE")
val = self.parsed.find("./ATTR_LIST/ATTR[NAME = 'AudioSampleFormatSize']/VALUE")
if val is not None:
return int(val.text)
@@ -98,8 +100,7 @@ class SteinbergMetadata:
"""
MediaCompany
"""
val = self.parsed.find(
"./ATTR_LIST/ATTR[NAME = 'MediaCompany']/VALUE")
val = self.parsed.find("./ATTR_LIST/ATTR[NAME = 'MediaCompany']/VALUE")
if val is not None:
return val.text
@@ -108,8 +109,7 @@ class SteinbergMetadata:
"""
MediaDropFrames
"""
val = self.parsed.find(
"./ATTR_LIST/ATTR[NAME = 'MediaDropFrames']/VALUE")
val = self.parsed.find("./ATTR_LIST/ATTR[NAME = 'MediaDropFrames']/VALUE")
if val is not None:
return val.text == "1"
@@ -118,8 +118,7 @@ class SteinbergMetadata:
"""
MediaDuration
"""
val = self.parsed.find(
"./ATTR_LIST/ATTR[NAME = 'MediaDuration']/VALUE")
val = self.parsed.find("./ATTR_LIST/ATTR[NAME = 'MediaDuration']/VALUE")
if val is not None:
return float(val.text)
@@ -192,13 +191,12 @@ class WavIXMLFormat:
:yields: `IXMLTrack` for each track.
"""
for track in self.parsed.find("./TRACK_LIST").iter():
if track.tag == 'TRACK':
if track.tag == "TRACK":
yield IXMLTrack(
channel_index=track.xpath('string(CHANNEL_INDEX/text())'),
interleave_index=track.xpath(
'string(INTERLEAVE_INDEX/text())'),
name=track.xpath('string(NAME/text())'),
function=track.xpath('string(FUNCTION/text())')
channel_index=track.xpath("string(CHANNEL_INDEX/text())"),
interleave_index=track.xpath("string(INTERLEAVE_INDEX/text())"),
name=track.xpath("string(NAME/text())"),
function=track.xpath("string(FUNCTION/text())"),
)
@property
@@ -258,6 +256,10 @@ class WavIXMLFormat:
def to_dict(self):
return dict(
track_list=list(map(lambda x: x._asdict(), self.track_list)),
project=self.project, scene=self.scene, take=self.take,
tape=self.tape, family_uid=self.family_uid,
family_name=self.family_name)
project=self.project,
scene=self.scene,
take=self.take,
tape=self.tape,
family_uid=self.family_uid,
family_name=self.family_name,
)
+87 -61
View File
@@ -38,7 +38,7 @@ class WavInfoReader:
Parse a WAV audio file for metadata.
"""
def __init__(self, path, info_encoding='latin_1', bext_encoding='ascii'):
def __init__(self, path, info_encoding="latin_1", bext_encoding="ascii"):
"""
Create a new reader object.
@@ -86,9 +86,9 @@ class WavInfoReader:
#: Sampler `smpl` metadata
self.smpl: Optional[WavSmplReader] = None
if hasattr(path, 'read'):
if hasattr(path, "read"):
self.get_wav_info(path)
self.url = 'about:blank'
self.url = "about:blank"
self.path = repr(path)
else:
@@ -99,7 +99,7 @@ class WavInfoReader:
self.path = absolute_path
with open(path, 'rb') as path:
with open(path, "rb") as path:
self.get_wav_info(path)
def get_wav_info(self, wavfile):
@@ -119,36 +119,43 @@ class WavInfoReader:
self.smpl = self._get_sampler_loops(wavfile)
self.data = self._describe_data()
def _find_chunk_data(self, ident, from_stream,
default_none=False) -> Optional[bytes]:
top_chunks = (chunk for chunk in self.main_list
if type(chunk) is ChunkDescriptor and
chunk.ident == ident)
def _find_chunk_data(
self, ident, from_stream, default_none=False
) -> Optional[bytes]:
top_chunks = (
chunk
for chunk in self.main_list
if type(chunk) is ChunkDescriptor and chunk.ident == ident
)
chunk_descriptor = next(top_chunks, None) \
if default_none else next(top_chunks)
chunk_descriptor = next(top_chunks, None) if default_none else next(top_chunks)
return chunk_descriptor.read_data(from_stream) \
if chunk_descriptor else None
return chunk_descriptor.read_data(from_stream) if chunk_descriptor else None
def _find_list_chunk(self, signature) -> Optional[ListChunkDescriptor]:
top_chunks = (chunk for chunk in self.main_list
if type(chunk) is ListChunkDescriptor and
chunk.signature == signature)
top_chunks = (
chunk
for chunk in self.main_list
if type(chunk) is ListChunkDescriptor and chunk.signature == signature
)
return next(top_chunks, None)
def _describe_data(self):
data_chunk = next(c for c in self.main_list
if type(c) is ChunkDescriptor and c.ident == b'data')
data_chunk = next(
c
for c in self.main_list
if type(c) is ChunkDescriptor and c.ident == b"data"
)
assert isinstance(self.fmt, WavAudioFormat)
return WavDataDescriptor(
byte_count=data_chunk.length,
frame_count=int(data_chunk.length / self.fmt.block_align))
frame_count=int(data_chunk.length / self.fmt.block_align),
)
def _get_format(self, f):
fmt_data = self._find_chunk_data(b'fmt ', f)
fmt_data = self._find_chunk_data(b"fmt ", f)
assert fmt_data is not None, "Fmt data not found, not a valid wav file"
packstring = "<HHIIHH"
@@ -156,62 +163,79 @@ class WavInfoReader:
unpacked = struct.unpack(packstring, fmt_data[:rest_starts])
return WavAudioFormat(audio_format=unpacked[0],
channel_count=unpacked[1],
sample_rate=unpacked[2],
byte_rate=unpacked[3],
block_align=unpacked[4],
bits_per_sample=unpacked[5]
)
return WavAudioFormat(
audio_format=unpacked[0],
channel_count=unpacked[1],
sample_rate=unpacked[2],
byte_rate=unpacked[3],
block_align=unpacked[4],
bits_per_sample=unpacked[5],
)
def _get_info(self, f, encoding):
finder = (chunk.signature for chunk in self.main_list
if type(chunk) is ListChunkDescriptor)
finder = (
chunk.signature
for chunk in self.main_list
if type(chunk) is ListChunkDescriptor
)
if b'INFO' in finder:
if b"INFO" in finder:
return WavInfoChunkReader(f, encoding)
def _get_bext(self, f, encoding):
bext_data = self._find_chunk_data(b'bext', f, default_none=True)
bext_data = self._find_chunk_data(b"bext", f, default_none=True)
return WavBextReader(bext_data, encoding) if bext_data else None
def _get_adm(self, f):
axml = self._find_chunk_data(b'axml', f, default_none=True)
chna = self._find_chunk_data(b'chna', f, default_none=True)
return WavADMReader(axml_data=axml, chna_data=chna) \
if axml and chna else None
axml = self._find_chunk_data(b"axml", f, default_none=True)
chna = self._find_chunk_data(b"chna", f, default_none=True)
return WavADMReader(axml_data=axml, chna_data=chna) if axml and chna else None
def _get_dbmd(self, f):
dbmd_data = self._find_chunk_data(b'dbmd', f, default_none=True)
return WavDolbyMetadataReader(dbmd_data=dbmd_data) \
if dbmd_data else None
dbmd_data = self._find_chunk_data(b"dbmd", f, default_none=True)
return WavDolbyMetadataReader(dbmd_data=dbmd_data) if dbmd_data else None
def _get_ixml(self, f):
ixml_data = self._find_chunk_data(b'iXML', f, default_none=True)
return WavIXMLFormat(ixml_data.rstrip(b'\0')) if ixml_data else None
ixml_data = self._find_chunk_data(b"iXML", f, default_none=True)
return WavIXMLFormat(ixml_data.rstrip(b"\0")) if ixml_data else None
def _get_cue(self, f):
cue = next((cue_chunk for cue_chunk in self.main_list if
type(cue_chunk) is ChunkDescriptor and
cue_chunk.ident == b'cue '), None)
cue = next(
(
cue_chunk
for cue_chunk in self.main_list
if type(cue_chunk) is ChunkDescriptor and cue_chunk.ident == b"cue "
),
None,
)
adtl = self._find_list_chunk(b'adtl')
adtl = self._find_list_chunk(b"adtl")
labls = []
ltxts = []
notes = []
if adtl is not None:
labls = [c for c in adtl.children
if type(c) is ChunkDescriptor and c.ident == b'labl']
ltxts = [c for c in adtl.children
if type(c) is ChunkDescriptor and c.ident == b'ltxt']
notes = [c for c in adtl.children
if type(c) is ChunkDescriptor and c.ident == b'note']
labls = [
c
for c in adtl.children
if type(c) is ChunkDescriptor and c.ident == b"labl"
]
ltxts = [
c
for c in adtl.children
if type(c) is ChunkDescriptor and c.ident == b"ltxt"
]
notes = [
c
for c in adtl.children
if type(c) is ChunkDescriptor and c.ident == b"note"
]
return WavCuesReader.read_all(f, cue, labls, ltxts, notes,
fallback_encoding=self.info_encoding)
return WavCuesReader.read_all(
f, cue, labls, ltxts, notes, fallback_encoding=self.info_encoding
)
def _get_sampler_loops(self, f):
sampler_data = self._find_chunk_data(b'smpl', f, default_none=True)
sampler_data = self._find_chunk_data(b"smpl", f, default_none=True)
return WavSmplReader(sampler_data) if sampler_data else None
# FIXME: this should probably be named "iter()"
@@ -225,22 +249,24 @@ class WavInfoReader:
"smpl".
"""
scopes = ('fmt', 'data', 'ixml', 'bext', 'info', 'adm', 'cues',
'dolby', 'smpl')
scopes = ("fmt", "data", "ixml", "bext", "info", "adm", "cues", "dolby", "smpl")
for scope in scopes:
if scope in ['fmt', 'data']:
if scope in ["fmt", "data"]:
attr = self.__getattribute__(scope)
for field in attr._fields:
yield scope, field, attr.__getattribute__(field)
else:
mdict = self.__getattribute__(scope).to_dict(
) if self.__getattribute__(scope) else {}
mdict = (
self.__getattribute__(scope).to_dict()
if self.__getattribute__(scope)
else {}
)
for key in mdict.keys():
yield scope, key, mdict[key]
def __repr__(self):
return 'WavInfoReader({}, {}, {})'.format(self.path,
self.info_encoding,
self.bext_encoding)
return "WavInfoReader({}, {}, {})".format(
self.path, self.info_encoding, self.bext_encoding
)
+37 -35
View File
@@ -13,30 +13,29 @@ class WaveSmplLoop(NamedTuple):
def loop_type_desc(self):
if self.loop_type == 0:
return 'FORWARD'
return "FORWARD"
elif self.loop_type == 1:
return 'FORWARD_BACKWARD'
return "FORWARD_BACKWARD"
elif self.loop_type == 2:
return 'BACKWARD'
return "BACKWARD"
elif 3 <= self.loop_type <= 31:
return 'RESERVED'
return "RESERVED"
else:
return 'VENDOR'
return "VENDOR"
def to_dict(self):
return {
'ident': self.ident,
'loop_type': self.loop_type,
'loop_type_description': self.loop_type_desc(),
'start_samples': self.start,
'end_samples': self.end,
'detune_cents': self.detune_cents,
'repetition_count': self.repetition_count,
"ident": self.ident,
"loop_type": self.loop_type,
"loop_type_description": self.loop_type_desc(),
"start_samples": self.start,
"end_samples": self.end,
"detune_cents": self.detune_cents,
"repetition_count": self.repetition_count,
}
class WavSmplReader:
def __init__(self, smpl_data: bytes):
"""
Read sampler metadata from smpl chunk.
@@ -47,8 +46,7 @@ class WavSmplReader:
header_size = struct.calcsize(header_field_fmt)
loop_size = struct.calcsize(loop_field_fmt)
unpacked_data = struct.unpack(header_field_fmt,
smpl_data[0:header_size])
unpacked_data = struct.unpack(header_field_fmt, smpl_data[0:header_size])
#: The MIDI Manufacturer's Association code for the sampler
#: manufactuer, or 0 if not specific.
@@ -80,35 +78,39 @@ class WavSmplReader:
#: List of loops in the file.
self.sample_loops: List[WaveSmplLoop] = []
loop_buffer = smpl_data[header_size:
header_size + loop_size * loop_count]
loop_buffer = smpl_data[header_size : header_size + loop_size * loop_count]
for unpacked_loop in struct.iter_unpack(loop_field_fmt, loop_buffer):
self.sample_loops.append(WaveSmplLoop(
ident=unpacked_loop[0],
loop_type=unpacked_loop[1],
start=unpacked_loop[2],
end=unpacked_loop[3],
detune_cents=unpacked_loop[4],
repetition_count=unpacked_loop[5]))
self.sample_loops.append(
WaveSmplLoop(
ident=unpacked_loop[0],
loop_type=unpacked_loop[1],
start=unpacked_loop[2],
end=unpacked_loop[3],
detune_cents=unpacked_loop[4],
repetition_count=unpacked_loop[5],
)
)
#: Sampler-specific user data.
self.sampler_udata: bytes | None = None
if sampler_udata_length > 0:
self.sampler_udata = smpl_data[
header_size + loop_size * loop_count:
header_size + loop_size * loop_count + sampler_udata_length]
header_size + loop_size * loop_count : header_size
+ loop_size * loop_count
+ sampler_udata_length
]
def to_dict(self):
return {
'manufactuer': self.manufacturer,
'product': self.product,
'sample_period_ns': self.sample_period_ns,
'midi_note': self.midi_note,
'midi_pitch_detune_cents': self.midi_pitch_detune_cents,
'smpte_format': self.smpte_format,
'smpte_offset': "%02i:%02i:%02i:%02i" % self.smpte_offset,
'loops': [x.to_dict() for x in self.sample_loops],
'sampler_user_data': self.sampler_udata,
"manufactuer": self.manufacturer,
"product": self.product,
"sample_period_ns": self.sample_period_ns,
"midi_note": self.midi_note,
"midi_pitch_detune_cents": self.midi_pitch_detune_cents,
"smpte_format": self.smpte_format,
"smpte_offset": "%02i:%02i:%02i:%02i" % self.smpte_offset,
"loops": [x.to_dict() for x in self.sample_loops],
"sampler_user_data": self.sampler_udata,
}
+10 -17
View File
@@ -1,6 +1,4 @@
"""
"""
""" """
from optparse import OptionParser, OptionGroup
import sys
@@ -9,24 +7,19 @@ import sys
def main():
parser = OptionParser()
parser.usage = ("wavfind [--scene=SCENE] [--take=TAKE] [--desc=DESC] "
"<PATH> +")
parser.usage = "wavfind [--scene=SCENE] [--take=TAKE] [--desc=DESC] <PATH> +"
primaries = OptionGroup(parser, title="Search Predicates",
description="Argument values can be globs, "
"and are logically-AND'ed.")
primaries = OptionGroup(
parser,
title="Search Predicates",
description="Argument values can be globs, and are logically-AND'ed.",
)
primaries.add_option("--scene",
help='Search for this scene',
metavar='SCENE')
primaries.add_option("--scene", help="Search for this scene", metavar="SCENE")
primaries.add_option("--take",
help='Search for this take',
metavar='TAKE')
primaries.add_option("--take", help="Search for this take", metavar="TAKE")
primaries.add_option("--desc",
help='Search descriptions',
metavar='DESC')
primaries.add_option("--desc", help="Search descriptions", metavar="DESC")
(options, args) = parser.parse_args(sys.argv)
+1 -3
View File
@@ -19,7 +19,7 @@ from contextlib import contextmanager
# @property
# def arch_path(self):
# return os.path.join(self.base_path, self.test_file_arch_name)
# return os.path.join(self.base_path, self.test_file_arch_name)
# @contextmanager
# def open(self, name) -> Generator[BytesIO]:
@@ -40,5 +40,3 @@ from contextlib import contextmanager
# p = os.path.join(root, name)
# zip.write(p)
# os.unlink(p)
+4 -5
View File
@@ -2,8 +2,8 @@ from unittest import TestCase
import wavinfo
class TestADMWave(TestCase):
class TestADMWave(TestCase):
def setUp(self) -> None:
self.protools_adm_wav = "tests/test_files/protools/Test_ADM_ProTools.wav"
return super().setUp()
@@ -24,7 +24,7 @@ class TestADMWave(TestCase):
assert adm is not None
dict = adm.to_dict()
self.assertIsNotNone(dict)
def test_programme(self):
info = wavinfo.WavInfoReader(self.protools_adm_wav)
adm = info.adm
@@ -32,8 +32,8 @@ class TestADMWave(TestCase):
pdict = adm.programme()
self.assertIn("programme_id", pdict.keys())
self.assertIn("programme_name", pdict.keys())
self.assertEqual(pdict['programme_id'], 'APR_1001')
self.assertEqual(pdict['programme_name'], 'Atmos_Master')
self.assertEqual(pdict["programme_id"], "APR_1001")
self.assertEqual(pdict["programme_name"], "Atmos_Master")
self.assertIn("contents", pdict.keys())
self.assertEqual(len(pdict["contents"]), 3)
@@ -51,4 +51,3 @@ class TestADMWave(TestCase):
t10 = adm.track_info(10)
self.assertTrue("content_name" in t10.keys())
self.assertEqual("Dialog", t10["content_name"])
+14 -15
View File
@@ -3,9 +3,10 @@ from glob import glob
import wavinfo
class TestCue(TestCase):
def setUp(self) -> None:
self.test_files = glob("tests/test_files/cue_chunks/*.wav")
self.test_files = glob("tests/test_files/cue_chunks/*.wav")
return super().setUp()
def test_enumerate(self):
@@ -14,7 +15,7 @@ class TestCue(TestCase):
self.assertIsNotNone(w1.cues)
assert w1.cues is not None
vals = list(w1.cues.each_cue())
self.assertEqual(vals, [(1,29616),(2,74592),(3,121200)])
self.assertEqual(vals, [(1, 29616), (2, 74592), (3, 121200)])
def test_labels_notes(self):
file = "tests/test_files/cue_chunks/izotoperx_cues_test.wav"
@@ -23,7 +24,7 @@ class TestCue(TestCase):
assert w1.cues is not None
for name, _ in w1.cues.each_cue():
self.assertIn(name,[1,2,3])
self.assertIn(name, [1, 2, 3])
label, note = w1.cues.label_and_note(name)
if name == 1:
self.assertEqual("Marker 1", label)
@@ -35,18 +36,20 @@ class TestCue(TestCase):
self.assertIsNotNone(w1.cues)
assert w1.cues is not None
self.assertEqual(w1.cues.range(3), 10000)
self.assertEqual(w1.cues.range(3), 10000)
def test_encoding_fallback(self):
"""
Added this after I noticed that iZotope RX seems to just encode "notes"
as utf-8 without bothering to dump this info into the ltxt or
as utf-8 without bothering to dump this info into the ltxt or
specifying an encoding by some other means.
"""
file = "tests/test_files/cue_chunks/izotoperx_cues_test.wav"
w = wavinfo.WavInfoReader(file, info_encoding='utf-8')
expected = ("Лорем ипсум долор сит амет, тимеам вивендум хас ет, "
"цу адолесценс дефинитионес еам.")
w = wavinfo.WavInfoReader(file, info_encoding="utf-8")
expected = (
"Лорем ипсум долор сит амет, тимеам вивендум хас ет, "
"цу адолесценс дефинитионес еам."
)
assert w.cues is not None
note = [n for n in w.cues.notes if n.name == 3]
@@ -56,20 +59,16 @@ class TestCue(TestCase):
def test_label(self):
file = "tests/test_files/cue_chunks/izotoperx_cues_test.wav"
w = wavinfo.WavInfoReader(file)
self.assertIsNotNone(w.cues)
assert w.cues is not None
assert w.cues is not None
self.assertEqual(len(w.cues.labels), 3)
for label in w.cues.labels:
self.assertIn(label.name, [1,2,3])
self.assertIn(label.name, [1, 2, 3])
if label.name == 1:
self.assertEqual(label.text, "Marker 1")
elif label.name == 2:
self.assertEqual(label.text, "Marker 2")
elif label.name == 3:
self.assertEqual(label.text, "Marker 3")
+19 -18
View File
@@ -3,6 +3,7 @@ from unittest import TestCase
import wavinfo
from wavinfo.wave_dbmd_reader import SegmentType, DolbyDigitalPlusMetadata
class TestDolby(TestCase):
def setUp(self):
self.test_file = "tests/test_files/protools/Test_ADM_ProTools.wav"
@@ -10,19 +11,17 @@ class TestDolby(TestCase):
def test_version(self):
t1 = wavinfo.WavInfoReader(self.test_file)
d = t1.dolby
assert d is not None
self.assertEqual((1,0,0,6), d.version)
self.assertEqual((1, 0, 0, 6), d.version)
def test_segments(self):
t1 = wavinfo.WavInfoReader(self.test_file)
d = t1.dolby
assert d is not None
ddp = [x for x in d.segment_list \
if x[0] == SegmentType.DolbyDigitalPlus]
atmos = [x for x in d.segment_list \
if x[0] == SegmentType.DolbyAtmos]
ddp = [x for x in d.segment_list if x[0] == SegmentType.DolbyDigitalPlus]
atmos = [x for x in d.segment_list if x[0] == SegmentType.DolbyAtmos]
self.assertEqual(len(ddp), 1)
self.assertEqual(len(atmos), 1)
@@ -40,21 +39,23 @@ class TestDolby(TestCase):
d = t1.dolby
assert d is not None
ddp = d.dolby_digital_plus()
self.assertEqual(len(ddp), 1,
("Failed to find exactly one Dolby Digital Plus "
"metadata segment")
)
self.assertEqual(
len(ddp),
1,
("Failed to find exactly one Dolby Digital Plus metadata segment"),
)
self.assertTrue(
ddp[0].audio_coding_mode,
DolbyDigitalPlusMetadata.AudioCodingMode.CH_ORD_3_2,
)
self.assertTrue(ddp[0].lfe_on)
self.assertTrue( ddp[0].audio_coding_mode,
DolbyDigitalPlusMetadata.AudioCodingMode.CH_ORD_3_2 )
self.assertTrue( ddp[0].lfe_on)
def test_atmos(self):
t1 = wavinfo.WavInfoReader(self.test_file)
d = t1.dolby
assert d is not None
atmos = d.dolby_atmos()
self.assertEqual(len(atmos), 1,
"Failed to find exactly one Atmos metadata segment")
self.assertEqual(
len(atmos), 1, "Failed to find exactly one Atmos metadata segment"
)
+9 -9
View File
@@ -7,10 +7,10 @@ from wavinfo.__main__ import main
import sys
import glob
class MainTest(unittest.TestCase):
def test_empty_argv(self):
with patch.object(sys, 'argv', []):
def test_empty_argv(self):
with patch.object(sys, "argv", []):
try:
main()
except:
@@ -18,17 +18,17 @@ class MainTest(unittest.TestCase):
def test_a_file(self):
for path in glob.glob("tests/test_files/**/*.wav"):
with patch.object(sys, 'argv', ["TEST", path]):
with patch.object(sys, "argv", ["TEST", path]):
try:
main()
except:
self.fail("main() throwing an exception")
self.fail("main() throwing an exception")
def test_ixml(self):
with patch.object(sys, 'argv',
['TEST', '--ixml',
'tests/test_files/sounddevices/A101_1.WAV']):
with patch.object(
sys, "argv", ["TEST", "--ixml", "tests/test_files/sounddevices/A101_1.WAV"]
):
try:
main()
except:
self.fail("main() throwing an exception")
self.fail("main() throwing an exception")
+1 -3
View File
@@ -9,17 +9,15 @@ from unittest import TestCase
import wavinfo
class TestRf64(TestCase):
class TestRf64(TestCase):
def setUp(self) -> None:
return super().setUp()
def test_open(self):
for path in glob("tests/test_files/rf64/*.wav.gz"):
gz = gzip.open(path)
wav_info = wavinfo.WavInfoReader(gz)
self.assertIsNotNone(wav_info)
# self.assertIsNotNone(wav_info.bext)
+1
View File
@@ -3,6 +3,7 @@ from glob import glob
import wavinfo
class TestSmpl(TestCase):
def setUp(self) -> None:
self.test_files = glob("tests/test_files/smpl/*.wav")
+11 -11
View File
@@ -3,35 +3,35 @@ import wavinfo
import glob
class TestWalk(unittest.TestCase):
def test_walk_metadata(self):
test_file = 'tests/test_files/protools/PT A101_4.A1.wav'
test_file = "tests/test_files/protools/PT A101_4.A1.wav"
info = wavinfo.WavInfoReader(test_file)
tested_data , tested_format = False, False
tested_data, tested_format = False, False
for scope, key, value in info.walk():
if scope == 'fmt':
if key == 'channel_count':
if scope == "fmt":
if key == "channel_count":
tested_format = True
self.assertEqual(value, 2)
if scope == 'data':
if key == 'frame_count':
if scope == "data":
if key == "frame_count":
tested_data = True
self.assertEqual(value, 144140)
self.assertTrue(tested_data and tested_format)
def test_walk_all(self):
for file in glob.glob('tests/test_files/**/*.wav'):
for file in glob.glob("tests/test_files/**/*.wav"):
info = wavinfo.WavInfoReader(file)
try:
for _, _, _ in info.walk():
pass
pass
except:
self.fail(f"Failed to walk metadata in file {file}")
if __name__ == '__main__':
if __name__ == "__main__":
unittest.main()
+113 -92
View File
@@ -13,9 +13,10 @@ class TestWaveInfo(TestCase):
def test_sanity(self):
for wav_file in all_files():
info = wavinfo.WavInfoReader(wav_file)
self.assertEqual(info.__repr__(),
'WavInfoReader({}, latin_1, ascii)'
.format(os.path.abspath(wav_file)))
self.assertEqual(
info.__repr__(),
"WavInfoReader({}, latin_1, ascii)".format(os.path.abspath(wav_file)),
)
self.assertIsNotNone(info)
def test_fmt_against_ffprobe(self):
@@ -26,31 +27,39 @@ class TestWaveInfo(TestCase):
assert info.fmt is not None
assert ffprobe_info is not None
self.assertEqual(info.fmt.channel_count,
ffprobe_info['streams'][0]['channels'])
self.assertEqual(info.fmt.sample_rate,
int(ffprobe_info['streams'][0]['sample_rate']))
self.assertEqual(info.fmt.bits_per_sample,
int(ffprobe_info['streams'][0]['bits_per_sample']
))
self.assertEqual(
info.fmt.channel_count, ffprobe_info["streams"][0]["channels"]
)
self.assertEqual(
info.fmt.sample_rate, int(ffprobe_info["streams"][0]["sample_rate"])
)
self.assertEqual(
info.fmt.bits_per_sample,
int(ffprobe_info["streams"][0]["bits_per_sample"]),
)
if info.fmt.audio_format == 1:
self.assertTrue(ffprobe_info['streams'][0]['codec_name']\
.startswith('pcm'))
streams = ffprobe_info['streams'][0]
byte_rate = int(streams['sample_rate']) * \
streams['channels'] * \
int(streams['bits_per_sample']) / 8
self.assertTrue(
ffprobe_info["streams"][0]["codec_name"].startswith("pcm")
)
streams = ffprobe_info["streams"][0]
byte_rate = (
int(streams["sample_rate"])
* streams["channels"]
* int(streams["bits_per_sample"])
/ 8
)
self.assertEqual(info.fmt.byte_rate, byte_rate)
def test_data_against_ffprobe(self):
for wav_file in all_files():
info = wavinfo.WavInfoReader(wav_file)
ffprobe_info = cast(Dict[str,Any], ffprobe(wav_file))
ffprobe_info = cast(Dict[str, Any], ffprobe(wav_file))
assert ffprobe_info is not None
assert info.data is not None
self.assertEqual(info.data.frame_count,
int(ffprobe_info['streams'][0]['duration_ts']))
self.assertEqual(
info.data.frame_count, int(ffprobe_info["streams"][0]["duration_ts"])
)
def test_bext_against_ffprobe(self):
for wav_file in all_files():
@@ -59,65 +68,78 @@ class TestWaveInfo(TestCase):
assert ffprobe_info is not None
if info.bext:
if 'comment' in ffprobe_info['format']['tags']:
self.assertEqual(info.bext.description,
ffprobe_info['format']['tags']\
['comment'])
if "comment" in ffprobe_info["format"]["tags"]:
self.assertEqual(
info.bext.description, ffprobe_info["format"]["tags"]["comment"]
)
else:
self.assertEqual(info.bext.description, '')
self.assertEqual(info.bext.description, "")
if 'encoded_by' in ffprobe_info['format']['tags']:
self.assertEqual(info.bext.originator,
ffprobe_info['format']['tags']\
['encoded_by'])
if "encoded_by" in ffprobe_info["format"]["tags"]:
self.assertEqual(
info.bext.originator,
ffprobe_info["format"]["tags"]["encoded_by"],
)
else:
self.assertEqual(info.bext.originator, '')
self.assertEqual(info.bext.originator, "")
if 'originator_reference' in ffprobe_info['format']['tags']:
self.assertEqual(info.bext.originator_ref,
ffprobe_info['format']['tags']\
['originator_reference'])
if "originator_reference" in ffprobe_info["format"]["tags"]:
self.assertEqual(
info.bext.originator_ref,
ffprobe_info["format"]["tags"]["originator_reference"],
)
else:
self.assertEqual(info.bext.originator_ref, '')
self.assertEqual(info.bext.originator_ref, "")
# these don't always reflect the bext info
# self.assertEqual(info.bext.originator_date,
# self.assertEqual(info.bext.originator_date,
# ffprobe_info['format']['tags']['date'])
# self.assertEqual(info.bext.originator_time,
# self.assertEqual(info.bext.originator_time,
# ffprobe_info['format']['tags']['creation_time'])
self.assertEqual(info.bext.time_reference,
int(ffprobe_info['format']['tags']\
['time_reference']))
self.assertEqual(
info.bext.time_reference,
int(ffprobe_info["format"]["tags"]["time_reference"]),
)
if 'coding_history' in ffprobe_info['format']['tags']:
self.assertEqual(info.bext.coding_history,
ffprobe_info['format']['tags']\
['coding_history'])
if "coding_history" in ffprobe_info["format"]["tags"]:
self.assertEqual(
info.bext.coding_history,
ffprobe_info["format"]["tags"]["coding_history"],
)
else:
self.assertEqual(info.bext.coding_history, '')
self.assertEqual(info.bext.coding_history, "")
def test_ixml(self):
expected = {'A101_4.WAV': {'project': 'BMH',
'scene': 'A101', 'take': '4',
'tape': '18Y12M31',
'family_uid':
'USSDVGR1112089007124015008231000'},
'A101_3.WAV': {'project': 'BMH',
'scene': 'A101', 'take': '3',
'tape': '18Y12M31',
'family_uid':
'USSDVGR1112089007124014008228300'},
'A101_2.WAV': {'project': 'BMH',
'scene': 'A101', 'take': '2',
'tape': '18Y12M31',
'family_uid':
'USSDVGR1112089007124004008218600'},
'A101_1.WAV': {'project': 'BMH',
'scene': 'A101', 'take': '1',
'tape': '18Y12M31',
'family_uid':
'USSDVGR1112089007124001008206300'},
}
expected = {
"A101_4.WAV": {
"project": "BMH",
"scene": "A101",
"take": "4",
"tape": "18Y12M31",
"family_uid": "USSDVGR1112089007124015008231000",
},
"A101_3.WAV": {
"project": "BMH",
"scene": "A101",
"take": "3",
"tape": "18Y12M31",
"family_uid": "USSDVGR1112089007124014008228300",
},
"A101_2.WAV": {
"project": "BMH",
"scene": "A101",
"take": "2",
"tape": "18Y12M31",
"family_uid": "USSDVGR1112089007124004008218600",
},
"A101_1.WAV": {
"project": "BMH",
"scene": "A101",
"take": "1",
"tape": "18Y12M31",
"family_uid": "USSDVGR1112089007124001008206300",
},
}
for wav_file in all_files():
basename = os.path.basename(wav_file)
@@ -125,29 +147,31 @@ class TestWaveInfo(TestCase):
info = wavinfo.WavInfoReader(wav_file)
e = expected[basename]
self.assertIsNotNone(info.ixml)
assert info.ixml is not None
self.assertEqual(e['project'], info.ixml.project)
self.assertEqual(e['scene'], info.ixml.scene)
self.assertEqual(e['take'], info.ixml.take)
self.assertEqual(e['tape'], info.ixml.tape)
self.assertEqual(e['family_uid'], info.ixml.family_uid)
assert info.ixml is not None
self.assertEqual(e["project"], info.ixml.project)
self.assertEqual(e["scene"], info.ixml.scene)
self.assertEqual(e["take"], info.ixml.take)
self.assertEqual(e["tape"], info.ixml.tape)
self.assertEqual(e["family_uid"], info.ixml.family_uid)
for track in info.ixml.track_list:
self.assertIsNotNone(track.channel_index)
if basename == 'A101_4.WAV' and track.channel_index == '1':
self.assertEqual(track.name, 'MKH516 A')
if basename == "A101_4.WAV" and track.channel_index == "1":
self.assertEqual(track.name, "MKH516 A")
def test_steinberg_ixml(self):
nuendo_files = 'tests/test_files/nuendo/*.wav'
nuendo_files = "tests/test_files/nuendo/*.wav"
for file in glob(nuendo_files):
info = wavinfo.WavInfoReader(file)
assert info.ixml is not None
assert info.ixml is not None
self.assertIsNotNone(info.ixml.steinberg)
assert info.ixml.steinberg is not None
self.assertIsNotNone(info.ixml.steinberg.audio_speaker_arrangement)
self.assertEqual(info.ixml.steinberg.sample_format_size, 3)
self.assertEqual(info.ixml.steinberg.media_company,
"https://github.com/iluvcapra/wavinfo")
self.assertEqual(info.ixml.steinberg.sample_format_size, 3)
self.assertEqual(
info.ixml.steinberg.media_company,
"https://github.com/iluvcapra/wavinfo",
)
self.assertFalse(info.ixml.steinberg.media_drop_frames)
self.assertEqual(info.ixml.steinberg.media_duration, 1200.0)
@@ -159,24 +183,21 @@ class TestWaveInfo(TestCase):
self.assertIsNone(info.ixml.steinberg)
def test_info_metadata(self):
file_with_metadata = \
'tests/test_files/sound_grinder_pro/new_camera bumb 1.wav'
file_with_metadata = "tests/test_files/sound_grinder_pro/new_camera bumb 1.wav"
self.assertTrue(os.path.exists(file_with_metadata))
info = wavinfo.WavInfoReader(file_with_metadata).info
assert info is not None
self.assertEqual(info.title, 'camera bumb 1')
self.assertEqual(info.artist, 'Jamie Hardt')
self.assertEqual(info.copyright, '© 2010 Jamie Hardt')
self.assertEqual(info.product, 'Test Sounds') # album
self.assertEqual(info.title, "camera bumb 1")
self.assertEqual(info.artist, "Jamie Hardt")
self.assertEqual(info.copyright, "© 2010 Jamie Hardt")
self.assertEqual(info.product, "Test Sounds") # album
self.assertEqual(info.album, info.product)
self.assertEqual(info.comment, 'Comments')
self.assertEqual(info.software, 'Sound Grinder Pro')
self.assertEqual(info.created_date, '2010-12-28')
self.assertEqual(info.engineer, 'JPH')
self.assertEqual(info.keywords,
'Sound Effect, movement, microphone, bump')
self.assertEqual(info.title, 'camera bumb 1')
self.assertEqual(info.comment, "Comments")
self.assertEqual(info.software, "Sound Grinder Pro")
self.assertEqual(info.created_date, "2010-12-28")
self.assertEqual(info.engineer, "JPH")
self.assertEqual(info.keywords, "Sound Effect, movement, microphone, bump")
self.assertEqual(info.title, "camera bumb 1")
self.assertEqual(type(info.to_dict()), dict)
self.assertEqual(type(info.__repr__()), str)
+9 -11
View File
@@ -4,35 +4,33 @@ import subprocess
from subprocess import PIPE
import json
FFPROBE = 'ffprobe'
FFPROBE = "ffprobe"
def ffprobe(path):
arguments = [FFPROBE, "-of", "json",
"-show_format", "-show_streams", path]
if int(sys.version[0]) < 3:
arguments = [FFPROBE, "-of", "json", "-show_format", "-show_streams", path]
if int(sys.version[0]) < 3:
process = subprocess.Popen(arguments, stdout=PIPE)
process.wait()
if process.returncode == 0:
output = process.communicate()[0]
if output:
output_str = output.decode('utf-8')
output_str = output.decode("utf-8")
return json.loads(output_str)
else:
return None
else:
process = subprocess.run(arguments, stdin=None,
stdout=PIPE, stderr=PIPE)
else:
process = subprocess.run(arguments, stdin=None, stdout=PIPE, stderr=PIPE)
if process.returncode == 0:
output_str = process.stdout.decode('utf-8')
output_str = process.stdout.decode("utf-8")
return json.loads(output_str)
else:
return None
def all_files():
for dirpath, _, filenames in os.walk('tests/test_files'):
for dirpath, _, filenames in os.walk("tests/test_files"):
for filename in filenames:
_, ext = os.path.splitext(filename)
if ext in ['.wav', '.WAV']:
if ext in [".wav", ".WAV"]:
yield os.path.join(dirpath, filename)