Spaces:
Configuration error
Configuration error
| import logging | |
| import math | |
| import re | |
| from datetime import datetime, timedelta, timezone | |
| from typing import List, Dict, Optional, Union | |
| from urllib.parse import urljoin | |
| import xmltodict | |
| logger = logging.getLogger(__name__) | |
| def resolve_url(base_url: str, relative_url: str) -> str: | |
| """ | |
| Resolve a relative URL against a base URL. | |
| Handles three cases: | |
| 1. Absolute URL (starts with http:// or https://) - return as-is | |
| 2. Absolute path (starts with /) - resolve against origin (scheme + host) | |
| 3. Relative path - resolve against base URL directory | |
| Args: | |
| base_url: The base URL (typically the MPD URL) | |
| relative_url: The URL to resolve | |
| Returns: | |
| The resolved absolute URL | |
| """ | |
| if not relative_url: | |
| return base_url | |
| # Already absolute URL | |
| if relative_url.startswith(("http://", "https://")): | |
| return relative_url | |
| # Use urljoin which correctly handles: | |
| # - Absolute paths (starting with /) -> resolves against origin | |
| # - Relative paths -> resolves against base URL | |
| return urljoin(base_url, relative_url) | |
| def parse_mpd(mpd_content: Union[str, bytes]) -> dict: | |
| """ | |
| Parses the MPD content into a dictionary. | |
| Args: | |
| mpd_content (Union[str, bytes]): The MPD content to parse. | |
| Returns: | |
| dict: The parsed MPD content as a dictionary. | |
| """ | |
| return xmltodict.parse(mpd_content) | |
| def parse_mpd_dict( | |
| mpd_dict: dict, mpd_url: str, parse_drm: bool = True, parse_segment_profile_id: Optional[str] = None | |
| ) -> dict: | |
| """ | |
| Parses the MPD dictionary and extracts relevant information. | |
| Args: | |
| mpd_dict (dict): The MPD content as a dictionary. | |
| mpd_url (str): The URL of the MPD manifest. | |
| parse_drm (bool, optional): Whether to parse DRM information. Defaults to True. | |
| parse_segment_profile_id (str, optional): The profile ID to parse segments for. Defaults to None. | |
| Returns: | |
| dict: The parsed MPD information including profiles and DRM info. | |
| This function processes the MPD dictionary to extract profiles, DRM information, and other relevant data. | |
| It handles both live and static MPD manifests. | |
| """ | |
| profiles = [] | |
| parsed_dict = {} | |
| is_live = mpd_dict["MPD"].get("@type", "static").lower() == "dynamic" | |
| parsed_dict["isLive"] = is_live | |
| media_presentation_duration = mpd_dict["MPD"].get("@mediaPresentationDuration") | |
| # Parse additional MPD attributes for live streams | |
| if is_live: | |
| parsed_dict["minimumUpdatePeriod"] = parse_duration(mpd_dict["MPD"].get("@minimumUpdatePeriod", "PT0S")) | |
| parsed_dict["timeShiftBufferDepth"] = parse_duration(mpd_dict["MPD"].get("@timeShiftBufferDepth", "PT2M")) | |
| parsed_dict["availabilityStartTime"] = datetime.fromisoformat( | |
| mpd_dict["MPD"]["@availabilityStartTime"].replace("Z", "+00:00") | |
| ) | |
| parsed_dict["publishTime"] = datetime.fromisoformat( | |
| mpd_dict["MPD"].get("@publishTime", "").replace("Z", "+00:00") | |
| ) | |
| periods = mpd_dict["MPD"]["Period"] | |
| periods = periods if isinstance(periods, list) else [periods] | |
| for period in periods: | |
| parsed_dict["PeriodStart"] = parse_duration(period.get("@start", "PT0S")) | |
| adaptation_sets = period["AdaptationSet"] | |
| adaptation_sets = adaptation_sets if isinstance(adaptation_sets, list) else [adaptation_sets] | |
| for adaptation in adaptation_sets: | |
| representations = adaptation["Representation"] | |
| representations = representations if isinstance(representations, list) else [representations] | |
| for representation in representations: | |
| profile = parse_representation( | |
| parsed_dict, | |
| representation, | |
| adaptation, | |
| mpd_url, | |
| media_presentation_duration, | |
| parse_segment_profile_id, | |
| ) | |
| if profile: | |
| profiles.append(profile) | |
| parsed_dict["profiles"] = profiles | |
| if parse_drm: | |
| drm_info = extract_drm_info(periods, mpd_url) | |
| else: | |
| drm_info = {} | |
| parsed_dict["drmInfo"] = drm_info | |
| return parsed_dict | |
| def pad_base64(encoded_key_id): | |
| """ | |
| Pads a base64 encoded key ID to make its length a multiple of 4. | |
| Args: | |
| encoded_key_id (str): The base64 encoded key ID. | |
| Returns: | |
| str: The padded base64 encoded key ID. | |
| """ | |
| return encoded_key_id + "=" * (4 - len(encoded_key_id) % 4) | |
| def extract_drm_info(periods: List[Dict], mpd_url: str) -> Dict: | |
| """ | |
| Extracts DRM information from the MPD periods. | |
| Args: | |
| periods (List[Dict]): The list of periods in the MPD. | |
| mpd_url (str): The URL of the MPD manifest. | |
| Returns: | |
| Dict: The extracted DRM information. | |
| This function processes the ContentProtection elements in the MPD to extract DRM system information, | |
| such as ClearKey, Widevine, and PlayReady. | |
| """ | |
| drm_info = {"isDrmProtected": False} | |
| for period in periods: | |
| adaptation_sets: Union[list[dict], dict] = period.get("AdaptationSet", []) | |
| if not isinstance(adaptation_sets, list): | |
| adaptation_sets = [adaptation_sets] | |
| for adaptation_set in adaptation_sets: | |
| # Check ContentProtection in AdaptationSet | |
| process_content_protection(adaptation_set.get("ContentProtection", []), drm_info) | |
| # Check ContentProtection inside each Representation | |
| representations: Union[list[dict], dict] = adaptation_set.get("Representation", []) | |
| if not isinstance(representations, list): | |
| representations = [representations] | |
| for representation in representations: | |
| process_content_protection(representation.get("ContentProtection", []), drm_info) | |
| # If we have a license acquisition URL, make sure it's absolute | |
| if "laUrl" in drm_info and not drm_info["laUrl"].startswith(("http://", "https://")): | |
| drm_info["laUrl"] = urljoin(mpd_url, drm_info["laUrl"]) | |
| return drm_info | |
| def process_content_protection(content_protection: Union[list[dict], dict], drm_info: dict): | |
| """ | |
| Processes the ContentProtection elements to extract DRM information. | |
| Args: | |
| content_protection (Union[list[dict], dict]): The ContentProtection elements. | |
| drm_info (dict): The dictionary to store DRM information. | |
| This function updates the drm_info dictionary with DRM system information found in the ContentProtection elements. | |
| """ | |
| if not isinstance(content_protection, list): | |
| content_protection = [content_protection] | |
| for protection in content_protection: | |
| drm_info["isDrmProtected"] = True | |
| scheme_id_uri = protection.get("@schemeIdUri", "").lower() | |
| if "clearkey" in scheme_id_uri: | |
| drm_info["drmSystem"] = "clearkey" | |
| if "clearkey:Laurl" in protection: | |
| la_url = protection["clearkey:Laurl"].get("#text") | |
| if la_url and "laUrl" not in drm_info: | |
| drm_info["laUrl"] = la_url | |
| elif "widevine" in scheme_id_uri or "edef8ba9-79d6-4ace-a3c8-27dcd51d21ed" in scheme_id_uri: | |
| drm_info["drmSystem"] = "widevine" | |
| pssh = protection.get("cenc:pssh", {}).get("#text") | |
| if pssh: | |
| drm_info["pssh"] = pssh | |
| elif "playready" in scheme_id_uri or "9a04f079-9840-4286-ab92-e65be0885f95" in scheme_id_uri: | |
| drm_info["drmSystem"] = "playready" | |
| if "@cenc:default_KID" in protection: | |
| key_id = protection["@cenc:default_KID"].replace("-", "") | |
| if "keyId" not in drm_info: | |
| drm_info["keyId"] = key_id | |
| if "ms:laurl" in protection: | |
| la_url = protection["ms:laurl"].get("@licenseUrl") | |
| if la_url and "laUrl" not in drm_info: | |
| drm_info["laUrl"] = la_url | |
| return drm_info | |
| def parse_representation( | |
| parsed_dict: dict, | |
| representation: dict, | |
| adaptation: dict, | |
| mpd_url: str, | |
| media_presentation_duration: str, | |
| parse_segment_profile_id: Optional[str], | |
| ) -> Optional[dict]: | |
| """ | |
| Parses a representation and extracts profile information. | |
| Args: | |
| parsed_dict (dict): The parsed MPD data. | |
| representation (dict): The representation data. | |
| adaptation (dict): The adaptation set data. | |
| mpd_url (str): The URL of the MPD manifest. | |
| media_presentation_duration (str): The media presentation duration. | |
| parse_segment_profile_id (str, optional): The profile ID to parse segments for. Defaults to None. | |
| Returns: | |
| Optional[dict]: The parsed profile information or None if not applicable. | |
| """ | |
| mime_type = _get_key(adaptation, representation, "@mimeType") or ( | |
| "video/mp4" if "avc" in representation["@codecs"] else "audio/mp4" | |
| ) | |
| if "video" not in mime_type and "audio" not in mime_type: | |
| return None | |
| profile = { | |
| "id": representation.get("@id") or adaptation.get("@id"), | |
| "mimeType": mime_type, | |
| "lang": representation.get("@lang") or adaptation.get("@lang"), | |
| "codecs": representation.get("@codecs") or adaptation.get("@codecs"), | |
| "bandwidth": int(representation.get("@bandwidth") or adaptation.get("@bandwidth")), | |
| "startWithSAP": (_get_key(adaptation, representation, "@startWithSAP") or "1") == "1", | |
| "mediaPresentationDuration": media_presentation_duration, | |
| } | |
| if "audio" in profile["mimeType"]: | |
| profile["audioSamplingRate"] = representation.get("@audioSamplingRate") or adaptation.get("@audioSamplingRate") | |
| profile["channels"] = representation.get("AudioChannelConfiguration", {}).get("@value", "2") | |
| else: | |
| # Handle video-specific attributes, making them optional with sensible defaults | |
| if "@width" in representation: | |
| profile["width"] = int(representation["@width"]) | |
| elif "@width" in adaptation: | |
| profile["width"] = int(adaptation["@width"]) | |
| else: | |
| profile["width"] = 0 # Default if width is missing | |
| if "@height" in representation: | |
| profile["height"] = int(representation["@height"]) | |
| elif "@height" in adaptation: | |
| profile["height"] = int(adaptation["@height"]) | |
| else: | |
| profile["height"] = 0 # Default if height is missing | |
| frame_rate = representation.get("@frameRate") or adaptation.get("@maxFrameRate") or "30000/1001" | |
| frame_rate = frame_rate if "/" in frame_rate else f"{frame_rate}/1" | |
| profile["frameRate"] = round(int(frame_rate.split("/")[0]) / int(frame_rate.split("/")[1]), 3) | |
| profile["sar"] = representation.get("@sar", "1:1") | |
| # Extract segment template start number for adaptive sequence calculation | |
| segment_template_data = adaptation.get("SegmentTemplate") or representation.get("SegmentTemplate") | |
| if segment_template_data: | |
| try: | |
| profile["segment_template_start_number"] = int(segment_template_data.get("@startNumber", 1)) | |
| except (ValueError, TypeError): | |
| profile["segment_template_start_number"] = 1 | |
| else: | |
| profile["segment_template_start_number"] = 1 | |
| # For SegmentBase profiles, we need to set initUrl even when not parsing segments | |
| # This is needed for the HLS playlist builder to reference the init URL | |
| segment_base_data = representation.get("SegmentBase") | |
| if segment_base_data and "initUrl" not in profile: | |
| base_url = representation.get("BaseURL", "") | |
| profile["initUrl"] = resolve_url(mpd_url, base_url) | |
| # Store initialization range if available | |
| if "Initialization" in segment_base_data: | |
| init_range = segment_base_data["Initialization"].get("@range") | |
| if init_range: | |
| profile["initRange"] = init_range | |
| # For SegmentList profiles, we also need to set initUrl even when not parsing segments | |
| segment_list_data = representation.get("SegmentList") or adaptation.get("SegmentList") | |
| if segment_list_data and "initUrl" not in profile: | |
| if "Initialization" in segment_list_data: | |
| init_data = segment_list_data["Initialization"] | |
| if "@sourceURL" in init_data: | |
| init_url = init_data["@sourceURL"] | |
| profile["initUrl"] = resolve_url(mpd_url, init_url) | |
| elif "@range" in init_data: | |
| base_url = representation.get("BaseURL", "") | |
| profile["initUrl"] = resolve_url(mpd_url, base_url) | |
| profile["initRange"] = init_data["@range"] | |
| if parse_segment_profile_id is None or profile["id"] != parse_segment_profile_id: | |
| return profile | |
| # Parse segments based on the addressing scheme used | |
| segment_template = adaptation.get("SegmentTemplate") or representation.get("SegmentTemplate") | |
| segment_list = adaptation.get("SegmentList") or representation.get("SegmentList") | |
| # Get BaseURL from representation (can be relative path like "a/b/c/") | |
| base_url = representation.get("BaseURL", "") | |
| if segment_template: | |
| profile["segments"] = parse_segment_template(parsed_dict, segment_template, profile, mpd_url, base_url) | |
| elif segment_list: | |
| # Get timescale from SegmentList or default to 1 | |
| timescale = int(segment_list.get("@timescale", 1)) | |
| profile["segments"] = parse_segment_list(adaptation, representation, profile, mpd_url, timescale) | |
| else: | |
| profile["segments"] = parse_segment_base(representation, profile, mpd_url) | |
| return profile | |
| def _get_key(adaptation: dict, representation: dict, key: str) -> Optional[str]: | |
| """ | |
| Retrieves a key from the representation or adaptation set. | |
| Args: | |
| adaptation (dict): The adaptation set data. | |
| representation (dict): The representation data. | |
| key (str): The key to retrieve. | |
| Returns: | |
| Optional[str]: The value of the key or None if not found. | |
| """ | |
| return representation.get(key, adaptation.get(key, None)) | |
| def parse_segment_template( | |
| parsed_dict: dict, item: dict, profile: dict, mpd_url: str, base_url: str = "" | |
| ) -> List[Dict]: | |
| """ | |
| Parses a segment template and extracts segment information. | |
| Args: | |
| parsed_dict (dict): The parsed MPD data. | |
| item (dict): The segment template data. | |
| profile (dict): The profile information. | |
| mpd_url (str): The URL of the MPD manifest. | |
| base_url (str): The BaseURL from the representation (optional, for per-representation paths). | |
| Returns: | |
| List[Dict]: The list of parsed segments. | |
| """ | |
| segments = [] | |
| timescale = int(item.get("@timescale", 1)) | |
| # Initialization | |
| if "@initialization" in item: | |
| media = item["@initialization"] | |
| media = media.replace("$RepresentationID$", profile["id"]) | |
| media = media.replace("$Bandwidth$", str(profile["bandwidth"])) | |
| # Combine base_url and media, then resolve against mpd_url | |
| if base_url: | |
| media = base_url + media | |
| profile["initUrl"] = resolve_url(mpd_url, media) | |
| # Segments | |
| if "SegmentTimeline" in item: | |
| segments.extend(parse_segment_timeline(parsed_dict, item, profile, mpd_url, timescale, base_url)) | |
| elif "@duration" in item: | |
| segments.extend(parse_segment_duration(parsed_dict, item, profile, mpd_url, timescale, base_url)) | |
| return segments | |
| def parse_segment_timeline( | |
| parsed_dict: dict, item: dict, profile: dict, mpd_url: str, timescale: int, base_url: str = "" | |
| ) -> List[Dict]: | |
| """ | |
| Parses a segment timeline and extracts segment information. | |
| Args: | |
| parsed_dict (dict): The parsed MPD data. | |
| item (dict): The segment timeline data. | |
| profile (dict): The profile information. | |
| mpd_url (str): The URL of the MPD manifest. | |
| timescale (int): The timescale for the segments. | |
| base_url (str): The BaseURL from the representation (optional, for per-representation paths). | |
| Returns: | |
| List[Dict]: The list of parsed segments. | |
| """ | |
| timelines = item["SegmentTimeline"]["S"] | |
| timelines = timelines if isinstance(timelines, list) else [timelines] | |
| period_start = parsed_dict.get("availabilityStartTime", datetime.fromtimestamp(0, tz=timezone.utc)) + timedelta( | |
| seconds=parsed_dict.get("PeriodStart", 0) | |
| ) | |
| presentation_time_offset = int(item.get("@presentationTimeOffset", 0)) | |
| start_number = int(item.get("@startNumber", 1)) | |
| segments = [ | |
| create_segment_data(timeline, item, profile, mpd_url, timescale, base_url) | |
| for timeline in preprocess_timeline(timelines, start_number, period_start, presentation_time_offset, timescale) | |
| ] | |
| return segments | |
| def preprocess_timeline( | |
| timelines: List[Dict], start_number: int, period_start: datetime, presentation_time_offset: int, timescale: int | |
| ) -> List[Dict]: | |
| """ | |
| Preprocesses the segment timeline data. | |
| Args: | |
| timelines (List[Dict]): The list of timeline segments. | |
| start_number (int): The starting segment number. | |
| period_start (datetime): The start time of the period. | |
| presentation_time_offset (int): The presentation time offset. | |
| timescale (int): The timescale for the segments. | |
| Returns: | |
| List[Dict]: The list of preprocessed timeline segments. | |
| """ | |
| processed_data = [] | |
| current_time = 0 | |
| for timeline in timelines: | |
| repeat = int(timeline.get("@r", 0)) | |
| duration = int(timeline["@d"]) | |
| start_time = int(timeline.get("@t", current_time)) | |
| for _ in range(repeat + 1): | |
| segment_start_time = period_start + timedelta(seconds=(start_time - presentation_time_offset) / timescale) | |
| segment_end_time = segment_start_time + timedelta(seconds=duration / timescale) | |
| presentation_time = start_time - presentation_time_offset | |
| processed_data.append( | |
| { | |
| "number": start_number, | |
| "start_time": segment_start_time, | |
| "end_time": segment_end_time, | |
| "duration": duration, | |
| "time": presentation_time, | |
| } | |
| ) | |
| start_time += duration | |
| start_number += 1 | |
| current_time = start_time | |
| return processed_data | |
| def parse_segment_duration( | |
| parsed_dict: dict, item: dict, profile: dict, mpd_url: str, timescale: int, base_url: str = "" | |
| ) -> List[Dict]: | |
| """ | |
| Parses segment duration and extracts segment information. | |
| This is used for static or live MPD manifests. | |
| Args: | |
| parsed_dict (dict): The parsed MPD data. | |
| item (dict): The segment duration data. | |
| profile (dict): The profile information. | |
| mpd_url (str): The URL of the MPD manifest. | |
| timescale (int): The timescale for the segments. | |
| base_url (str): The BaseURL from the representation (optional, for per-representation paths). | |
| Returns: | |
| List[Dict]: The list of parsed segments. | |
| """ | |
| duration = int(item["@duration"]) | |
| start_number = int(item.get("@startNumber", 1)) | |
| segment_duration_sec = duration / timescale | |
| if parsed_dict["isLive"]: | |
| segments = generate_live_segments(parsed_dict, segment_duration_sec, start_number) | |
| else: | |
| segments = generate_vod_segments(profile, duration, timescale, start_number) | |
| return [create_segment_data(seg, item, profile, mpd_url, timescale, base_url) for seg in segments] | |
| def generate_live_segments(parsed_dict: dict, segment_duration_sec: float, start_number: int) -> List[Dict]: | |
| """ | |
| Generates live segments based on the segment duration and start number. | |
| This is used for live MPD manifests. | |
| Args: | |
| parsed_dict (dict): The parsed MPD data. | |
| segment_duration_sec (float): The segment duration in seconds. | |
| start_number (int): The starting segment number. | |
| Returns: | |
| List[Dict]: The list of generated live segments. | |
| """ | |
| time_shift_buffer_depth = timedelta(seconds=parsed_dict.get("timeShiftBufferDepth", 60)) | |
| segment_count = math.ceil(time_shift_buffer_depth.total_seconds() / segment_duration_sec) | |
| current_time = datetime.now(tz=timezone.utc) | |
| earliest_segment_number = max( | |
| start_number | |
| + math.floor((current_time - parsed_dict["availabilityStartTime"]).total_seconds() / segment_duration_sec) | |
| - segment_count, | |
| start_number, | |
| ) | |
| return [ | |
| { | |
| "number": number, | |
| "start_time": parsed_dict["availabilityStartTime"] | |
| + timedelta(seconds=(number - start_number) * segment_duration_sec), | |
| "duration": segment_duration_sec, | |
| } | |
| for number in range(earliest_segment_number, earliest_segment_number + segment_count) | |
| ] | |
| def generate_vod_segments(profile: dict, duration: int, timescale: int, start_number: int) -> List[Dict]: | |
| """ | |
| Generates VOD segments based on the segment duration and start number. | |
| This is used for static MPD manifests. | |
| Args: | |
| profile (dict): The profile information. | |
| duration (int): The segment duration. | |
| timescale (int): The timescale for the segments. | |
| start_number (int): The starting segment number. | |
| Returns: | |
| List[Dict]: The list of generated VOD segments. | |
| """ | |
| total_duration = profile.get("mediaPresentationDuration") or 0 | |
| if isinstance(total_duration, str): | |
| total_duration = parse_duration(total_duration) | |
| segment_count = math.ceil(total_duration * timescale / duration) | |
| return [{"number": start_number + i, "duration": duration / timescale} for i in range(segment_count)] | |
| def create_segment_data( | |
| segment: Dict, item: dict, profile: dict, mpd_url: str, timescale: Optional[int] = None, base_url: str = "" | |
| ) -> Dict: | |
| """ | |
| Creates segment data based on the segment information. This includes the segment URL and metadata. | |
| Args: | |
| segment (Dict): The segment information. | |
| item (dict): The segment template data. | |
| profile (dict): The profile information. | |
| mpd_url (str): The URL of the MPD manifest. | |
| timescale (int, optional): The timescale for the segments. Defaults to None. | |
| base_url (str): The BaseURL from the representation (optional, for per-representation paths). | |
| Returns: | |
| Dict: The created segment data. | |
| """ | |
| media_template = item["@media"] | |
| media = media_template.replace("$RepresentationID$", profile["id"]) | |
| media = media.replace("$Number%04d$", f"{segment['number']:04d}") | |
| media = media.replace("$Number$", str(segment["number"])) | |
| media = media.replace("$Bandwidth$", str(profile["bandwidth"])) | |
| if "time" in segment and timescale is not None: | |
| media = media.replace("$Time$", str(int(segment["time"]))) | |
| # Combine base_url and media, then resolve against mpd_url | |
| if base_url: | |
| media = base_url + media | |
| media = resolve_url(mpd_url, media) | |
| segment_data = { | |
| "type": "segment", | |
| "media": media, | |
| "number": segment["number"], | |
| } | |
| # Add time and duration metadata for adaptive sequence calculation | |
| if "time" in segment: | |
| segment_data["time"] = segment["time"] | |
| if "duration" in segment: | |
| segment_data["duration_mpd_timescale"] = segment["duration"] | |
| if "start_time" in segment and "end_time" in segment: | |
| segment_data.update( | |
| { | |
| "start_time": segment["start_time"], | |
| "end_time": segment["end_time"], | |
| "extinf": (segment["end_time"] - segment["start_time"]).total_seconds(), | |
| "program_date_time": segment["start_time"].isoformat() + "Z", | |
| } | |
| ) | |
| elif "start_time" in segment and "duration" in segment: | |
| # duration here is in timescale units (from timeline segments) | |
| duration_seconds = segment["duration"] / timescale if timescale else segment["duration"] | |
| segment_data.update( | |
| { | |
| "start_time": segment["start_time"], | |
| "end_time": segment["start_time"] + timedelta(seconds=duration_seconds), | |
| "extinf": duration_seconds, | |
| "program_date_time": segment["start_time"].isoformat() + "Z", | |
| } | |
| ) | |
| elif "duration" in segment: | |
| # duration from generate_vod_segments and generate_live_segments is already in seconds | |
| segment_data["extinf"] = segment["duration"] | |
| return segment_data | |
| def parse_segment_list( | |
| adaptation: dict, representation: dict, profile: dict, mpd_url: str, timescale: int | |
| ) -> List[Dict]: | |
| """ | |
| Parses SegmentList element with explicit SegmentURL entries. | |
| SegmentList MPDs explicitly list each segment URL, unlike SegmentTemplate which uses | |
| URL patterns. This is less common but used by some packagers. | |
| Args: | |
| adaptation (dict): The adaptation set data. | |
| representation (dict): The representation data. | |
| profile (dict): The profile information. | |
| mpd_url (str): The URL of the MPD manifest. | |
| timescale (int): The timescale for duration calculations. | |
| Returns: | |
| List[Dict]: The list of parsed segments. | |
| """ | |
| # SegmentList can be at AdaptationSet or Representation level | |
| segment_list = representation.get("SegmentList") or adaptation.get("SegmentList", {}) | |
| segments = [] | |
| # Handle Initialization element | |
| if "Initialization" in segment_list: | |
| init_data = segment_list["Initialization"] | |
| if "@sourceURL" in init_data: | |
| init_url = init_data["@sourceURL"] | |
| profile["initUrl"] = resolve_url(mpd_url, init_url) | |
| elif "@range" in init_data: | |
| # Initialization by byte range on the BaseURL | |
| base_url = representation.get("BaseURL", "") | |
| profile["initUrl"] = resolve_url(mpd_url, base_url) | |
| profile["initRange"] = init_data["@range"] | |
| # Get segment duration from SegmentList attributes | |
| duration = int(segment_list.get("@duration", 0)) | |
| list_timescale = int(segment_list.get("@timescale", timescale or 1)) | |
| segment_duration_sec = duration / list_timescale if list_timescale else 0 | |
| # Parse SegmentURL elements | |
| segment_urls = segment_list.get("SegmentURL", []) | |
| if not isinstance(segment_urls, list): | |
| segment_urls = [segment_urls] | |
| for i, seg_url in enumerate(segment_urls): | |
| if seg_url is None: | |
| continue | |
| # Get media URL - can be @media attribute or use BaseURL with @mediaRange | |
| media_url = seg_url.get("@media", "") | |
| media_range = seg_url.get("@mediaRange") | |
| if media_url: | |
| media_url = resolve_url(mpd_url, media_url) | |
| else: | |
| # Use BaseURL with byte range | |
| base_url = representation.get("BaseURL", "") | |
| media_url = resolve_url(mpd_url, base_url) | |
| segment_data = { | |
| "type": "segment", | |
| "media": media_url, | |
| "number": i + 1, | |
| "extinf": segment_duration_sec if segment_duration_sec > 0 else 1.0, | |
| } | |
| # Include media range if specified | |
| if media_range: | |
| segment_data["mediaRange"] = media_range | |
| segments.append(segment_data) | |
| return segments | |
| def parse_segment_base(representation: dict, profile: dict, mpd_url: str) -> List[Dict]: | |
| """ | |
| Parses segment base information and extracts segment data. This is used for single-segment representations | |
| (SegmentBase MPDs, typically GPAC-generated on-demand profiles). | |
| For SegmentBase, the entire media file is treated as a single segment. The initialization data | |
| is specified by the Initialization element's range, and the segment index (SIDX) is at indexRange. | |
| Args: | |
| representation (dict): The representation data. | |
| profile (dict): The profile information. | |
| mpd_url (str): The URL of the MPD manifest. | |
| Returns: | |
| List[Dict]: The list of parsed segments. | |
| """ | |
| segment = representation.get("SegmentBase", {}) | |
| base_url = representation.get("BaseURL", "") | |
| # Build the full media URL | |
| media_url = resolve_url(mpd_url, base_url) | |
| # Set initUrl for SegmentBase - this is the URL with the initialization range | |
| # The initialization segment contains codec/track info needed before playing media | |
| profile["initUrl"] = media_url | |
| # For SegmentBase, we need to specify byte ranges for init and media segments | |
| init_range = None | |
| if "Initialization" in segment: | |
| init_range = segment["Initialization"].get("@range") | |
| # Store initialization range in profile for segment endpoint to use | |
| if init_range: | |
| profile["initRange"] = init_range | |
| # Get the index range which points to SIDX box | |
| index_range = segment.get("@indexRange", "") | |
| # Calculate total duration from profile's mediaPresentationDuration | |
| total_duration = profile.get("mediaPresentationDuration") | |
| if isinstance(total_duration, str): | |
| total_duration = parse_duration(total_duration) | |
| elif total_duration is None: | |
| total_duration = 0 | |
| # For SegmentBase, we return a single segment representing the entire media | |
| # The media URL is the same as initUrl but will be accessed with different byte ranges | |
| return [ | |
| { | |
| "type": "segment", | |
| "media": media_url, | |
| "number": 1, | |
| "extinf": total_duration if total_duration > 0 else 1.0, | |
| "indexRange": index_range, | |
| "initRange": init_range, | |
| } | |
| ] | |
| def parse_duration(duration_str: str) -> float: | |
| """ | |
| Parses a duration ISO 8601 string into seconds. | |
| Args: | |
| duration_str (str): The duration string to parse. | |
| Returns: | |
| float: The parsed duration in seconds. | |
| """ | |
| pattern = re.compile(r"P(?:(\d+)Y)?(?:(\d+)M)?(?:(\d+)D)?T?(?:(\d+)H)?(?:(\d+)M)?(?:(\d+(?:\.\d+)?)S)?") | |
| match = pattern.match(duration_str) | |
| if not match: | |
| raise ValueError(f"Invalid duration format: {duration_str}") | |
| years, months, days, hours, minutes, seconds = [float(g) if g else 0 for g in match.groups()] | |
| return years * 365 * 24 * 3600 + months * 30 * 24 * 3600 + days * 24 * 3600 + hours * 3600 + minutes * 60 + seconds | |