diff --git a/dashlive/mpeg/dash/period.py b/dashlive/mpeg/dash/period.py index 09649808c..bafd2b537 100644 --- a/dashlive/mpeg/dash/period.py +++ b/dashlive/mpeg/dash/period.py @@ -24,6 +24,9 @@ from typing import Any, ClassVar import urllib.parse +from dashlive.drm.keymaterial import KeyMaterial +from dashlive.mpeg.dash.event_stream import EventStream +from dashlive.server.options.container import OptionsContainer import flask from dashlive.utils.list_of import ListOf @@ -38,8 +41,10 @@ class Period(ObjectWithFields): """ id: str adaptationSets: list[AdaptationSet] + event_streams: list[EventStream] start: datetime.timedelta duration: datetime.timedelta | None = None + mpdDuration: datetime.timedelta | None = None # value used in manifest time_offset: datetime.timedelta OBJECT_FIELDS: ClassVar[dict[str, Any]] = { @@ -51,7 +56,7 @@ class Period(ObjectWithFields): 'id': 'p0', } - def __init__(self, **kwargs): + def __init__(self, **kwargs) -> None: super().__init__(**kwargs) defaults = { 'adaptationSets': [], @@ -61,7 +66,7 @@ def __init__(self, **kwargs): } self.apply_defaults(defaults) - def key_ids(self): + def key_ids(self) -> set[KeyMaterial]: kids = set() for adp in self.adaptationSets: kids.update(adp.key_ids()) @@ -87,21 +92,24 @@ def maxSegmentDuration(self) -> float: return max([a.maxSegmentDuration for a in self.adaptationSets]) def finish_setup(self, - mode: str, timing: DashTiming | None, base_url: str, - use_base_urls: bool) -> None: - if use_base_urls: + options: OptionsContainer) -> None: + if options.useBaseUrls: self.baseURL = urllib.parse.urljoin(flask.request.host_url, base_url) + if options.mode == 'live' and not options.forcePeriodDurations: + self.mpdDuration = None + else: + self.mpdDuration = self.duration for adp in self.adaptationSets: - if mode == 'odvod': + if options.mode == 'odvod': for rep in adp.representations: rep.baseURL = f"{rep.id}.{adp.fileSuffix}" - if not use_base_urls: + if not options.useBaseUrls: rep.baseURL = f"{base_url}{rep.baseURL}" - if not use_base_urls: - if mode != 'odvod': + if not options.useBaseUrls: + if options.mode != 'odvod': adp.initURL = f"{base_url}{adp.initURL}" adp.mediaURL = f"{base_url}{adp.mediaURL}" if timing: diff --git a/dashlive/mpeg/dash/validator/content_protection.py b/dashlive/mpeg/dash/validator/content_protection.py index 7fb28eb2e..63c76d5b1 100644 --- a/dashlive/mpeg/dash/validator/content_protection.py +++ b/dashlive/mpeg/dash/validator/content_protection.py @@ -90,8 +90,8 @@ def validate_playready_pro(self, pro: PlayReadyRecord) -> None: self.elt.check_includes( ["4.0.0.0", "4.1.0.0", "4.2.0.0", "4.3.0.0"], xml.attrib['version']) - if 'playready_version' in self.mpd.params: - version = float(self.mpd.params['playready_version']) + if 'playready__version' in self.mpd.params: + version = float(self.mpd.params['playready__version']) if version < 2.0: self.elt.check_equal(xml.attrib['version'], "4.0.0.0") self.elt.check_equal( diff --git a/dashlive/mpeg/dash/validator/errors.py b/dashlive/mpeg/dash/validator/errors.py index a5f21a3fe..e9fe5c725 100644 --- a/dashlive/mpeg/dash/validator/errors.py +++ b/dashlive/mpeg/dash/validator/errors.py @@ -209,7 +209,7 @@ def check_greater_or_equal(self, a: Any, b: Any, result = a >= b if template is None: template = f'{0} should be >= {1}' - return self.check_true(result, a, b, **kwargs) + return self.check_true(result, a, b, template=template, **kwargs) def check_starts_with(self, text: str, prefix: str, template: str | None = None, **kwargs) -> bool: diff --git a/dashlive/mpeg/dash/validator/init_segment.py b/dashlive/mpeg/dash/validator/init_segment.py index 9d5867917..4c9143e4c 100644 --- a/dashlive/mpeg/dash/validator/init_segment.py +++ b/dashlive/mpeg/dash/validator/init_segment.py @@ -275,9 +275,9 @@ def validate_pssh(self, pssh: mp4.ContentProtectionSpecificBox) -> None: version: str | None = root.get("version") self.elt.check_includes( {"4.0.0.0", "4.1.0.0", "4.2.0.0", "4.3.0.0"}, version) - if 'playready_version' not in self.mpd.params: + if 'playready__version' not in self.mpd.params: continue - version = float(self.mpd.params['playready_version']) + version = float(self.mpd.params['playready__version']) if version < 2.0: self.elt.check_equal(root.attrib['version'], "4.0.0.0") elif version < 3.0: diff --git a/dashlive/mpeg/dash/validator/manifest.py b/dashlive/mpeg/dash/validator/manifest.py index deefe040d..a7c85fafc 100644 --- a/dashlive/mpeg/dash/validator/manifest.py +++ b/dashlive/mpeg/dash/validator/manifest.py @@ -81,6 +81,13 @@ def set_target_durations(self) -> None: return todo = datetime.timedelta(seconds=self.options.duration) for idx, period in enumerate(self.periods): + if period.duration is None and idx < len(self.periods) - 1: + next_period = self.periods[idx + 1] + if next_period.start is not None and period.start is not None: + period.duration = next_period.start - period.start + self.log.debug( + '%d: Setting duration of period %s to %s based on next period start time', + idx, period.id, toIsoDuration(period.duration)) if period.duration is None: self.log.debug( '%d: Using target duration %s for last period %s', idx, todo, period.id) @@ -209,19 +216,21 @@ def validate_self(self): self.elt.check_equal( self.patches, [], msg='PatchLocation elements should only be used in live streams') - start: datetime.timedelta | None = datetime.timedelta() - if self.periods: - start = self.periods[0].start + start: datetime.timedelta | None = datetime.timedelta() if self.mode == 'vod' else None for period in self.periods: + if start is not None and period.start is not None: + period.attrs.check_almost_equal( + period.start.total_seconds(), + start.total_seconds(), + delta=0.2, + msg=( + f"Expected Period@start {toIsoDuration(start)} " + + f"but found {toIsoDuration(period.start)}")) + if start is None: + start = period.start if not period.attrs.check_not_none( - start, 'Previous Period@duration was absent'): + start, 'Period@start is missing, but previous Period did not have a duration'): continue - period.attrs.check_almost_equal( - period.start.total_seconds(), - start.total_seconds(), - delta=0.2, - msg=(f"Expected Period@start {toIsoDuration(start)} " + - f"but found {toIsoDuration(period.start)}")) if period.duration is None: start = None else: diff --git a/dashlive/mpeg/dash/validator/period.py b/dashlive/mpeg/dash/validator/period.py index 708693079..18afce957 100644 --- a/dashlive/mpeg/dash/validator/period.py +++ b/dashlive/mpeg/dash/validator/period.py @@ -26,7 +26,7 @@ class Period(DashElement["Manifest"]): event_streams: list[EventStream] start: datetime.timedelta | None duration: datetime.timedelta | None - target_duration: datetime.timedelta | None = None + target_duration: datetime.timedelta | None = None # the amount of this period to verify attributes = [ ('id', str, None), diff --git a/dashlive/mpeg/dash/validator/representation.py b/dashlive/mpeg/dash/validator/representation.py index b820f240e..8e6db3df6 100644 --- a/dashlive/mpeg/dash/validator/representation.py +++ b/dashlive/mpeg/dash/validator/representation.py @@ -78,6 +78,10 @@ def __init__(self, elt: ET.ElementBase, parent: DashElement) -> None: def period(self) -> "Period": return cast("Period", self.parent.parent) + @property + def expected_duration(self) -> datetime.timedelta | None: + return self.period.duration + @property def target_duration(self) -> datetime.timedelta | None: return self.period.target_duration @@ -330,10 +334,10 @@ def generate_segments_using_segment_timeline(self, frameRate: float) -> None: self.id, idx + 1, need_duration) break if self.mode == 'live': - if self.target_duration is None or self.target_duration >= self.mpd.timeShiftBufferDepth: - tsb = self.mpd.timeShiftBufferDepth.total_seconds() * self.dash_timescale() + if self.expected_duration is not None: + expected_dur = timedelta_to_timecode(self.expected_duration, self.dash_timescale()) self.elt.check_greater_or_equal( - total_duration, tsb, + total_duration, expected_dur, template=r'SegmentTimeline has duration {0}, expected {1} based upon timeshiftbufferdepth') def presentation_time_offset(self) -> int: diff --git a/dashlive/server/events/base.py b/dashlive/server/events/base.py index 894775ac7..e3193a85e 100644 --- a/dashlive/server/events/base.py +++ b/dashlive/server/events/base.py @@ -1,19 +1,5 @@ ############################################################################# # -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# -############################################################################# -# # Project Name : Simulated MPEG DASH service # # Author : Alex Ashley @@ -22,10 +8,17 @@ from abc import abstractmethod import datetime -from typing import Any, Callable from dashlive.mpeg.mp4 import EventMessageBox -from dashlive.server.options.dash_option import DashOption +from dashlive.server.options.dash_option import ( + BoolDashOption, + CgiChoiceType, + DashOption, + DateTimeDashOption, + IntOrNoneDashOption, + StringDashOption, + StringOrNoneDashOption, +) from dashlive.server.options.types import OptionUsage from dashlive.utils.object_with_fields import ObjectWithFields @@ -40,6 +33,14 @@ class EventBase(ObjectWithFields): 'value': '0', 'version': 0, } + count: int + duration: int + inband: bool + interval: int + start: int + timescale: int + value: str + version: int @abstractmethod def create_manifest_context(self, context: dict) -> dict: @@ -49,54 +50,41 @@ def create_manifest_context(self, context: dict) -> dict: def create_emsg_boxes(self, **kwargs) -> list[EventMessageBox]: ... - @staticmethod - def int_or_default_from_string(default: int) -> Callable[[str], int]: - def int_or_default(value: str): - value = DashOption.int_or_none_from_string(value) - if value is None: - return default - return value - return int_or_default - @classmethod def get_dash_options(cls) -> list[DashOption]: """ Get a list of all DASH options for this event """ - def default_to_string(val: Any) -> str: - return str(val) result: list[DashOption] = [] for key, dflt in cls.DEFAULT_VALUES.items(): name: str = f"{cls.PREFIX}__{key}" short_name: str = f"{cls.PREFIX[:3]}{key.title()[:4]}" - cgi_choices = None + cgi_choices: tuple[CgiChoiceType, ...] | None = None + cgi_type: str | None = None input_type: str = 'text' - to_string = default_to_string - if isinstance(dflt, bool): - from_string = DashOption.bool_from_string - to_string = DashOption.bool_to_string + OptionType: type[DashOption] = StringDashOption + if dflt is None or dflt == "": + OptionType = StringOrNoneDashOption + elif isinstance(dflt, bool): + OptionType = BoolDashOption cgi_type = '(0|1)' input_type = 'checkbox' cgi_choices = (str(dflt), str(not dflt)) elif isinstance(dflt, int): - from_string = cls.int_or_default_from_string(dflt) + OptionType = IntOrNoneDashOption input_type = 'number' cgi_type = '' cgi_choices = tuple([str(dflt)]) elif isinstance(dflt, ( datetime.date, datetime.datetime, datetime.time, datetime.timedelta)): - from_string = DashOption.datetime_or_none_from_string - to_string = DashOption.datetime_or_none_to_string + OptionType = DateTimeDashOption cgi_type = '' cgi_choices = tuple([str(dflt)]) else: - from_string = default_to_string - cgi_type = None - if dflt is not None: - cgi_choices = tuple(str(dflt)) - opt = DashOption( + cgi_choices = tuple(str(dflt)) + opt = OptionType( usage=(OptionUsage.MANIFEST + OptionUsage.AUDIO + OptionUsage.VIDEO), short_name=short_name, full_name=key, @@ -106,8 +94,6 @@ def default_to_string(val: Any) -> str: input_type=input_type, cgi_name=name, cgi_type=cgi_type, - cgi_choices=cgi_choices, - from_string=from_string, - to_string=to_string) + cgi_choices=cgi_choices) result.append(opt) return result diff --git a/dashlive/server/events/factory.py b/dashlive/server/events/factory.py index 520b0b63d..b803848b4 100644 --- a/dashlive/server/events/factory.py +++ b/dashlive/server/events/factory.py @@ -1,34 +1,22 @@ ############################################################################# # -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# -############################################################################# -# # Project Name : Simulated MPEG DASH service # # Author : Alex Ashley # ############################################################################# -from typing import ClassVar +from typing import TYPE_CHECKING, ClassVar from dashlive.server.options.dash_option import DashOption -from dashlive.server.options.container import OptionsContainer from .base import EventBase from .ping_pong import PingPongEvents from .scte35_events import Scte35Events +if TYPE_CHECKING: + from dashlive.server.options.container import OptionsContainer + class EventFactory: EVENT_TYPES: ClassVar[dict[str, EventBase]] = { PingPongEvents.PREFIX: PingPongEvents, @@ -36,7 +24,7 @@ class EventFactory: } @classmethod - def create_event_generators(cls, options: OptionsContainer) -> list[EventBase]: + def create_event_generators(cls, options: "OptionsContainer") -> list[EventBase]: retval = [] for name in options.eventTypes: try: @@ -44,7 +32,7 @@ def create_event_generators(cls, options: OptionsContainer) -> list[EventBase]: except KeyError as err: print(f'Unknown event class "{name}": {err}') continue - args = options[EventClazz.PREFIX].toJSON(exclude={'_type'}) + args = getattr(options, EventClazz.PREFIX).toJSON() retval.append(EventClazz(**args)) return retval diff --git a/dashlive/server/manifests.py b/dashlive/server/manifests.py index 40091f40f..f2278156d 100644 --- a/dashlive/server/manifests.py +++ b/dashlive/server/manifests.py @@ -1,19 +1,5 @@ ############################################################################# # -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# -############################################################################# -# # Project Name : Simulated MPEG DASH service # # Author : Alex Ashley @@ -86,7 +72,6 @@ def cgi_query_combinations(self) -> Iterator[str]: """ Returns an iterator that yields of all possible combinations of CGI query parameters """ - defaults = OptionsRepository.get_default_options() indexes = [0] * len(self.options) done = False num_options = len(self.options) @@ -105,9 +90,9 @@ def cgi_query_combinations(self) -> Iterator[str]: break params[name] = val if allowed: - candidate = OptionsRepository.convert_cgi_options(params, defaults) + candidate = OptionsRepository.convert_cgi_options(params) candidate.update(**self.kwargs) - candidate.remove_unused_parameters(self.mode) + candidate.reset_unused_parameters(self.mode) cgi_str = candidate.generate_cgi_parameters_string() digest: bytes = hashlib.sha1(bytes(cgi_str, 'ascii')).digest() if digest not in checked: diff --git a/dashlive/server/options/all_options.py b/dashlive/server/options/all_options.py new file mode 100644 index 000000000..4e6427c8d --- /dev/null +++ b/dashlive/server/options/all_options.py @@ -0,0 +1,26 @@ +############################################################################# +# +# Project Name : Simulated MPEG DASH service +# +# Author : Alex Ashley +# +############################################################################# +from .audio_options import audio_options +from .dash_option import DashOption +from .drm_options import drm_options +from .event_options import event_options +from .manifest_options import manifest_options +from .player_options import player_options +from .text_options import text_options +from .video_options import video_options +from .utc_time_options import time_options + +ALL_OPTIONS: list[DashOption] = ( + audio_options + + drm_options + + event_options + + manifest_options + + player_options + + video_options + + text_options + + time_options) diff --git a/dashlive/server/options/audio_options.py b/dashlive/server/options/audio_options.py index eca4253f7..45fce316b 100644 --- a/dashlive/server/options/audio_options.py +++ b/dashlive/server/options/audio_options.py @@ -6,11 +6,11 @@ # ############################################################################# -from .dash_option import DashOption +from .dash_option import StringDashOption, StringOrNoneDashOption from .http_error import AudioHttpError from .types import OptionUsage -AudioCodec = DashOption( +AudioCodec = StringDashOption( usage=(OptionUsage.MANIFEST | OptionUsage.AUDIO), short_name='ac', full_name='audioCodec', @@ -25,7 +25,7 @@ input_type='select', featured=True) -AudioDescriptionTrack = DashOption( +AudioDescriptionTrack = StringOrNoneDashOption( usage=OptionUsage.MANIFEST, short_name='ad', full_name='audioDescription', @@ -35,7 +35,7 @@ cgi_name='ad_audio', input_type='audio_representation') -MainAudioTrack = DashOption( +MainAudioTrack = StringOrNoneDashOption( usage=OptionUsage.MANIFEST, short_name='ma', full_name='mainAudio', diff --git a/dashlive/server/options/container.py b/dashlive/server/options/container.py index 563a57874..eaa2be809 100644 --- a/dashlive/server/options/container.py +++ b/dashlive/server/options/container.py @@ -1,19 +1,5 @@ ############################################################################# # -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# -############################################################################# -# # Project Name : Simulated MPEG DASH service # # Author : Alex Ashley @@ -21,35 +7,29 @@ ############################################################################# from collections import defaultdict +import copy +import dataclasses +import logging from typing import AbstractSet, Any, Optional from dashlive.drm.location import DrmLocation from dashlive.drm.system import DrmSystem from dashlive.server.options.drm_options import ( + ALL_DRM_LOCATIONS, DrmLocationOption, DrmSelectionTuple ) from dashlive.components.field_group import InputFieldGroup from dashlive.server.options.form_input_field import FormInputContext +from dashlive.server.options.name_maps import DashOptionNameMaps +from dashlive.server.options.options_types import OptionsContainerType from dashlive.utils.json_object import JsonObject -from dashlive.utils.object_with_fields import ObjectWithFields from dashlive.utils.objects import dict_to_cgi_params from .dash_option import DashOption from .types import OptionUsage -class OptionsContainer(ObjectWithFields): - OBJECT_FIELDS = {} - _parameter_map: Optional[dict[str, DashOption]] - _defaults: Optional["OptionsContainer"] - - def __init__(self, - parameter_map: Optional[dict[str, DashOption]] = None, - defaults: Optional["OptionsContainer"] = None, - **kwargs) -> None: - super().__init__(**kwargs) - self._parameter_map = parameter_map - self._defaults = defaults +class OptionsContainer(OptionsContainerType): @property def encrypted(self) -> bool: @@ -59,74 +39,122 @@ def encrypted(self) -> bool: return False def clone(self, **kwargs) -> "OptionsContainer": - args = { - 'parameter_map': self._parameter_map, - } - for key in self._fields: - if key[0] == '_': - continue - ours = getattr(self, key) - value = kwargs.get(key, ours) - if isinstance(value, OptionsContainer) or isinstance(ours, OptionsContainer): - if ours is None: - ours = {} - elif isinstance(ours, OptionsContainer): - ours = ours.toJSON() - if value is None: - theirs = {} - elif isinstance(value, OptionsContainer): - theirs = value.toJSON() - else: - theirs = value - value = { - **ours, - **theirs, - } - dflt = None - if self._defaults is not None: - dflt = self._defaults[key] - value = self.__class__( - parameter_map=self._parameter_map, defaults=dflt, **value) - args[key] = value - for key, value in kwargs.items(): - if key in self._fields or key[0] == '_': - continue - args[key] = value - if 'defaults' not in args: - args['defaults'] = self._defaults - return OptionsContainer(**args) + result: OptionsContainer = copy.deepcopy(self) + result.update(**kwargs) + return result def update(self, **kwargs) -> None: """ - Apply the provided values to this options container + Apply the provided values to this options container. Each item in kwargs + must be a full-name field name. If a sub-option group is provided, this function + will recursively apply the provided values to update the sub-option. """ + parameter_map = DashOptionNameMaps.get_parameter_map() for key, value in kwargs.items(): - self.add_field(key, value) + assert key in self.__dict__, f"Invalid option name: {key}" + if key not in parameter_map: # must be a sub-option group + assert isinstance(value, dict) + dest = getattr(self, key) + for k2, v2 in value.items(): + opt: DashOption = parameter_map[f'{key}.{k2}'] + self.set_with_type_coercion(dest, opt, k2, v2) + elif key == 'drmSelection': + # special handling for drmSelection to translate to DrmLocation enum values + # TODO: find a general way to handle this kind of field + if value is None: + value = [] + new_value: list[DrmSelectionTuple] = [] + for name, loc_val in value: + locs: set[DrmLocation] = set() + if loc_val is None: + locs = ALL_DRM_LOCATIONS + elif isinstance(loc_val, str): + locs.add(DrmLocation.from_string(loc_val)) + else: + for it in loc_val: + if isinstance(it, str): + locs.add(DrmLocation.from_string(it)) + else: + locs.add(it) + new_value.append((name, locs)) + self.drmSelection = new_value + else: + opt: DashOption = parameter_map[key] + self.set_with_type_coercion(self, opt, key, value) + + @staticmethod + def set_with_type_coercion(dest: object, opt: DashOption, key: str, value: Any) -> None: + ours = getattr(dest, key) + if ours == value: + return + if ours is not None and value is not None and not isinstance(value, type(ours)): + if isinstance(value, str): + value = opt.from_string(value) + if not isinstance(value, type(ours)): + py_type: str | None = opt.python_type_hint() + if py_type is None or value.__class__.__name__ not in py_type: + raise ValueError( + f"Invalid type for field {key}: expected {type(ours)}, got {type(value)}. Allowed types: {py_type}") + setattr(dest, key, value) + + def apply_options(self, + params: dict[str, str], + is_cgi: bool) -> None: + """ + Apply the provided CGI parameters (or short name parameters) to this options container + """ + param_map: dict[str, DashOption] + if is_cgi: + param_map = DashOptionNameMaps.get_cgi_map() + else: + param_map = DashOptionNameMaps.get_short_param_map() + for key, value in params.items(): + try: + opt: DashOption = param_map[key] + value = opt.from_string(value) + default = opt.default_value() + if value is None and default is not None: + value = default + if opt.prefix: + dest = getattr(self, opt.prefix) + setattr(dest, opt.full_name, value) + else: + setattr(self, opt.full_name, value) + except KeyError as err: + logging.warning(r'Invalid parameter name %s is_cgi=%s: %s', key, is_cgi, err) + print(f"Invalid parameter name {key} is_cgi={is_cgi}: {err}") def _convert_sub_options(self, destination: dict[str, str], + is_cgi: bool, prefix: str, - sub_opts: dict[str, Any], + sub_opts: object, use: OptionUsage | None, exclude: AbstractSet | None, - remove_defaults: bool) -> None: - defaults = ObjectWithFields() - if self._defaults is not None and prefix in self._defaults._fields: - defaults = self._defaults[prefix] - for key, value in sub_opts.items(): - name: str = f'{prefix}.{key}' - opt: DashOption = self._parameter_map[name] - if use is not None and (opt.usage & use) == 0: - continue + defaults: object | None) -> None: + if exclude is None: + exclude = set() + assert dataclasses.is_dataclass(sub_opts) + parameter_map = DashOptionNameMaps.get_parameter_map() + for field in dataclasses.fields(sub_opts): + name: str = f'{prefix}.{field.name}' if name in exclude: continue - try: - dft_val = defaults[key] - if remove_defaults and value == dft_val: - continue - except KeyError: - pass - destination[opt.cgi_name] = opt.to_string(value) + opt: DashOption = parameter_map[name] + skip: bool = use is not None and (opt.usage & use) == 0 + value: Any = getattr(sub_opts, field.name) + if defaults is not None: + try: + dft_val = getattr(defaults, field.name) + if value == dft_val: + skip = True + except AttributeError: + pass + if not skip: + if is_cgi: + destination[opt.cgi_name] = opt.to_string(value) + else: + destination[opt.short_name] = value def generate_cgi_parameters(self, destination: dict[str, str] | None = None, @@ -138,7 +166,7 @@ def generate_cgi_parameters(self, Any option that matches its default is excluded. """ return self._generate_parameters_dict( - 'cgi_name', destination=destination, use=use, exclude=exclude, + is_cgi=True, destination=destination, use=use, exclude=exclude, remove_defaults=remove_defaults) def generate_short_parameters(self, @@ -151,11 +179,11 @@ def generate_short_parameters(self, Any option that matches its default is excluded. """ return self._generate_parameters_dict( - 'short_name', destination=destination, use=use, exclude=exclude, + is_cgi=False, destination=destination, use=use, exclude=exclude, remove_defaults=remove_defaults) def _generate_parameters_dict(self, - attr_name: str, + is_cgi: bool, destination: dict[str, str] | None, use: OptionUsage | None, exclude: AbstractSet | None, @@ -164,52 +192,58 @@ def _generate_parameters_dict(self, Produces a dictionary of parameters that represent these options. Any option that matches its default is excluded if :remove_defaults: is True """ + attr_name: str = 'cgi_name' if is_cgi else 'short_name' if exclude is None: exclude = {'encrypted', 'mode'} if destination is None: destination = {} - assert self._parameter_map is not None - for key, value in self.items(): - if isinstance(value, OptionsContainer): - self._convert_sub_options(destination, key, value, use, exclude, remove_defaults) + parameter_map = DashOptionNameMaps.get_parameter_map() + defaults = OptionsContainer() + for field in dataclasses.fields(self): + if field.name in exclude: continue - if key in exclude: + + value = getattr(self, field.name) + + if dataclasses.is_dataclass(value): + sub_defaults = getattr(defaults, field.name) if remove_defaults else None + self._convert_sub_options( + destination=destination, prefix=field.name, sub_opts=value, use=use, + exclude=exclude, defaults=sub_defaults, is_cgi=is_cgi) continue - if remove_defaults and self._defaults is not None: - if key in self._defaults._fields: - dft_val = getattr(self._defaults, key) - if value == dft_val: - continue - opt: DashOption = self._parameter_map[key] + + skip: bool = False + if remove_defaults: + try: + dft_val = getattr(defaults, field.name) + skip = value == dft_val + except AttributeError: + pass + opt: DashOption = parameter_map[field.name] if use is not None and (opt.usage & use) == 0: - continue - destination[getattr(opt, attr_name)] = opt.to_string(value) + skip = True + if not skip: + destination[getattr(opt, attr_name)] = opt.to_string(value) return destination - def remove_default_values(self, defaults: Optional["OptionsContainer"] = None) -> JsonObject: - if defaults is None: - defaults = self._defaults + def json_without_default_values(self, defaults: Optional["OptionsContainer"] = None) -> JsonObject: if defaults is None: - return self.toJSON() + defaults = OptionsContainer() result: JsonObject = {} - for key, value in self.items(): - try: - dflt = defaults[key] - except KeyError: - continue - if isinstance(value, OptionsContainer): - sub_result = {} - for k, v in value.items(): - if k not in dflt: - continue - d = dflt[k] + for field in dataclasses.fields(self): + value = getattr(self, field.name) + dflt = getattr(defaults, field.name) + if dataclasses.is_dataclass(value): + sub_result: JsonObject = {} + for it in dataclasses.fields(value): + v = getattr(value, it.name) + d = getattr(dflt, it.name) if d != v: - sub_result[k] = v + sub_result[it.name] = v if sub_result: - result[key] = sub_result - continue - if value != dflt: - result[key] = value + result[field.name] = sub_result + elif value != dflt: + result[field.name] = value return result def generate_cgi_parameters_string(self, @@ -219,56 +253,75 @@ def generate_cgi_parameters_string(self, use=use, exclude=exclude)) def remove_unsupported_features(self, supported_features: AbstractSet[str]) -> None: - todo = { + todo: set[str] = { 'abr', 'audioCodec', 'useBaseUrls', 'drmSelection', 'eventTypes', 'minimumUpdatePeriod', 'segmentTimeline', 'utcMethod' } todo.difference_update(supported_features) + defaults = OptionsContainer() for name in todo: - if self._defaults is None: - self.remove_field(name) - else: - setattr(self, name, getattr(self._defaults, name)) + setattr(self, name, getattr(defaults, name)) - def remove_unused_parameters(self, mode: str, encrypted: bool | None = None, - use: OptionUsage | None = None) -> None: + def reset_unused_parameters( + self, + mode: str, + encrypted: bool | None = None, + use: OptionUsage | None = None) -> None: + """ + Reset to default all values that are not relevant based upon selected mode. + """ if encrypted is None: encrypted = self.encrypted todo: list[str] = [] if mode != 'live': - todo += {'availabilityStartTime', 'minimumUpdatePeriod', + todo += ['availabilityStartTime', 'minimumUpdatePeriod', 'ntpSources', 'timeShiftBufferDepth', 'utcMethod', - 'utcValue', 'patch'} + 'utcValue', 'patch'] if encrypted: - drms = {item[0] for item in self.drmSelection} + drms: set[str] = {item[0] for item in self.drmSelection} if 'playready' not in drms: - todo += {'playreadyLicenseUrl', 'playreadyPiff', 'playreadyVersion'} + todo += ['playready.licenseUrl', 'playready.piff', 'playready.version'] if 'marlin' not in drms: - todo.append('marlinLicenseUrl') + todo.append('marlin.licenseUrl') + if 'clearkey' not in drms: + todo.append('clearkey.licenseUrl') else: - todo += {'marlinLicenseUrl', 'playreadyLicenseUrl', 'playreadyPiff', - 'playreadyVersion'} + todo += ['marlin.licenseUrl', 'playready.licenseUrl', 'playready.piff', + 'playready.version', 'clearkey.licenseUrl'] if use is not None: - fields = set(self._fields) - fields.discard(set(todo)) + fields: set[str] = set() + for field in dataclasses.fields(self): + if dataclasses.is_dataclass(field.type): + for it in dataclasses.fields(field.type): + fields.add(f"{field.name}.{it.name}") + else: + fields.add(field.name) + fields -= set(todo) + parameter_map = DashOptionNameMaps.get_parameter_map() for name in fields: try: - opt = self._parameter_map[name] + opt = parameter_map[name] + if (opt.usage & use) == 0: + todo.append(name) except KeyError: - continue - if (opt.usage & use) == 0: - todo.append(name) + pass + defaults = OptionsContainer() for name in todo: - self.remove_field(name) + if '.' in name: + prefix, key = name.split('.') + val = getattr(getattr(defaults, prefix), key) + setattr(getattr(self, prefix), key, val) + else: + setattr(self, name, getattr(defaults, name)) def generate_input_field_groups( self, field_choices: dict, exclude: AbstractSet | None = None) -> list[InputFieldGroup]: sections: dict[str, list[FormInputContext]] = defaultdict(list) for field in self.generate_input_fields(field_choices, exclude): - group: str = field['prefix'] + group: str = field.get('prefix', '') if group == "": - group = "general" if field["featured"] else "advanced" + group = "general" if field.get("featured", False) else "advanced" sections[group].append(field) result: list[InputFieldGroup] = [ InputFieldGroup("general", "General Options", sections["general"], show=True), @@ -283,20 +336,22 @@ def generate_input_fields(self, field_choices: dict, fields: list[FormInputContext] = [] if exclude is None: exclude = set() - for key, value in self.items(): - if key in exclude: + parameter_map = DashOptionNameMaps.get_parameter_map() + for field in dataclasses.fields(self): + if field.name in exclude: continue - if isinstance(value, OptionsContainer): - for ok, ov in value.items(): - name: str = f'{key}.{ok}' + value = getattr(self, field.name) + if dataclasses.is_dataclass(value): + for it in dataclasses.fields(value): + name: str = f'{field.name}.{it.name}' if name in exclude: continue - op: DashOption = self._parameter_map[name] + op: DashOption = parameter_map[name] fields.append( - op.input_field(ov, field_choices)) + op.input_field(getattr(value, it.name), field_choices)) continue try: - opt = self._parameter_map[key] + opt = parameter_map[field.name] except KeyError: continue if opt.full_name == 'drmSelection': diff --git a/dashlive/server/options/dash_option.py b/dashlive/server/options/dash_option.py index 840417b8b..ea19d135a 100644 --- a/dashlive/server/options/dash_option.py +++ b/dashlive/server/options/dash_option.py @@ -1,42 +1,28 @@ ############################################################################# # -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# -############################################################################# -# # Project Name : Simulated MPEG DASH service # # Author : Alex Ashley # ############################################################################# +from abc import abstractmethod from dataclasses import asdict, dataclass, field -import datetime -from typing import Any, Union +from datetime import datetime, timedelta +from typing import Any, Generic, TypeVar, Union, cast -from collections.abc import Callable import urllib.parse -from dashlive.utils.date_time import from_isodatetime, to_iso_datetime -from dashlive.utils.objects import flatten +from dashlive.utils.date_time import from_isodatetime, to_iso_datetime, toIsoDuration -from .form_input_field import FormInputContext +from .form_input_field import FieldOption, FormInputContext from .types import CgiOption, CgiOptionChoice, OptionUsage CgiChoiceType = Union[tuple[str, str], str, None] +T = TypeVar('T') @dataclass(slots=True, frozen=True) -class DashOption: +class DashOption(Generic[T]): usage: OptionUsage short_name: str full_name: str @@ -46,12 +32,38 @@ class DashOption: cgi_choices: tuple[CgiChoiceType, ...] | None = field(default=None) cgi_type: str | None = None input_type: str = '' - from_string: Callable[[str], Any] = field(default_factory=lambda: DashOption.string_or_none) - to_string: Callable[[str], Any] = field(default_factory=lambda: flatten) prefix: str = field(default='') featured: bool = False html: str | None = None + @abstractmethod + def from_string(self, value: str) -> T: + raise NotImplementedError(f"{__class__}.from_string() not implemented") + + @abstractmethod + def to_string(self, value: T) -> str: + raise NotImplementedError(f"{__class__}.to_string() not implemented") + + def html_input_type(self) -> str: + return self.input_type + + def python_type_hint(self) -> str | None: + return None + + def default_value(self) -> T | None: + if self.cgi_choices is None: + return None + if len(self.cgi_choices) == 0: + return None + first = self.cgi_choices[0] + if isinstance(first, tuple): + if first[1] is None: + return None + return self.from_string(first[1]) + if isinstance(first, str): + return self.from_string(first) + return cast(T, first) + def get_cgi_option(self, omit_empty: bool = True) -> CgiOption | None: """ Get a description of the CGI values that are allowed for this option @@ -107,7 +119,7 @@ def input_field(self, value: Any, field_choices: dict) -> FormInputContext: "title": self.title, "value": value, "text": self.description, - "type": self.input_type, + "type": self.html_input_type(), "prefix": self.prefix, "fullName": self.full_name, "shortName": self.short_name, @@ -124,44 +136,42 @@ def input_field(self, value: Any, field_choices: dict) -> FormInputContext: title = '--' if val is None: val = '' - input['options'].append({ - "value": val, - "title": title, - "selected": value == val - }) + input['options'].append(FieldOption( + value=val, + title=title, + selected=(value == val) + )) if input['type'] == '': - if isinstance(value, bool) or self.to_string == DashOption.bool_to_string: + if isinstance(value, bool): input['type'] = 'bool' - elif isinstance(value, int) or self.to_string == DashOption.int_or_none_from_string: + elif isinstance(value, (int, float)): input['type'] = 'number' elif self.cgi_choices and len(self.cgi_choices) > 1: input['type'] = 'select' if input['type'] == 'multipleSelect': input['type'] = 'select' input['multiple'] = True - for val in value: - for ch in input['options']: - if ch['value'] == val: - ch['selected'] = True + options: list[FieldOption] = input.get('options', []) + for val in cast(list, value): + for idx, ch in enumerate(options): # pyright: ignore[reportTypedDictNotRequiredAccess] + if ch.value == val: + options[idx] = FieldOption( + value=ch.value, title=ch.title, selected=True) + input['options'] = options elif input['type'] == 'bool': if value is None: input['type'] = 'select' - input['options'] = [{ - "value": '', - "title": '--', - "selected": value is None, - }, { - "value": '1', - "title": 'True', - "selected": value is True, - }, { - "value": '0', - "title": 'False', - "selected": value is False, - }] + input['options'] = [ + FieldOption(value='', title='--', selected=(value is None)), + FieldOption(value='1', title='True', selected=(value is True)), + FieldOption(value='0', title='False', selected=(value is False)), + ] else: input['type'] = 'checkbox' - del input['options'] + try: + del input['options'] + except KeyError: + pass elif input['type'] == 'numberList': input['datalist_type'] = 'number' input['type'] = 'datalist' @@ -178,44 +188,101 @@ def replace(self, **kwargs) -> "DashOption": new_kwargs.update(kwargs) return DashOption(**new_kwargs) - @staticmethod - def bool_from_string(value: str) -> bool: + +class BoolDashOption(DashOption[bool]): + def from_string(self, value: str) -> bool: return value.lower() in {'true', '1', 'on'} - @staticmethod - def bool_to_string(value: bool) -> str: + def to_string(self, value: bool) -> str: if value: return '1' return '0' - @staticmethod - def int_or_none_from_string(value: str) -> int | None: - if value in {None, '', 'none'}: + def html_input_type(self) -> str: + return 'bool' + + def python_type_hint(self) -> str | None: + return 'bool' + + +class IntOrNoneDashOption(DashOption[int | None]): + def from_string(self, value: str) -> int | None: + if value == '' or value.lower() == 'none': return None return int(value, 10) - @staticmethod - def float_or_none_from_string(value: str) -> float | None: + def to_string(self, value: int | None) -> str: + if value is None: + return '' + return f"{value:d}" + + def html_input_type(self) -> str: + return 'number' + + def python_type_hint(self) -> str | None: + if self.default_value() is not None: + return 'int' + return 'int | None' + +class FloatOrNoneDashOption(DashOption[float | None]): + def from_string(self, value: str) -> float | None: if value in {None, '', 'none'}: return None return float(value) - @staticmethod - def datetime_or_none_from_string(value: str) -> datetime.datetime | datetime.timedelta | None: - if value in {None, '', 'none'}: + def to_string(self, value: float | None) -> str: + if value is None: + return '' + return f'{value:f}' + + def python_type_hint(self) -> str | None: + return 'float | None' + + +class StringDashOption(DashOption[str]): + def from_string(self, value: str) -> str: + if value.lower() in ['', 'none']: + return '' + return value + + def to_string(self, value: str) -> str: + return value + + def python_type_hint(self) -> str | None: + return 'str' + + +class StringOrNoneDashOption(DashOption[str | None]): + def from_string(self, value: str) -> str | None: + if value.lower() in ['', 'none']: return None - return from_isodatetime(value) + return value - @staticmethod - def datetime_or_none_to_string(value: datetime.datetime | None) -> str | None: + def to_string(self, value: str | None) -> str: if value is None: + return '' + return value + + def python_type_hint(self) -> str | None: + return 'str | None' + + +class UrlOrNoneDashOption(DashOption[str | None]): + def from_string(self, value: str) -> str | None: + if value.lower() in ['', 'none']: return None - return to_iso_datetime(value) + return urllib.parse.unquote_plus(value) - @staticmethod - def list_without_none_from_string(value: str | None) -> list[str]: + def to_string(self, value: str | None) -> str: if value is None: - return [] + return '' + return urllib.parse.quote_plus(value) + + def python_type_hint(self) -> str | None: + return 'str | None' + +class StringListDashOption(DashOption[list[str]]): + def from_string(self, value: str) -> list[str]: if value.lower() in {'', 'none'}: return [] rv = [] @@ -224,20 +291,32 @@ def list_without_none_from_string(value: str | None) -> list[str]: rv.append(item) return rv - @staticmethod - def string_or_none(value: str) -> str | None: - if value.lower() in ['', 'none']: - return None - return value + def to_string(self, value: list[str]) -> str: + return ','.join(value) - @staticmethod - def unquoted_url_or_none_from_string(value: str): - if value.lower() in ['', 'none']: + def python_type_hint(self) -> str | None: + return 'list[str]' + + def default_value(self) -> list[str] | None: + if not self.cgi_choices: + return [] + default: str | None = cast(str | None, self.cgi_choices[0]) + if default is None: + return [] + return [default] + +class DateTimeDashOption(DashOption[datetime | timedelta | None]): + def from_string(self, value: str) -> datetime | timedelta | None: + if value in {None, '', 'none'}: return None - return urllib.parse.unquote_plus(value) + return from_isodatetime(value) - @staticmethod - def quoted_url_or_none_to_string(value: str | None): + def to_string(self, value: datetime | timedelta | None) -> str: if value is None: - return None - return urllib.parse.quote_plus(value) + return '' + if isinstance(value, timedelta): + return toIsoDuration(value) + return to_iso_datetime(value) + + def python_type_hint(self) -> str | None: + return 'datetime.datetime | datetime.timedelta | None' diff --git a/dashlive/server/options/drm_options.py b/dashlive/server/options/drm_options.py index 2934d4507..32ee99439 100644 --- a/dashlive/server/options/drm_options.py +++ b/dashlive/server/options/drm_options.py @@ -1,42 +1,33 @@ ############################################################################# # -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# -############################################################################# -# # Project Name : Simulated MPEG DASH service # # Author : Alex Ashley # ############################################################################# -from typing import TypeAlias +from typing import TypeAlias, cast from dashlive.drm.location import DrmLocation from dashlive.drm.system import DrmSystem -from .dash_option import DashOption +from .dash_option import ( + BoolDashOption, + CgiChoiceType, + DashOption, + FloatOrNoneDashOption, + StringListDashOption, + UrlOrNoneDashOption, +) from .types import OptionUsage -ClearkeyLicenseUrl = DashOption( +ClearkeyLicenseUrl = UrlOrNoneDashOption( usage=(OptionUsage.MANIFEST | OptionUsage.AUDIO | OptionUsage.VIDEO), short_name='clu', full_name='licenseUrl', prefix='clearkey', title='Clearkey LA_URL', description='Override the Clearkey license URL field', - from_string=DashOption.unquoted_url_or_none_from_string, - to_string=DashOption.quoted_url_or_none_to_string, cgi_name='clearkey__la_url', cgi_type='') @@ -59,16 +50,15 @@

For example: drm=playready-pro-cenc,clearkey-moov

''' -DrmLocationOption = DashOption( +DrmLocationOption = StringListDashOption( usage=(OptionUsage.MANIFEST | OptionUsage.AUDIO | OptionUsage.VIDEO | OptionUsage.TEXT), short_name='dloc', full_name='drmLocation', title='DRM location', description='Location to place DRM data', - from_string=DashOption.list_without_none_from_string, cgi_name='drmloc', cgi_choices=( - ('All locations', None), + cast(CgiChoiceType, ('All locations', None,)), ('mspr:pro element in MPD', 'pro'), ('dash:cenc element in MPD', 'cenc'), ('PSSH in init segment', 'moov'), @@ -85,106 +75,114 @@ DrmSelectionTuple: TypeAlias = tuple[str, set[DrmLocation]] -def _drm_selection_from_string(value: str) -> list[DrmSelectionTuple]: - locations: set[DrmLocation] - - value = value.lower() - if value.startswith('none') or value == '': +class DrmSelectionOption(DashOption[list[DrmSelectionTuple]]): + def __init__(self) -> None: + super().__init__( + usage=(OptionUsage.MANIFEST | OptionUsage.AUDIO | OptionUsage.VIDEO | OptionUsage.TEXT), + short_name='drm', + full_name='drmSelection', + title='Encryption', + description=( + 'Comma separated list of DRM names to enable. ' + + 'Optionally each DRM name can contain a hyphen separated list of locations for the DRM data'), + html=HTML_DESCRIPTION, + cgi_name='drm', + cgi_type=',.. or -,..', + input_type='multipleSelect', + cgi_choices=tuple([None, 'all'] + DrmSystem.values()), + featured=True) + + def from_string(self, value: str) -> list[DrmSelectionTuple]: + locations: set[DrmLocation] + + value = value.lower() + if value.startswith('none') or value == '': + return [] + if value.startswith('all'): + if '-' in value: + locations = {DrmLocation.from_string(loc) for loc in value.split('-')[1:]} + else: + locations = ALL_DRM_LOCATIONS + return [(drm, locations) for drm in DrmSystem.values()] + + result: list[DrmSelectionTuple] = [] + for item in value.split(','): + if '-' in item: + parts = item.split('-') + drm = parts[0] + locations = {DrmLocation(loc) for loc in parts[1:]} + else: + drm = item + locations = ALL_DRM_LOCATIONS + result.append((drm, locations)) + return result + + def to_string(self, value: list[DrmSelectionTuple]) -> str: + result: list[str] = [] + for drm, locations in value: + if locations == ALL_DRM_LOCATIONS: + result.append(drm) + else: + locs: list[str] = [] + for it in locations: + if isinstance(it, DrmLocation): + locs.append(it.to_json()) + elif isinstance(it, str): + locs.append(it) + locs.sort() + parts: list[str] = [drm] + locs + result.append('-'.join(parts)) + if set(result) == ALL_DRM_NAMES: + return 'all' + return ','.join(result) + + def default_value(self) -> list[tuple[str, set[DrmLocation]]] | None: return [] - if value.startswith('all'): - if '-' in value: - locations = {DrmLocation.from_string(loc) for loc in value.split('-')[1:]} - else: - locations = ALL_DRM_LOCATIONS - return [(drm, locations) for drm in DrmSystem.values()] - result: list[DrmSelectionTuple] = [] - for item in value.split(','): - if '-' in item: - parts = item.split('-') - drm = parts[0] - locations = {DrmLocation(loc) for loc in parts[1:]} - else: - drm = item - locations = ALL_DRM_LOCATIONS - result.append((drm, locations)) - return result - - -def _drm_selection_to_string(value: list[DrmSelectionTuple]) -> str: - result: list[str] = [] - for drm, locations in value: - if locations == ALL_DRM_LOCATIONS: - result.append(drm) - else: - parts = [drm] + sorted([loc.to_json() for loc in locations]) - result.append('-'.join(parts)) - if set(result) == ALL_DRM_NAMES: - return 'all' - return ','.join(result) - - -DrmSelection = DashOption( - usage=(OptionUsage.MANIFEST | OptionUsage.AUDIO | OptionUsage.VIDEO | OptionUsage.TEXT), - short_name='drm', - full_name='drmSelection', - title='Encryption', - description=( - 'Comma separated list of DRM names to enable. ' + - 'Optionally each DRM name can contain a hyphen separated list of locations for the DRM data'), - html=HTML_DESCRIPTION, - from_string=_drm_selection_from_string, - to_string=_drm_selection_to_string, - cgi_name='drm', - cgi_type=',.. or -,..', - input_type='multipleSelect', - cgi_choices=tuple([None, 'all'] + DrmSystem.values()), - featured=True) -MarlinLicenseUrl = DashOption( + def python_type_hint(self) -> str | None: + return 'list[tuple]' + + +DrmSelection = DrmSelectionOption() + +MarlinLicenseUrl = UrlOrNoneDashOption( usage=(OptionUsage.MANIFEST | OptionUsage.AUDIO | OptionUsage.VIDEO), short_name='mlu', full_name='licenseUrl', prefix='marlin', title='Marlin LA_URL', description='Override the Marlin S-URL field', - from_string=DashOption.unquoted_url_or_none_from_string, - to_string=DashOption.quoted_url_or_none_to_string, cgi_name='marlin__la_url', cgi_type='') -PlayreadyLicenseUrl = DashOption( +PlayreadyLicenseUrl = UrlOrNoneDashOption( usage=(OptionUsage.MANIFEST | OptionUsage.AUDIO | OptionUsage.VIDEO), short_name='plu', full_name='licenseUrl', title='Playready LA_URL', description='Override the Playready LA_URL field', - from_string=DashOption.unquoted_url_or_none_from_string, - to_string=DashOption.quoted_url_or_none_to_string, cgi_name='playready__la_url', cgi_type='', prefix='playready') -PlayreadyPiff = DashOption( +PlayreadyPiff = BoolDashOption( usage=(OptionUsage.MANIFEST | OptionUsage.AUDIO | OptionUsage.VIDEO), short_name='pff', full_name='piff', title='Playready PIFF', prefix='playready', description='Include PIFF sample encryption data', - from_string=DashOption.bool_from_string, - to_string=DashOption.bool_to_string, input_type='checkbox', cgi_name='playready__piff', cgi_choices=('1', '0')) -PlayreadyVersion = DashOption( +PlayreadyVersion = FloatOrNoneDashOption( usage=(OptionUsage.MANIFEST | OptionUsage.AUDIO | OptionUsage.VIDEO), short_name='pvn', full_name='version', prefix='playready', title='Playready Version', description='Set the PlayReady version compatibility for this stream', - from_string=DashOption.float_or_none_from_string, input_type='select', cgi_name='playready__version', cgi_choices=(None, '1.0', '2.0', '3.0', '4.0')) diff --git a/dashlive/server/options/event_options.py b/dashlive/server/options/event_options.py index be3948ba5..ca223759f 100644 --- a/dashlive/server/options/event_options.py +++ b/dashlive/server/options/event_options.py @@ -8,7 +8,7 @@ from dashlive.server.events.factory import EventFactory -from .dash_option import DashOption +from .dash_option import StringListDashOption from .types import OptionUsage EV_HTML = ''' @@ -25,14 +25,12 @@ ''' -EventSelection = DashOption( +EventSelection = StringListDashOption( usage=(OptionUsage.MANIFEST | OptionUsage.AUDIO | OptionUsage.VIDEO), short_name='evs', full_name='eventTypes', title='DASH events', description='A comma separated list of event formats', - from_string=DashOption.list_without_none_from_string, - to_string=lambda evs: ','.join(evs), html=EV_HTML, cgi_name='events', cgi_type=',..', diff --git a/dashlive/server/options/form_input_field.py b/dashlive/server/options/form_input_field.py index a008ffe94..7a7d9a1e8 100644 --- a/dashlive/server/options/form_input_field.py +++ b/dashlive/server/options/form_input_field.py @@ -1,10 +1,24 @@ -from typing import Any, NamedTuple, NotRequired, TypedDict +from dataclasses import dataclass +from typing import AbstractSet, Any, NamedTuple, NotRequired, TypedDict -class FieldOption(NamedTuple): +class FieldOptionJson(TypedDict): + title: str + value: str | int + selected: NotRequired[bool] + +@dataclass(slots=True) +class FieldOption: title: str value: str | int selected: bool + def toJSON(self, pure: bool = False, exclude: AbstractSet[str] | None = None) -> FieldOptionJson: + return { + 'title': self.title, + 'value': self.value, + 'selected': self.selected + } + class ColumnClassNames(NamedTuple): left: str middle: str @@ -17,6 +31,7 @@ class FormInputContext(TypedDict): disabled: NotRequired[bool] error: NotRequired[str] featured: NotRequired[bool] + fullName: str href: NotRequired[str] link_title: NotRequired[str] max: NotRequired[int] @@ -31,6 +46,7 @@ class FormInputContext(TypedDict): prefix: NotRequired[str] required: NotRequired[bool] rowClass: NotRequired[str] + shortName: str spellcheck: NotRequired[bool] step: NotRequired[int] title: NotRequired[str] diff --git a/dashlive/server/options/generate_types.py b/dashlive/server/options/generate_types.py new file mode 100644 index 000000000..01ff10cce --- /dev/null +++ b/dashlive/server/options/generate_types.py @@ -0,0 +1,314 @@ +############################################################################# +# +# Project Name : Simulated MPEG DASH service +# +# Author : Alex Ashley +# +############################################################################# + +from dataclasses import InitVar, dataclass, field +from operator import attrgetter +from pathlib import Path +import re +from typing import Any, ClassVar + +from dashlive.drm.system import DrmSystem + +from .all_options import ALL_OPTIONS +from .dash_option import DashOption + +@dataclass +class DashOptionTypeHint: + name: str + short: str + cgi: str + py_type: str + ts_type: str + default: InitVar[Any] = None + factory: InitVar[str | None] = None + py_default: str = field(init=False) + + def __post_init__(self, default: Any, factory: str | None) -> None: + py_default: str = '' + if factory is not None: + py_default = f' = field(default_factory={factory})' + elif default is not None: + if isinstance(default, list): + if len(default) > 0: + py_default = f' = field(default_factory=lambda: {default} )' + else: + py_default = ' = field(default_factory=list)' + elif isinstance(default, str): + py_default = f' = field(default="{default}")' + else: + py_default = f" = {default}" + else: + if self.py_type.startswith('list'): + py_default = ' = field(default_factory=list)' + elif 'None' in self.py_type: + py_default = ' = None' + self.py_default = py_default + + def to_python_hint(self, form: str) -> str: + """ + Generates the parameter name, type hint and its default value. + :form: one of "name", "short" or "cgi" + """ + assert form in {"name", "short", "cgi"} + name: str = getattr(self, form) + if form != "name" and "OptionsType" in self.py_type: + return f'{name}: {form.title()}{self.py_type}{self.py_default}' + + return f'{name}: {self.py_type}{self.py_default}' + + +class OptionsTypesGenerator: + @staticmethod + def python_type_from_value(value: Any) -> str: + if value is None: + return 'str | None' + if isinstance(value, bool): + return 'bool' + if isinstance(value, int): + return 'int' + if isinstance(value, float): + return 'float' + if isinstance(value, list): + if len(value) > 0: + return f"list[{OptionsTypesGenerator.guess_python_type(value[0])}]" + return 'list' + return 'Any' + + @staticmethod + def guess_python_type(opt: DashOption) -> str: + hint = opt.python_type_hint() + if hint is not None: + return hint + value: Any = '' + if opt.cgi_choices: + value = opt.cgi_choices[0] + if isinstance(value, tuple): + items: set[str] = set() + for it in value: + if isinstance(it, tuple): + items.add(OptionsTypesGenerator.python_type_from_value(it[1])) + else: + items.add(OptionsTypesGenerator.python_type_from_value(it)) + return f"{'| '.join(items)}" + if value is None: + value = 'none' + value = opt.from_string(value) + return OptionsTypesGenerator.python_type_from_value(value) + + @staticmethod + def typescript_type_from_value(value: Any) -> str: + if value is None: + return 'string | null | undefined' + if isinstance(value, bool): + return 'boolean' + if isinstance(value, (int, float)): + return 'number' + if isinstance(value, list): + if len(value) > 0: + return f"{OptionsTypesGenerator.guess_python_type(value[0])}[]" + return 'unknown[]' + return 'unknown' + + PYTHON_TO_TS_TYPES: ClassVar[dict[str, str]] = { + "int": "number", + "float": "number", + "bool": "boolean", + "str": "string", + "None": "null", + "datetime.datetime": "Date", + "tuple": "unknown", + } + PY_TO_TS_RE = re.compile(r'(\s*)([A-Za-z.]+)($|,|\s|\[)') # '|'.join([f"({t})" for t in PYTHON_TO_TS_TYPES.keys()])) + + @staticmethod + def python_to_ts(m: re.Match[str]) -> str: + try: + ts_type: str = OptionsTypesGenerator.PYTHON_TO_TS_TYPES[m.group(2)] + return f"{m.group(1)}{ts_type}{m.group(3)}" + except KeyError: + return m.group(0) + + @staticmethod + def translate_tuple(m: re.Match[str]) -> str: + fields: str = re.sub(OptionsTypesGenerator.PY_TO_TS_RE, OptionsTypesGenerator.python_to_ts, m.group(1)) + return f"[{fields}]" + + @staticmethod + def guess_typescript_type(opt: DashOption) -> str: + hint = opt.python_type_hint() + if hint == 'str' and opt.cgi_choices is not None: + hint = None + if hint is not None: + if hint.startswith('list['): + hint = f"{hint[5:-1]}[]" + hint = re.sub(r"tuple\[([^]]+)\]", OptionsTypesGenerator.translate_tuple, hint) + hint = re.sub(OptionsTypesGenerator.PY_TO_TS_RE, OptionsTypesGenerator.python_to_ts, hint) + return hint + value: Any = '' + if opt.cgi_choices: + options: list[str | None] = [] + for ch in opt.cgi_choices: + if ch is None: + options.append(None) + elif isinstance(ch, tuple): + options.append(ch[1]) + else: + options.append(ch) + if None in options or len(options) < 2: + return 'string' + return ' | '.join([f'"{o}"' for o in options]) + if value is None: + value = 'none' + value = opt.from_string(value) + return OptionsTypesGenerator.typescript_type_from_value(value) + + @classmethod + def guess_options_container_types(cls) -> tuple[list[DashOptionTypeHint], dict[str, list[DashOptionTypeHint]]]: + primary_options: list[DashOptionTypeHint] = [] + sub_options: dict[str, list[DashOptionTypeHint]] = {} + for opt in ALL_OPTIONS: + if opt.prefix: + name: str = f"{opt.prefix.title()}OptionsType" + sub: list[DashOptionTypeHint] + try: + sub = sub_options[name] + except KeyError: + sub = [] + sub_options[name] = sub + primary_options.append(DashOptionTypeHint( + name=opt.prefix, + short=opt.prefix, + cgi=opt.prefix, + default=None, + factory=name, + py_type=name, + ts_type=name)) + sub.append(DashOptionTypeHint( + name=opt.full_name, + short=opt.short_name, + cgi=opt.cgi_name, + default=opt.default_value(), + py_type=cls.guess_python_type(opt), + ts_type=cls.guess_typescript_type(opt))) + else: + primary_options.append(DashOptionTypeHint( + name=opt.full_name, + short=opt.short_name, + cgi=opt.cgi_name, + default=opt.default_value(), + py_type=cls.guess_python_type(opt), + ts_type=cls.guess_typescript_type(opt))) + + primary_options.sort(key=attrgetter('name')) + for key in sub_options.keys(): + sub_options[key].sort(key=attrgetter('name')) + return (primary_options, sub_options,) + + @classmethod + def create_options_container_types_files(cls, py_dest: Path, ts_dest: Path) -> None: + primary_options, sub_options = cls.guess_options_container_types() + cls.create_python_types_file(py_dest, primary_options, sub_options) + cls.create_typescript_types_file(ts_dest, primary_options, sub_options) + + @classmethod + def create_python_types_file(cls, + py_dest: Path, + primary_options: list[DashOptionTypeHint], + sub_options: dict[str, list[DashOptionTypeHint]]) -> None: + print(f"Creating {py_dest}") + with py_dest.open('wt', encoding='utf-8') as dest: + dest.write('# this file is auto-generated, do not edit!\n') + dest.write('# to re-generate this file, use the command:\n') + dest.write('# uv run -m dashlive.server.options.generate_types\n\n') + dest.write('import datetime\n') + dest.write('from dataclasses import dataclass, field\n') + dest.write('from .options_group import OptionsGroup\n\n') + for name, options in sub_options.items(): + dest.write(f'@dataclass\nclass {name}(OptionsGroup):\n') + for opt in options: + dest.write(f' {opt.to_python_hint("name")}\n') + dest.write('\n\n') + dest.write('SUB_OPTION_PREFIX_MAP: dict[str, type] = {\n') + for name in sub_options.keys(): + for opt in primary_options: + if opt.py_type == name: + dest.write(f' "{opt.name}": {name},\n') + dest.write('}\n\n') + dest.write('@dataclass\nclass OptionsContainerType(OptionsGroup):\n') + for opt in primary_options: + dest.write(f' {opt.to_python_hint("name")}\n') + dest.write('\n\n@dataclass\nclass ShortOptionsContainerType(OptionsGroup):\n') + for opt in primary_options: + if "OptionsType" in opt.py_type: + for name, options in sub_options.items(): + if name not in opt.py_type: + continue + for short_opt in options: + dest.write(f' {short_opt.to_python_hint("short")}\n') + else: + dest.write(f' {opt.to_python_hint("short")}\n') + dest.write('\n\n@dataclass\nclass CgiOptionsContainerType(OptionsGroup):\n') + for opt in primary_options: + if "OptionsType" in opt.py_type: + for name, options in sub_options.items(): + if name not in opt.py_type: + continue + for cgi_opt in options: + dest.write(f' {cgi_opt.to_python_hint("cgi")}\n') + else: + dest.write(f' {opt.to_python_hint("cgi")}\n') + + @classmethod + def create_typescript_types_file(cls, + ts_dest: Path, + primary_options: list[DashOptionTypeHint], + sub_options: dict[str, list[DashOptionTypeHint]]) -> None: + print(f"Creating {ts_dest}") + with ts_dest.open('wt', encoding='utf-8') as dest: + dest.write('// this file is auto-generated, do not edit!\n\n') + dest.write('// to re-generate this file, use the command:\n') + dest.write('// uv run -m dashlive.server.options.generate_types\n\n') + drm_names: list[str] = [f'"{d.lower()}"' for d in DrmSystem.keys()] + dest.write(f'export type DrmSystemType = {" | ".join(drm_names)};\n\n') + for name, options in sub_options.items(): + dest.write(f'export type {name} = {{\n') + for opt in options: + dest.write(f' {opt.name}: {opt.ts_type};\n') + dest.write('}\n\n') + dest.write('export type OptionsContainerType = {\n') + for opt in primary_options: + dest.write(f' {opt.name}: {opt.ts_type};\n') + dest.write('}\n\n') + dest.write('export type ShortOptionsContainerType = {\n') + for opt in primary_options: + if "OptionsType" in opt.ts_type: + for name, options in sub_options.items(): + if name not in opt.ts_type: + continue + for short_opt in options: + dest.write(f' {short_opt.short}: {short_opt.ts_type};\n') + else: + dest.write(f' {opt.short}: {opt.ts_type};\n') + dest.write('}\n\n') + dest.write('export type CgiOptionsContainerType = {\n') + for opt in primary_options: + if "OptionsType" in opt.ts_type: + for name, options in sub_options.items(): + if name not in opt.ts_type: + continue + for cgi_opt in options: + dest.write(f' {cgi_opt.cgi}: {cgi_opt.ts_type};\n') + else: + dest.write(f' {opt.cgi}: {opt.ts_type};\n') + dest.write('}\n') + + +if __name__ == "__main__": + py_dest: Path = Path(__file__).parent / "options_types.py" + ts_dest: Path = Path(__file__).parent.parent.parent.parent / "frontend" / "@types" / "@dashlive" / "dash-options.d.ts" + OptionsTypesGenerator.create_options_container_types_files(py_dest, ts_dest) diff --git a/dashlive/server/options/http_error.py b/dashlive/server/options/http_error.py index 520c6d996..5f616d29c 100644 --- a/dashlive/server/options/http_error.py +++ b/dashlive/server/options/http_error.py @@ -1,19 +1,5 @@ ############################################################################# # -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# -############################################################################# -# # Project Name : Simulated MPEG DASH service # # Author : Alex Ashley @@ -22,44 +8,54 @@ from dashlive.utils.date_time import from_isodatetime -from .dash_option import DashOption +from .dash_option import DashOption, IntOrNoneDashOption from .types import OptionUsage -def _errors_from_string(value: str) -> list[tuple[int, str]]: - if value.lower() in ['', 'none']: - return [] - items: list[tuple] = [] - for val in value.split(','): - code, pos = val.split('=') - try: - pos = int(pos, 10) - except ValueError: - pos = from_isodatetime(pos) - items.append((int(code, 10), pos)) - return items +class HttpErrorOption(DashOption[list[tuple[int, str]]]): + def __init__(self, use: str, description: str) -> None: + prefix: str = use[0] + super().__init__( + usage=OptionUsage.from_string(use), + short_name=f'{prefix}he', + full_name=f'{use}Errors', + title=f'{description} HTTP errors', + description=f'Cause an HTTP error to be generated when requesting {description}', + cgi_name=f'{prefix}err', + cgi_type='=,..' + ) + + def from_string(self, value: str) -> list[tuple[int, str]]: + if value.lower() in ['', 'none']: + return [] + items: list[tuple] = [] + for val in value.split(','): + code, pos = val.split('=') + try: + pos = int(pos, 10) + except ValueError: + pos = from_isodatetime(pos) + items.append((int(code, 10), pos)) + return items + + def to_string(self, value: list[tuple[int, str]]) -> str: + return str(value) -def http_error_factory(use: str, description: str): - prefix = use[0] - return DashOption( - usage=OptionUsage.from_string(use), - short_name=f'{prefix}he', - full_name=f'{use}Errors', - title=f'{description} HTTP errors', - description=f'Cause an HTTP error to be generated when requesting {description}', - from_string=_errors_from_string, - cgi_name=f'{prefix}err', - cgi_type='=,..') + def python_type_hint(self) -> str | None: + return 'list[tuple[int, str]]' + + def default_value(self) -> list[tuple[int, str]] | None: + return [] -ManifestHttpError = http_error_factory('manifest', 'Manifest') +ManifestHttpError = HttpErrorOption('manifest', 'Manifest') -VideoHttpError = http_error_factory('video', 'Video fragments') +VideoHttpError = HttpErrorOption('video', 'Video fragments') -AudioHttpError = http_error_factory('audio', 'Audio fragments') +AudioHttpError = HttpErrorOption('audio', 'Audio fragments') -TextHttpError = http_error_factory('text', 'Text fragments') +TextHttpError = HttpErrorOption('text', 'Text fragments') -FailureCount = DashOption( +FailureCount = IntOrNoneDashOption( usage=(OptionUsage.MANIFEST | OptionUsage.AUDIO | OptionUsage.VIDEO | OptionUsage.TEXT), short_name='hfc', full_name='failureCount', @@ -68,6 +64,5 @@ def http_error_factory(use: str, description: str): 'Number of times to respond with a 5xx error before ' + 'accepting the request. Only relevant in combination ' + 'with one of the error injection parameters (e.g. v503, m503).'), - from_string=DashOption.int_or_none_from_string, cgi_name='failures', cgi_type='') diff --git a/dashlive/server/options/manifest_options.py b/dashlive/server/options/manifest_options.py index 7733aa633..d667b1088 100644 --- a/dashlive/server/options/manifest_options.py +++ b/dashlive/server/options/manifest_options.py @@ -6,21 +6,26 @@ # ############################################################################# import datetime -import logging +from typing import ClassVar, cast from dashlive.utils.date_time import from_isodatetime, to_iso_datetime -from .dash_option import DashOption +from .dash_option import ( + BoolDashOption, + CgiChoiceType, + DashOption, + IntOrNoneDashOption, + StringDashOption, + StringListDashOption, +) from .http_error import FailureCount, ManifestHttpError from .types import OptionUsage -AbrControl = DashOption( +AbrControl = BoolDashOption( usage=OptionUsage.MANIFEST, short_name='ab', full_name='abr', title='Adaptive bitrate', description='Enable or disable adaptive bitrate', - from_string=DashOption.bool_from_string, - to_string=DashOption.bool_to_string, cgi_name='abr', cgi_choices=( ('Enabled', '1'), @@ -29,7 +34,8 @@ input_type='checkbox', featured=True) -AST_HTML = ''' +class AvailabilityStartTimeDashOption(DashOption[datetime.datetime | str]): + AST_HTML: ClassVar[str] = '''

Specify availabilityStartTime as "today", "now", "year", "month", "epoch" or an ISO datetime (YYYY-MM-DDTHH:MM:SSZ). @@ -41,48 +47,48 @@

''' -SPECIAL_AST_VALUES = {'now', 'today', 'month', 'year', 'epoch'} - -def ast_from_string(value: str) -> datetime.datetime | str: - if value in SPECIAL_AST_VALUES: - return value - try: - value = from_isodatetime(value) - except ValueError as err: - logging.warning('Failed to parse availabilityStartTime: %s', err) - raise err - return value - -def ast_to_string(value: datetime.datetime | str | None) -> str: - if value in SPECIAL_AST_VALUES: - return value - if value is None: - return '' - return to_iso_datetime(value) - - -AvailabilityStartTime = DashOption( - usage=OptionUsage.MANIFEST + OptionUsage.VIDEO + OptionUsage.AUDIO + OptionUsage.TEXT, - short_name='ast', - full_name='availabilityStartTime', - title='Availability start time', - description='Sets availabilityStartTime for live streams', - from_string=ast_from_string, - to_string=ast_to_string, - cgi_name='start', - cgi_type='(today|month|year|epoch|now|)', - cgi_choices=('year', 'today', 'month', 'epoch', 'now'), - html=AST_HTML, - input_type='textList') - -UseBaseUrl = DashOption( + SPECIAL_AST_VALUES: ClassVar[set[str]] = {'now', 'today', 'month', 'year', 'epoch'} + + def __init__(self) -> None: + super().__init__( + usage=OptionUsage.MANIFEST + OptionUsage.VIDEO + OptionUsage.AUDIO + OptionUsage.TEXT, + short_name='ast', + full_name='availabilityStartTime', + title='Availability start time', + description='Sets availabilityStartTime for live streams', + cgi_name='start', + cgi_type='(today|month|year|epoch|now|)', + cgi_choices=('year', 'today', 'month', 'epoch', 'now'), + html=AvailabilityStartTimeDashOption.AST_HTML, + input_type='textList' + ) + + def from_string(self, value: str) -> datetime.datetime | str: + if value == '' and self.default is not None: + return self.default + if value in AvailabilityStartTimeDashOption.SPECIAL_AST_VALUES: + return value + return cast(datetime.datetime, from_isodatetime(value)) + + def to_string(self, value: datetime.datetime | str | None) -> str: + if value is None: + return '' + if value in AvailabilityStartTimeDashOption.SPECIAL_AST_VALUES: + return cast(str, value) + return to_iso_datetime(value) + + def python_type_hint(self) -> str: + return 'datetime.datetime | str' + + +AvailabilityStartTime = AvailabilityStartTimeDashOption() + +UseBaseUrl = BoolDashOption( usage=OptionUsage.MANIFEST, short_name='base', full_name='useBaseUrls', title='Use BaseURLs', description='Include a BaseURL element?', - from_string=DashOption.bool_from_string, - to_string=DashOption.bool_to_string, input_type='checkbox', cgi_name='base', cgi_choices=( @@ -90,29 +96,26 @@ def ast_to_string(value: datetime.datetime | str | None) -> str: ('No', '0') )) -Bugs = DashOption( +Bugs = StringListDashOption( usage=OptionUsage.MANIFEST + OptionUsage.VIDEO + OptionUsage.AUDIO + OptionUsage.TEXT, short_name='bug', full_name='bugCompatibility', title='Bug compatibility', description='Produce a stream with known bugs', - from_string=DashOption.list_without_none_from_string, - to_string=lambda bugs: ','.join(bugs), cgi_name='bugs', cgi_choices=(None, 'saio')) -Leeway = DashOption( +Leeway = IntOrNoneDashOption( usage=OptionUsage.MANIFEST + OptionUsage.VIDEO + OptionUsage.AUDIO + OptionUsage.TEXT, short_name='lee', full_name='leeway', title='Fragment expiration leeway', description='Number of seconds after a fragment has expired before it becomes unavailable', - from_string=DashOption.int_or_none_from_string, cgi_name='leeway', input_type='numberList', cgi_choices=('16', '60', '0')) -OperatingMode = DashOption( +OperatingMode = StringDashOption( usage=OptionUsage.MANIFEST, short_name='md', full_name='mode', @@ -126,16 +129,15 @@ def ast_to_string(value: datetime.datetime | str | None) -> str: ), featured=True) -MinimumUpdatePeriod = DashOption( +MinimumUpdatePeriod = IntOrNoneDashOption( usage=OptionUsage.MANIFEST, short_name='mup', full_name='minimumUpdatePeriod', title='Minimum update period', description='Specify minimumUpdatePeriod (in seconds) or -1 to disable updates', - from_string=DashOption.int_or_none_from_string, cgi_name='mup', cgi_choices=( - ('Every 2 fragments', None), + cast(CgiChoiceType, ('Every 2 fragments', None)), ('Never', '-1'), ('Every fragment', '4'), ('Every 30 seconds', '30'), @@ -143,14 +145,12 @@ def ast_to_string(value: datetime.datetime | str | None) -> str: cgi_type='', input_type='numberList') -SegmentTimeline = DashOption( +SegmentTimeline = BoolDashOption( usage=OptionUsage.MANIFEST, short_name='st', full_name='segmentTimeline', title='Segment timeline', description='Enable or disable segment timeline', - from_string=DashOption.bool_from_string, - to_string=DashOption.bool_to_string, input_type='checkbox', cgi_name='timeline', cgi_choices=( @@ -158,36 +158,32 @@ def ast_to_string(value: datetime.datetime | str | None) -> str: ('Yes (use $Time$)', '1')), featured=True) -TimeshiftBufferDepth = DashOption( +TimeshiftBufferDepth = IntOrNoneDashOption( usage=OptionUsage.MANIFEST + OptionUsage.VIDEO + OptionUsage.AUDIO + OptionUsage.TEXT, short_name='tbd', full_name='timeShiftBufferDepth', title='timeShiftBufferDepth size', description='Number of seconds for timeShiftBufferDepth', - from_string=DashOption.int_or_none_from_string, cgi_name='depth', cgi_type='', cgi_choices=('1800', '30'), input_type='numberList') -UpdateCount = DashOption( +UpdateCount = IntOrNoneDashOption( usage=OptionUsage.MANIFEST, short_name='uc', full_name='updateCount', title='Manifest update count', description='Counter of manifest reloads', - from_string=DashOption.int_or_none_from_string, cgi_name='update', cgi_type='') -UsePatches = DashOption( +UsePatches = BoolDashOption( usage=OptionUsage.MANIFEST, short_name='patch', full_name='patch', title='Use MPD patches', description='Use MPD patches for live streams', - from_string=DashOption.bool_from_string, - to_string=DashOption.bool_to_string, input_type='checkbox', cgi_name='patch', cgi_choices=( @@ -195,11 +191,25 @@ def ast_to_string(value: datetime.datetime | str | None) -> str: ('Yes', '1'), )) +ForcePeriodDurations = BoolDashOption( + usage=OptionUsage.MANIFEST, + short_name='fpd', + full_name='forcePeriodDurations', + title='Forced Period durations', + description='Always add a duration attribute to Period elements', + input_type='checkbox', + cgi_name='periodDur', + cgi_choices=( + ('No', '0'), + ('Yes', '1'), + )) + manifest_options = [ AbrControl, AvailabilityStartTime, Bugs, FailureCount, + ForcePeriodDurations, Leeway, ManifestHttpError, MinimumUpdatePeriod, diff --git a/dashlive/server/options/name_maps.py b/dashlive/server/options/name_maps.py new file mode 100644 index 000000000..defdc2fd9 --- /dev/null +++ b/dashlive/server/options/name_maps.py @@ -0,0 +1,60 @@ +############################################################################# +# +# Project Name : Simulated MPEG DASH service +# +# Author : Alex Ashley +# +############################################################################# +from typing import ClassVar + +from .all_options import ALL_OPTIONS +from .dash_option import DashOption + +class DashOptionNameMaps: + _cgi_map: ClassVar[dict[str, DashOption] | None] = None + _short_name_map: ClassVar[dict[str, DashOption] | None] = None + _param_map: ClassVar[dict[str, DashOption] | None] = None + + @classmethod + def get_parameter_map(cls) -> dict[str, DashOption]: + """ + Returns a dictionary that maps from the full parameter name + to its DashOption entry + """ + if cls._param_map is not None: + return cls._param_map + cls._param_map = {} + for opt in ALL_OPTIONS: + if opt.prefix: + cls._param_map[f'{opt.prefix}.{opt.short_name}'] = opt + name = f'{opt.prefix}.{opt.full_name}' + else: + name = opt.full_name + cls._param_map[name] = opt + return cls._param_map + + @classmethod + def get_cgi_map(cls) -> dict[str, DashOption]: + """ + Returns a dictionary that maps from the CGI parameter used in a + URL to its DashOption entry + """ + if cls._cgi_map is not None: + return cls._cgi_map + cls._cgi_map = {} + for opt in ALL_OPTIONS: + cls._cgi_map[opt.cgi_name] = opt + return cls._cgi_map + + @classmethod + def get_short_param_map(cls) -> dict[str, DashOption]: + """ + Returns a dictionary that maps from the short parameter used in a + stream defaults + """ + if cls._short_name_map is not None: + return cls._short_name_map + cls._short_name_map = {} + for opt in ALL_OPTIONS: + cls._short_name_map[opt.short_name] = opt + return cls._short_name_map diff --git a/dashlive/server/options/options_group.py b/dashlive/server/options/options_group.py new file mode 100644 index 000000000..1be1cf4ef --- /dev/null +++ b/dashlive/server/options/options_group.py @@ -0,0 +1,20 @@ +import dataclasses +from typing import AbstractSet + +from dashlive.utils.json_object import JsonObject + +class OptionsGroup: + @classmethod + def classname(cls) -> str: + if cls.__module__.startswith('__'): + return cls.__name__ + return cls.__module__ + '.' + cls.__name__ + + def toJSON(self, exclude: AbstractSet[str] | None = None) -> JsonObject: + rv: JsonObject = dataclasses.asdict(self) + if exclude is None: + exclude = set() + for k in list(rv.keys()): + if k in exclude or k[0] == '_': + del rv[k] + return rv diff --git a/dashlive/server/options/options_types.py b/dashlive/server/options/options_types.py new file mode 100644 index 000000000..43fc62bf5 --- /dev/null +++ b/dashlive/server/options/options_types.py @@ -0,0 +1,222 @@ +# this file is auto-generated, do not edit! +# to re-generate this file, use the command: +# uv run -m dashlive.server.options.generate_types + +import datetime +from dataclasses import dataclass, field +from .options_group import OptionsGroup + +@dataclass +class ClearkeyOptionsType(OptionsGroup): + licenseUrl: str | None = None + + +@dataclass +class MarlinOptionsType(OptionsGroup): + licenseUrl: str | None = None + + +@dataclass +class PlayreadyOptionsType(OptionsGroup): + licenseUrl: str | None = None + piff: bool = True + version: float | None = None + + +@dataclass +class PingOptionsType(OptionsGroup): + count: int = 0 + duration: int = 200 + inband: bool = True + interval: int = 1000 + start: int = 0 + timescale: int = 100 + value: str = field(default="0") + version: int = 0 + + +@dataclass +class Scte35OptionsType(OptionsGroup): + count: int = 0 + duration: int = 200 + inband: bool = True + interval: int = 1000 + program_id: int = 1620 + start: int = 0 + timescale: int = 100 + value: str | None = None + version: int = 0 + + +SUB_OPTION_PREFIX_MAP: dict[str, type] = { + "clearkey": ClearkeyOptionsType, + "marlin": MarlinOptionsType, + "playready": PlayreadyOptionsType, + "ping": PingOptionsType, + "scte35": Scte35OptionsType, +} + +@dataclass +class OptionsContainerType(OptionsGroup): + abr: bool = True + audioCodec: str = field(default="mp4a") + audioDescription: str | None = None + audioErrors: list[tuple[int, str]] = field(default_factory=list) + availabilityStartTime: datetime.datetime | str = field(default="year") + bugCompatibility: list[str] = field(default_factory=list) + clearkey: ClearkeyOptionsType = field(default_factory=ClearkeyOptionsType) + clockDrift: int | None = None + dashjsVersion: str | None = None + drmSelection: list[tuple] = field(default_factory=list) + eventTypes: list[str] = field(default_factory=list) + failureCount: int | None = None + forcePeriodDurations: bool = False + leeway: int = 16 + mainAudio: str | None = None + mainText: str | None = None + manifestErrors: list[tuple[int, str]] = field(default_factory=list) + marlin: MarlinOptionsType = field(default_factory=MarlinOptionsType) + minimumUpdatePeriod: int | None = None + mode: str = field(default="vod") + ntpSources: list[str] = field(default_factory=list) + patch: bool = False + ping: PingOptionsType = field(default_factory=PingOptionsType) + playready: PlayreadyOptionsType = field(default_factory=PlayreadyOptionsType) + scte35: Scte35OptionsType = field(default_factory=Scte35OptionsType) + segmentTimeline: bool = False + shakaVersion: str | None = None + textCodec: str | None = None + textErrors: list[tuple[int, str]] = field(default_factory=list) + textLanguage: str | None = None + textPreference: str | None = None + timeShiftBufferDepth: int = 1800 + updateCount: int | None = None + useBaseUrls: bool = True + utcMethod: str | None = None + utcValue: str | None = None + videoCorruption: list[str] = field(default_factory=list) + videoCorruptionFrameCount: int | None = None + videoErrors: list[tuple[int, str]] = field(default_factory=list) + videoPlayer: str | None = field(default="native") + + +@dataclass +class ShortOptionsContainerType(OptionsGroup): + ab: bool = True + ac: str = field(default="mp4a") + ad: str | None = None + ahe: list[tuple[int, str]] = field(default_factory=list) + ast: datetime.datetime | str = field(default="year") + bug: list[str] = field(default_factory=list) + clu: str | None = None + dft: int | None = None + djVer: str | None = None + drm: list[tuple] = field(default_factory=list) + evs: list[str] = field(default_factory=list) + hfc: int | None = None + fpd: bool = False + lee: int = 16 + ma: str | None = None + mt: str | None = None + mhe: list[tuple[int, str]] = field(default_factory=list) + mlu: str | None = None + mup: int | None = None + md: str = field(default="vod") + ntps: list[str] = field(default_factory=list) + patch: bool = False + pinCoun: int = 0 + pinDura: int = 200 + pinInba: bool = True + pinInte: int = 1000 + pinStar: int = 0 + pinTime: int = 100 + pinValu: str = field(default="0") + pinVers: int = 0 + plu: str | None = None + pff: bool = True + pvn: float | None = None + sctCoun: int = 0 + sctDura: int = 200 + sctInba: bool = True + sctInte: int = 1000 + sctProg: int = 1620 + sctStar: int = 0 + sctTime: int = 100 + sctValu: str | None = None + sctVers: int = 0 + st: bool = False + skVer: str | None = None + tc: str | None = None + the: list[tuple[int, str]] = field(default_factory=list) + tl: str | None = None + ptxLang: str | None = None + tbd: int = 1800 + uc: int | None = None + base: bool = True + utc: str | None = None + utv: str | None = None + vcor: list[str] = field(default_factory=list) + vcfc: int | None = None + vhe: list[tuple[int, str]] = field(default_factory=list) + vp: str | None = field(default="native") + + +@dataclass +class CgiOptionsContainerType(OptionsGroup): + abr: bool = True + acodec: str = field(default="mp4a") + ad_audio: str | None = None + aerr: list[tuple[int, str]] = field(default_factory=list) + start: datetime.datetime | str = field(default="year") + bugs: list[str] = field(default_factory=list) + clearkey__la_url: str | None = None + drift: int | None = None + dashjs: str | None = None + drm: list[tuple] = field(default_factory=list) + events: list[str] = field(default_factory=list) + failures: int | None = None + periodDur: bool = False + leeway: int = 16 + main_audio: str | None = None + main_text: str | None = None + merr: list[tuple[int, str]] = field(default_factory=list) + marlin__la_url: str | None = None + mup: int | None = None + mode: str = field(default="vod") + ntp_servers: list[str] = field(default_factory=list) + patch: bool = False + ping__count: int = 0 + ping__duration: int = 200 + ping__inband: bool = True + ping__interval: int = 1000 + ping__start: int = 0 + ping__timescale: int = 100 + ping__value: str = field(default="0") + ping__version: int = 0 + playready__la_url: str | None = None + playready__piff: bool = True + playready__version: float | None = None + scte35__count: int = 0 + scte35__duration: int = 200 + scte35__inband: bool = True + scte35__interval: int = 1000 + scte35__program_id: int = 1620 + scte35__start: int = 0 + scte35__timescale: int = 100 + scte35__value: str | None = None + scte35__version: int = 0 + timeline: bool = False + shaka: str | None = None + tcodec: str | None = None + terr: list[tuple[int, str]] = field(default_factory=list) + tlang: str | None = None + text_pref: str | None = None + depth: int = 1800 + update: int | None = None + base: bool = True + time: str | None = None + time_value: str | None = None + vcorrupt: list[str] = field(default_factory=list) + frames: int | None = None + verr: list[tuple[int, str]] = field(default_factory=list) + player: str | None = field(default="native") diff --git a/dashlive/server/options/player_options.py b/dashlive/server/options/player_options.py index 46d8cccf9..6e3188513 100644 --- a/dashlive/server/options/player_options.py +++ b/dashlive/server/options/player_options.py @@ -1,7 +1,7 @@ -from .dash_option import DashOption +from .dash_option import DashOption, StringOrNoneDashOption from .types import OptionUsage -DashjsVersion = DashOption( +DashjsVersion = StringOrNoneDashOption( usage=OptionUsage.HTML, short_name='djVer', full_name='dashjsVersion', @@ -20,7 +20,7 @@ ''' -NativePlayback = DashOption( +NativePlayback = StringOrNoneDashOption( usage=OptionUsage.HTML, short_name='vp', full_name='videoPlayer', @@ -36,7 +36,7 @@ input_type='select', featured=True) -ShakaVersion = DashOption( +ShakaVersion = StringOrNoneDashOption( usage=OptionUsage.HTML, short_name='skVer', full_name='shakaVersion', @@ -46,7 +46,7 @@ cgi_choices=(None, '4.11.2', '4.3.8', '2.5.4',), input_type='datalist') -TextLanguage = DashOption( +TextLanguage = StringOrNoneDashOption( usage=OptionUsage.HTML, featured=True, short_name='ptxLang', diff --git a/dashlive/server/options/repository.py b/dashlive/server/options/repository.py index a3ce43c5c..8672982f5 100644 --- a/dashlive/server/options/repository.py +++ b/dashlive/server/options/repository.py @@ -1,55 +1,20 @@ ############################################################################# # -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# -############################################################################# -# # Project Name : Simulated MPEG DASH service # # Author : Alex Ashley # ############################################################################# -import logging -from typing import AbstractSet, ClassVar +from typing import AbstractSet + +from dashlive.server.options.all_options import ALL_OPTIONS -from .audio_options import audio_options from .container import OptionsContainer from .dash_option import CgiOption, DashOption -from .drm_options import drm_options -from .event_options import event_options -from .manifest_options import manifest_options -from .player_options import player_options -from .text_options import text_options from .types import OptionUsage -from .video_options import video_options -from .utc_time_options import time_options class OptionsRepository: - _cgi_map: ClassVar[dict[str, DashOption] | None] = None - _short_name_map: ClassVar[dict[str, DashOption] | None] = None - _param_map: ClassVar[dict[str, DashOption] | None] = None - _global_default_options: ClassVar[OptionsContainer | None] = None - _all_options: ClassVar[list[DashOption]] = ( - audio_options + - drm_options + - event_options + - manifest_options + - player_options + - video_options + - text_options + - time_options) - @classmethod def get_dash_options(cls, use: OptionUsage | None = None, @@ -59,12 +24,12 @@ def get_dash_options(cls, Returns a list of all the options applicable to "use", or all options if use is None. """ - result = [] + result: list[DashOption] = [] if use is None: use = 0xFF if exclude is None: exclude = set() - for opt in cls._all_options: + for opt in ALL_OPTIONS: if (opt.usage & use) == 0: continue if only is not None and opt.full_name not in only: @@ -113,137 +78,19 @@ def matches(item: DashOption) -> bool: return result @classmethod - def get_cgi_map(cls) -> dict[str, str]: - """ - Returns a dictionary that maps from the CGI parameter used in a - URL to its DashOption entry - """ - if cls._cgi_map is not None: - return cls._cgi_map - cls._cgi_map = {} - for opt in cls.get_dash_options(): - cls._cgi_map[opt.cgi_name] = opt - return cls._cgi_map - - @classmethod - def get_short_param_map(cls) -> dict[str, str]: - """ - Returns a dictionary that maps from the short parameter used in a - stream defaults - """ - if cls._short_name_map is not None: - return cls._short_name_map - cls._short_name_map = {} - for opt in cls.get_dash_options(): - cls._short_name_map[opt.short_name] = opt - return cls._short_name_map - - @classmethod - def get_parameter_map(cls) -> dict[str, str]: - """ - Returns a dictionary that maps from the full parameter name - to its DashOption entry - """ - if cls._param_map is not None: - return cls._param_map - cls._param_map = {} - for opt in cls.get_dash_options(): - if opt.prefix: - cls._param_map[f'{opt.prefix}.{opt.short_name}'] = opt - name = f'{opt.prefix}.{opt.full_name}' - else: - name = opt.full_name - cls._param_map[name] = opt - return cls._param_map - - @classmethod - def get_default_options(cls, use: OptionUsage | None = None) -> OptionsContainer: - """ - Returns a dictionary containing the global defaults for every option - """ - if cls._global_default_options is not None: - return cls._global_default_options - result = OptionsContainer(cls.get_parameter_map(), None) - cls._global_default_options = result - for opt in cls.get_dash_options(): - if opt.cgi_choices: - value = opt.cgi_choices[0] - if isinstance(value, tuple): - value = value[1] - else: - value = '' - if value is None: - value = 'none' - value = opt.from_string(value) - if opt.prefix: - try: - dest = result[opt.prefix] - except KeyError: - dest = OptionsContainer(cls.get_parameter_map(), None) - result.add_field(opt.prefix, dest) - else: - dest = result - dest.add_field(opt.full_name, value) - return result - - @classmethod - def convert_cgi_options(cls, params: dict[str, str], - defaults: OptionsContainer | None = None) -> OptionsContainer: + def convert_cgi_options(cls, params: dict[str, str]) -> OptionsContainer: """ Convert a dictionary of CGI parameters to an OptionsContainer object """ - return cls.convert_options(params, True, defaults=defaults) + result = OptionsContainer() + result.apply_options(params, True) + return result @classmethod - def convert_short_name_options( - cls, params: dict[str, str], - defaults: OptionsContainer | None = None) -> OptionsContainer: + def convert_short_name_options(cls, params: dict[str, str]) -> OptionsContainer: """ Convert a dictionary of CGI parameters to an OptionsContainer object """ - return cls.convert_options(params, False, defaults=defaults) - - @classmethod - def convert_options(cls, - params: dict[str, str], - is_cgi: bool, - defaults: OptionsContainer | None = None) -> OptionsContainer: - result: OptionsContainer - if defaults is not None: - result = defaults.clone( - parameter_map=cls.get_parameter_map(), - defaults=defaults) - else: - result = OptionsContainer(cls.get_parameter_map(), defaults) - if is_cgi: - param_map: dict[str, DashOption] = cls.get_cgi_map() - else: - param_map = cls.get_short_param_map() - for key, value in params.items(): - try: - opt = param_map[key] - value = opt.from_string(value) - if opt.prefix: - try: - dest = result[opt.prefix] - except KeyError: - dflt = None - if defaults is not None: - dflt = getattr(defaults, opt.prefix) - dest = OptionsContainer( - cls.get_parameter_map(), dflt) - result.add_field(opt.prefix, dest) - dest.add_field(opt.full_name, value) - else: - result.add_field(opt.full_name, value) - except KeyError: - logging.warning(f'Invalid parameter {key} cgi={is_cgi}') - continue + result = OptionsContainer() + result.apply_options(params, False) return result - - -# manually set OptionsContainer.OBJECT_FIELDS to avoid a circular -# import reference -for opt in OptionsRepository._all_options: - if opt.prefix and opt.prefix not in OptionsContainer.OBJECT_FIELDS: - OptionsContainer.OBJECT_FIELDS[opt.prefix] = OptionsContainer diff --git a/dashlive/server/options/text_options.py b/dashlive/server/options/text_options.py index 33261a62a..19125d5af 100644 --- a/dashlive/server/options/text_options.py +++ b/dashlive/server/options/text_options.py @@ -6,39 +6,37 @@ # ############################################################################# -from .dash_option import DashOption +from typing import cast +from .dash_option import CgiChoiceType, StringOrNoneDashOption from .http_error import TextHttpError from .types import OptionUsage -TextCodec = DashOption( +TextCodec = StringOrNoneDashOption( usage=OptionUsage.MANIFEST, short_name='tc', full_name='textCodec', title='Text Codec', description='Filter text adaptation sets by text codec', - from_string=DashOption.string_or_none, cgi_name='tcodec', cgi_choices=( - ('Any codec', None), + cast(CgiChoiceType, ('Any codec', None)), ('im1t codec', 'im1t|etd1'), )) -TextLanguage = DashOption( +TextLanguage = StringOrNoneDashOption( usage=OptionUsage.MANIFEST, short_name='tl', full_name='textLanguage', title='Text Language', description='Filter text adaptation sets by language', - from_string=DashOption.string_or_none, cgi_name='tlang') -MainTextTrack = DashOption( +MainTextTrack = StringOrNoneDashOption( usage=OptionUsage.MANIFEST, short_name='mt', full_name='mainText', title='Main text track', description='Select text AdaptationSet that will be given the "main" role', - from_string=DashOption.string_or_none, cgi_name='main_text', cgi_type='()', input_type='text_representation') diff --git a/dashlive/server/options/utc_time_options.py b/dashlive/server/options/utc_time_options.py index 4d4366203..6ea043468 100644 --- a/dashlive/server/options/utc_time_options.py +++ b/dashlive/server/options/utc_time_options.py @@ -6,40 +6,37 @@ # ############################################################################# -from .dash_option import DashOption +from .dash_option import IntOrNoneDashOption, StringListDashOption, StringOrNoneDashOption from .types import OptionUsage -ClockDrift = DashOption( +ClockDrift = IntOrNoneDashOption( usage=(OptionUsage.MANIFEST | OptionUsage.TIME), short_name='dft', full_name='clockDrift', title='Clock drift', description='Number of seconds of delay to add to wall clock time', - from_string=DashOption.int_or_none_from_string, input_type='number', cgi_name='drift', cgi_type='', cgi_choices=(None, '10')) -UTCMethod = DashOption( +UTCMethod = StringOrNoneDashOption( usage=OptionUsage.MANIFEST, short_name='utc', full_name='utcMethod', title='UTC timing method', description='Select UTCTiming element method.', - from_string=DashOption.string_or_none, input_type='select', cgi_name='time', cgi_choices=(None, 'direct', 'head', 'http-ntp', 'iso', 'ntp', 'sntp', 'xsd'), featured=True) -UTCValue = DashOption( +UTCValue = StringOrNoneDashOption( usage=(OptionUsage.MANIFEST | OptionUsage.TIME), short_name='utv', full_name='utcValue', title='UTC value', description='Sets the value attribute of the UTCTiming element', - from_string=DashOption.string_or_none, cgi_name='time_value', cgi_type='') @@ -56,14 +53,12 @@ POOL_NAMES: list[str] = sorted(list(NTP_POOLS.keys())) -NTPSources = DashOption( +NTPSources = StringListDashOption( usage=OptionUsage.MANIFEST, short_name='ntps', full_name='ntpSources', title='NTP time servers', description='List of servers to use for NTP requests', - from_string=DashOption.list_without_none_from_string, - to_string=lambda servers: ','.join(servers), input_type='select', cgi_name='ntp_servers', cgi_type=f'({"|".join(POOL_NAMES)}|,..)', diff --git a/dashlive/server/options/video_corrupt.py b/dashlive/server/options/video_corrupt.py index 2a6268c2a..258d5846a 100644 --- a/dashlive/server/options/video_corrupt.py +++ b/dashlive/server/options/video_corrupt.py @@ -1,29 +1,15 @@ ############################################################################# # -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# -############################################################################# -# # Project Name : Simulated MPEG DASH service # # Author : Alex Ashley # ############################################################################# -from .dash_option import DashOption +from .dash_option import IntOrNoneDashOption, StringListDashOption from .types import OptionUsage -VideoCorruption = DashOption( +VideoCorruption = StringListDashOption( usage=(OptionUsage.MANIFEST | OptionUsage.VIDEO), short_name='vcor', full_name='videoCorruption', @@ -32,12 +18,11 @@ 'Cause video corruption to be generated when requesting a fragment at the given time. ' + 'Invalid data is placed inside NAL packets of video frames. ' + 'Each time must be in the form HH:MM:SSZ.'), - from_string=DashOption.list_without_none_from_string, cgi_name='vcorrupt', cgi_type='