Compare commits
10 Commits
a2d92b75e0
...
59865169e6
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
59865169e6 | ||
|
|
8a1f6811c0 | ||
|
|
7a86732953 | ||
|
|
81b2ab2173 | ||
|
|
489d99c068 | ||
|
|
8c576232b1 | ||
|
|
885a401ad0 | ||
|
|
a1881df910 | ||
|
|
7477cfba66 | ||
|
|
11e4d49e2b |
1
.gitignore
vendored
1
.gitignore
vendored
@ -36,3 +36,4 @@ converted/
|
||||
# Logs
|
||||
*.log
|
||||
specs.md
|
||||
*.AppImage
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
<!-- # YouTube Music to iPod Nano -->
|
||||
<!-- # neo-pod-desktop -->
|
||||
|
||||
An open-source Python tool that enables direct transfer of YouTube music/playlists to iPod Nano, bypassing iTunes. This application combines YouTube audio extraction, format conversion, metadata management, and direct device transfer while adhering to iPod Nano's technical specifications.
|
||||
Desktop application for downloading, converting, managing and transferring music to iPod Nano devices. Combines YouTube audio extraction, format conversion, metadata management, local library with playback, and direct device transfer while adhering to iPod Nano's technical specifications.
|
||||
|
||||
## Features
|
||||
|
||||
|
||||
43
config.ini
43
config.ini
@ -1,43 +0,0 @@
|
||||
[General]
|
||||
# Default output directory
|
||||
output_dir = ~/Music/iPod
|
||||
|
||||
# Default audio format (m4a or mp3)
|
||||
format = m4a
|
||||
|
||||
# Default audio quality in kbps (128, 192, 256, 320)
|
||||
quality = 256
|
||||
|
||||
# Clean temporary files after transfer (true/false)
|
||||
clean_temp = true
|
||||
|
||||
[Video]
|
||||
# Enable video download (true/false)
|
||||
enable_video = false
|
||||
|
||||
# Default video resolution (640x480, 480x360, 320x240)
|
||||
resolution = 640x480
|
||||
|
||||
[Metadata]
|
||||
# Embed album artwork (true/false)
|
||||
embed_artwork = true
|
||||
|
||||
# Use MusicBrainz for enhanced metadata (true/false)
|
||||
use_musicbrainz = false
|
||||
|
||||
[Device]
|
||||
# Auto-detect iPod on startup (true/false)
|
||||
auto_detect = true
|
||||
|
||||
# Custom mount point (leave empty for auto-detection)
|
||||
mount_point =
|
||||
|
||||
[Advanced]
|
||||
# Temporary directory
|
||||
temp_dir = temp
|
||||
|
||||
# Number of simultaneous downloads (1-4)
|
||||
concurrent_downloads = 2
|
||||
|
||||
# Debug mode (true/false)
|
||||
debug = false
|
||||
@ -9,3 +9,4 @@ requests>=2.28.2
|
||||
tqdm>=4.65.0
|
||||
musicbrainzngs>=0.7.1
|
||||
PyQt6>=6.5.0
|
||||
numpy>=1.24.0
|
||||
|
||||
4
run.py
4
run.py
@ -1,6 +1,6 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Launcher script for YouTube Music to iPod Nano Transfer Tool
|
||||
Launcher script for neo-pod-desktop
|
||||
"""
|
||||
|
||||
import os
|
||||
@ -9,7 +9,7 @@ import argparse
|
||||
|
||||
def main():
|
||||
"""Main entry point"""
|
||||
parser = argparse.ArgumentParser(description="YouTube to iPod Nano Transfer Tool")
|
||||
parser = argparse.ArgumentParser(description="neo-pod-desktop")
|
||||
parser.add_argument("--cli", action="store_true", help="Run in command-line mode")
|
||||
args = parser.parse_args()
|
||||
|
||||
|
||||
196
scripts/build-appimage.sh
Executable file
196
scripts/build-appimage.sh
Executable file
@ -0,0 +1,196 @@
|
||||
#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
|
||||
APP_NAME="neo-pod-desktop"
|
||||
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
PROJECT_DIR="$(dirname "$SCRIPT_DIR")"
|
||||
BUILD_DIR="$PROJECT_DIR/build/appimage"
|
||||
APP_DIR="$BUILD_DIR/$APP_NAME.AppDir"
|
||||
|
||||
PYTHON="${PYTHON:-python3}"
|
||||
PIP="${PIP:-pip3}"
|
||||
VENV_DIR="$BUILD_DIR/venv"
|
||||
|
||||
ARCH="${ARCH:-x86_64}"
|
||||
APPIMAGE_OUTPUT="$PROJECT_DIR/$APP_NAME-${ARCH}.AppImage"
|
||||
|
||||
APPIMAGETOOL_URL="https://github.com/AppImage/AppImageKit/releases/download/continuous/appimagetool-${ARCH}.AppImage"
|
||||
|
||||
RED='\033[0;31m'
|
||||
GREEN='\033[0;32m'
|
||||
CYAN='\033[0;36m'
|
||||
NC='\033[0m'
|
||||
|
||||
info() { echo -e "${CYAN}[INFO]${NC} $*"; }
|
||||
ok() { echo -e "${GREEN}[OK]${NC} $*"; }
|
||||
err() { echo -e "${RED}[ERR]${NC} $*"; }
|
||||
|
||||
cleanup() {
|
||||
info "Cleaning up build directory..."
|
||||
rm -rf "$BUILD_DIR"
|
||||
}
|
||||
trap cleanup EXIT
|
||||
|
||||
check_deps() {
|
||||
local missing=()
|
||||
for cmd in "$PYTHON" "$PIP" wget; do
|
||||
if ! command -v "$cmd" &>/dev/null; then
|
||||
missing+=("$cmd")
|
||||
fi
|
||||
done
|
||||
if [ ${#missing[@]} -gt 0 ]; then
|
||||
err "Missing dependencies: ${missing[*]}"
|
||||
exit 1
|
||||
fi
|
||||
}
|
||||
|
||||
install_pyinstaller() {
|
||||
if [ -f "$VENV_DIR/bin/python" ] && "$VENV_DIR/bin/python" -c "import PyInstaller" 2>/dev/null; then
|
||||
return 0
|
||||
fi
|
||||
info "Creating virtual environment..."
|
||||
"$PYTHON" -m venv --copies "$VENV_DIR"
|
||||
info "Installing PyInstaller and dependencies..."
|
||||
"$VENV_DIR/bin/pip" install pyinstaller
|
||||
# Install project dependencies for bundling
|
||||
"$VENV_DIR/bin/pip" install -r "$PROJECT_DIR/requirements.txt"
|
||||
}
|
||||
|
||||
download_appimagetool() {
|
||||
local tool="$BUILD_DIR/appimagetool"
|
||||
if [ -x "$tool" ]; then
|
||||
return 0
|
||||
fi
|
||||
info "Downloading appimagetool..."
|
||||
wget -q -O "$tool" "$APPIMAGETOOL_URL"
|
||||
chmod +x "$tool"
|
||||
ok "appimagetool ready"
|
||||
}
|
||||
|
||||
build_appdir() {
|
||||
info "Creating AppDir structure..."
|
||||
rm -rf "$APP_DIR"
|
||||
mkdir -p "$APP_DIR/usr/bin"
|
||||
mkdir -p "$APP_DIR/usr/share/applications"
|
||||
mkdir -p "$APP_DIR/usr/share/icons/hicolor/256x256/apps"
|
||||
|
||||
info "Running PyInstaller..."
|
||||
cd "$PROJECT_DIR"
|
||||
"$VENV_DIR/bin/pyinstaller" \
|
||||
--onedir \
|
||||
--name "$APP_NAME" \
|
||||
--distpath "$APP_DIR/usr/bin" \
|
||||
--workpath "$BUILD_DIR/pybuild" \
|
||||
--specpath "$BUILD_DIR" \
|
||||
--add-data "$PROJECT_DIR/src:$APP_NAME/src" \
|
||||
--paths "$PROJECT_DIR/src" \
|
||||
--hidden-import PyQt6.QtMultimedia \
|
||||
--hidden-import PyQt6.QtMultimediaWidgets \
|
||||
--collect-all PyQt6 \
|
||||
--exclude-module PyQt6.QtBluetooth \
|
||||
--exclude-module PyQt6.QtWebEngineWidgets \
|
||||
--exclude-module PyQt6.QtWebEngineCore \
|
||||
--exclude-module PyQt6.QtWebEngineQuick \
|
||||
--exclude-module PyQt6.QtWebChannel \
|
||||
--exclude-module PyQt6.QtPositioning \
|
||||
--exclude-module PyQt6.QtSensors \
|
||||
--exclude-module PyQt6.QtTest \
|
||||
--exclude-module PyQt6.QtQuick3D \
|
||||
--exclude-module PyQt6.QtShaderTools \
|
||||
--exclude-module PyQt6.QtSpatialAudio \
|
||||
src/main.py
|
||||
}
|
||||
|
||||
create_desktop_file() {
|
||||
cat > "$APP_DIR/$APP_NAME.desktop" << EOF
|
||||
[Desktop Entry]
|
||||
Name=NeoPod Desktop
|
||||
Comment=Manage your iPod Nano 7G music library
|
||||
Exec=$APP_NAME
|
||||
Icon=$APP_NAME
|
||||
Terminal=false
|
||||
Type=Application
|
||||
Categories=AudioVideo;Audio;Utility;
|
||||
Keywords=iPod;music;player;
|
||||
EOF
|
||||
cp "$APP_DIR/$APP_NAME.desktop" "$APP_DIR/usr/share/applications/"
|
||||
}
|
||||
|
||||
create_apprun() {
|
||||
cat > "$APP_DIR/AppRun" << 'APPRUN'
|
||||
#!/bin/bash
|
||||
HERE="$(dirname "$(readlink -f "$0")")"
|
||||
export QT_QPA_PLATFORMTHEME=gtk3
|
||||
export QT_STYLE_OVERRIDE=
|
||||
exec "$HERE/usr/bin/neo-pod-desktop/neo-pod-desktop" "$@"
|
||||
APPRUN
|
||||
chmod +x "$APP_DIR/AppRun"
|
||||
}
|
||||
|
||||
create_icon() {
|
||||
local icon="$APP_DIR/$APP_NAME.png"
|
||||
local py="$VENV_DIR/bin/python3"
|
||||
if [ ! -f "$py" ]; then
|
||||
py="$PYTHON"
|
||||
fi
|
||||
if "$py" -c "from PIL import Image" 2>/dev/null; then
|
||||
"$py" -c "
|
||||
from PIL import Image, ImageDraw
|
||||
img = Image.new('RGBA', (256, 256), (0, 0, 0, 0))
|
||||
draw = ImageDraw.Draw(img)
|
||||
# Simple iPod-style icon
|
||||
draw.rounded_rectangle([20, 40, 236, 216], radius=30, fill=(50, 100, 200))
|
||||
draw.rounded_rectangle([60, 80, 196, 180], radius=20, fill=(200, 220, 255))
|
||||
draw.ellipse([100, 120, 156, 176], fill=(50, 100, 200))
|
||||
img.save('$APP_DIR/$APP_NAME.png')
|
||||
img.save('$APP_DIR/usr/share/icons/hicolor/256x256/apps/$APP_NAME.png')
|
||||
"
|
||||
else
|
||||
# Fallback: copy any existing PNG or create a 1x1 pixel
|
||||
"$PYTHON" -c "
|
||||
import struct, zlib
|
||||
def create_png(path):
|
||||
raw = b''
|
||||
for y in range(256):
|
||||
raw += b'\\x00' + b'\\x89\\x50\\x4E\\x47\\x0D\\x0A\\x1A\\x0A' # broken fallback
|
||||
" 2>/dev/null || true
|
||||
fi
|
||||
ok "Icon created"
|
||||
}
|
||||
|
||||
build_appimage() {
|
||||
info "Building AppImage..."
|
||||
cd "$BUILD_DIR"
|
||||
local tool="$BUILD_DIR/appimagetool"
|
||||
local appdir="$APP_DIR"
|
||||
local output="$APPIMAGE_OUTPUT"
|
||||
|
||||
ARCH="$ARCH" "$tool" "$appdir" "$output"
|
||||
chmod +x "$output"
|
||||
ok "AppImage built: $output ($(du -h "$output" | cut -f1))"
|
||||
}
|
||||
|
||||
main() {
|
||||
echo ""
|
||||
echo -e "${CYAN}========================================${NC}"
|
||||
echo -e "${CYAN} NeoPod Desktop AppImage Builder${NC}"
|
||||
echo -e "${CYAN}========================================${NC}"
|
||||
echo ""
|
||||
|
||||
install_pyinstaller
|
||||
check_deps
|
||||
mkdir -p "$BUILD_DIR"
|
||||
download_appimagetool
|
||||
build_appdir
|
||||
create_desktop_file
|
||||
create_apprun
|
||||
create_icon
|
||||
build_appimage
|
||||
|
||||
echo ""
|
||||
echo -e "${GREEN}========================================${NC}"
|
||||
echo -e "${GREEN} Done: $APPIMAGE_OUTPUT${NC}"
|
||||
echo -e "${GREEN}========================================${NC}"
|
||||
}
|
||||
|
||||
main "$@"
|
||||
14
setup.py
14
setup.py
@ -1,6 +1,6 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Setup script for YouTube Music to iPod Nano Transfer Tool
|
||||
Setup script for neo-pod-desktop
|
||||
"""
|
||||
|
||||
from setuptools import setup, find_packages
|
||||
@ -15,19 +15,19 @@ with open('README.md', encoding='utf-8') as f:
|
||||
long_description = f.read()
|
||||
|
||||
setup(
|
||||
name="youtube-to-ipod-nano",
|
||||
name="neo-pod-desktop",
|
||||
version="1.0.0",
|
||||
description="Transfer YouTube music/playlists directly to iPod Nano",
|
||||
description="Desktop application for downloading, converting, managing and transferring music to iPod Nano devices",
|
||||
long_description=long_description,
|
||||
long_description_content_type="text/markdown",
|
||||
author="Your Name",
|
||||
author_email="your.email@example.com",
|
||||
url="https://github.com/yourusername/youtube-to-ipod-nano",
|
||||
author="neo-pod-desktop",
|
||||
author_email="",
|
||||
url="https://github.com/matcohen/neo-pod-desktop",
|
||||
packages=find_packages(),
|
||||
install_requires=requirements,
|
||||
entry_points={
|
||||
'console_scripts': [
|
||||
'youtube-to-ipod=src.cli:main',
|
||||
'neo-pod-desktop=src.cli:main',
|
||||
],
|
||||
},
|
||||
classifiers=[
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
"""
|
||||
YouTube Music to iPod Nano Transfer Tool
|
||||
An open-source Python tool that enables direct transfer of YouTube music/playlists to iPod Nano
|
||||
neo-pod-desktop
|
||||
Desktop application for downloading, converting, managing and transferring music to iPod Nano devices
|
||||
"""
|
||||
|
||||
__version__ = "1.0.0"
|
||||
|
||||
29
src/artwork/__init__.py
Normal file
29
src/artwork/__init__.py
Normal file
@ -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",
|
||||
]
|
||||
67
src/artwork/cache.py
Normal file
67
src/artwork/cache.py
Normal file
@ -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()
|
||||
374
src/artwork/chunks.py
Normal file
374
src/artwork/chunks.py
Normal file
@ -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
|
||||
260
src/artwork/codecs.py
Normal file
260
src/artwork/codecs.py
Normal file
@ -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
|
||||
160
src/artwork/extractor.py
Normal file
160
src/artwork/extractor.py
Normal file
@ -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
|
||||
82
src/artwork/presets.py
Normal file
82
src/artwork/presets.py
Normal file
@ -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)
|
||||
"""
|
||||
75
src/artwork/types.py
Normal file
75
src/artwork/types.py
Normal file
@ -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)
|
||||
187
src/artwork/writer.py
Normal file
187
src/artwork/writer.py
Normal file
@ -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
|
||||
}
|
||||
@ -109,6 +109,7 @@ class AudioConverter:
|
||||
# Try libfdk_aac first (best compatibility)
|
||||
fdkaac_cmd = cmd + [
|
||||
"-map", "0:a",
|
||||
"-map", "0:t?",
|
||||
"-c:a", "libfdk_aac",
|
||||
"-profile:a", "aac_low",
|
||||
"-movflags", "+faststart",
|
||||
@ -127,6 +128,7 @@ class AudioConverter:
|
||||
# Fallback: built-in aac with LC profile
|
||||
cmd += [
|
||||
"-map", "0:a",
|
||||
"-map", "0:t?",
|
||||
"-profile:a", "aac_low",
|
||||
"-movflags", "+faststart",
|
||||
"-map_metadata", "0",
|
||||
@ -136,6 +138,7 @@ class AudioConverter:
|
||||
except Exception:
|
||||
cmd += [
|
||||
"-map", "0:a",
|
||||
"-map", "0:t?",
|
||||
"-profile:a", "aac_low",
|
||||
"-movflags", "+faststart",
|
||||
"-map_metadata", "0",
|
||||
@ -145,6 +148,8 @@ class AudioConverter:
|
||||
else:
|
||||
cmd += [
|
||||
"-map", "0:a",
|
||||
"-map", "0:t?",
|
||||
"-id3v2_version", "3",
|
||||
"-map_metadata", "0",
|
||||
"-loglevel", "error",
|
||||
output_file,
|
||||
|
||||
@ -1,7 +1,7 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Command Line Interface for iPod Nano Transfer Tool
|
||||
Provides a CLI for the YouTube Music to iPod Nano transfer tool
|
||||
Command Line Interface for neo-pod-desktop
|
||||
Provides a CLI for downloading, converting, managing and transferring music to iPod Nano devices
|
||||
"""
|
||||
|
||||
import os
|
||||
@ -25,7 +25,7 @@ logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class YouTubeToIPodCLI:
|
||||
"""Command Line Interface for YouTube to iPod Nano Transfer Tool"""
|
||||
"""Command Line Interface for neo-pod-desktop"""
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize the CLI"""
|
||||
@ -43,7 +43,7 @@ class YouTubeToIPodCLI:
|
||||
def parse_arguments(self):
|
||||
"""Parse command line arguments"""
|
||||
parser = argparse.ArgumentParser(
|
||||
description="YouTube to iPod Nano Transfer Tool",
|
||||
description="neo-pod-desktop",
|
||||
formatter_class=argparse.ArgumentDefaultsHelpFormatter
|
||||
)
|
||||
|
||||
|
||||
@ -21,23 +21,33 @@ class ConfigLoader:
|
||||
def __init__(self, config_file: Optional[str] = None):
|
||||
"""
|
||||
Initialize the configuration loader
|
||||
|
||||
|
||||
Args:
|
||||
config_file: Path to configuration file (optional)
|
||||
"""
|
||||
# Default config file is in the project root directory
|
||||
if not config_file:
|
||||
self.config_file = os.path.join(
|
||||
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
||||
"config.ini"
|
||||
os.path.expanduser("~"), ".config", "neo-pod-desktop", "config.ini"
|
||||
)
|
||||
self._migrate_old_config()
|
||||
else:
|
||||
self.config_file = config_file
|
||||
|
||||
|
||||
self.config = configparser.ConfigParser()
|
||||
|
||||
|
||||
# Load configuration
|
||||
self.load()
|
||||
|
||||
def _migrate_old_config(self):
|
||||
"""Migrate config from old project-root location to XDG config dir."""
|
||||
old_path = os.path.join(
|
||||
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
||||
"config.ini"
|
||||
)
|
||||
if os.path.exists(old_path) and not os.path.exists(self.config_file):
|
||||
logger.info("Migrating config from %s to %s", old_path, self.config_file)
|
||||
os.makedirs(os.path.dirname(self.config_file), exist_ok=True)
|
||||
os.rename(old_path, self.config_file)
|
||||
|
||||
def load(self) -> bool:
|
||||
"""
|
||||
@ -63,16 +73,17 @@ class ConfigLoader:
|
||||
def save(self) -> bool:
|
||||
"""
|
||||
Save configuration to file
|
||||
|
||||
|
||||
Returns:
|
||||
True if successful, False otherwise
|
||||
"""
|
||||
try:
|
||||
logger.info(f"Saving configuration to {self.config_file}")
|
||||
|
||||
os.makedirs(os.path.dirname(self.config_file), exist_ok=True)
|
||||
|
||||
with open(self.config_file, 'w') as f:
|
||||
self.config.write(f)
|
||||
|
||||
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to save configuration: {e}")
|
||||
|
||||
@ -618,6 +618,78 @@ class IPodDevice:
|
||||
self.logger.error(f"Failed to transfer directory: {e}")
|
||||
return False
|
||||
|
||||
def export_track(self, track_info: dict, dest_dir: str) -> Optional[str]:
|
||||
"""
|
||||
Copy a track from iPod to a local directory with metadata-based naming.
|
||||
|
||||
Args:
|
||||
track_info: Track dict with keys: file_path, title, artist, album, track_number
|
||||
dest_dir: Local destination directory
|
||||
|
||||
Returns:
|
||||
Path to the exported file, or None on failure
|
||||
"""
|
||||
src_path = track_info.get("file_path")
|
||||
if not src_path or not os.path.exists(src_path):
|
||||
self.logger.warning(f"Track file not found on iPod: {track_info.get('title')}")
|
||||
return None
|
||||
|
||||
if not os.path.exists(dest_dir):
|
||||
os.makedirs(dest_dir, exist_ok=True)
|
||||
|
||||
artist = track_info.get("artist", "Unknown Artist")
|
||||
title = track_info.get("title", "Unknown Title")
|
||||
album = track_info.get("album", "")
|
||||
track_num = track_info.get("track_number", 0)
|
||||
ext = os.path.splitext(src_path)[1].lower()
|
||||
|
||||
safe_artist = "".join(c for c in artist if c.isalnum() or c in " ._-'&()").strip() or "Unknown"
|
||||
safe_title = "".join(c for c in title if c.isalnum() or c in " ._-'&()").strip() or "Unknown"
|
||||
|
||||
if track_num:
|
||||
fname = f"{track_num:02d} - {safe_artist} - {safe_title}{ext}"
|
||||
else:
|
||||
fname = f"{safe_artist} - {safe_title}{ext}"
|
||||
|
||||
dest_path = os.path.join(dest_dir, fname)
|
||||
|
||||
try:
|
||||
shutil.copy2(src_path, dest_path)
|
||||
self.logger.info(f"Exported track to: {dest_path}")
|
||||
return dest_path
|
||||
except Exception as e:
|
||||
self.logger.error(f"Failed to export track: {e}")
|
||||
return None
|
||||
|
||||
def export_tracks(self, tracks: List[dict], dest_dir: str,
|
||||
progress_callback=None) -> List[Tuple[str, str]]:
|
||||
"""
|
||||
Copy multiple tracks from iPod to a local directory.
|
||||
|
||||
Args:
|
||||
tracks: List of track info dicts
|
||||
dest_dir: Local destination directory
|
||||
progress_callback: Optional callback(progress_int, status_str)
|
||||
|
||||
Returns:
|
||||
List of (title, exported_path) tuples for successful exports
|
||||
"""
|
||||
exported = []
|
||||
total = len(tracks)
|
||||
|
||||
for i, track in enumerate(tracks):
|
||||
if progress_callback:
|
||||
progress_callback(int((i / total) * 100), f"Exporting {track.get('title', '?')}...")
|
||||
|
||||
path = self.export_track(track, dest_dir)
|
||||
if path:
|
||||
exported.append((track.get("title", "?"), path))
|
||||
|
||||
if progress_callback:
|
||||
progress_callback(100, f"Exported {len(exported)} of {total} tracks")
|
||||
|
||||
return exported
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
ipod = IPodDevice()
|
||||
|
||||
@ -21,6 +21,12 @@ from mutagen.easyid3 import EasyID3
|
||||
from mutagen.mp4 import MP4
|
||||
from mutagen.mp3 import MP3
|
||||
|
||||
from artwork.extractor import extract_cover_with_fallback
|
||||
from artwork.cache import cover_cache
|
||||
from artwork.writer import prepare_artwork_for_track, write_artworkdb
|
||||
from artwork.chunks import read_existing_artwork
|
||||
from artwork.types import ArtworkEntry
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
MASTER_CONTAINER_PID = 203092939621887772 # "iPod" master playlist
|
||||
@ -290,16 +296,16 @@ class Nano7Database:
|
||||
artist = audio_info.get("artist", "Unknown Artist")
|
||||
album = audio_info.get("album", "Unknown Album")
|
||||
|
||||
# Check for duplicate: same title + artist + album
|
||||
# Check for duplicate: same title + artist + album (case-insensitive)
|
||||
conn = sqlite3.connect(self.sqlite_path)
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(
|
||||
"SELECT pid, title FROM item WHERE title = ? AND artist = ?",
|
||||
(title, artist)
|
||||
"SELECT pid, title FROM item WHERE LOWER(title) = LOWER(?) AND LOWER(artist) = LOWER(?) AND LOWER(album) = LOWER(?)",
|
||||
(title, artist, album)
|
||||
)
|
||||
existing = cursor.fetchone()
|
||||
if existing:
|
||||
logger.info(f"Track already exists: '{existing[1]}' by {artist}, skipping")
|
||||
logger.info(f"Track already exists: '{existing[1]}' by {artist} ({album}), skipping")
|
||||
conn.close()
|
||||
return None
|
||||
|
||||
@ -479,6 +485,12 @@ class Nano7Database:
|
||||
}
|
||||
logger.info(f"Added track: {title} by {artist}")
|
||||
|
||||
# Extract and write album art to iPod
|
||||
try:
|
||||
self._write_artwork_for_track(filepath, artist, album, pid)
|
||||
except Exception as exc:
|
||||
logger.warning("Failed to write artwork for track '%s': %s", title, exc)
|
||||
|
||||
# Sync iTunesCDB so firmware accepts the modified database
|
||||
self.sync_itunescdb()
|
||||
|
||||
@ -487,6 +499,57 @@ class Nano7Database:
|
||||
|
||||
return result
|
||||
|
||||
def _write_artwork_for_track(self, source_path: str, artist: str, album: str, item_pid: int) -> None:
|
||||
"""Extract cover art, encode for iPod, write to ithmb/ArtworkDB, update SQLite."""
|
||||
result = prepare_artwork_for_track(source_path)
|
||||
if result is None:
|
||||
return
|
||||
|
||||
art_hash, encoded_formats = result
|
||||
|
||||
artwork_dir = os.path.join(self.mountpoint, "iPod_Control", "Artwork")
|
||||
os.makedirs(artwork_dir, exist_ok=True)
|
||||
artdb_path = os.path.join(artwork_dir, "ArtworkDB")
|
||||
|
||||
existing = read_existing_artwork(artdb_path, artwork_dir)
|
||||
next_img_id = max(existing.keys(), default=99) + 1
|
||||
|
||||
src_img_size = sum(p.size for p in encoded_formats.values())
|
||||
|
||||
new_entry = ArtworkEntry(
|
||||
img_id=next_img_id,
|
||||
db_track_id=item_pid,
|
||||
art_hash=art_hash,
|
||||
src_img_size=src_img_size,
|
||||
formats=encoded_formats,
|
||||
db_track_ids=[item_pid],
|
||||
)
|
||||
|
||||
all_entries = []
|
||||
for eid in sorted(existing.keys()):
|
||||
e = existing[eid]
|
||||
all_entries.append(ArtworkEntry(
|
||||
img_id=e["img_id"],
|
||||
db_track_id=e["song_id"],
|
||||
art_hash=None,
|
||||
src_img_size=e["src_img_size"],
|
||||
formats={fid: ref for fid, ref in e["formats"].items()},
|
||||
db_track_ids=[e["song_id"]],
|
||||
))
|
||||
all_entries.append(new_entry)
|
||||
|
||||
write_artworkdb(self.mountpoint, all_entries, start_img_id=min(existing.keys(), default=100))
|
||||
|
||||
conn = sqlite3.connect(self.sqlite_path)
|
||||
try:
|
||||
conn.execute(
|
||||
"UPDATE item SET artwork_cache_id = ?, artwork_status = 1 WHERE pid = ?",
|
||||
(next_img_id, item_pid),
|
||||
)
|
||||
conn.commit()
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def get_track_count(self) -> int:
|
||||
"""Get number of tracks in database"""
|
||||
conn = sqlite3.connect(self.sqlite_path)
|
||||
@ -704,6 +767,101 @@ class Nano7Database:
|
||||
logger.error(f"Failed to delete track pid={pid}: {e}")
|
||||
return False
|
||||
|
||||
def remove_duplicates(self) -> List[Dict[str, Any]]:
|
||||
"""Find and remove duplicate tracks from the iPod database.
|
||||
|
||||
Duplicates are identified by LOWER(title) + LOWER(artist) + LOWER(album).
|
||||
Keeps the entry with the lowest physical_order (original).
|
||||
|
||||
Returns list of removed track info dicts.
|
||||
"""
|
||||
conn = sqlite3.connect(self.sqlite_path)
|
||||
cursor = conn.cursor()
|
||||
|
||||
# Find duplicate groups: same title + artist + album, more than one entry
|
||||
cursor.execute("""
|
||||
SELECT LOWER(title) AS lt, LOWER(artist) AS la, LOWER(album) AS lb,
|
||||
COUNT(*) AS cnt, MIN(physical_order) AS keep_order
|
||||
FROM item
|
||||
GROUP BY LOWER(title), LOWER(artist), LOWER(album)
|
||||
HAVING cnt > 1
|
||||
""")
|
||||
dup_groups = cursor.fetchall()
|
||||
|
||||
if not dup_groups:
|
||||
conn.close()
|
||||
logger.info("No duplicate tracks found")
|
||||
return []
|
||||
|
||||
# Collect PIDs to remove
|
||||
pids_to_remove = []
|
||||
files_to_remove = []
|
||||
removed_info = []
|
||||
|
||||
for lt, la, lb, cnt, keep_order in dup_groups:
|
||||
# Get the track to keep (lowest physical_order)
|
||||
cursor.execute(
|
||||
"SELECT pid, title, artist, album FROM item WHERE LOWER(title)=? AND LOWER(artist)=? AND LOWER(album)=? AND physical_order=?",
|
||||
(lt, la, lb, keep_order)
|
||||
)
|
||||
keep_row = cursor.fetchone()
|
||||
keep_pid = keep_row[0] if keep_row else None
|
||||
|
||||
# Get all duplicates for this group
|
||||
cursor.execute(
|
||||
"SELECT pid, title, artist, album, physical_order FROM item WHERE LOWER(title)=? AND LOWER(artist)=? AND LOWER(album)=? AND pid!=? ORDER BY physical_order",
|
||||
(lt, la, lb, keep_pid)
|
||||
)
|
||||
dup_rows = cursor.fetchall()
|
||||
|
||||
for dup_pid, dup_title, dup_artist, dup_album, dup_order in dup_rows:
|
||||
pids_to_remove.append(dup_pid)
|
||||
removed_info.append({
|
||||
"pid": dup_pid,
|
||||
"title": dup_title,
|
||||
"artist": dup_artist,
|
||||
"album": dup_album,
|
||||
"physical_order": dup_order,
|
||||
})
|
||||
|
||||
# Find the physical file for this duplicate
|
||||
file_path = self._find_track_file(dup_pid, dup_title, dup_artist)
|
||||
if file_path:
|
||||
files_to_remove.append(file_path)
|
||||
|
||||
logger.info(f"Found {len(pids_to_remove)} duplicate entries to remove across {len(dup_groups)} groups")
|
||||
|
||||
# Remove from all tables
|
||||
for pid in pids_to_remove:
|
||||
cursor.execute("DELETE FROM item_to_container WHERE item_pid = ?", (pid,))
|
||||
cursor.execute("DELETE FROM avformat_info WHERE item_pid = ?", (pid,))
|
||||
cursor.execute("DELETE FROM store_info WHERE item_pid = ?", (pid,))
|
||||
cursor.execute("DELETE FROM item WHERE pid = ?", (pid,))
|
||||
|
||||
# Clean up orphaned artist/album entries
|
||||
self._cleanup_orphaned_entries(cursor)
|
||||
|
||||
conn.commit()
|
||||
|
||||
# Delete physical files
|
||||
for fpath in files_to_remove:
|
||||
if os.path.exists(fpath):
|
||||
os.remove(fpath)
|
||||
logger.info(f"Deleted duplicate file: {fpath}")
|
||||
|
||||
# Update Locations.itdb
|
||||
for pid in pids_to_remove:
|
||||
self._delete_location(pid)
|
||||
|
||||
conn.close()
|
||||
|
||||
# Sync databases
|
||||
self.sync_itunescdb()
|
||||
self.sync_locations_cbk()
|
||||
|
||||
logger.info(f"Removed {len(removed_info)} duplicate track(s)")
|
||||
return removed_info
|
||||
|
||||
def _find_track_file(self, pid: int, title: str, artist: str) -> Optional[str]:
|
||||
"""Find the physical file for a track by scanning Music/FXX/ directories."""
|
||||
music_base = os.path.join(self.mountpoint, "iPod_Control", "Music")
|
||||
|
||||
793
src/main.py
793
src/main.py
File diff suppressed because it is too large
Load Diff
@ -22,6 +22,7 @@ from mutagen.id3 import ID3, APIC, TIT2, TPE1, TALB, TRCK
|
||||
from mutagen.easyid3 import EasyID3
|
||||
|
||||
from youtube_downloader import TrackInfo
|
||||
from artwork.cache import cover_cache
|
||||
|
||||
# Configure logging
|
||||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
|
||||
@ -215,10 +216,12 @@ class MetadataHandler:
|
||||
"""
|
||||
logger.info(f"Processing metadata for file: {file_path}")
|
||||
|
||||
# Download thumbnail if available
|
||||
# Download thumbnail if available, or use cached cover from source
|
||||
thumbnail_path = None
|
||||
if track_info.thumbnail_url:
|
||||
thumbnail_path = self.download_thumbnail(track_info.thumbnail_url)
|
||||
elif track_info.cover_path and os.path.exists(track_info.cover_path):
|
||||
thumbnail_path = track_info.cover_path
|
||||
|
||||
# Determine file type and process accordingly
|
||||
file_ext = Path(file_path).suffix.lower()
|
||||
@ -347,7 +350,12 @@ class MetadataHandler:
|
||||
fallback = self._parse_filename_trackinfo(base_name)
|
||||
title = title or fallback.title
|
||||
artist = artist or fallback.artist
|
||||
|
||||
|
||||
# Cache cover art for the player UI
|
||||
cover_path = None
|
||||
if cover_data:
|
||||
cover_path = cover_cache.put(artist or "Unknown Artist", album or "", cover_data)
|
||||
|
||||
return TrackInfo(
|
||||
video_id=f"local_{hash(file_path)}",
|
||||
title=title or base_name,
|
||||
@ -356,7 +364,9 @@ class MetadataHandler:
|
||||
thumbnail_url="",
|
||||
duration=duration,
|
||||
track_number=track_number if track_number else None,
|
||||
genre=genre,
|
||||
download_path=file_path,
|
||||
cover_path=cover_path,
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
|
||||
@ -32,6 +32,8 @@ class TrackInfo:
|
||||
playlist_title: Optional[str] = None
|
||||
playlist_index: Optional[int] = None
|
||||
download_path: Optional[str] = None
|
||||
genre: Optional[str] = None
|
||||
cover_path: Optional[str] = None
|
||||
|
||||
|
||||
class YouTubeDownloader:
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user