Add cover art pipeline (extract/encode/cache) and iTunes-style player
- New src/artwork/ module with iPod Nano 7G cover art support - codecs: RGB565/RGB555/UYVY/JPEG encode/decode via numpy - presets: 50+ iPod artwork format definitions - chunks: ArtworkDB binary parser/writer (MHFD→MHSD→MHLI→MHII→MHNI) - writer: ithmb file generation + ArtworkDB assembly - extractor: cover extraction from audio files (APIC/covr/FLAC pictures) - cache: local JPEG cache for player UI - iTunes-style playback bar with 60x60 cover art display - Scan library now caches cover art by artist+album - iPod Nano7 transfer writes artwork to ArtworkDB + ithmb - FFmpeg conversion preserves embedded cover art (-map 0:t?) - One-time reembed_covers.py script in /tmp - MetadataHandler process_file() fallback to cover_path - Bug fixes: missing read_existing_artwork import, signed PID in Q format
This commit is contained in:
Maksim Totmin
@@ -0,0 +1,29 @@
|
||||
"""Artwork module — iPod album art pipeline.
|
||||
|
||||
Public API:
|
||||
process_artwork_for_track(source_path) -> bytes | None
|
||||
extract_cover(file_path) -> bytes | None
|
||||
encode_for_ipod(art_bytes, format_id) -> EncodedFormatPayload
|
||||
write_artwork_to_ipod(ipod_path, entries, formats_map)
|
||||
get_local_cover(cache_key) -> str | None
|
||||
cache_cover(cache_key, art_bytes) -> str
|
||||
"""
|
||||
|
||||
from .presets import ArtworkFormat, NANO_7G_COVER_ART_FORMATS
|
||||
from .types import ArtworkEntry, IthmbLocation, EncodedFormatPayload
|
||||
from .extractor import extract_cover
|
||||
from .codecs import encode_for_ipod, decode_from_ipod
|
||||
from .writer import write_artworkdb, prepare_artwork_for_track
|
||||
|
||||
__all__ = [
|
||||
"ArtworkFormat",
|
||||
"ArtworkEntry",
|
||||
"IthmbLocation",
|
||||
"EncodedFormatPayload",
|
||||
"NANO_7G_COVER_ART_FORMATS",
|
||||
"extract_cover",
|
||||
"encode_for_ipod",
|
||||
"decode_from_ipod",
|
||||
"write_artworkdb",
|
||||
"prepare_artwork_for_track",
|
||||
]
|
||||
@@ -0,0 +1,67 @@
|
||||
"""Local cover art cache for the player UI.
|
||||
|
||||
Caches extracted cover art as JPEG files in a local directory
|
||||
so the player can display them without re-extracting from audio.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import logging
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class CoverCache:
|
||||
"""Manages a local cache of cover art images for the player UI.
|
||||
|
||||
The cache stores JPEG thumbnails keyed by (artist, album) hash.
|
||||
"""
|
||||
|
||||
def __init__(self, cache_dir: str | None = None):
|
||||
self.cache_dir = cache_dir or os.path.join(
|
||||
tempfile.gettempdir(), "neo-pod-covers"
|
||||
)
|
||||
os.makedirs(self.cache_dir, exist_ok=True)
|
||||
|
||||
def _key(self, artist: str, album: str) -> str:
|
||||
raw = f"{artist.lower()}|{album.lower()}"
|
||||
return hashlib.md5(raw.encode()).hexdigest()
|
||||
|
||||
def get(self, artist: str, album: str) -> str | None:
|
||||
"""Return path to cached cover, or None."""
|
||||
path = os.path.join(self.cache_dir, f"{self._key(artist, album)}.jpg")
|
||||
if os.path.exists(path):
|
||||
return path
|
||||
return None
|
||||
|
||||
def put(self, artist: str, album: str, art_bytes: bytes) -> str | None:
|
||||
"""Cache cover art bytes and return the file path."""
|
||||
from PIL import Image
|
||||
from io import BytesIO
|
||||
|
||||
key = self._key(artist, album)
|
||||
path = os.path.join(self.cache_dir, f"{key}.jpg")
|
||||
try:
|
||||
img = Image.open(BytesIO(art_bytes))
|
||||
img = img.convert("RGB")
|
||||
img.thumbnail((300, 300), Image.Resampling.LANCZOS)
|
||||
img.save(path, "JPEG", quality=85)
|
||||
return path
|
||||
except Exception as exc:
|
||||
logger.warning("Failed to cache cover art: %s", exc)
|
||||
return None
|
||||
|
||||
def clear(self) -> None:
|
||||
"""Remove all cached covers."""
|
||||
for fname in os.listdir(self.cache_dir):
|
||||
if fname.endswith(".jpg"):
|
||||
try:
|
||||
os.remove(os.path.join(self.cache_dir, fname))
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
cover_cache = CoverCache()
|
||||
@@ -0,0 +1,374 @@
|
||||
"""Binary ArtworkDB chunk serialization.
|
||||
|
||||
Ported from iOpenPod ArtworkDB_Writer/artworkdb_chunks.py.
|
||||
Builds the MHFD → MHSD → MHLI/MHLA/MHLF → MHII → MHOD → MHNI → MHOD(ithmb)
|
||||
hierarchy that the iPod firmware parses.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import struct
|
||||
from collections.abc import Mapping
|
||||
|
||||
from .types import ArtworkEntry, ArtworkFormatPayload, ExistingFormatRef, IthmbLocation
|
||||
from .codecs import expected_size_bytes
|
||||
|
||||
MHFD_HEADER_SIZE = 132
|
||||
MHSD_HEADER_SIZE = 96
|
||||
MHLI_HEADER_SIZE = 92
|
||||
MHLA_HEADER_SIZE = 92
|
||||
MHLF_HEADER_SIZE = 92
|
||||
MHII_HEADER_SIZE = 152
|
||||
MHOD_HEADER_SIZE = 24
|
||||
MHNI_HEADER_SIZE = 76
|
||||
MHIF_HEADER_SIZE = 124
|
||||
|
||||
|
||||
def _default_ithmb_filename(format_id: int) -> str:
|
||||
return f"F{int(format_id)}_1.ithmb"
|
||||
|
||||
|
||||
def _normalize_ithmb_filename(format_id: int, filename: str | None) -> str:
|
||||
name = (filename or "").strip().replace("\\", "/")
|
||||
if ":" in name:
|
||||
name = name.split(":")[-1]
|
||||
if "/" in name:
|
||||
name = name.rsplit("/", 1)[-1]
|
||||
return name or _default_ithmb_filename(format_id)
|
||||
|
||||
|
||||
def build_artworkdb(
|
||||
entries: list[ArtworkEntry],
|
||||
format_locations_map: Mapping[int, Mapping[int, IthmbLocation | tuple[str, int] | int]],
|
||||
format_ids: list[int],
|
||||
image_sizes: dict[int, int],
|
||||
next_mhii_id: int,
|
||||
reference_mhfd: bytes | None = None,
|
||||
) -> bytes:
|
||||
"""Serialize a complete ArtworkDB binary."""
|
||||
ds1 = _write_mhsd(1, _write_mhli(entries, format_locations_map))
|
||||
ds2 = _write_mhsd(2, _write_mhla())
|
||||
ds3 = _write_mhsd(3, _write_mhlf(format_ids, image_sizes))
|
||||
return _write_mhfd([ds1, ds2, ds3], next_mhii_id, reference_mhfd)
|
||||
|
||||
|
||||
def _write_mhfd(datasets: list[bytes], next_mhii_id: int, reference_mhfd: bytes | None = None) -> bytes:
|
||||
all_data = b"".join(datasets)
|
||||
total_len = MHFD_HEADER_SIZE + len(all_data)
|
||||
header = bytearray(MHFD_HEADER_SIZE)
|
||||
header[0:4] = b"mhfd"
|
||||
struct.pack_into("<I", header, 4, MHFD_HEADER_SIZE)
|
||||
struct.pack_into("<I", header, 8, total_len)
|
||||
struct.pack_into("<I", header, 16, 2)
|
||||
struct.pack_into("<I", header, 20, len(datasets))
|
||||
struct.pack_into("<I", header, 28, next_mhii_id)
|
||||
if reference_mhfd and len(reference_mhfd) >= 48:
|
||||
header[32:48] = reference_mhfd[32:48]
|
||||
struct.pack_into("<I", header, 48, 2)
|
||||
if reference_mhfd and len(reference_mhfd) >= 68:
|
||||
header[60:68] = reference_mhfd[60:68]
|
||||
return bytes(header) + all_data
|
||||
|
||||
|
||||
def _write_mhsd(ds_type: int, child_data: bytes) -> bytes:
|
||||
total_len = MHSD_HEADER_SIZE + len(child_data)
|
||||
header = bytearray(MHSD_HEADER_SIZE)
|
||||
header[0:4] = b"mhsd"
|
||||
struct.pack_into("<I", header, 4, MHSD_HEADER_SIZE)
|
||||
struct.pack_into("<I", header, 8, total_len)
|
||||
struct.pack_into("<H", header, 12, ds_type)
|
||||
return bytes(header) + child_data
|
||||
|
||||
|
||||
def _write_mhli(entries: list[ArtworkEntry], format_locations_map) -> bytes:
|
||||
children_data = b"".join(
|
||||
_write_mhii(entry, format_locations_map[entry.img_id])
|
||||
for entry in entries
|
||||
)
|
||||
header = bytearray(MHLI_HEADER_SIZE)
|
||||
header[0:4] = b"mhli"
|
||||
struct.pack_into("<I", header, 4, MHLI_HEADER_SIZE)
|
||||
struct.pack_into("<I", header, 8, len(entries))
|
||||
return bytes(header) + children_data
|
||||
|
||||
|
||||
def _write_mhla() -> bytes:
|
||||
header = bytearray(MHLA_HEADER_SIZE)
|
||||
header[0:4] = b"mhla"
|
||||
struct.pack_into("<I", header, 4, MHLA_HEADER_SIZE)
|
||||
struct.pack_into("<I", header, 8, 0)
|
||||
return bytes(header)
|
||||
|
||||
|
||||
def _write_mhlf(format_ids: list[int], image_sizes: dict[int, int]) -> bytes:
|
||||
children_data = b"".join(
|
||||
_write_mhif(fmt_id, image_sizes[fmt_id])
|
||||
for fmt_id in format_ids
|
||||
)
|
||||
header = bytearray(MHLF_HEADER_SIZE)
|
||||
header[0:4] = b"mhlf"
|
||||
struct.pack_into("<I", header, 4, MHLF_HEADER_SIZE)
|
||||
struct.pack_into("<I", header, 8, len(format_ids))
|
||||
return bytes(header) + children_data
|
||||
|
||||
|
||||
def _write_mhif(format_id: int, image_size: int) -> bytes:
|
||||
header = bytearray(MHIF_HEADER_SIZE)
|
||||
header[0:4] = b"mhif"
|
||||
struct.pack_into("<I", header, 4, MHIF_HEADER_SIZE)
|
||||
struct.pack_into("<I", header, 8, MHIF_HEADER_SIZE)
|
||||
struct.pack_into("<I", header, 16, format_id)
|
||||
struct.pack_into("<I", header, 20, image_size)
|
||||
return bytes(header)
|
||||
|
||||
|
||||
def _write_mhii(entry: ArtworkEntry, format_locations: Mapping[int, IthmbLocation | tuple[str, int] | int]) -> bytes:
|
||||
children = []
|
||||
for fmt_id in sorted(entry.formats.keys()):
|
||||
payload = entry.formats[fmt_id]
|
||||
location = _coerce_ithmb_location(fmt_id, format_locations.get(fmt_id, 0))
|
||||
children.append(_write_mhod_container(2, _write_mhni(fmt_id, location, payload)))
|
||||
|
||||
children_data = b"".join(children)
|
||||
total_len = MHII_HEADER_SIZE + len(children_data)
|
||||
header = bytearray(MHII_HEADER_SIZE)
|
||||
header[0:4] = b"mhii"
|
||||
struct.pack_into("<I", header, 4, MHII_HEADER_SIZE)
|
||||
struct.pack_into("<I", header, 8, total_len)
|
||||
struct.pack_into("<I", header, 12, len(children))
|
||||
struct.pack_into("<I", header, 16, entry.img_id)
|
||||
struct.pack_into("<q", header, 20, entry.db_track_id)
|
||||
struct.pack_into("<I", header, 48, entry.src_img_size)
|
||||
return bytes(header) + children_data
|
||||
|
||||
|
||||
def _write_mhod_container(mhod_type: int, mhni_data: bytes) -> bytes:
|
||||
total_len = MHOD_HEADER_SIZE + len(mhni_data)
|
||||
header = bytearray(MHOD_HEADER_SIZE)
|
||||
header[0:4] = b"mhod"
|
||||
struct.pack_into("<I", header, 4, MHOD_HEADER_SIZE)
|
||||
struct.pack_into("<I", header, 8, total_len)
|
||||
struct.pack_into("<H", header, 12, mhod_type)
|
||||
return bytes(header) + mhni_data
|
||||
|
||||
|
||||
def _write_mhni(format_id: int, location: IthmbLocation, payload: ArtworkFormatPayload) -> bytes:
|
||||
filename = _normalize_ithmb_filename(format_id, location.filename)
|
||||
mhod3 = _write_mhod_string(3, f":{filename}")
|
||||
total_len = MHNI_HEADER_SIZE + len(mhod3)
|
||||
|
||||
visible_h = int(payload.height)
|
||||
visible_w = int(payload.width)
|
||||
img_size = int(payload.size)
|
||||
stride = max(visible_w, int(payload.stride_pixels))
|
||||
|
||||
header = bytearray(MHNI_HEADER_SIZE)
|
||||
header[0:4] = b"mhni"
|
||||
struct.pack_into("<I", header, 4, MHNI_HEADER_SIZE)
|
||||
struct.pack_into("<I", header, 8, total_len)
|
||||
struct.pack_into("<I", header, 12, 1)
|
||||
struct.pack_into("<I", header, 16, format_id)
|
||||
struct.pack_into("<I", header, 20, int(location.offset))
|
||||
struct.pack_into("<I", header, 24, img_size)
|
||||
struct.pack_into("<h", header, 28, 0)
|
||||
struct.pack_into("<h", header, 30, 0)
|
||||
struct.pack_into("<H", header, 32, visible_h)
|
||||
struct.pack_into("<H", header, 34, visible_w)
|
||||
struct.pack_into("<I", header, 40, img_size)
|
||||
return bytes(header) + mhod3
|
||||
|
||||
|
||||
def _write_mhod_string(mhod_type: int, string: str) -> bytes:
|
||||
if mhod_type == 3:
|
||||
encoded = string.encode("utf-16-le")
|
||||
encoding_byte = 2
|
||||
else:
|
||||
encoded = string.encode("utf-8")
|
||||
encoding_byte = 1
|
||||
str_len = len(encoded)
|
||||
padding = (4 - (str_len % 4)) % 4
|
||||
body = struct.pack("<I", str_len)
|
||||
body += struct.pack("<B", encoding_byte)
|
||||
body += b"\x00" * 3
|
||||
body += b"\x00" * 4
|
||||
body += encoded
|
||||
body += b"\x00" * padding
|
||||
total_len = MHOD_HEADER_SIZE + len(body)
|
||||
header = bytearray(MHOD_HEADER_SIZE)
|
||||
header[0:4] = b"mhod"
|
||||
struct.pack_into("<I", header, 4, MHOD_HEADER_SIZE)
|
||||
struct.pack_into("<I", header, 8, total_len)
|
||||
struct.pack_into("<H", header, 12, mhod_type)
|
||||
return bytes(header) + body
|
||||
|
||||
|
||||
def _coerce_ithmb_location(format_id: int, location) -> IthmbLocation:
|
||||
if isinstance(location, IthmbLocation):
|
||||
return IthmbLocation(
|
||||
_normalize_ithmb_filename(format_id, location.filename),
|
||||
int(location.offset),
|
||||
)
|
||||
if isinstance(location, tuple):
|
||||
return IthmbLocation(
|
||||
_normalize_ithmb_filename(format_id, location[0]),
|
||||
int(location[1]),
|
||||
)
|
||||
return IthmbLocation(_default_ithmb_filename(format_id), int(location or 0))
|
||||
|
||||
|
||||
def read_existing_artwork(artworkdb_path: str, artwork_dir: str) -> dict[int, dict]:
|
||||
"""Read existing ArtworkDB entries, returning {img_id: entry_dict}."""
|
||||
if not os.path.exists(artworkdb_path):
|
||||
return {}
|
||||
|
||||
try:
|
||||
with open(artworkdb_path, "rb") as f:
|
||||
data = f.read()
|
||||
except Exception:
|
||||
return {}
|
||||
|
||||
if len(data) < 32 or data[:4] != b"mhfd":
|
||||
return {}
|
||||
|
||||
entries = {}
|
||||
mhfd_header_size = struct.unpack_from("<I", data, 4)[0]
|
||||
child_count = struct.unpack_from("<I", data, 20)[0]
|
||||
if mhfd_header_size < 32 or mhfd_header_size > len(data):
|
||||
return {}
|
||||
|
||||
offset = mhfd_header_size
|
||||
for _ in range(child_count):
|
||||
if offset + 14 > len(data) or data[offset:offset + 4] != b"mhsd":
|
||||
break
|
||||
mhsd_header = struct.unpack_from("<I", data, offset + 4)[0]
|
||||
mhsd_total = struct.unpack_from("<I", data, offset + 8)[0]
|
||||
ds_type = struct.unpack_from("<H", data, offset + 12)[0]
|
||||
if mhsd_header < 14 or mhsd_total < mhsd_header or offset + mhsd_total > len(data):
|
||||
break
|
||||
|
||||
if ds_type == 1:
|
||||
dataset_end = offset + mhsd_total
|
||||
mhli_offset = offset + mhsd_header
|
||||
if mhli_offset + 12 <= dataset_end and data[mhli_offset:mhli_offset + 4] == b"mhli":
|
||||
mhli_header = struct.unpack_from("<I", data, mhli_offset + 4)[0]
|
||||
mhii_count = struct.unpack_from("<I", data, mhli_offset + 8)[0]
|
||||
if mhli_header < 12 or mhli_offset + mhli_header > dataset_end:
|
||||
break
|
||||
mhii_offset = mhli_offset + mhli_header
|
||||
for _ in range(mhii_count):
|
||||
if mhii_offset + 52 > dataset_end or data[mhii_offset:mhii_offset + 4] != b"mhii":
|
||||
break
|
||||
mhii_total = struct.unpack_from("<I", data, mhii_offset + 8)[0]
|
||||
if mhii_total < 52 or mhii_offset + mhii_total > dataset_end:
|
||||
break
|
||||
entry = _parse_mhii_existing(data, mhii_offset, mhii_total, artwork_dir)
|
||||
if entry:
|
||||
entries[entry["img_id"]] = entry
|
||||
mhii_offset += mhii_total
|
||||
offset += mhsd_total
|
||||
|
||||
return entries
|
||||
|
||||
|
||||
def _parse_mhii_existing(data: bytes, offset: int, total_len: int, artwork_dir: str) -> dict | None:
|
||||
"""Parse one MHII entry from an existing ArtworkDB."""
|
||||
entry_end = offset + total_len
|
||||
if offset + 52 > entry_end:
|
||||
return None
|
||||
|
||||
header_size = struct.unpack_from("<I", data, offset + 4)[0]
|
||||
child_count = struct.unpack_from("<I", data, offset + 12)[0]
|
||||
img_id = struct.unpack_from("<I", data, offset + 16)[0]
|
||||
song_id = struct.unpack_from("<q", data, offset + 20)[0]
|
||||
src_img_size = struct.unpack_from("<I", data, offset + 48)[0]
|
||||
if header_size < 52 or header_size > total_len:
|
||||
return None
|
||||
|
||||
formats: dict[int, ExistingFormatRef] = {}
|
||||
child_offset = offset + header_size
|
||||
for _ in range(child_count):
|
||||
if child_offset + 14 > entry_end or data[child_offset:child_offset + 4] != b"mhod":
|
||||
break
|
||||
mhod_header = struct.unpack_from("<I", data, child_offset + 4)[0]
|
||||
mhod_total = struct.unpack_from("<I", data, child_offset + 8)[0]
|
||||
mhod_type = struct.unpack_from("<H", data, child_offset + 12)[0]
|
||||
if mhod_header < 14 or mhod_total < mhod_header or child_offset + mhod_total > entry_end:
|
||||
break
|
||||
|
||||
if mhod_type == 2:
|
||||
mhni_offset = child_offset + mhod_header
|
||||
child_end = child_offset + mhod_total
|
||||
if mhni_offset + MHNI_HEADER_SIZE <= child_end and data[mhni_offset:mhni_offset + 4] == b"mhni":
|
||||
fmt_id = struct.unpack_from("<I", data, mhni_offset + 16)[0]
|
||||
ithmb_offset = struct.unpack_from("<I", data, mhni_offset + 20)[0]
|
||||
img_size = struct.unpack_from("<I", data, mhni_offset + 24)[0]
|
||||
fname = _parse_mhni_filename(data, mhni_offset, child_end)
|
||||
fname = _normalize_ithmb_filename(fmt_id, fname)
|
||||
ipath = os.path.join(artwork_dir, fname)
|
||||
if os.path.exists(ipath) and img_size > 0:
|
||||
formats[fmt_id] = ExistingFormatRef(
|
||||
path=ipath,
|
||||
ithmb_offset=ithmb_offset,
|
||||
size=img_size,
|
||||
width=max(1, struct.unpack_from("<H", data, mhni_offset + 34)[0]),
|
||||
height=max(1, struct.unpack_from("<H", data, mhni_offset + 32)[0]),
|
||||
hpad=max(0, struct.unpack_from("<h", data, mhni_offset + 30)[0]),
|
||||
vpad=max(0, struct.unpack_from("<h", data, mhni_offset + 28)[0]),
|
||||
ithmb_filename=fname,
|
||||
)
|
||||
child_offset += mhod_total
|
||||
|
||||
if not formats:
|
||||
return None
|
||||
return {"img_id": img_id, "song_id": song_id, "src_img_size": src_img_size, "formats": formats}
|
||||
|
||||
|
||||
def _parse_mhni_filename(data: bytes, mhni_offset: int, container_end: int) -> str | None:
|
||||
"""Read the MHOD type=3 filename child from an MHNI chunk."""
|
||||
if mhni_offset + 12 > container_end:
|
||||
return None
|
||||
mhni_header = struct.unpack_from("<I", data, mhni_offset + 4)[0]
|
||||
mhni_total = struct.unpack_from("<I", data, mhni_offset + 8)[0]
|
||||
if mhni_header < MHNI_HEADER_SIZE:
|
||||
mhni_header = MHNI_HEADER_SIZE
|
||||
mhni_end = min(container_end, mhni_offset + mhni_total)
|
||||
child_offset = mhni_offset + mhni_header
|
||||
while child_offset + MHOD_HEADER_SIZE <= mhni_end:
|
||||
if data[child_offset:child_offset + 4] != b"mhod":
|
||||
break
|
||||
mhod_header = struct.unpack_from("<I", data, child_offset + 4)[0]
|
||||
mhod_total = struct.unpack_from("<I", data, child_offset + 8)[0]
|
||||
mhod_type = struct.unpack_from("<H", data, child_offset + 12)[0]
|
||||
if mhod_header < MHOD_HEADER_SIZE or mhod_total < mhod_header:
|
||||
break
|
||||
if child_offset + mhod_total > mhni_end:
|
||||
break
|
||||
if mhod_type == 3:
|
||||
return _parse_mhod_string(data, child_offset, mhod_total)
|
||||
child_offset += mhod_total
|
||||
return None
|
||||
|
||||
|
||||
def _parse_mhod_string(data: bytes, offset: int, total_len: int) -> str | None:
|
||||
"""Parse an ArtworkDB string MHOD body."""
|
||||
if offset + MHOD_HEADER_SIZE + 12 > offset + total_len:
|
||||
return None
|
||||
header_size = struct.unpack_from("<I", data, offset + 4)[0]
|
||||
if header_size < MHOD_HEADER_SIZE or header_size > total_len:
|
||||
return None
|
||||
body_offset = offset + header_size
|
||||
body_end = offset + total_len
|
||||
if body_offset + 12 > body_end:
|
||||
return None
|
||||
str_len = struct.unpack_from("<I", data, body_offset)[0]
|
||||
encoding_byte = data[body_offset + 4]
|
||||
raw_start = body_offset + 12
|
||||
raw_end = min(body_end, raw_start + str_len)
|
||||
raw = data[raw_start:raw_end]
|
||||
try:
|
||||
if encoding_byte == 2:
|
||||
return raw.decode("utf-16-le", errors="replace").rstrip("\x00")
|
||||
return raw.decode("utf-8", errors="replace").rstrip("\x00")
|
||||
except UnicodeError:
|
||||
return None
|
||||
@@ -0,0 +1,260 @@
|
||||
"""Format-aware ithmb encode/decode for iPod artwork.
|
||||
|
||||
Ported from iOpenPod ArtworkDB_Writer/ithmb_codecs.py.
|
||||
Supports RGB565_LE, RGB565_BE, RGB565_BE_90, RGB555_LE, RGB555_BE,
|
||||
REC_RGB555_LE, UYVY, I420_LE, JPEG.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import logging
|
||||
|
||||
import numpy as np
|
||||
from PIL import Image
|
||||
|
||||
from .presets import ARTWORK_FORMATS_BY_ID, ArtworkFormat
|
||||
from .types import EncodedFormatPayload
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _fmt(format_id: int, fmt_override: ArtworkFormat | None = None) -> ArtworkFormat | None:
|
||||
return fmt_override if fmt_override is not None else ARTWORK_FORMATS_BY_ID.get(format_id)
|
||||
|
||||
|
||||
def format_pixel_format(format_id: int, fmt_override: ArtworkFormat | None = None) -> str:
|
||||
fmt = _fmt(format_id, fmt_override)
|
||||
return fmt.pixel_format if fmt is not None else "UNKNOWN"
|
||||
|
||||
|
||||
def format_dimensions(
|
||||
format_id: int,
|
||||
fallback_w: int,
|
||||
fallback_h: int,
|
||||
fmt_override: ArtworkFormat | None = None,
|
||||
) -> tuple[int, int]:
|
||||
fmt = _fmt(format_id, fmt_override)
|
||||
if fmt is None:
|
||||
return fallback_w, fallback_h
|
||||
return int(fmt.width), int(fmt.height)
|
||||
|
||||
|
||||
def default_stride_pixels(format_id: int, width: int, fmt_override: ArtworkFormat | None = None) -> int:
|
||||
fmt = _fmt(format_id, fmt_override)
|
||||
if fmt is None:
|
||||
return width
|
||||
pf = fmt.pixel_format
|
||||
if pf in ("RGB565_LE", "RGB565_BE", "RGB565_BE_90", "RGB555_LE", "RGB555_BE"):
|
||||
return max(width, int(fmt.row_bytes // 2) if fmt.row_bytes else width)
|
||||
if pf.startswith("REC_RGB555"):
|
||||
return max(width, int(fmt.row_bytes // 2) if fmt.row_bytes else width)
|
||||
if pf == "UYVY":
|
||||
return max(width, int(fmt.row_bytes // 2) if fmt.row_bytes else width)
|
||||
return width
|
||||
|
||||
|
||||
def expected_size_bytes(
|
||||
format_id: int,
|
||||
width: int,
|
||||
height: int,
|
||||
stride_pixels: int | None = None,
|
||||
fmt_override: ArtworkFormat | None = None,
|
||||
) -> int:
|
||||
pf = format_pixel_format(format_id, fmt_override=fmt_override)
|
||||
stride = stride_pixels if stride_pixels is not None else default_stride_pixels(
|
||||
format_id, width, fmt_override=fmt_override,
|
||||
)
|
||||
if pf in ("RGB565_LE", "RGB565_BE", "RGB565_BE_90", "RGB555_LE", "RGB555_BE", "UYVY") or pf.startswith("REC_RGB555"):
|
||||
return int(stride) * int(height) * 2
|
||||
if pf == "I420_LE":
|
||||
w = int(width) & ~1
|
||||
h = int(height) & ~1
|
||||
return (w * h * 3) // 2
|
||||
if pf in ("JPEG", "UNKNOWN"):
|
||||
return 0
|
||||
return int(stride) * int(height) * 2
|
||||
|
||||
|
||||
def _rgb565_array_from_image(img: Image.Image) -> np.ndarray:
|
||||
arr = np.array(img.convert("RGB"), dtype=np.uint32)
|
||||
r = (arr[:, :, 0] >> 3) & 0x1F
|
||||
g = (arr[:, :, 1] >> 2) & 0x3F
|
||||
b = (arr[:, :, 2] >> 3) & 0x1F
|
||||
return ((r << 11) | (g << 5) | b).astype(np.uint16)
|
||||
|
||||
|
||||
def _rgb565_to_rgb(arr16: np.ndarray) -> np.ndarray:
|
||||
r = ((arr16 >> 11) & 0x1F).astype(np.uint8)
|
||||
g = ((arr16 >> 5) & 0x3F).astype(np.uint8)
|
||||
b = (arr16 & 0x1F).astype(np.uint8)
|
||||
return np.stack(((r << 3) | (r >> 2), (g << 2) | (g >> 4), (b << 3) | (b >> 2)), axis=2)
|
||||
|
||||
|
||||
def _rgb555_array_from_image(img: Image.Image) -> np.ndarray:
|
||||
arr = np.array(img.convert("RGB"), dtype=np.uint32)
|
||||
r = (arr[:, :, 0] >> 3) & 0x1F
|
||||
g = (arr[:, :, 1] >> 3) & 0x1F
|
||||
b = (arr[:, :, 2] >> 3) & 0x1F
|
||||
return ((r << 10) | (g << 5) | b).astype(np.uint16)
|
||||
|
||||
|
||||
def _rgb555_to_rgb(arr16: np.ndarray) -> np.ndarray:
|
||||
r = ((arr16 >> 10) & 0x1F).astype(np.uint8)
|
||||
g = ((arr16 >> 5) & 0x1F).astype(np.uint8)
|
||||
b = (arr16 & 0x1F).astype(np.uint8)
|
||||
return np.stack(((r << 3) | (r >> 2), (g << 3) | (g >> 2), (b << 3) | (b >> 2)), axis=2)
|
||||
|
||||
|
||||
def _pad_packed_rows(arr16: np.ndarray, stride_pixels: int) -> np.ndarray:
|
||||
stride = max(1, int(stride_pixels))
|
||||
height, width = arr16.shape[:2]
|
||||
if stride <= width:
|
||||
return arr16
|
||||
padded = np.zeros((height, stride), dtype=arr16.dtype)
|
||||
padded[:, :width] = arr16
|
||||
return padded
|
||||
|
||||
|
||||
def encode_for_ipod(
|
||||
source_img: Image.Image,
|
||||
format_id: int,
|
||||
target_width: int | None = None,
|
||||
target_height: int | None = None,
|
||||
fmt_override: ArtworkFormat | None = None,
|
||||
) -> EncodedFormatPayload:
|
||||
"""Encode a PIL Image into an iPod-native format payload."""
|
||||
pf = format_pixel_format(format_id, fmt_override=fmt_override)
|
||||
w, h = format_dimensions(
|
||||
format_id,
|
||||
int(target_width or source_img.width),
|
||||
int(target_height or source_img.height),
|
||||
fmt_override=fmt_override,
|
||||
)
|
||||
stride = default_stride_pixels(format_id, w, fmt_override=fmt_override)
|
||||
base = source_img.convert("RGB").resize((w, h), Image.Resampling.LANCZOS)
|
||||
|
||||
if pf == "RGB565_BE_90":
|
||||
rotated = base.transpose(Image.Transpose.ROTATE_270)
|
||||
arr16 = _rgb565_array_from_image(rotated)
|
||||
arr16 = _pad_packed_rows(arr16, stride)
|
||||
raw = arr16.astype(">u2").tobytes()
|
||||
return EncodedFormatPayload(raw, w, h, len(raw), stride, pixel_format=pf)
|
||||
|
||||
if pf == "RGB565_BE":
|
||||
arr16 = _rgb565_array_from_image(base)
|
||||
arr16 = _pad_packed_rows(arr16, stride)
|
||||
raw = arr16.astype(">u2").tobytes()
|
||||
return EncodedFormatPayload(raw, w, h, len(raw), stride, pixel_format=pf)
|
||||
|
||||
if pf == "RGB555_BE":
|
||||
arr16 = _rgb555_array_from_image(base)
|
||||
arr16 = _pad_packed_rows(arr16, stride)
|
||||
raw = arr16.astype(">u2").tobytes()
|
||||
return EncodedFormatPayload(raw, w, h, len(raw), stride, pixel_format=pf)
|
||||
|
||||
if pf in ("RGB555_LE", "REC_RGB555_LE"):
|
||||
arr16 = _rgb555_array_from_image(base)
|
||||
arr16 = _pad_packed_rows(arr16, stride)
|
||||
raw = arr16.astype("<u2").tobytes()
|
||||
return EncodedFormatPayload(raw, w, h, len(raw), stride, pixel_format=pf)
|
||||
|
||||
if pf == "JPEG":
|
||||
out = io.BytesIO()
|
||||
base.save(out, format="JPEG", quality=92, optimize=False)
|
||||
raw = out.getvalue()
|
||||
return EncodedFormatPayload(raw, w, h, len(raw), stride, pixel_format=pf)
|
||||
|
||||
if pf == "UYVY":
|
||||
if w % 2 != 0:
|
||||
w -= 1
|
||||
base = base.resize((w, h), Image.Resampling.LANCZOS)
|
||||
arr = np.array(base, dtype=np.float32)
|
||||
r, g, b = arr[:, :, 0], arr[:, :, 1], arr[:, :, 2]
|
||||
y = np.clip(0.257 * r + 0.504 * g + 0.098 * b + 16, 0, 255).astype(np.uint8)
|
||||
u = np.clip(-0.148 * r - 0.291 * g + 0.439 * b + 128, 0, 255)
|
||||
v = np.clip(0.439 * r - 0.368 * g - 0.071 * b + 128, 0, 255)
|
||||
u2 = ((u[:, 0::2] + u[:, 1::2]) * 0.5).astype(np.uint8)
|
||||
v2 = ((v[:, 0::2] + v[:, 1::2]) * 0.5).astype(np.uint8)
|
||||
packed = np.empty((h, w * 2), dtype=np.uint8)
|
||||
packed[:, 0::4] = u2
|
||||
packed[:, 1::4] = y[:, 0::2]
|
||||
packed[:, 2::4] = v2
|
||||
packed[:, 3::4] = y[:, 1::2]
|
||||
raw = packed.tobytes()
|
||||
return EncodedFormatPayload(raw, w, h, len(raw), stride, pixel_format=pf)
|
||||
|
||||
if pf == "I420_LE":
|
||||
w_even, h_even = w & ~1, h & ~1
|
||||
if w_even != w or h_even != h:
|
||||
w, h = w_even, h_even
|
||||
base = base.resize((w, h), Image.Resampling.LANCZOS)
|
||||
arr = np.array(base, dtype=np.float32)
|
||||
r, g, b = arr[:, :, 0], arr[:, :, 1], arr[:, :, 2]
|
||||
y = np.clip(0.257 * r + 0.504 * g + 0.098 * b + 16, 0, 255).astype(np.uint8)
|
||||
u = np.clip(-0.148 * r - 0.291 * g + 0.439 * b + 128, 0, 255)
|
||||
v = np.clip(0.439 * r - 0.368 * g - 0.071 * b + 128, 0, 255)
|
||||
u420 = ((u[0::2, 0::2] + u[0::2, 1::2] + u[1::2, 0::2] + u[1::2, 1::2]) * 0.25).astype(np.uint8)
|
||||
v420 = ((v[0::2, 0::2] + v[0::2, 1::2] + v[1::2, 0::2] + v[1::2, 1::2]) * 0.25).astype(np.uint8)
|
||||
raw = y.tobytes() + u420.tobytes() + v420.tobytes()
|
||||
return EncodedFormatPayload(raw, w, h, len(raw), stride, pixel_format=pf)
|
||||
|
||||
arr16 = _rgb565_array_from_image(base)
|
||||
arr16 = _pad_packed_rows(arr16, stride)
|
||||
raw = arr16.astype("<u2").tobytes()
|
||||
return EncodedFormatPayload(raw, w, h, len(raw), stride, "RGB565_LE")
|
||||
|
||||
|
||||
def decode_from_ipod(
|
||||
pixel_bytes: bytes,
|
||||
format_id: int,
|
||||
width: int,
|
||||
height: int,
|
||||
hpad: int = 0,
|
||||
vpad: int = 0,
|
||||
fmt_override: ArtworkFormat | None = None,
|
||||
) -> Image.Image | None:
|
||||
"""Decode iPod-native pixel data back to a PIL Image."""
|
||||
pf = format_pixel_format(format_id, fmt_override=fmt_override)
|
||||
width = max(1, int(width))
|
||||
height = max(1, int(height))
|
||||
|
||||
if pf in ("RGB565_LE", "RGB565_BE", "RGB565_BE_90"):
|
||||
dtype = "<u2" if pf == "RGB565_LE" else ">u2"
|
||||
stride = default_stride_pixels(format_id, width, fmt_override=fmt_override)
|
||||
stored_h = height
|
||||
needed = stride * stored_h * 2
|
||||
if len(pixel_bytes) < needed:
|
||||
return None
|
||||
arr = np.frombuffer(pixel_bytes[:needed], dtype=dtype)
|
||||
if arr.size != stride * stored_h:
|
||||
return None
|
||||
arr = arr.reshape((stored_h, stride))
|
||||
arr = arr[:, :width]
|
||||
rgb = _rgb565_to_rgb(arr)
|
||||
if pf == "RGB565_BE_90":
|
||||
rgb = np.rot90(rgb, k=1)
|
||||
return Image.fromarray(rgb)
|
||||
|
||||
if pf in ("RGB555_LE", "RGB555_BE", "REC_RGB555_LE"):
|
||||
dtype = "<u2" if pf != "RGB555_BE" else ">u2"
|
||||
stride = default_stride_pixels(format_id, width, fmt_override=fmt_override)
|
||||
stored_h = height
|
||||
needed = stride * stored_h * 2
|
||||
if len(pixel_bytes) < needed:
|
||||
return None
|
||||
arr = np.frombuffer(pixel_bytes[:needed], dtype=dtype)
|
||||
if arr.size != stride * stored_h:
|
||||
return None
|
||||
arr = arr.reshape((stored_h, stride))
|
||||
arr = arr[:, :width]
|
||||
rgb = _rgb555_to_rgb(arr)
|
||||
return Image.fromarray(rgb)
|
||||
|
||||
if pf == "JPEG":
|
||||
try:
|
||||
return Image.open(io.BytesIO(pixel_bytes)).convert("RGB")
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
return None
|
||||
@@ -0,0 +1,160 @@
|
||||
"""Extract embedded album art from media files using mutagen.
|
||||
|
||||
Ported from iOpenPod ArtworkDB_Writer/art_extractor.py.
|
||||
Supports: MP3, M4A/AAC, M4B, M4V/MP4, FLAC, OGG, OPUS, AIFF/WAV.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
import mutagen
|
||||
|
||||
|
||||
def extract_cover(file_path: str) -> bytes | None:
|
||||
"""Extract the first embedded album art image from a media file.
|
||||
|
||||
Returns raw image bytes (JPEG/PNG) or None.
|
||||
"""
|
||||
path = Path(file_path)
|
||||
ext = path.suffix.lower()
|
||||
|
||||
try:
|
||||
if ext in _IMAGE_EXTS:
|
||||
return path.read_bytes()
|
||||
if ext == ".mp3":
|
||||
return _extract_mp3(file_path)
|
||||
elif ext in (".m4a", ".m4p", ".m4b", ".aac", ".alac"):
|
||||
return _extract_mp4(file_path)
|
||||
elif ext in (".m4v", ".mp4", ".mov"):
|
||||
art = _extract_mp4(file_path)
|
||||
if art:
|
||||
return art
|
||||
return None
|
||||
elif ext == ".flac":
|
||||
return _extract_flac(file_path)
|
||||
elif ext == ".ogg":
|
||||
return _extract_ogg(file_path)
|
||||
elif ext == ".opus":
|
||||
return _extract_opus(file_path)
|
||||
elif ext in (".aif", ".aiff"):
|
||||
return _extract_aiff(file_path)
|
||||
else:
|
||||
return _extract_generic(file_path)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def find_folder_art(file_path: str) -> str | None:
|
||||
"""Return path of folder artwork image next to *file_path*, or None."""
|
||||
directory = str(Path(file_path).parent)
|
||||
try:
|
||||
entries = os.listdir(directory)
|
||||
except OSError:
|
||||
return None
|
||||
lower_map = {e.lower(): e for e in entries}
|
||||
for stem in _FOLDER_ART_NAMES:
|
||||
for ext in _IMAGE_EXTS:
|
||||
if (stem + ext) in lower_map:
|
||||
return os.path.join(directory, lower_map[stem + ext])
|
||||
return None
|
||||
|
||||
|
||||
def extract_cover_with_fallback(file_path: str) -> bytes | None:
|
||||
"""Extract cover art, falling back to folder.jpg if no embedded art."""
|
||||
art = extract_cover(file_path)
|
||||
if art is not None:
|
||||
return art
|
||||
folder_img = find_folder_art(file_path)
|
||||
if folder_img:
|
||||
try:
|
||||
return Path(folder_img).read_bytes()
|
||||
except OSError:
|
||||
pass
|
||||
return None
|
||||
|
||||
|
||||
_FOLDER_ART_NAMES = ("cover", "folder", "album", "front", "artwork", "thumb")
|
||||
_IMAGE_EXTS = (".jpg", ".jpeg", ".png", ".bmp", ".gif", ".webp")
|
||||
|
||||
|
||||
def _extract_mp3(path: str) -> bytes | None:
|
||||
from mutagen.mp3 import MP3
|
||||
audio = MP3(path)
|
||||
if audio.tags is None:
|
||||
return None
|
||||
for key in audio.tags:
|
||||
if key.startswith("APIC"):
|
||||
return audio.tags[key].data
|
||||
return None
|
||||
|
||||
|
||||
def _extract_mp4(path: str) -> bytes | None:
|
||||
from mutagen.mp4 import MP4
|
||||
audio = MP4(path)
|
||||
if audio.tags is None:
|
||||
return None
|
||||
covers = audio.tags.get("covr", [])
|
||||
if covers:
|
||||
return bytes(covers[0])
|
||||
return None
|
||||
|
||||
|
||||
def _extract_flac(path: str) -> bytes | None:
|
||||
from mutagen.flac import FLAC
|
||||
audio = FLAC(path)
|
||||
if audio.pictures:
|
||||
return audio.pictures[0].data
|
||||
return None
|
||||
|
||||
|
||||
def _extract_ogg(path: str) -> bytes | None:
|
||||
from mutagen.oggvorbis import OggVorbis
|
||||
audio = OggVorbis(path)
|
||||
return _extract_vorbis_picture(audio)
|
||||
|
||||
|
||||
def _extract_opus(path: str) -> bytes | None:
|
||||
from mutagen.oggopus import OggOpus
|
||||
audio = OggOpus(path)
|
||||
return _extract_vorbis_picture(audio)
|
||||
|
||||
|
||||
def _extract_vorbis_picture(audio) -> bytes | None:
|
||||
import base64
|
||||
pictures = audio.get("metadata_block_picture", [])
|
||||
if pictures:
|
||||
try:
|
||||
from mutagen.flac import Picture
|
||||
pic = Picture(base64.b64decode(pictures[0]))
|
||||
return pic.data
|
||||
except Exception:
|
||||
pass
|
||||
return None
|
||||
|
||||
|
||||
def _extract_aiff(path: str) -> bytes | None:
|
||||
from mutagen.aiff import AIFF
|
||||
audio = AIFF(path)
|
||||
if audio.tags is None:
|
||||
return None
|
||||
for key in audio.tags:
|
||||
if key.startswith("APIC"):
|
||||
return audio.tags[key].data
|
||||
return None
|
||||
|
||||
|
||||
def _extract_generic(path: str) -> bytes | None:
|
||||
audio = mutagen.File(path)
|
||||
if audio is None or audio.tags is None:
|
||||
return None
|
||||
for key in audio.tags:
|
||||
if hasattr(key, "startswith") and key.startswith("APIC"):
|
||||
frame = audio.tags[key]
|
||||
if hasattr(frame, "data"):
|
||||
return frame.data
|
||||
covers = audio.tags.get("covr", [])
|
||||
if covers:
|
||||
return bytes(covers[0])
|
||||
return None
|
||||
@@ -0,0 +1,82 @@
|
||||
"""Canonical ithmb artwork format definitions for iPod devices.
|
||||
|
||||
Ported from iOpenPod ArtworkDB_Writer/artwork_presets.py.
|
||||
For Nano 7G the relevant formats are 1010 (240x240), 1013 (50x50),
|
||||
1015 (58x58), and 1016 (57x57) — all RGB565_LE.
|
||||
"""
|
||||
|
||||
from dataclasses import dataclass
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ArtworkFormat:
|
||||
format_id: int
|
||||
width: int
|
||||
height: int
|
||||
row_bytes: int
|
||||
pixel_format: str = "RGB565_LE"
|
||||
role: str = "cover"
|
||||
description: str = ""
|
||||
|
||||
|
||||
ARTWORK_FORMATS_BY_ID: dict[int, ArtworkFormat] = {
|
||||
1005: ArtworkFormat(1005, 80, 80, 160, "RGB565_LE", "photo_thumb", "Nano 7G photo thumbnail"),
|
||||
1007: ArtworkFormat(1007, 480, 864, 960, "RGB565_LE", "photo_full", "Nano 7G photo full screen"),
|
||||
1009: ArtworkFormat(1009, 42, 30, 84, "RGB565_LE", "photo_list", "Photo list thumbnail"),
|
||||
1010: ArtworkFormat(1010, 240, 240, 480, "RGB565_LE", "cover_large", "Nano 7G album art large"),
|
||||
1013: ArtworkFormat(1013, 220, 176, 440, "RGB565_BE_90", "photo_full", "Photo full screen (rotated)"),
|
||||
1015: ArtworkFormat(1015, 130, 88, 260, "RGB565_LE", "photo_preview", "Photo/Video preview"),
|
||||
1016: ArtworkFormat(1016, 140, 140, 280, "RGB565_LE", "cover_large", "Photo album art large"),
|
||||
1017: ArtworkFormat(1017, 56, 56, 112, "RGB565_LE", "cover_small", "Photo album art small"),
|
||||
1019: ArtworkFormat(1019, 720, 480, 1440, "UYVY", "tv_out", "Photo/Video NTSC TV output"),
|
||||
1020: ArtworkFormat(1020, 220, 176, 440, "RGB565_BE_90", "photo_full", "Photo full screen (alt rotated)"),
|
||||
1023: ArtworkFormat(1023, 176, 132, 352, "RGB565_BE", "photo_full", "Nano full screen"),
|
||||
1024: ArtworkFormat(1024, 320, 240, 640, "RGB565_LE", "photo_full", "320x240 photo full screen"),
|
||||
1027: ArtworkFormat(1027, 100, 100, 200, "RGB565_LE", "cover_large", "Nano album art large"),
|
||||
1028: ArtworkFormat(1028, 100, 100, 200, "RGB565_LE", "cover_small", "Video album art small"),
|
||||
1029: ArtworkFormat(1029, 200, 200, 400, "RGB565_LE", "cover_large", "Video album art large"),
|
||||
1031: ArtworkFormat(1031, 42, 42, 84, "RGB565_LE", "cover_small", "Nano album art small"),
|
||||
1032: ArtworkFormat(1032, 42, 37, 84, "RGB565_LE", "photo_list", "Nano list thumbnail"),
|
||||
1036: ArtworkFormat(1036, 50, 41, 100, "RGB565_LE", "photo_list", "Video list thumbnail"),
|
||||
1044: ArtworkFormat(1044, 128, 128, 256, "RGB565_LE", "cover_medium", "Classic album art medium"),
|
||||
1055: ArtworkFormat(1055, 128, 128, 256, "RGB565_LE", "cover_medium", "Classic album art medium"),
|
||||
1056: ArtworkFormat(1056, 128, 128, 256, "RGB565_LE", "cover_medium_alt", "128x128 cover art (alternate)"),
|
||||
1060: ArtworkFormat(1060, 320, 320, 640, "RGB565_LE", "cover_large", "Classic album art large"),
|
||||
1061: ArtworkFormat(1061, 56, 56, 112, "RGB565_LE", "cover_small", "Classic album art small"),
|
||||
1066: ArtworkFormat(1066, 64, 64, 128, "RGB565_LE", "photo_thumb", "Classic photo thumbnail"),
|
||||
1067: ArtworkFormat(1067, 720, 480, 1080, "I420_LE", "tv_out", "Classic TV output (YUV)"),
|
||||
1068: ArtworkFormat(1068, 128, 128, 256, "RGB565_LE", "cover_medium_alt", "Classic album art medium (alt 2)"),
|
||||
1071: ArtworkFormat(1071, 240, 240, 480, "RGB565_LE", "cover_large", "Nano 4G album art large"),
|
||||
1073: ArtworkFormat(1073, 240, 240, 480, "RGB565_LE", "cover_large", "Nano 5G/6G album art large"),
|
||||
1074: ArtworkFormat(1074, 50, 50, 100, "RGB565_LE", "cover_xsmall", "Nano album art tiny"),
|
||||
1078: ArtworkFormat(1078, 80, 80, 160, "RGB565_LE", "cover_small", "Nano 4G/5G album art small"),
|
||||
1079: ArtworkFormat(1079, 80, 80, 160, "RGB565_LE", "photo_thumb", "Nano 4G/5G photo thumbnail"),
|
||||
1081: ArtworkFormat(1081, 640, 480, 0, "JPEG", "photo_full", "JPEG photo format (experimental/legacy)"),
|
||||
1083: ArtworkFormat(1083, 240, 320, 480, "RGB565_LE", "photo_full", "Nano 4G photo full screen (portrait)"),
|
||||
1084: ArtworkFormat(1084, 240, 240, 480, "RGB565_LE", "cover_large_alt", "Nano 4G album art (alt)"),
|
||||
1085: ArtworkFormat(1085, 88, 88, 176, "RGB565_LE", "cover_medium", "Nano 6G album art medium"),
|
||||
1087: ArtworkFormat(1087, 384, 384, 768, "RGB565_LE", "photo_large", "Nano 5G photo large"),
|
||||
1089: ArtworkFormat(1089, 58, 58, 116, "RGB565_LE", "cover_small", "Nano 6G album art small"),
|
||||
1092: ArtworkFormat(1092, 80, 80, 160, "RGB565_LE", "photo_thumb", "Nano 6G photo thumbnail"),
|
||||
1093: ArtworkFormat(1093, 512, 512, 1024, "RGB565_LE", "photo_full", "Nano 6G photo full screen"),
|
||||
2002: ArtworkFormat(2002, 50, 50, 100, "RGB565_BE", "cover_small", "iPod Mobile cover art small"),
|
||||
2003: ArtworkFormat(2003, 150, 150, 300, "RGB565_BE", "cover_large", "iPod Mobile cover art large"),
|
||||
3001: ArtworkFormat(3001, 256, 256, 512, "REC_RGB555_LE", "cover_large", "iPod touch cover art large"),
|
||||
3002: ArtworkFormat(3002, 128, 128, 256, "REC_RGB555_LE", "cover_medium", "iPod touch cover art medium"),
|
||||
3003: ArtworkFormat(3003, 64, 64, 128, "REC_RGB555_LE", "cover_small", "iPod touch cover art small"),
|
||||
3005: ArtworkFormat(3005, 320, 320, 640, "RGB555_LE", "cover_xlarge", "iPod touch cover art xlarge"),
|
||||
}
|
||||
|
||||
|
||||
NANO_7G_COVER_ART_FORMATS = (
|
||||
ARTWORK_FORMATS_BY_ID[1010],
|
||||
ArtworkFormat(1013, 50, 50, 100, "RGB565_LE", "cover_xsmall", "Nano 7G album art tiny"),
|
||||
ArtworkFormat(1015, 58, 58, 116, "RGB565_LE", "cover_small", "Nano 7G album art small"),
|
||||
ArtworkFormat(1016, 57, 57, 116, "RGB565_LE", "cover_small_alt", "Nano 7G album art small (aligned)"),
|
||||
)
|
||||
"""Nano 7G cover art format overrides:
|
||||
- 1010: 240x240 (large)
|
||||
- 1013: 50x50 (tiny, overrides global 220x176)
|
||||
- 1015: 58x58 (small, overrides global 130x88)
|
||||
- 1016: 57x57 (small alt, overrides global 140x140)
|
||||
"""
|
||||
@@ -0,0 +1,75 @@
|
||||
"""Typed models for the artwork pipeline.
|
||||
|
||||
Ported from iOpenPod ArtworkDB_Writer/artwork_types.py, simplified.
|
||||
"""
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class IthmbLocation:
|
||||
filename: str
|
||||
offset: int
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ExistingFormatRef:
|
||||
path: str
|
||||
ithmb_offset: int
|
||||
size: int
|
||||
width: int
|
||||
height: int
|
||||
hpad: int = 0
|
||||
vpad: int = 0
|
||||
ithmb_filename: str = ""
|
||||
|
||||
@property
|
||||
def stride_pixels(self) -> int:
|
||||
return max(1, self.width + self.hpad)
|
||||
|
||||
@property
|
||||
def stored_height(self) -> int:
|
||||
return max(1, self.height + self.vpad)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class EncodedFormatPayload:
|
||||
data: bytes
|
||||
width: int
|
||||
height: int
|
||||
size: int
|
||||
stride_pixels: int
|
||||
hpad: int = 0
|
||||
vpad: int = 0
|
||||
pixel_format: str | None = None
|
||||
|
||||
@classmethod
|
||||
def from_existing_ref(cls, ref: ExistingFormatRef, data: bytes) -> "EncodedFormatPayload":
|
||||
return cls(
|
||||
data=data,
|
||||
width=ref.width,
|
||||
height=ref.height,
|
||||
size=ref.size,
|
||||
stride_pixels=ref.stride_pixels,
|
||||
hpad=ref.hpad,
|
||||
vpad=ref.vpad,
|
||||
)
|
||||
|
||||
|
||||
ArtworkFormatPayload = EncodedFormatPayload | ExistingFormatRef
|
||||
|
||||
|
||||
@dataclass
|
||||
class ArtworkPayload:
|
||||
formats: dict[int, ArtworkFormatPayload] = field(default_factory=dict)
|
||||
src_img_size: int = 0
|
||||
|
||||
|
||||
@dataclass
|
||||
class ArtworkEntry:
|
||||
img_id: int
|
||||
db_track_id: int
|
||||
art_hash: str | None
|
||||
src_img_size: int
|
||||
formats: dict[int, ArtworkFormatPayload] = field(default_factory=dict)
|
||||
db_track_ids: list[int] = field(default_factory=list)
|
||||
@@ -0,0 +1,187 @@
|
||||
"""Write artwork (ithmb + ArtworkDB) to iPod.
|
||||
|
||||
Ported from iOpenPod ArtworkDB_Writer/artwork_writer.py, simplified
|
||||
for our use case (Nano 7, always-new tracks, no incremental sync).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import logging
|
||||
import os
|
||||
|
||||
from PIL import Image
|
||||
|
||||
from .presets import NANO_7G_COVER_ART_FORMATS, ArtworkFormat
|
||||
from .types import ArtworkEntry, EncodedFormatPayload, ExistingFormatRef, IthmbLocation
|
||||
from .codecs import encode_for_ipod
|
||||
from .chunks import build_artworkdb
|
||||
from .extractor import extract_cover_with_fallback
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
ITHMB_MAX_SIZE_BYTES = 256 * 1000 * 1000
|
||||
|
||||
|
||||
def _ithmb_filename(format_id: int, index: int = 1) -> str:
|
||||
return f"F{int(format_id)}_{int(index)}.ithmb"
|
||||
|
||||
|
||||
def _art_hash(art_bytes: bytes) -> str:
|
||||
return hashlib.md5(art_bytes).hexdigest()
|
||||
|
||||
|
||||
def image_from_bytes(art_bytes: bytes) -> Image.Image | None:
|
||||
"""Load a PIL Image from raw bytes (JPEG/PNG), ensuring RGB mode."""
|
||||
from io import BytesIO
|
||||
try:
|
||||
img = Image.open(BytesIO(art_bytes))
|
||||
if img.mode != "RGB":
|
||||
img = img.convert("RGBA").convert("RGB")
|
||||
return img
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def prepare_artwork_for_track(source_path: str) -> tuple[str, dict[int, EncodedFormatPayload]] | None:
|
||||
"""Extract cover art from a source file and encode as iPod formats.
|
||||
|
||||
Returns (art_hash, {format_id: EncodedFormatPayload}) or None.
|
||||
"""
|
||||
art_bytes = extract_cover_with_fallback(source_path)
|
||||
if art_bytes is None:
|
||||
return None
|
||||
|
||||
img = image_from_bytes(art_bytes)
|
||||
if img is None:
|
||||
return None
|
||||
|
||||
formats: dict[int, EncodedFormatPayload] = {}
|
||||
for fmt in NANO_7G_COVER_ART_FORMATS:
|
||||
try:
|
||||
encoded = encode_for_ipod(img, fmt.format_id, fmt_override=fmt)
|
||||
formats[fmt.format_id] = encoded
|
||||
except Exception as exc:
|
||||
logger.warning("Failed to encode format %d: %s", fmt.format_id, exc)
|
||||
|
||||
if not formats:
|
||||
return None
|
||||
|
||||
return _art_hash(art_bytes), formats
|
||||
|
||||
|
||||
def write_artworkdb(
|
||||
ipod_path: str,
|
||||
entries: list[ArtworkEntry],
|
||||
start_img_id: int = 100,
|
||||
) -> dict[int, tuple[int, int]]:
|
||||
"""Write ithmb files and ArtworkDB binary to the iPod.
|
||||
|
||||
Args:
|
||||
ipod_path: iPod mount point
|
||||
entries: List of ArtworkEntry (one per unique album art)
|
||||
start_img_id: Starting image ID
|
||||
|
||||
Returns:
|
||||
Dict mapping db_track_id → (img_id, src_img_size)
|
||||
"""
|
||||
artwork_dir = os.path.join(ipod_path, "iPod_Control", "Artwork")
|
||||
os.makedirs(artwork_dir, exist_ok=True)
|
||||
|
||||
# Collect all unique format IDs across all entries
|
||||
format_ids = sorted({fmt_id for entry in entries for fmt_id in entry.formats.keys()})
|
||||
writable_format_ids = sorted({
|
||||
fmt_id for entry in entries
|
||||
for fmt_id, payload in entry.formats.items()
|
||||
if isinstance(payload, EncodedFormatPayload)
|
||||
})
|
||||
|
||||
# Write ithmb files
|
||||
ithmb_state: dict[int, dict[str, int]] = {
|
||||
fmt_id: {"index": 0, "offset": 0}
|
||||
for fmt_id in writable_format_ids
|
||||
}
|
||||
ithmb_files: dict[int, object] = {}
|
||||
format_locations_map: dict[int, dict[int, IthmbLocation]] = {}
|
||||
image_sizes: dict[int, int] = {}
|
||||
|
||||
def _close_current(fmt_id: int) -> None:
|
||||
handle = ithmb_files.pop(fmt_id, None)
|
||||
if handle is not None:
|
||||
handle.close()
|
||||
|
||||
def _open_next(fmt_id: int) -> None:
|
||||
_close_current(fmt_id)
|
||||
state = ithmb_state[fmt_id]
|
||||
state["index"] += 1
|
||||
state["offset"] = 0
|
||||
filename = _ithmb_filename(fmt_id, state["index"])
|
||||
ithmb_files[fmt_id] = open(os.path.join(artwork_dir, filename), "wb")
|
||||
|
||||
def _write_payload(fmt_id: int, data: bytes) -> IthmbLocation:
|
||||
state = ithmb_state[fmt_id]
|
||||
if fmt_id not in ithmb_files:
|
||||
_open_next(fmt_id)
|
||||
elif state["offset"] > 0 and state["offset"] + len(data) > ITHMB_MAX_SIZE_BYTES:
|
||||
_open_next(fmt_id)
|
||||
filename = _ithmb_filename(fmt_id, state["index"])
|
||||
offset = state["offset"]
|
||||
ithmb_files[fmt_id].write(data)
|
||||
state["offset"] += len(data)
|
||||
return IthmbLocation(filename, offset)
|
||||
|
||||
try:
|
||||
for entry in entries:
|
||||
locations: dict[int, IthmbLocation] = {}
|
||||
for fmt_id in format_ids:
|
||||
if fmt_id not in entry.formats:
|
||||
continue
|
||||
payload = entry.formats[fmt_id]
|
||||
if isinstance(payload, EncodedFormatPayload):
|
||||
locations[fmt_id] = _write_payload(fmt_id, payload.data)
|
||||
elif isinstance(payload, ExistingFormatRef):
|
||||
fname = _ithmb_filename(fmt_id)
|
||||
locations[fmt_id] = IthmbLocation(fname, payload.ithmb_offset)
|
||||
format_locations_map[entry.img_id] = locations
|
||||
finally:
|
||||
for fmt_id in list(ithmb_files.keys()):
|
||||
_close_current(fmt_id)
|
||||
|
||||
# Calculate image sizes for MHIF
|
||||
for fmt_id in format_ids:
|
||||
sizes = [
|
||||
int(entry.formats[fmt_id].size)
|
||||
for entry in entries
|
||||
if fmt_id in entry.formats and int(entry.formats[fmt_id].size) > 0
|
||||
]
|
||||
if sizes:
|
||||
from collections import Counter
|
||||
image_sizes[fmt_id] = Counter(sizes).most_common(1)[0][0]
|
||||
|
||||
# Build and write ArtworkDB binary
|
||||
ref_mhfd = None
|
||||
existing_artdb = os.path.join(artwork_dir, "ArtworkDB")
|
||||
if os.path.exists(existing_artdb):
|
||||
with open(existing_artdb, "rb") as f:
|
||||
ref_mhfd = f.read()
|
||||
|
||||
next_id = start_img_id + len(entries)
|
||||
db_data = build_artworkdb(
|
||||
entries,
|
||||
format_locations_map,
|
||||
format_ids,
|
||||
image_sizes,
|
||||
next_id,
|
||||
ref_mhfd,
|
||||
)
|
||||
|
||||
with open(existing_artdb, "wb") as f:
|
||||
f.write(db_data)
|
||||
f.flush()
|
||||
os.fsync(f.fileno())
|
||||
|
||||
# Return track_id → (img_id, src_img_size) mapping
|
||||
return {
|
||||
entry.db_track_id: (entry.img_id, entry.src_img_size)
|
||||
for entry in entries
|
||||
}
|
||||
Reference in New Issue
Block a user