import subprocess from collections.abc import Callable from decimal import Decimal, InvalidOperation from typing import Any, Final from pydantic import AnyUrl, BaseModel, ConfigDict, Field, ValidationError from src.atmos.models import ( AtmosValidationBuilder, AtmosValidationFindingCode, AtmosValidationMetadata, MediainfoOutput, ) from src.types import MaybeInt, MaybeStr class AssetFetchError(Exception): pass class MediainfoUnavailable(Exception): pass class MediainfoError(Exception): pass class MediainfoOutputError(Exception): pass class MediainfoCrashError(Exception): pass class _MediainfoTrack(BaseModel): model_config = ConfigDict(frozen=True) type: str = Field(alias="@type") format: MaybeStr = Field(default=None, alias="Format") codec_id: MaybeStr = Field(default=None, alias="CodecID") channels: MaybeInt = Field(default=None, alias="Channels") sampling_rate: MaybeInt = Field(default=None, alias="SamplingRate") bit_depth: MaybeInt = Field(default=None, alias="BitDepth") duration: MaybeStr = Field(default=None, alias="Duration") extra: dict[str, Any] = Field(default_factory=dict) class _MediainfoMedia(BaseModel): model_config = ConfigDict(frozen=True) track: list[_MediainfoTrack] = Field(min_length=1) class _MediainfoResponse(BaseModel): model_config = ConfigDict(frozen=True) media: _MediainfoMedia | None = None def run_mediainfo(media_url: AnyUrl) -> MediainfoOutput: try: mediainfo_process = subprocess.run( ["mediainfo", "--Output=JSON", str(media_url)], capture_output=True, text=True, check=False, ) except OSError as e: error_message = f"could not execute mediainfo: {e}" raise MediainfoUnavailable(error_message) from e if mediainfo_process.returncode < 0: error_message = ( f"mediainfo killed by signal {-mediainfo_process.returncode}: {mediainfo_process.stderr.strip()}" ) raise MediainfoCrashError(error_message) if mediainfo_process.returncode != 0: error_message = f"mediainfo exited with code {mediainfo_process.returncode}: {mediainfo_process.stderr.strip()}" raise MediainfoError(error_message) try: response = _MediainfoResponse.model_validate_json(mediainfo_process.stdout) except ValidationError as e: error_message = f"mediainfo output unexpected: {e}" raise MediainfoOutputError(error_message) from e if response.media is None: detail = mediainfo_process.stderr.strip() or "(no stderr detail)" raise AssetFetchError(f"mediainfo could not fetch asset: {detail}") return _parse_mediainfo(response.media) def _parse_mediainfo(media: _MediainfoMedia) -> MediainfoOutput: general_track = next((t for t in media.track if t.type == "General"), None) if general_track is None: raise MediainfoOutputError("mediainfo output missing General track") audio_track = next((t for t in media.track if t.type == "Audio"), None) if audio_track is None: raise MediainfoOutputError("mediainfo output missing Audio track") return MediainfoOutput( container=general_track.format, codec=audio_track.format, codec_id=audio_track.codec_id, channels=audio_track.channels, sample_rate_hz=audio_track.sampling_rate, bits_per_sample=audio_track.bit_depth, is_truncated=general_track.extra.get("IsTruncated") == "Yes", adm_profile=audio_track.extra.get("AdmProfile_Format"), duration_ms=_parse_duration_ms(general_track.duration), ) def _parse_duration_ms(duration_seconds_str: str | None) -> int | None: # mediainfo emits General.Duration as a string of seconds with up to 3 decimal places # (e.g. "499.840"). Decimal avoids float rounding error in the seconds→milliseconds scaling. if duration_seconds_str is None: return None try: return int(Decimal(duration_seconds_str) * 1000) except InvalidOperation: return None def check_mediainfo(media_url: AnyUrl, atmos_validation_builder: AtmosValidationBuilder) -> None: output = run_mediainfo(media_url) atmos_validation_builder.metadata = AtmosValidationMetadata(**output.model_dump()) for check in _CHECKS: check(output, atmos_validation_builder) def _check_truncation(output: MediainfoOutput, atmos_validation_builder: AtmosValidationBuilder) -> None: if not output.is_truncated: return atmos_validation_builder.error( AtmosValidationFindingCode.ATMOS_FILE_TRUNCATED, "File is truncated — actual size is less than the size declared in the header", ) def _check_container(output: MediainfoOutput, atmos_validation_builder: AtmosValidationBuilder) -> None: required: Final = "Wave" if output.container == required: return description = f"'{output.container}'" if output.container is not None else "not reported" atmos_validation_builder.error( AtmosValidationFindingCode.NOT_WAVE, f"Container format is {description}, must be '{required}'", ) def _check_pcm(output: MediainfoOutput, atmos_validation_builder: AtmosValidationBuilder) -> None: required_codec: Final = "PCM" if output.codec == required_codec: return description = f"'{output.codec}'" if output.codec is not None else "not reported" atmos_validation_builder.error( AtmosValidationFindingCode.NOT_PCM, f"Audio codec is {description}, must be '{required_codec}'", ) def _check_lpcm(output: MediainfoOutput, atmos_validation_builder: AtmosValidationBuilder) -> None: # Only meaningful when codec is PCM. The Dolby Atmos ADM Profile spec states: # "The audio format shall be PCM (formatType value = 0x0001 (WAVE_FORMAT_PCM))." # https://professionalsupport.dolby.com/s/article/Dolby-Atmos-ADM-Profile-specification?language=en_US # WAVEFORMATEXTENSIBLE (codec_id "65534" or the PCM subformat GUID) is rejected even # though it also encodes linear PCM data. if output.codec != "PCM": return required_codec_id: Final = "1" if output.codec_id == required_codec_id: return description = f"'{output.codec_id}'" if output.codec_id is not None else "not reported" atmos_validation_builder.error( AtmosValidationFindingCode.NOT_LPCM, f"Audio codec_id is {description}, must be '{required_codec_id}' (WAVE_FORMAT_PCM)", ) def _check_bit_depth(output: MediainfoOutput, atmos_validation_builder: AtmosValidationBuilder) -> None: required_bits: Final = 24 if output.bits_per_sample == required_bits: return description = f"{output.bits_per_sample} bit" if output.bits_per_sample is not None else "not reported" atmos_validation_builder.error( AtmosValidationFindingCode.WRONG_BIT_DEPTH, f"Bit depth is {description}, must be {required_bits} bit", ) def _check_sample_rate(output: MediainfoOutput, atmos_validation_builder: AtmosValidationBuilder) -> None: required_hz: Final = 48000 if output.sample_rate_hz == required_hz: return description = f"{output.sample_rate_hz} Hz" if output.sample_rate_hz is not None else "not reported" atmos_validation_builder.error( AtmosValidationFindingCode.WRONG_SAMPLE_RATE, f"Sample rate is {description}, must be {required_hz} Hz", ) def _check_dolby_atmos(output: MediainfoOutput, atmos_validation_builder: AtmosValidationBuilder) -> None: required_profile: Final = "Dolby Atmos Master" if output.adm_profile == required_profile: return description = f"'{output.adm_profile}'" if output.adm_profile is not None else "not reported" atmos_validation_builder.error( AtmosValidationFindingCode.NOT_DOLBY_ATMOS, f"ADM profile format is {description}, must be '{required_profile}'", ) _CHECKS: Final[tuple[Callable[[MediainfoOutput, AtmosValidationBuilder], None], ...]] = ( _check_truncation, _check_container, _check_pcm, _check_lpcm, _check_bit_depth, _check_sample_rate, _check_dolby_atmos, )