Compare commits

...

55 Commits

Author SHA1 Message Date
jamie 739e574a10 Removed dead line from pyproject.py 2026-09-17 14:28:31 -07:00
jamie b3d5377f7b Ruff: Fixed FA100 2026-09-17 14:28:14 -07:00
jamie f3547c8bc5 Ruff: Fixed UP022 2026-09-17 14:13:04 -07:00
jamie 5d481e2960 Ruff: Fixed SIM115 2026-09-17 14:10:25 -07:00
jamie 94aec13d00 Ruff: Fixed C408 and C417 2026-09-17 14:08:22 -07:00
jamie 0060212083 Ruff: Fixed C408 2026-09-17 14:00:33 -07:00
Jamie Hardt 90d15c78c0 Update pyproject.toml
Nudged version number
2026-09-17 11:34:16 -07:00
Jamie Hardt ae79a32d4a Merge pull request #42 from iluvcapra/bug-41-audiostreamformat-refs
Added tests to audioStreamFormat parsing. Also some code style changes.
2026-09-17 11:33:33 -07:00
jamie 50dbc35595 Ruff 2026-09-17 11:28:36 -07:00
jamie 6f5cb795a3 Ruff 2026-09-17 11:15:09 -07:00
jamie 7d5bb29a42 Tweaked pyproject for ruff settings 2026-09-17 10:43:17 -07:00
jamie f9de312610 Added test file, Ruff pass 2026-09-17 10:30:32 -07:00
jamie e6320e84c6 Ruffification and fixed bug 2026-09-17 10:19:11 -07:00
jamie 13d9cd4914 Added tests to audioStreamFormat parsing
ASF's may have zero or one channelformatrefs or packformatrefs. We were
expecting one.
2026-09-17 09:57:52 -07:00
Jamie Hardt 742c0b91fb Update CONTRIBUTING.md
Updated agents policy
2026-05-03 12:18:23 -07:00
Jamie Hardt 8e56a1aeb2 Update python-package.yml 2026-04-03 10:41:38 -07:00
Jamie Hardt 6bc3636814 Update .readthedocs.yaml
Updating RTD build to `ubuntu-lts-latest`
2026-03-30 11:57:04 -07:00
Jamie Hardt e12ae4519e Update contribution guidelines regarding ML systems 2026-02-15 15:48:07 -08:00
Jamie Hardt 8926274c50 Update CONTRIBUTING.md
No LLM/ML contributions
2026-02-14 06:30:13 -08:00
jamie 95ba34187a Docs 2025-10-09 23:22:26 -07:00
jamie 4d2dfcd370 Docs 2025-10-09 23:21:43 -07:00
jamie 717b6a4117 Docs 2025-10-09 23:19:05 -07:00
jamie 7393afac95 Docs 2025-10-09 23:17:46 -07:00
jamie e9450cd65a Docs 2025-10-09 23:16:11 -07:00
jamie 51b1a2e8b4 Docs 2025-10-09 23:15:19 -07:00
jamie 53303232b4 Docs 2025-10-09 23:11:00 -07:00
jamie 3519a0251f Docs 2025-10-09 23:09:51 -07:00
jamie 79ec1649c4 Docs 2025-10-09 23:09:04 -07:00
jamie 7c3f4c9b5e Docs 2025-10-09 23:08:32 -07:00
jamie 2e8224cb3c Docs fix 2025-10-09 23:07:39 -07:00
jamie 8ef4186b4c Merge branch 'master' of https://github.com/iluvcapra/wavinfo 2025-10-09 23:02:32 -07:00
jamie cd5346c1cb Docs fixups 2025-10-09 23:02:15 -07:00
Jamie Hardt 925bf4f8a6 Update pythonpublish.yml 2025-10-09 22:55:13 -07:00
Jamie Hardt e98ba0bf07 Update pythonpublish.yml 2025-10-09 22:48:40 -07:00
jamie afca634dc3 modernized build action 2025-10-09 22:44:58 -07:00
Jamie Hardt df9ae0f4d6 Update pyproject.toml 2025-10-09 22:34:45 -07:00
Jamie Hardt 2f23bcb982 Merge pull request #40 from iluvcapra/maint-uv
Build modernization
2025-10-09 22:33:55 -07:00
jamie 5e641b0963 Removing .flake8 file 2025-10-09 22:32:10 -07:00
jamie 1b57ad0fac Typo 2025-10-09 22:26:04 -07:00
jamie c7a34e0064 Added 3.14 to test matrix and dropped 3.8 2025-10-09 22:24:45 -07:00
jamie 6b788484da Typo 2025-10-09 22:22:12 -07:00
jamie 76905f1a40 Updated name of lint workflow 2025-10-09 22:21:11 -07:00
jamie 9ac06040a2 Making changes to the workflows 2025-10-09 22:16:12 -07:00
jamie 1b78f5b821 Reorganized pyproject, nudged version 2025-10-09 22:08:36 -07:00
jamie 03d718b4ad Fixed dumb typo 2025-10-09 22:06:48 -07:00
jamie 61f79760e6 Initial work on uv build system
Moved module into src/ and modernized pyproject.toml
2025-10-09 21:49:33 -07:00
Jamie Hardt afe5ea9ed3 Update pythonpublish.yml
Updated publish action to latest version
2025-09-08 12:49:05 -07:00
Jamie Hardt c1205d52e8 Merge pull request #38 from iluvcapra/maint-no-mastodon
Workflow Spruce-up
2024-11-26 12:00:18 -08:00
Jamie Hardt b82b6b6d43 Fixed publish to Bluesky worksflow 2024-11-26 11:56:51 -08:00
Jamie Hardt 8ef664266f Updated flake8 step to use python 3.13 2024-11-26 10:30:24 -08:00
Jamie Hardt dfb7e34fc7 Update pythonpublish.yml
Updated `checkout` and `setup-python` versions
2024-11-25 18:41:36 -08:00
Jamie Hardt 2ebdefaab5 Update pythonpublish.yml 2024-11-25 18:37:15 -08:00
Jamie Hardt c609e22270 Update pythonpublish.yml 2024-11-25 18:33:53 -08:00
Jamie Hardt ef9c39f1b6 Update pythonpublish.yml
Adding posting to Bluesky
2024-11-25 18:32:19 -08:00
Jamie Hardt cc9d884ea8 Update pythonpublish.yml
Removed Mastodon notification step
2024-11-25 18:07:17 -08:00
41 changed files with 1239 additions and 1148 deletions
-3
View File
@@ -1,3 +0,0 @@
[flake8]
per-file-ignores =
wavinfo/__init__.py: F401
+6 -6
View File
@@ -16,21 +16,21 @@ jobs:
strategy:
fail-fast: false
matrix:
python-version: ["3.8", "3.9", "3.10", "3.11", "3.12", "3.13"]
python-version: ["3.9", "3.10", "3.11", "3.12", "3.13", "3.14"]
steps:
- uses: actions/checkout@v2.5.0
- uses: actions/checkout@v6.0.2
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v4.3.0
uses: actions/setup-python@v6.2.0
with:
python-version: ${{ matrix.python-version }}
- name: Install dependencies
run: |
python -m pip install --upgrade pip
python -m pip install pytest
python -m pip install -e .
python -m pip install --group dev
python -m pip install .
- name: Setup FFmpeg
uses: FedericoCarboni/setup-ffmpeg@v2
uses: federicocarboni/setup-ffmpeg@v3.1
- name: Test with pytest
run: |
pytest
@@ -1,7 +1,7 @@
# This workflow will install Python dependencies, run tests and lint with a variety of Python versions
# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-python
name: Flake8
name: Lint with Ruff
on:
push:
@@ -11,12 +11,11 @@ on:
jobs:
build:
runs-on: ubuntu-latest
strategy:
fail-fast: false
matrix:
python-version: ["3.11"]
python-version: ["3.13", "3.14"]
steps:
- uses: actions/checkout@v2.5.0
@@ -27,14 +26,8 @@ jobs:
- name: Install dependencies
run: |
python -m pip install --upgrade pip
python -m pip install flake8
python -m pip install -e .
- name: Lint with flake8
python -m pip install --group dev
python -m pip install .
- name: Lint with ruff
run: |
# stop the build if there are Python syntax errors or undefined names
flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
# exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide
flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
- name: Lint with flake8
run: |
flake8 wavinfo
ruff check src
+22 -18
View File
@@ -8,29 +8,33 @@ jobs:
deploy:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v1
- uses: actions/checkout@v4.2.2
- name: Set up Python
uses: actions/setup-python@v1
uses: actions/setup-python@v5.3.0
with:
python-version: '3.x'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install setuptools build wheel twine lxml
- name: Build and publish
env:
TWINE_USERNAME: __token__
TWINE_PASSWORD: ${{ secrets.PYPI_APIKEY }}
- name: Setup uv and Handle Its Cache
# You may pin to the exact commit or the version.
# uses: hynek/setup-cached-uv@757bedc3f972eb7227a1aa657651f15a8527c817
uses: hynek/setup-cached-uv@v2.3.0
- name: Build
run: |
python -m build .
twine upload dist/*
- name: Report to Mastodon
uses: cbrgm/mastodon-github-action@v1.0.1
uv build --wheel
- name: Publish to Pypi
uses: pypa/gh-action-pypi-publish@v1.13.0
with:
message: |
I just released a new version of wavinfo, my library for reading WAVE file metadata!
#sounddesign #filmmaking #audio #python
${{ github.server_url }}/${{ github.repository }}
env:
MASTODON_URL: ${{ secrets.MASTODON_URL }}
MASTODON_ACCESS_TOKEN: ${{ secrets.MASTODON_ACCESS_TOKEN }}
password: ${{ secrets.PYPI_APIKEY }}
# - name: Send Bluesky Post
# uses: myConsciousness/bluesky-post@v5
# with:
# text: |
# I've released a new version of wavinfo, my module for
# reading WAVE metadata.
# link-preview-url: ${{ github.server_url }}/${{ github.repository }}
# identifier: ${{ secrets.BLUESKY_APP_USER }}
# password: ${{ secrets.BLUESKY_APP_PASSWORD }}
# service: bsky.social
# retry-count: 1
+6 -8
View File
@@ -7,13 +7,13 @@ version: 2
# Set the version of Python and other tools you might need
build:
os: ubuntu-20.04
os: ubuntu-lts-latest
tools:
python: "3.10"
# You can also specify other tool versions:
# nodejs: "16"
# rust: "1.55"
# golang: "1.17"
python: "3.13"
jobs:
install:
- pip install --upgrade pip
- pip install --group 'doc'
# Build documentation in the docs/ directory with Sphinx
sphinx:
@@ -28,5 +28,3 @@ python:
install:
- method: pip
path: .
extra_requirements:
- doc
+32
View File
@@ -13,3 +13,35 @@ If you discover a bug or would like better support for a feature, please do the
review it as soon as I can. There's a `.devcontainer` available so you can creates commits
on this project in a GitHub codespace.
## Regarding use of Agents
`wavinfo` is an open-source project that is offered free for no commerical gain, and
is developed and maintained for educational and creative reasons.
If you use an agent or LLM to produce code for it you are missing out on the benefits
of contributing to an open-source project, particularly community, collaboration with
other developers and designers, and being able to learn and experiment without the
burden of deadlines or worrying about business cases or profits.
This project is supposed to be fun, do not let machines have fun for you.
We can't prevent you from using LLMs to contribute to this project but we ask you
abide by the following eitiquette when doing so:
* All communication with the maintainers must be written by a human in their own
voice. Never use an LLM to craft thread comments, discussion posts, issues, emails
or other correspondence with other developers or the maintainers.
* PRs must be submitted by a person. Do not allow an agent to submit its own PRs to
this project.
* Especially if you are a new contributor to this project, please submit only one PR
at a time and please restrict the subject matter of the PR to a specific unit,
module or tool. All submissions have to be reviewed and understood by the
maintainers before they can be merged.
Obviously we can't verify if you follow all of these rules but certain telltale
traits of LLM-predicted text or code will raise a flag: lack of brevity in
descriptions or code comments, large amounts of text describing your process or
steps that add little to understanding the changes you've made, use of an
obsequious tone or being excessively accomodating, immediately doing requests
without further discussion or clarifications.
+1 -1
View File
@@ -2,7 +2,7 @@
![GitHub last commit](https://img.shields.io/github/last-commit/iluvcapra/wavinfo) [![Documentation Status](https://readthedocs.org/projects/wavinfo/badge/?version=latest)](https://wavinfo.readthedocs.io/en/latest/?badge=latest) ![](https://img.shields.io/github/license/iluvcapra/wavinfo.svg)
[![Tests](https://github.com/iluvcapra/wavinfo/actions/workflows/python-package.yml/badge.svg)](https://github.com/iluvcapra/wavinfo/actions/workflows/python-package.yml)
[![Flake8](https://github.com/iluvcapra/wavinfo/actions/workflows/python-flake8.yml/badge.svg)](https://github.com/iluvcapra/wavinfo/actions/workflows/python-flake8.yml)
[![Ruff](https://github.com/iluvcapra/wavinfo/actions/workflows/python-ruff.yml/badge.svg)](https://github.com/iluvcapra/wavinfo/actions/workflows/python-ruff.yml)
[![codecov](https://codecov.io/gh/iluvcapra/wavinfo/branch/master/graph/badge.svg?token=9DZQfZENYv)](https://codecov.io/gh/iluvcapra/wavinfo)
# wavinfo
+33 -35
View File
@@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
#
# Configuration file for the Sphinx documentation builder.
#
@@ -12,25 +11,25 @@
# add these directories to sys.path here. If the directory is relative to the
# documentation root, use os.path.abspath to make it absolute, like shown here.
#
import importlib
# import importlib
import os
import sys
sys.path.insert(0, os.path.abspath('../..'))
sys.path.insert(0, os.path.abspath("../../.."))
print(sys.path)
import importlib
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-2024, Jamie Hardt'
author = u'Jamie Hardt'
project = "wavinfo"
copyright = "2018-2025, Jamie Hardt"
author = "Jamie Hardt"
# The short X.Y version
version = "3.1"
version = "4.0"
# The full version, including alpha/beta/rc tags
release = importlib.metadata.version("wavinfo")
release = "4.0.0"
# release = importlib.metadata.version("wavinfo")
# -- General configuration ---------------------------------------------------
@@ -43,34 +42,34 @@ release = importlib.metadata.version("wavinfo")
# 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
@@ -81,7 +80,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
@@ -92,7 +91,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.
@@ -108,7 +107,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 ------------------------------------------------
@@ -117,15 +116,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',
@@ -135,8 +131,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"),
]
@@ -144,10 +139,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 ----------------------------------------------
@@ -156,9 +148,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",
),
]
@@ -177,7 +175,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",
+32 -28
View File
@@ -1,27 +1,26 @@
# https://python-poetry.org/docs/pyproject/
[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"
requires = ["uv_build>=0.8.18,<0.9.0"]
build-backend = "uv_build"
[tool.poetry]
[project]
name = "wavinfo"
version = "3.1.0"
version = "4.0.1"
description = "Probe WAVE files for all metadata"
authors = ["Jamie Hardt <jamiehardt@me.com>"]
authors = [{ name = "Jamie Hardt", email = "jamiehardt@me.com"}]
license = "MIT"
readme = "README.md"
requires-python = ">=3.8"
classifiers = [
'Development Status :: 5 - Production/Stable',
'License :: OSI Approved :: MIT License',
'Topic :: Multimedia',
'Topic :: Multimedia :: Sound/Audio',
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13"
"Programming Language :: Python :: 3.13",
"Programming Language :: Python :: 3.14"
]
homepage = "https://github.com/iluvcapra/wavinfo"
repository = "https://github.com/iluvcapra/wavinfo.git"
@@ -39,29 +38,34 @@ keywords = [
'broadcast'
]
[tool.poetry.extras]
doc = ['sphinx', 'sphinx_rtd_theme']
dependencies = [
"lxml>=6.0.2",
]
[tool.poetry.scripts]
wavinfo = 'wavinfo.__main__:main'
[dependency-groups]
dev = [
"pytest>=8.3.5",
"ruff>=0.16.8",
]
doc = [
"sphinx>=7.1.2",
"sphinx-rtd-theme>=3.0.2",
]
[tool.poetry.dependencies]
python = "^3.8"
lxml = "~= 5.3.0"
sphinx_rtd_theme = {version= '>= 1.1.1', optional=true}
sphinx = {version= '>= 5.3.0', optional=true}
[project.scripts]
wavinfo = "wavinfo:__main__.main"
[tool.pyright]
typeCheckingMode = "basic"
[tool.pylint]
max-line-length = 88
disable = [
"C0103", # (invalid-name)
"C0114", # (missing-module-docstring)
"C0115", # (missing-class-docstring)
"C0116", # (missing-function-docstring)
"R0903", # (too-few-public-methods)
"R0913", # (too-many-arguments)
"W0105", # (pointless-string-statement)
[tool.ruff]
line-length = 88
indent-width = 4
[tool.ruff.lint]
fixable = ['ALL']
ignore = [
'UP031', #Use format specifiers instead of percent format
]
@@ -2,5 +2,7 @@
Probe WAVE Files for iXML, Broadcast-WAVE and other metadata.
"""
from .wave_reader import WavInfoReader
__all__ = ["WavInfoEOFError", "WavInfoReader"]
from .riff_parser import WavInfoEOFError
from .wave_reader import WavInfoReader
+52 -46
View File
@@ -1,16 +1,17 @@
from . import WavInfoReader
from __future__ import annotations
import datetime
from optparse import OptionParser
import sys
import os
import json
from enum import Enum
import importlib.metadata
import json
import os
import sys
from base64 import b64encode
from cmd import Cmd
from enum import Enum
from optparse import OptionParser
from shlex import split
from typing import List, Dict, Union
from . import WavInfoReader
class MyJSONEncoder(json.JSONEncoder):
@@ -18,7 +19,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)
@@ -30,12 +31,16 @@ class MissingDataError(RuntimeError):
class MetaBrowser(Cmd):
prompt = "(wavinfo) "
metadata: Union[List, Dict]
path: List[str] = []
metadata: list | dict
path: list[str]
def preloop(self) -> None:
self.path = []
return super().preloop()
@property
def cwd(self):
root: List | Dict = self.metadata
root: list | dict = self.metadata
for key in self.path:
if isinstance(root, list):
root = root[int(key)]
@@ -50,7 +55,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 +68,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):
@@ -91,10 +96,10 @@ class MetaBrowser(Cmd):
else:
print(f"Index {argv[0]} does not exist")
elif isinstance(self.cwd, dict):
if argv[0] in self.cwd.keys():
if argv[0] in self.cwd:
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 +107,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',
parser.add_option(
"-i",
help="Read metadata with an interactive prompt",
default=False,
action='store_true')
action="store_true",
)
(options, args) = parser.parse_args(sys.argv)
@@ -149,6 +153,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,29 +181,30 @@ 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(
tz=datetime.timezone.utc
).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"]:
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
if len(interactive_dict) > 0:
cli = MetaBrowser()
@@ -1,26 +1,28 @@
from __future__ import annotations
import struct
# from collections import namedtuple
from typing import NamedTuple, Dict
from typing import NamedTuple
from . import riff_parser
class RF64Context(NamedTuple):
sample_count: int
bigchunk_table: Dict[str, int]
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
@@ -34,14 +36,13 @@ def parse_rf64(stream, signature=b'RF64') -> RF64Context:
# 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)
@@ -1,7 +1,9 @@
# from optparse import Option
from __future__ import annotations
import struct
from .rf64_parser import parse_rf64, RF64Context
from typing import NamedTuple, Union, List, Optional
from typing import NamedTuple
from .rf64_parser import RF64Context, parse_rf64
class WavInfoEOFError(EOFError):
@@ -12,14 +14,14 @@ class WavInfoEOFError(EOFError):
class ListChunkDescriptor(NamedTuple):
signature: bytes
children: List[Union['ChunkDescriptor', 'ListChunkDescriptor']]
children: list[ChunkDescriptor | ListChunkDescriptor]
class ChunkDescriptor(NamedTuple):
ident: bytes
start: int
length: int
rf64_context: Optional[RF64Context]
rf64_context: RF64Context | None
def read_data(self, from_stream) -> bytes:
from_stream.seek(self.start)
@@ -48,14 +50,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 +66,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
)
+212
View File
@@ -0,0 +1,212 @@
"""
ADM Reader
"""
from __future__ import annotations
from collections import namedtuple
from io import BytesIO
from struct import calcsize, unpack, unpack_from
from lxml import etree as ET
ChannelEntry = namedtuple("ChannelEntry", "track_index uid track_ref pack_ref")
class WavADMReader:
"""
Reads XML data from an EBU ADM (Audio Definiton Model) WAV File.
"""
def __init__(self, axml_data: bytes, chna_data: bytes):
header_fmt = "<HH"
uid_fmt = "<H12s14s11sx"
#: An :mod:`lxml.etree` of the ADM XML document
self.axml = ET.parse(BytesIO(axml_data))
_, uid_count = unpack(header_fmt, chna_data[0:4])
self.channel_uids = []
offset = calcsize(header_fmt)
for _ in range(uid_count):
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"),
)
)
offset += calcsize(uid_fmt)
def xml_str(self) -> str:
"""ADM XML as a string"""
return ET.tostring(self.axml).decode("utf-8")
def programme(self) -> dict:
"""
Read the ADM `audioProgramme` data structure and some of its reference
properties.
"""
ret_dict = {}
nsmap = self.axml.getroot().nsmap
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"] = []
for content_ref in program.findall("audioContentIDRef", namespaces=nsmap):
content_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"] = []
for object_ref in content.findall("audioObjectIDRef", namespaces=nsmap):
object_dict = {}
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
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)
ret_dict["contents"].append(content_dict)
return ret_dict
def track_info(self, index) -> dict | None:
"""
Information about a track in the WAV file.
:param index: index of audio track (indexed from zero)
:returns: a dictionary with *content_name*, *content_id*,
*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
)
if channel_info is None:
return None
ret_dict = {}
nsmap = self.axml.getroot().nsmap
afext = self.axml.find(".//audioFormatExtended", namespaces=nsmap)
trackformat_elem = afext.find(
"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 is not None:
channelformat_id = channelformatref_elem.text
else:
channelformat_id = None
packformatref_elem = afext.find(
("audioStreamFormat[@audioStreamFormatID='%s']/audioPackFormatIDRef")
% stream_id,
namespaces=nsmap,
)
if packformatref_elem is not None:
packformat_id = packformatref_elem.text
else:
packformat_id = None
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
)
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
)
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
)
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
def to_dict(self) -> dict: # FIXME should be "asdict"
"""
Get ADM metadata as a dictionary.
"""
def make_entry(channel_uid_rec):
rd = channel_uid_rec._asdict()
rd.update(self.track_info(channel_uid_rec.track_index))
return rd
return {
'channel_entries': [make_entry(z) for z in self.channel_uids],
'programme': self.programme(),
}
@@ -1,7 +1,8 @@
import struct
# from .umid_parser import UMIDParser
from __future__ import annotations
from typing import Optional
import struct
# from .umid_parser import UMIDParser
class WavBextReader:
@@ -13,16 +14,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
@@ -56,22 +59,22 @@ class WavBextReader:
#: SMPTE 330M UMID of this audio file, 64 bytes are allocated though
#: the UMID may only be 32 bytes long.
self.umid: Optional[bytes] = None
self.umid: bytes | None = None
#: EBU R128 Integrated loudness, in LUFS.
self.loudness_value: Optional[float] = None
self.loudness_value: float | None = None
#: EBU R128 Loudness range, in LUFS.
self.loudness_range: Optional[float] = None
self.loudness_range: float | None = None
#: True peak level, in dBFS TP
self.max_true_peak: Optional[float] = None
self.max_true_peak: float | None = None
#: EBU R128 Maximum momentary loudness, in LUFS
self.max_momentary_loudness: Optional[float] = None
self.max_momentary_loudness: float | None = None
#: EBU R128 Maximum short-term loudness, in LUFS.
self.max_shortterm_loudness: Optional[float] = None
self.max_shortterm_loudness: float | None = None
if self.version > 0:
self.umid = unpacked[7]
@@ -91,18 +94,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,
}
@@ -7,11 +7,14 @@ 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 __future__ import annotations
from dataclasses import dataclass
from struct import calcsize, unpack
from typing import Any, Generator, NamedTuple
from .riff_parser import ChunkDescriptor
#: Country Codes used in the RIFF standard to resolve locale. These codes
#: appear in CSET and LTXT metadata.
@@ -100,6 +103,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 +122,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 +162,7 @@ class RangeLabel(NamedTuple):
"""
A ``ltxt`` structure.
"""
name: int
length: int
purpose: str
@@ -165,38 +178,47 @@ class RangeLabel(NamedTuple):
parsed = unpack(leader_struct_fmt, data[0 : calcsize(leader_struct_fmt)])
text_data = data[calcsize(leader_struct_fmt) :]
purpose_str = parsed[2].decode('ascii')
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=purpose_str,
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]
cues: list[CueEntry]
#: Every ``labl`` in the file
labels: List[LabelEntry]
labels: list[LabelEntry]
#: Every ``ltxt`` in the file
ranges: List[RangeLabel]
ranges: list[RangeLabel]
#: Every ``note`` in the file
notes: List[NoteEntry]
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: ChunkDescriptor | None,
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)
@@ -212,28 +234,26 @@ class WavCuesReader:
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]:
def each_cue(self) -> Generator[tuple[int, int], None, None]:
"""
Iterate through each cue.
@@ -242,8 +262,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[str | None, str | None]:
"""
Get the label and note (extended comment) for a cue.
@@ -251,36 +270,35 @@ 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]:
def range(self, cue_ident: int) -> int | None:
"""
Get the length of the time range for a cue, if it has one.
: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()
def to_dict(self) -> dict[str, Any]:
retval = {}
for n, t in self.each_cue():
retval[n] = dict()
retval[n]['frame'] = t
retval[n] = {}
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
@@ -7,18 +7,20 @@ Unless otherwise stated, all § references here are to
.. _EBU Tech 3285 Supplement 6: https://tech.ebu.ch/docs/tech/tech3285s6.pdf
"""
from enum import IntEnum, Enum
from struct import unpack
from dataclasses import dataclass, asdict
from typing import List, Tuple, Any, Union
from __future__ import annotations
from dataclasses import asdict, dataclass
from enum import Enum, IntEnum
from io import BytesIO
from struct import unpack
from typing import Any
class SegmentType(IntEnum):
"""
Metadata segment type.
"""
EndMarker = 0x0
DolbyE = 0x1
# Reserved2 = 0x2
@@ -29,7 +31,7 @@ class SegmentType(IntEnum):
DolbyDigitalPlus = 0x7
AudioInfo = 0x8
DolbyAtmos = 0x9
DolbyAtmosSupplemental = 0xa
DolbyAtmosSupplemental = 0xA
@classmethod
def _missing_(cls, val):
@@ -82,6 +84,7 @@ class DolbyDigitalPlusMetadata:
"""
Dolby surround endcoding mode.
"""
RESERVED = 0b11
IN_USE = 0b10
NOT_IN_USE = 0b01
@@ -126,6 +129,7 @@ class DolbyDigitalPlusMetadata:
Dolby Digital Plus `acmod` field
§ 4.3.2.3
"""
RESERVED = 0b000
CH_ORD_1_0 = 0b001
"Mono"
@@ -163,6 +167,7 @@ class DolbyDigitalPlusMetadata:
Dolby Digital Plus `surmixlev` field
§ 4.3.3.2
"""
DOWN_3DB = 0b00
DOWN_6DB = 0b01
MUTE = 0b10
@@ -174,24 +179,22 @@ 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 +206,7 @@ class DolbyDigitalPlusMetadata:
should downmix.
§ 4.3.8.1
"""
NOT_INDICATED = 0b00
PRO_LOGIC = 0b01
STEREO = 0b10
@@ -213,6 +217,7 @@ class DolbyDigitalPlusMetadata:
Dolby Surround-EX mode.
`dsurexmod` § 4.3.9.1
"""
NOT_INDICATED = 0b00
NOT_SEX = 0b01
SEX = 0b10
@@ -222,6 +227,7 @@ class DolbyDigitalPlusMetadata:
"""
`dheadphonmod` § 4.3.9.2
"""
NOT_INDICATED = 0b00
NOT_DOLBY_HEADPHONE = 0b01
DOLBY_HEADPHONE = 0b10
@@ -246,6 +252,7 @@ class DolbyDigitalPlusMetadata:
`compr1` RF compression profile
§ 4.3.10 (fig 42)
"""
NONE = 0
FILM_STANDARD = 1
FILM_LIGHT = 2
@@ -334,16 +341,20 @@ class DolbyDigitalPlusMetadata:
@staticmethod
def load(buffer: bytes):
assert len(buffer) == 96, "Dolby Digital Plus segment incorrect size, "
assert len(buffer) == 96, (
"Dolby Digital Plus segment incorrect size, "
"expected 96 got %i" % len(buffer)
)
def program_id(b) -> int:
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 +362,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
@@ -392,13 +413,13 @@ class DolbyDigitalPlusMetadata:
return DolbyDigitalPlusMetadata.RFCompressionProfile(b)
def dynrng1(b):
DolbyDigitalPlusMetadata.RFCompressionProfile(b)
return DolbyDigitalPlusMetadata.RFCompressionProfile(b)
def ddplus_reserved3(_):
pass
def ddplus_info1(b):
return DolbyDigitalPlusMetadata.StreamDependency(b & 0xc >> 2)
return DolbyDigitalPlusMetadata.StreamDependency(b & 0xC >> 2)
def ddplus_reserved4(_):
pass
@@ -412,19 +433,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,7 +462,8 @@ class DolbyDigitalPlusMetadata:
reserved(buffer[27:69])
return DolbyDigitalPlusMetadata(
program_id=pid, lfe_on=lfe_on,
program_id=pid,
lfe_on=lfe_on,
bitstream_mode=bitstream_mode,
audio_coding_mode=audio_coding_mode,
center_downmix_level=center_downmix_level,
@@ -461,7 +488,8 @@ class DolbyDigitalPlusMetadata:
compression_profile=compression,
dynamic_range=dynamic_range,
stream_dependency=stream_info,
datarate_kbps=data_rate)
datarate_kbps=data_rate,
)
@dataclass
@@ -480,7 +508,7 @@ class DolbyAtmosMetadata:
NOT_INDICATED = 0x04
tool_name: str
tool_version: Tuple[int, int, int]
tool_version: tuple[int, int, int]
warp_mode: WarpMode
SEGMENT_LENGTH = 248
@@ -488,7 +516,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 +525,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 +535,11 @@ class DolbyAtmosMetadata:
a_val = unpack("B", h.read(1))[0]
warp_mode = a_val & 0x7
return DolbyAtmosMetadata(tool_name=toolname,
return DolbyAtmosMetadata(
tool_name=toolname,
tool_version=(major, minor, fix),
warp_mode=DolbyAtmosMetadata
.WarpMode(warp_mode))
warp_mode=DolbyAtmosMetadata.WarpMode(warp_mode),
)
@dataclass
@@ -531,15 +559,14 @@ class DolbyAtmosSupplementalMetadata:
NOT_INDICATED = 0x04
object_count: int
render_modes: List['DolbyAtmosSupplementalMetadata.BinauralRenderMode']
trim_modes: List[int]
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 +591,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:
@@ -580,18 +607,18 @@ class WavDolbyMetadataReader:
#: indicating if the segment's checksum was valid, and the
#: segment's parsed dataclass (or a `bytes` array if it was
#: not recognized).
segment_list: List[Tuple[Union[SegmentType, int], bool, Any]]
segment_list: list[tuple[SegmentType | int, bool, Any]]
version: Tuple[int, int, int, int]
version: tuple[int, int, int, int]
@staticmethod
def segment_checksum(bs: bytes, size: int):
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 +641,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 +654,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]:
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]:
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]:
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())
ddp = [asdict(x) for x in self.dolby_digital_plus()]
atmos = [asdict(x) for x in 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 {"dolby_digital_plus": ddp, "dolby_atmos": atmos}
+102
View File
@@ -0,0 +1,102 @@
from __future__ import annotations
from .riff_parser import ListChunkDescriptor, parse_chunk
class WavInfoChunkReader:
def __init__(self, f, encoding):
self.encoding = encoding
f.seek(0)
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
]
self.info_chunk = next(
(chunk for chunk in list_chunks if chunk.signature == b"INFO"), None
)
#: 'ICOP' Copyright
self.copyright: str | None = self._get_field(f, b"ICOP")
#: 'IPRD' Product
self.product: str | None = self._get_field(f, b"IPRD")
self.album: str | None = self.product
#: 'IGNR' Genre
self.genre: str | None = self._get_field(f, b"IGNR")
#: 'ISBJ' Subject
self.subject: str | None = self._get_field(f, b"ISBJ")
#: 'IART' Artist, composer, author
self.artist: str | None = self._get_field(f, b"IART")
#: 'ICMT' Comment
self.comment: str | None = self._get_field(f, b"ICMT")
#: 'ISFT' Software, encoding application
self.software: str | None = self._get_field(f, b"ISFT")
#: 'ICRD' Created date
self.created_date: str | None = self._get_field(f, b"ICRD")
#: 'IENG' Engineer
self.engineer: str | None = self._get_field(f, b"IENG")
#: 'ITCH' Technician
self.technician: str | None = self._get_field(f, b"ITCH")
#: 'IKEY' Keywords, keyword list
self.keywords: str | None = self._get_field(f, b"IKEY")
#: 'INAM' Name, title
self.title: str | None = self._get_field(f, b"INAM")
#: 'ISRC' Source
self.source: str | None = self._get_field(f, b"ISRC")
#: 'TAPE' Tape
self.tape: str | None = self._get_field(f, b"TAPE")
#: 'IARL' Archival Location
self.archival_location: str | None = self._get_field(f, b"IARL")
#: 'ICSM' Commissioned
self.commissioned: str | None = self._get_field(f, b"ICMS")
def _get_field(self, f, field_ident) -> str | 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")
else:
return None
def to_dict(self) -> dict: # FIXME should be asdict
"""
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,
}
def __repr__(self):
return_val = self.to_dict()
return_val.update({"encoding": self.encoding})
return str(return_val)
@@ -1,10 +1,13 @@
from lxml import etree as ET
from __future__ import annotations
import io
# from collections import namedtuple
from typing import Optional
from enum import IntEnum
# from collections import namedtuple
from typing import NamedTuple
from lxml import etree as ET
class IXMLTrack(NamedTuple):
channel_index: int
@@ -22,6 +25,7 @@ class SteinbergMetadata:
"""
Steinberg speaker format enumeration.
"""
MONO = 0
STEREO = 1
LRC = 10
@@ -74,82 +78,75 @@ class SteinbergMetadata:
self.parsed = xml.find(self.Steinberg_xpath)
@property
def audio_speaker_arrangement(self) -> Optional[AudioSpeakerArrangement]:
def audio_speaker_arrangement(self) -> AudioSpeakerArrangement | None:
"""
`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))
@property
def sample_format_size(self) -> Optional[int]:
def sample_format_size(self) -> int | None:
"""
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)
@property
def media_company(self) -> Optional[str]:
def media_company(self) -> str | None:
"""
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
@property
def media_drop_frames(self) -> Optional[bool]:
def media_drop_frames(self) -> bool | None:
"""
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"
@property
def media_duration(self) -> Optional[float]:
def media_duration(self) -> float | None:
"""
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)
@property
def media_start_time(self) -> Optional[float]:
"""
MediaStartTime
"""
pass
# @property
# def media_start_time(self) -> float | None:
# """
# MediaStartTime
# """
@property
def media_track_title(self) -> Optional[str]:
"""
MediaTrackTitle
"""
pass
# @property
# def media_track_title(self) -> str | None:
# """
# MediaTrackTitle
# """
@property
def program_name(self) -> Optional[str]:
"""
ProgramName
"""
pass
# @property
# def program_name(self) -> str | None:
# """
# ProgramName
# """
@property
def program_version(self) -> Optional[str]:
"""
ProgramVersion
"""
pass
# @property
# def program_version(self) -> str | None:
# """
# ProgramVersion
# """
class WavIXMLFormat:
@@ -167,7 +164,7 @@ class WavIXMLFormat:
parser = ET.XMLParser(recover=True)
self.parsed: ET.ElementTree = ET.parse(xml_bytes, parser=parser)
def _get_text_value(self, xpath) -> Optional[str]:
def _get_text_value(self, xpath) -> str | None:
e = self.parsed.find("./" + xpath)
if e is not None:
return e.text
@@ -192,45 +189,44 @@ 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
def project(self) -> Optional[str]:
def project(self) -> str | None:
"""
The project/film name entered for the recording.
"""
return self._get_text_value("PROJECT")
@property
def scene(self) -> Optional[str]:
def scene(self) -> str | None:
"""
Scene/slate.
"""
return self._get_text_value("SCENE")
@property
def take(self) -> Optional[str]:
def take(self) -> str | None:
"""
Take number.
"""
return self._get_text_value("TAKE")
@property
def tape(self) -> Optional[str]:
def tape(self) -> str | None:
"""
Tape name.
"""
return self._get_text_value("TAPE")
@property
def family_uid(self) -> Optional[str]:
def family_uid(self) -> str | None:
"""
The globally-unique ID for this file family. This may be in the format
of a GUID, or an EBU Rec 9 source identifier, or some other dumb
@@ -239,14 +235,14 @@ class WavIXMLFormat:
return self._get_text_value("FILE_SET/FAMILY_UID")
@property
def family_name(self) -> Optional[str]:
def family_name(self) -> str | None:
"""
The name of this file's file family.
"""
return self._get_text_value("FILE_SET/FAMILY_NAME")
@property
def steinberg(self) -> Optional[SteinbergMetadata]:
def steinberg(self) -> SteinbergMetadata | None:
"""
Steinberg vendor iXML metadata if present.
"""
@@ -256,8 +252,12 @@ class WavIXMLFormat:
return None
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)
return {
"track_list": [x._asdict() for x in 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,
}
@@ -1,18 +1,17 @@
# -*- coding: utf-8 -*-
import struct
from __future__ import annotations
import os
from typing import Optional, Generator, Any, NamedTuple
import pathlib
import struct
from typing import Any, Generator, NamedTuple
from .riff_parser import parse_chunk, ChunkDescriptor, ListChunkDescriptor
from .wave_ixml_reader import WavIXMLFormat
from .wave_bext_reader import WavBextReader
from .wave_info_reader import WavInfoChunkReader
from .riff_parser import ChunkDescriptor, ListChunkDescriptor, parse_chunk
from .wave_adm_reader import WavADMReader
from .wave_dbmd_reader import WavDolbyMetadataReader
from .wave_bext_reader import WavBextReader
from .wave_cues_reader import WavCuesReader
from .wave_dbmd_reader import WavDolbyMetadataReader
from .wave_info_reader import WavInfoChunkReader
from .wave_ixml_reader import WavIXMLFormat
from .wave_smpl_reader import WavSmplReader
#: Calculated statistics about the audio data.
@@ -38,7 +37,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.
@@ -60,35 +59,35 @@ class WavInfoReader:
self.bext_encoding = bext_encoding
#: Wave audio data format.
self.fmt: Optional[WavAudioFormat] = None
self.fmt: WavAudioFormat | None = None
#: Statistics of the `data` section.
self.data: Optional[WavDataDescriptor] = None
self.data: WavDataDescriptor | None = None
#: Broadcast-Wave metadata.
self.bext: Optional[WavBextReader] = None
self.bext: WavBextReader | None = None
#: iXML metadata.
self.ixml: Optional[WavIXMLFormat] = None
self.ixml: WavIXMLFormat | None = None
#: ADM Audio Definiton Model metadata.
self.adm: Optional[WavADMReader] = None
self.adm: WavADMReader | None = None
#: Dolby bitstream metadata.
self.dolby: Optional[WavDolbyMetadataReader] = None
self.dolby: WavDolbyMetadataReader | None = None
#: RIFF INFO metadata.
self.info: Optional[WavInfoChunkReader] = None
self.info: WavInfoChunkReader | None = None
#: RIFF cues markers, labels, and notes.
self.cues: Optional[WavCuesReader] = None
self.cues: WavCuesReader | None = None
#: Sampler `smpl` metadata
self.smpl: Optional[WavSmplReader] = None
self.smpl: WavSmplReader | None = 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,8 +98,8 @@ class WavInfoReader:
self.path = absolute_path
with open(path, 'rb') as path:
self.get_wav_info(path)
with open(path, "rb") as file:
self.get_wav_info(file)
def get_wav_info(self, wavfile):
chunks = parse_chunk(wavfile)
@@ -119,36 +118,41 @@ 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) -> bytes | None:
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)
def _find_list_chunk(self, signature) -> ListChunkDescriptor | None:
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 +160,79 @@ class WavInfoReader:
unpacked = struct.unpack(packstring, fmt_data[:rest_starts])
return WavAudioFormat(audio_format=unpacked[0],
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]
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 +246,22 @@ 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 {}
for key in mdict.keys():
mdict = (
self.__getattribute__(scope).to_dict()
if self.__getattribute__(scope)
else {}
)
for key in mdict:
yield scope, key, mdict[key]
def __repr__(self):
return 'WavInfoReader({}, {}, {})'.format(self.path,
self.info_encoding,
self.bext_encoding)
return f"WavInfoReader({self.path}, {self.info_encoding}, {self.bext_encoding})"
@@ -1,6 +1,7 @@
import struct
from __future__ import annotations
from typing import Tuple, NamedTuple, List
import struct
from typing import NamedTuple
class WaveSmplLoop(NamedTuple):
@@ -13,30 +14,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 +47,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.
@@ -72,43 +71,47 @@ class WavSmplReader:
#: The SMPTE offset to apply, as a tuple of four ints representing
#: hh, mm, ss, ff
self.smpte_offset: Tuple[int, int, int, int] = unpacked_data[6:10]
self.smpte_offset: tuple[int, int, int, int] = unpacked_data[6:10]
loop_count = unpacked_data[10]
sampler_udata_length = unpacked_data[11]
#: List of loops in the file.
self.sample_loops: List[WaveSmplLoop] = []
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(
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]))
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,
}
+31
View File
@@ -0,0 +1,31 @@
"""
wavfind.py
Under construction! Do Not Use!
"""
import sys
from optparse import OptionGroup, OptionParser
def main():
parser = OptionParser()
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.add_option("--scene", help="Search for this scene", metavar="SCENE")
primaries.add_option("--take", help="Search for this take", metavar="TAKE")
primaries.add_option("--desc", help="Search descriptions", metavar="DESC")
(_options, _args) = parser.parse_args(sys.argv)
if __name__ == "__main__":
main()
-44
View File
@@ -1,44 +0,0 @@
from . import test_wave_parsing
from io import BytesIO
from typing import Generator
import zipfile as zf
import os.path
import os
from contextlib import contextmanager
# class TestFileLoader:
# """
# This guy manages the test_files archive.
# """
# def __init__(self, test_file_arch_name = 'archive.zip') -> None:
# self.base_path = os.path.join(os.path.dirname(__file__), "test_files")
# self.test_file_arch_name = test_file_arch_name
# self._gather_test_files_into_archive()
# @property
# def arch_path(self):
# return os.path.join(self.base_path, self.test_file_arch_name)
# @contextmanager
# def open(self, name) -> Generator[BytesIO]:
# z = zf.ZipFile(self.arch_path, 'r')
# member = z.open(name, 'r')
# try:
# yield member
# finally:
# zf.close()
# def _gather_test_files_into_archive(self):
# with zf.ZipFile(self.arch_path, 'a') as zip:
# for root, _, files in os.walk(self.base_path):
# for name in files:
# if root == self.base_path and name == self.test_file_arch_name:
# continue
# else:
# p = os.path.join(root, name)
# zip.write(p)
# os.unlink(p)
+8 -7
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()
@@ -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)
@@ -42,13 +42,14 @@ class TestADMWave(TestCase):
adm = info.adm
assert adm is not None
t1 = adm.track_info(0)
self.assertTrue("channel_format_name" in t1.keys())
assert t1 is not None
self.assertTrue("channel_format_name" in t1)
self.assertEqual("RoomCentricLeft", t1["channel_format_name"])
self.assertTrue("pack_format_name" in t1.keys())
self.assertTrue("pack_format_name" in t1)
self.assertEqual("AtmosCustomPackFormat1", t1["pack_format_name"])
t10 = adm.track_info(10)
self.assertTrue("content_name" in t10.keys())
assert t10 is not None
self.assertTrue("content_name" in t10)
self.assertEqual("Dialog", t10["content_name"])
+7 -8
View File
@@ -1,8 +1,9 @@
from unittest import TestCase
from glob import glob
from unittest import TestCase
import wavinfo
class TestCue(TestCase):
def setUp(self) -> None:
self.test_files = glob("tests/test_files/cue_chunks/*.wav")
@@ -44,9 +45,11 @@ class TestCue(TestCase):
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]
@@ -69,7 +72,3 @@ class TestCue(TestCase):
self.assertEqual(label.text, "Marker 2")
elif label.name == 3:
self.assertEqual(label.text, "Marker 3")
+15 -14
View File
@@ -1,7 +1,8 @@
from unittest import TestCase
import wavinfo
from wavinfo.wave_dbmd_reader import SegmentType, DolbyDigitalPlusMetadata
from wavinfo.wave_dbmd_reader import DolbyDigitalPlusMetadata, SegmentType
class TestDolby(TestCase):
def setUp(self):
@@ -19,10 +20,8 @@ class TestDolby(TestCase):
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,13 +39,16 @@ 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].audio_coding_mode,
DolbyDigitalPlusMetadata.AudioCodingMode.CH_ORD_3_2,
)
self.assertTrue(ddp[0].lfe_on)
def test_atmos(self):
@@ -54,7 +56,6 @@ class TestDolby(TestCase):
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"
)
Binary file not shown.
+7 -18
View File
@@ -1,34 +1,23 @@
import glob
import sys
import unittest
from unittest.mock import patch
from wavinfo.__main__ import main
import sys
import glob
class MainTest(unittest.TestCase):
def test_empty_argv(self):
with patch.object(sys, 'argv', []):
try:
with patch.object(sys, "argv", []):
main()
except:
self.fail("main() throwing an exception")
def test_a_file(self):
for path in glob.glob("tests/test_files/**/*.wav"):
with patch.object(sys, 'argv', ["TEST", path]):
try:
with patch.object(sys, "argv", ["TEST", path]):
main()
except:
self.fail("main() throwing an exception")
def test_ixml(self):
with patch.object(sys, 'argv',
['TEST', '--ixml',
'tests/test_files/sounddevices/A101_1.WAV']):
try:
with patch.object(
sys, "argv", ["TEST", "--ixml", "tests/test_files/sounddevices/A101_1.WAV"]
):
main()
except:
self.fail("main() throwing an exception")
+3 -8
View File
@@ -1,25 +1,20 @@
# import os.path
import gzip
from glob import glob
# from typing import Dict, Any, cast
# from typing import Dict, Any, cast
from unittest import TestCase
# from .utils import all_files, ffprobe
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)
with gzip.open(path) as gz:
wav_info = wavinfo.WavInfoReader(gz)
self.assertIsNotNone(wav_info)
# self.assertIsNotNone(wav_info.bext)
+2 -1
View File
@@ -1,8 +1,9 @@
from unittest import TestCase
from glob import glob
from unittest import TestCase
import wavinfo
class TestSmpl(TestCase):
def setUp(self) -> None:
self.test_files = glob("tests/test_files/smpl/*.wav")
+10 -12
View File
@@ -1,37 +1,35 @@
import glob
import unittest
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
for scope, key, value in info.walk():
if scope == 'fmt':
if key == 'channel_count':
if scope == "fmt" and key == "channel_count":
tested_format = True
self.assertEqual(value, 2)
if scope == 'data':
if key == 'frame_count':
if scope == "data" and 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:
# try:
for _, _, _ in info.walk():
pass
except:
self.fail(f"Failed to walk metadata in file {file}")
# except Exception as e:
# self.fail(f"Failed to walk metadata in file {file}: {e}")
if __name__ == '__main__':
if __name__ == "__main__":
unittest.main()
+109 -89
View File
@@ -1,21 +1,21 @@
import os.path
from glob import glob
from typing import Dict, Any, cast
from typing import Any, Dict, cast
from unittest import TestCase
from .utils import all_files, ffprobe
import wavinfo
from .utils import all_files, ffprobe
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__(),
f"WavInfoReader({os.path.abspath(wav_file)}, latin_1, ascii)",
)
self.assertIsNotNone(info)
def test_fmt_against_ffprobe(self):
@@ -26,21 +26,28 @@ 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):
@@ -49,8 +56,9 @@ class TestWaveInfo(TestCase):
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,64 +67,77 @@ 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,
# ffprobe_info['format']['tags']['date'])
# 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():
@@ -126,19 +147,19 @@ class TestWaveInfo(TestCase):
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)
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
@@ -146,8 +167,10 @@ class TestWaveInfo(TestCase):
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.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 +182,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)
+1 -7
View File
@@ -1,12 +1,6 @@
import os.path
import sys
import json
import subprocess
from subprocess import PIPE
from unittest import TestCase
import wavinfo
# import wavinfo
class TestZoomF8(TestCase):
+8 -23
View File
@@ -1,38 +1,23 @@
import os.path
import sys
import subprocess
from subprocess import PIPE
import json
import os.path
import subprocess
FFPROBE = 'ffprobe'
FFPROBE = "ffprobe"
def ffprobe(path):
arguments = [FFPROBE, "-of", "json",
"-show_format", "-show_streams", path]
if int(sys.version[0]) < 3:
process = subprocess.Popen(arguments, stdout=PIPE)
process.wait()
arguments = [FFPROBE, "-of", "json", "-show_format", "-show_streams", path]
process = subprocess.run(arguments, check=True, capture_output=True)
if process.returncode == 0:
output = process.communicate()[0]
if output:
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)
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)
-189
View File
@@ -1,189 +0,0 @@
"""
ADM Reader
"""
from struct import unpack, unpack_from, calcsize
from io import BytesIO
from collections import namedtuple
from typing import Optional
from lxml import etree as ET
ChannelEntry = namedtuple('ChannelEntry', "track_index uid track_ref pack_ref")
class WavADMReader:
"""
Reads XML data from an EBU ADM (Audio Definiton Model) WAV File.
"""
def __init__(self, axml_data: bytes, chna_data: bytes):
header_fmt = "<HH"
uid_fmt = "<H12s14s11sx"
#: An :mod:`lxml.etree` of the ADM XML document
self.axml = ET.parse(BytesIO(axml_data))
_, uid_count = unpack(header_fmt, chna_data[0:4])
self.channel_uids = []
offset = calcsize(header_fmt)
for _ in range(uid_count):
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')
)
)
offset += calcsize(uid_fmt)
def xml_str(self) -> str:
"""ADM XML as a string"""
return ET.tostring(self.axml).decode("utf-8")
def programme(self) -> dict:
"""
Read the ADM `audioProgramme` data structure and some of its reference
properties.
"""
ret_dict = dict()
nsmap = self.axml.getroot().nsmap
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'] = []
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'] = []
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)
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
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)
ret_dict['contents'].append(content_dict)
return ret_dict
def track_info(self, index) -> Optional[dict]:
"""
Information about a track in the WAV file.
:param index: index of audio track (indexed from zero)
:returns: a dictionary with *content_name*, *content_id*,
*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)
if channel_info is None:
return None
ret_dict = {}
nsmap = self.axml.getroot().nsmap
afext = self.axml.find(".//audioFormatExtended",
namespaces=nsmap)
trackformat_elem = afext.find(
"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)
channelformat_id = channelformatref_elem.text
packformatref_elem = afext.find(
("audioStreamFormat[@audioStreamFormatID='%s']"
"/audioPackFormatIDRef") % stream_id,
namespaces=nsmap)
packformat_id = packformatref_elem.text
channelformat_elem = afext\
.find("audioChannelFormat[@audioChannelFormatID='%s']"
% channelformat_id,
namespaces=nsmap)
ret_dict['channel_format_name'] = channelformat_elem.get(
"audioChannelFormatName")
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")
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
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")
return ret_dict
def to_dict(self) -> dict: # FIXME should be "asdict"
"""
Get ADM metadata as a dictionary.
"""
def make_entry(channel_uid_rec):
rd = channel_uid_rec._asdict()
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())
-94
View File
@@ -1,94 +0,0 @@
from .riff_parser import parse_chunk, ListChunkDescriptor
from typing import Optional
class WavInfoChunkReader:
def __init__(self, f, encoding):
self.encoding = encoding
f.seek(0)
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]
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')
#: 'IPRD' Product
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')
#: 'ISBJ' Subject
self.subject: Optional[str] = self._get_field(f, b'ISBJ')
#: 'IART' Artist, composer, author
self.artist: Optional[str] = self._get_field(f, b'IART')
#: 'ICMT' Comment
self.comment: Optional[str] = self._get_field(f, b'ICMT')
#: 'ISFT' Software, encoding application
self.software: Optional[str] = self._get_field(f, b'ISFT')
#: 'ICRD' Created date
self.created_date: Optional[str] = self._get_field(f, b'ICRD')
#: 'IENG' Engineer
self.engineer: Optional[str] = self._get_field(f, b'IENG')
#: 'ITCH' Technician
self.technician: Optional[str] = self._get_field(f, b'ITCH')
#: 'IKEY' Keywords, keyword list
self.keywords: Optional[str] = self._get_field(f, b'IKEY')
#: 'INAM' Name, title
self.title: Optional[str] = self._get_field(f, b'INAM')
#: 'ISRC' Source
self.source: Optional[str] = self._get_field(f, b'ISRC')
#: 'TAPE' 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')
#: 'ICSM' Commissioned
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)
if search is not None:
f.seek(search[0])
data = f.read(search[1])
return data.decode(self.encoding).rstrip('\0')
else:
return None
def to_dict(self) -> dict: # FIXME should be asdict
"""
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
}
def __repr__(self):
return_val = self.to_dict()
return_val.update({'encoding': self.encoding})
return str(return_val)
-35
View File
@@ -1,35 +0,0 @@
"""
"""
from optparse import OptionParser, OptionGroup
import sys
def main():
parser = OptionParser()
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.add_option("--scene",
help='Search for this scene',
metavar='SCENE')
primaries.add_option("--take",
help='Search for this take',
metavar='TAKE')
primaries.add_option("--desc",
help='Search descriptions',
metavar='DESC')
(options, args) = parser.parse_args(sys.argv)
if __name__ == "__main__":
main()