From 23920055679cc065d2ec84b1495db918a7fa6186 Mon Sep 17 00:00:00 2001 From: Alex Ashley Date: Fri, 13 Feb 2026 14:37:40 +0000 Subject: [PATCH 01/11] Period@duration is optional The duration attribute of a Period element is optional. When it is not present, the duration can be calculated from the start of the next period, or is undefined if this is the last Period. The validator was incorrectly requiring a duration element. In addition, when checking SegmentTimeline within a Representation, it was using total timeshiftBufferDepth, rather than the expected duration of the Period. --- dashlive/mpeg/dash/validator/errors.py | 2 +- dashlive/mpeg/dash/validator/manifest.py | 29 ++++++++++++------- dashlive/mpeg/dash/validator/period.py | 2 +- .../mpeg/dash/validator/representation.py | 10 +++++-- 4 files changed, 28 insertions(+), 15 deletions(-) diff --git a/dashlive/mpeg/dash/validator/errors.py b/dashlive/mpeg/dash/validator/errors.py index a5f21a3f..e9fe5c72 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/manifest.py b/dashlive/mpeg/dash/validator/manifest.py index deefe040..a7c85faf 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 70869307..18afce95 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 b820f240..8e6db3df 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: From 3fdc6f887aaecd47cb0d44af68223697a996a88e Mon Sep 17 00:00:00 2001 From: Alex Ashley Date: Fri, 13 Feb 2026 14:40:18 +0000 Subject: [PATCH 02/11] only add Period@duration if forcePeriodDurations is True In most cases, there is no need to put in a Period@duration attribute, as the start of the next Period can be used to calculate a Period's duration. A new forcePeriodDurations option ("periodDur" CGI parameter) is added that allows overriding this default of omitting duration attributes. --- dashlive/mpeg/dash/period.py | 26 ++++++++++++------- dashlive/server/options/manifest_options.py | 16 ++++++++++++ .../server/requesthandler/manifest_context.py | 4 +-- templates/manifests/hand_made.mpd | 2 +- tests/fixtures/hand_made_mps_live.mpd | 14 +++++----- tests/test_cgi_options.py | 2 ++ 6 files changed, 44 insertions(+), 20 deletions(-) diff --git a/dashlive/mpeg/dash/period.py b/dashlive/mpeg/dash/period.py index 09649808..bafd2b53 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/server/options/manifest_options.py b/dashlive/server/options/manifest_options.py index 7733aa63..0519e077 100644 --- a/dashlive/server/options/manifest_options.py +++ b/dashlive/server/options/manifest_options.py @@ -195,11 +195,27 @@ def ast_to_string(value: datetime.datetime | str | None) -> str: ('Yes', '1'), )) +ForcePeriodDurations = DashOption( + usage=OptionUsage.MANIFEST, + short_name='fpd', + full_name='forcePeriodDurations', + title='Forced Period durations', + description='Always add a duration attribute to Period elements', + from_string=DashOption.bool_from_string, + to_string=DashOption.bool_to_string, + 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/requesthandler/manifest_context.py b/dashlive/server/requesthandler/manifest_context.py index 3c4365ec..b35cb129 100644 --- a/dashlive/server/requesthandler/manifest_context.py +++ b/dashlive/server/requesthandler/manifest_context.py @@ -338,9 +338,7 @@ def create_period(self, base_url = flask.url_for( 'dash-media-base-url', mode=opts.mode, stream=stream.directory) - period.finish_setup( - mode=opts.mode, timing=timing, base_url=base_url, - use_base_urls=opts.useBaseUrls) + period.finish_setup(timing=timing, base_url=base_url, options=opts) if is_https_request(): period.baseURL = period.baseURL.replace('http://', 'https://') for adp in period.adaptationSets: diff --git a/templates/manifests/hand_made.mpd b/templates/manifests/hand_made.mpd index dde29fa1..7a279ea2 100644 --- a/templates/manifests/hand_made.mpd +++ b/templates/manifests/hand_made.mpd @@ -36,7 +36,7 @@ {% endif -%} {%- for period in mpd.periods %} {%- if period.baseURL %}{{ period.baseURL }}{% endif %} {% include "events/period.xml" %} diff --git a/tests/fixtures/hand_made_mps_live.mpd b/tests/fixtures/hand_made_mps_live.mpd index 01b7ce35..b7210fb5 100644 --- a/tests/fixtures/hand_made_mps_live.mpd +++ b/tests/fixtures/hand_made_mps_live.mpd @@ -17,7 +17,7 @@ Example multi-period stream http://unit.test/mps/live/testmps/hand_made.mpd?start=2024-10-07T00:00:00Z - http://unit.test/mps/live/testmps/1/ - http://unit.test/mps/live/testmps/2/ - http://unit.test/mps/live/testmps/1/ - http://unit.test/mps/live/testmps/2/ - http://unit.test/mps/live/testmps/1/ - http://unit.test/mps/live/testmps/2/ - \ No newline at end of file + diff --git a/tests/test_cgi_options.py b/tests/test_cgi_options.py index cb6b587b..eeb59a91 100644 --- a/tests/test_cgi_options.py +++ b/tests/test_cgi_options.py @@ -227,6 +227,7 @@ def test_default_values(self) -> None: 'drmSelection': [], 'eventTypes': [], 'failureCount': None, + 'forcePeriodDurations': False, 'leeway': 16, 'mainAudio': None, 'mode': 'vod', @@ -395,6 +396,7 @@ def test_convert_stream_default_options(self) -> None: "ping" ], "failureCount": None, + 'forcePeriodDurations': False, "leeway": 16, "mainAudio": None, "mainText": None, From 065042bbe95345b59974016abb99f50e5d98df57 Mon Sep 17 00:00:00 2001 From: Alex Ashley Date: Tue, 17 Feb 2026 11:24:36 +0000 Subject: [PATCH 03/11] fix incorrect class name for CGI tests the class name had been copied and pasted from the server tests --- tests/test_cgi_options.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_cgi_options.py b/tests/test_cgi_options.py index eeb59a91..82c36704 100644 --- a/tests/test_cgi_options.py +++ b/tests/test_cgi_options.py @@ -11,7 +11,7 @@ from .mixins.mock_time import MockTime from .mixins.mixin import TestCaseMixin -class TestServerOptions(TestCaseMixin, unittest.TestCase): +class TestCgiOptions(TestCaseMixin, unittest.TestCase): def test_option_usage_from_string(self) -> None: test_cases = [ ('manifest', OptionUsage.MANIFEST), From ce55ab705f55f825869c8bffd8971681d11f286b Mon Sep 17 00:00:00 2001 From: Alex Ashley Date: Tue, 17 Feb 2026 11:25:18 +0000 Subject: [PATCH 04/11] auto-generate Python and TypeScript types for DashOptions rather than using generic dictionary and mapping types to describe all of the DashOptions, generate Python and Typescript type definitions. --- dashlive/server/options/repository.py | 120 ++++++++++++++++++++++---- 1 file changed, 102 insertions(+), 18 deletions(-) diff --git a/dashlive/server/options/repository.py b/dashlive/server/options/repository.py index a3ce43c5..e597c41d 100644 --- a/dashlive/server/options/repository.py +++ b/dashlive/server/options/repository.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,7 +7,9 @@ ############################################################################# import logging -from typing import AbstractSet, ClassVar +from operator import attrgetter +from pathlib import Path +from typing import AbstractSet, Any, ClassVar, NamedTuple from .audio_options import audio_options from .container import OptionsContainer @@ -35,6 +23,11 @@ from .video_options import video_options from .utc_time_options import time_options +class DashOptionTypeHint(NamedTuple): + name: str + py_type: str + ts_type: str + class OptionsRepository: _cgi_map: ClassVar[dict[str, DashOption] | None] = None _short_name_map: ClassVar[dict[str, DashOption] | None] = None @@ -126,7 +119,7 @@ def get_cgi_map(cls) -> dict[str, str]: return cls._cgi_map @classmethod - def get_short_param_map(cls) -> dict[str, str]: + def get_short_param_map(cls) -> dict[str, DashOption]: """ Returns a dictionary that maps from the short parameter used in a stream defaults @@ -139,7 +132,7 @@ def get_short_param_map(cls) -> dict[str, str]: return cls._short_name_map @classmethod - def get_parameter_map(cls) -> dict[str, str]: + def get_parameter_map(cls) -> dict[str, DashOption]: """ Returns a dictionary that maps from the full parameter name to its DashOption entry @@ -157,7 +150,7 @@ def get_parameter_map(cls) -> dict[str, str]: return cls._param_map @classmethod - def get_default_options(cls, use: OptionUsage | None = None) -> OptionsContainer: + def get_default_options(cls) -> OptionsContainer: """ Returns a dictionary containing the global defaults for every option """ @@ -185,6 +178,93 @@ def get_default_options(cls, use: OptionUsage | None = None) -> OptionsContainer dest = result dest.add_field(opt.full_name, value) return result + + @staticmethod + def guess_python_type(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[{OptionsRepository.guess_python_type(value[0])}]" + return 'list' + return 'Any' + + @staticmethod + def guess_typescript_type(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"{OptionsRepository.guess_python_type(value[0])}[]" + return 'unknown[]' + return 'unknown' + + @classmethod + def create_python_options_container_types(cls, py_dest: Path, ts_dest: Path) -> None: + primary_options: list[DashOptionTypeHint] = [] + sub_options: dict[str, list[DashOptionTypeHint]] = {} + defaults = cls.get_default_options() + for key, value in defaults.items(): + if key == "_type": + continue + if isinstance(value, OptionsContainer): + sub: list[DashOptionTypeHint] = [] + for k2, v2 in value.items(): + if k2 == "_type": + continue + sub.append(DashOptionTypeHint( + name=k2, + py_type=cls.guess_python_type(v2), + ts_type=cls.guess_typescript_type(v2))) + sub.sort(key=attrgetter('name')) + name: str = f"{key.title()}OptionsType" + sub_options[name] = sub + primary_options.append(DashOptionTypeHint( + name=key, + py_type=name, + ts_type=name)) + else: + primary_options.append(DashOptionTypeHint( + name=key, + py_type=cls.guess_python_type(value), + ts_type=cls.guess_typescript_type(value))) + primary_options.sort(key=attrgetter('name')) + print(sub_options) + print(primary_options) + with py_dest.open('wt', encoding='utf-8') as dest: + dest.write('# this file is auto-generated, do not edit!\n\n') + dest.write('from typing import Any\n') + dest.write('from dataclasses import dataclass\n\n') + for name, options in sub_options.items(): + dest.write(f'@dataclass\nclass {name}:\n') + for opt in options: + dest.write(f' {opt.name}: {opt.py_type}\n') + dest.write('\n\n') + dest.write(f'@dataclass\nclass OptionsContainerType:\n') + for opt in primary_options: + dest.write(f' {opt.name}: {opt.py_type}\n') + dest.write('\n\n') + with ts_dest.open('wt', encoding='utf-8') as dest: + dest.write('// this file is auto-generated, do not edit!\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(f'export type OptionsContainerType = {{\n') + for opt in primary_options: + dest.write(f' {opt.name}: {opt.ts_type};\n') + dest.write('}\n') @classmethod def convert_cgi_options(cls, params: dict[str, str], @@ -247,3 +327,7 @@ def convert_options(cls, for opt in OptionsRepository._all_options: if opt.prefix and opt.prefix not in OptionsContainer.OBJECT_FIELDS: OptionsContainer.OBJECT_FIELDS[opt.prefix] = OptionsContainer + +if __name__ == "__main__": + OptionsRepository.create_python_options_container_types( + Path("/tmp/options_types.py"), Path("/tmp/options_types.d.ts")) \ No newline at end of file From e3efa44ebb7b071b21696b95b1f3a50fa5c32804 Mon Sep 17 00:00:00 2001 From: Alex Ashley Date: Wed, 18 Feb 2026 11:25:02 +0000 Subject: [PATCH 05/11] refactor DashOptions into type-specific classes Rather than providing a from_string and to_string lambda in each DashOption, use a set of option classes, that inherit from DashOption, which perform the conversion to and from strings. This allows DashOption to use Generic type hints, making it easier for the type metadata to correctly describe the type of each option. --- dashlive/server/events/base.py | 60 ++---- dashlive/server/options/audio_options.py | 9 +- dashlive/server/options/dash_option.py | 201 +++++++++++--------- dashlive/server/options/drm_options.py | 161 +++++++--------- dashlive/server/options/event_options.py | 6 +- dashlive/server/options/form_input_field.py | 2 + dashlive/server/options/http_error.py | 91 ++++----- dashlive/server/options/manifest_options.py | 121 ++++++------ dashlive/server/options/player_options.py | 10 +- dashlive/server/options/repository.py | 100 +--------- dashlive/server/options/text_options.py | 14 +- dashlive/server/options/utc_time_options.py | 15 +- dashlive/server/options/video_corrupt.py | 22 +-- tests/test_cgi_options.py | 61 +++--- 14 files changed, 366 insertions(+), 507 deletions(-) diff --git a/dashlive/server/events/base.py b/dashlive/server/events/base.py index 894775ac..aa36098d 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,16 @@ 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, + StringOrNoneDashOption, +) from dashlive.server.options.types import OptionUsage from dashlive.utils.object_with_fields import ObjectWithFields @@ -49,54 +41,39 @@ 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 + OptionType: type[DashOption] = StringOrNoneDashOption if isinstance(dflt, bool): - from_string = DashOption.bool_from_string - to_string = DashOption.bool_to_string + 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( + elif dflt is not None: + cgi_choices = tuple(str(dflt)) + opt = OptionType( usage=(OptionUsage.MANIFEST + OptionUsage.AUDIO + OptionUsage.VIDEO), short_name=short_name, full_name=key, @@ -107,7 +84,6 @@ def default_to_string(val: Any) -> str: cgi_name=name, cgi_type=cgi_type, cgi_choices=cgi_choices, - from_string=from_string, - to_string=to_string) + default=dflt) result.append(opt) return result diff --git a/dashlive/server/options/audio_options.py b/dashlive/server/options/audio_options.py index eca4253f..aba85f74 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', @@ -23,9 +23,10 @@ ('Any codec', 'any'), ), input_type='select', + default='mp4a', featured=True) -AudioDescriptionTrack = DashOption( +AudioDescriptionTrack = StringOrNoneDashOption( usage=OptionUsage.MANIFEST, short_name='ad', full_name='audioDescription', @@ -35,7 +36,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/dash_option.py b/dashlive/server/options/dash_option.py index 840417b8..fa20768e 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,11 +32,21 @@ 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 + default: T | 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 get_cgi_option(self, omit_empty: bool = True) -> CgiOption | None: """ @@ -107,7 +103,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 +120,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 +172,82 @@ 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: + def html_input_type(self) -> str: + return 'bool' + + +class IntOrNoneDashOption(DashOption[int | None]): + def from_string(self, value: str) -> int | None: if value in {None, '', 'none'}: - return None + return self.default 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' + + +class FloatOrNoneDashOption(DashOption[float | None]): + def from_string(self, value: str) -> float | None: if value in {None, '', 'none'}: - return None + return self.default return float(value) - @staticmethod - def datetime_or_none_from_string(value: str) -> datetime.datetime | datetime.timedelta | None: - if value in {None, '', 'none'}: - return None - return from_isodatetime(value) + def to_string(self, value: float | None) -> str: + if value is None: + return '' + return f'{value:f}' + + +class StringDashOption(DashOption[str]): + def from_string(self, value: str) -> str: + if value.lower() in ['', 'none']: + if self.default is None: + return '' + return self.default + return value - @staticmethod - def datetime_or_none_to_string(value: datetime.datetime | None) -> str | None: + def to_string(self, value: str) -> str: + return value + +class StringOrNoneDashOption(DashOption[str | None]): + def from_string(self, value: str) -> str | None: + if value.lower() in ['', 'none']: + return self.default + return value + + def to_string(self, value: str | None) -> str: if value is None: - return None - return to_iso_datetime(value) + return '' + return value - @staticmethod - def list_without_none_from_string(value: str | None) -> list[str]: +class UrlOrNoneDashOption(DashOption[str | None]): + def from_string(self, value: str) -> str | None: + if value.lower() in ['', 'none']: + return self.default + return urllib.parse.unquote_plus(value) + + def to_string(self, value: str | None) -> str: if value is None: - return [] + return '' + return urllib.parse.quote_plus(value) + +class StringListDashOption(DashOption[list[str]]): + def from_string(self, value: str) -> list[str]: if value.lower() in {'', 'none'}: return [] rv = [] @@ -224,20 +256,19 @@ 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']: - return None - return urllib.parse.unquote_plus(value) - @staticmethod - def quoted_url_or_none_to_string(value: str | None): +class DateTimeDashOption(DashOption[datetime | timedelta | None]): + def from_string(self, value: str) -> datetime | timedelta | None: + if value in {None, '', 'none'}: + return self.default + return from_isodatetime(value) + + 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) diff --git a/dashlive/server/options/drm_options.py b/dashlive/server/options/drm_options.py index 2934d450..7add5d37 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,101 @@ 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 == '': - 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( +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: + 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 = 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 be3948ba..ca223759 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 a008ffe9..2a89e0c6 100644 --- a/dashlive/server/options/form_input_field.py +++ b/dashlive/server/options/form_input_field.py @@ -17,6 +17,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 +32,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/http_error.py b/dashlive/server/options/http_error.py index 520c6d99..b99c9e6e 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,48 @@ 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 - -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='=,..') - - -ManifestHttpError = http_error_factory('manifest', 'Manifest') - -VideoHttpError = http_error_factory('video', 'Video fragments') - -AudioHttpError = http_error_factory('audio', 'Audio fragments') - -TextHttpError = http_error_factory('text', 'Text fragments') - -FailureCount = DashOption( +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) + + +ManifestHttpError = HttpErrorOption('manifest', 'Manifest') + +VideoHttpError = HttpErrorOption('video', 'Video fragments') + +AudioHttpError = HttpErrorOption('audio', 'Audio fragments') + +TextHttpError = HttpErrorOption('text', 'Text fragments') + +FailureCount = IntOrNoneDashOption( usage=(OptionUsage.MANIFEST | OptionUsage.AUDIO | OptionUsage.VIDEO | OptionUsage.TEXT), short_name='hfc', full_name='failureCount', @@ -68,6 +58,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 0519e077..a36f14ae 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,45 @@

''' -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) + + +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 +93,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 +126,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 +142,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 +155,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,14 +188,12 @@ def ast_to_string(value: datetime.datetime | str | None) -> str: ('Yes', '1'), )) -ForcePeriodDurations = DashOption( +ForcePeriodDurations = BoolDashOption( usage=OptionUsage.MANIFEST, short_name='fpd', full_name='forcePeriodDurations', title='Forced Period durations', description='Always add a duration attribute to Period elements', - from_string=DashOption.bool_from_string, - to_string=DashOption.bool_to_string, input_type='checkbox', cgi_name='periodDur', cgi_choices=( diff --git a/dashlive/server/options/player_options.py b/dashlive/server/options/player_options.py index 46d8cccf..6e318851 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 e597c41d..4ea44eb5 100644 --- a/dashlive/server/options/repository.py +++ b/dashlive/server/options/repository.py @@ -7,9 +7,7 @@ ############################################################################# import logging -from operator import attrgetter -from pathlib import Path -from typing import AbstractSet, Any, ClassVar, NamedTuple +from typing import AbstractSet, ClassVar, NamedTuple from .audio_options import audio_options from .container import OptionsContainer @@ -106,7 +104,7 @@ def matches(item: DashOption) -> bool: return result @classmethod - def get_cgi_map(cls) -> dict[str, str]: + 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 @@ -178,93 +176,6 @@ def get_default_options(cls) -> OptionsContainer: dest = result dest.add_field(opt.full_name, value) return result - - @staticmethod - def guess_python_type(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[{OptionsRepository.guess_python_type(value[0])}]" - return 'list' - return 'Any' - - @staticmethod - def guess_typescript_type(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"{OptionsRepository.guess_python_type(value[0])}[]" - return 'unknown[]' - return 'unknown' - - @classmethod - def create_python_options_container_types(cls, py_dest: Path, ts_dest: Path) -> None: - primary_options: list[DashOptionTypeHint] = [] - sub_options: dict[str, list[DashOptionTypeHint]] = {} - defaults = cls.get_default_options() - for key, value in defaults.items(): - if key == "_type": - continue - if isinstance(value, OptionsContainer): - sub: list[DashOptionTypeHint] = [] - for k2, v2 in value.items(): - if k2 == "_type": - continue - sub.append(DashOptionTypeHint( - name=k2, - py_type=cls.guess_python_type(v2), - ts_type=cls.guess_typescript_type(v2))) - sub.sort(key=attrgetter('name')) - name: str = f"{key.title()}OptionsType" - sub_options[name] = sub - primary_options.append(DashOptionTypeHint( - name=key, - py_type=name, - ts_type=name)) - else: - primary_options.append(DashOptionTypeHint( - name=key, - py_type=cls.guess_python_type(value), - ts_type=cls.guess_typescript_type(value))) - primary_options.sort(key=attrgetter('name')) - print(sub_options) - print(primary_options) - with py_dest.open('wt', encoding='utf-8') as dest: - dest.write('# this file is auto-generated, do not edit!\n\n') - dest.write('from typing import Any\n') - dest.write('from dataclasses import dataclass\n\n') - for name, options in sub_options.items(): - dest.write(f'@dataclass\nclass {name}:\n') - for opt in options: - dest.write(f' {opt.name}: {opt.py_type}\n') - dest.write('\n\n') - dest.write(f'@dataclass\nclass OptionsContainerType:\n') - for opt in primary_options: - dest.write(f' {opt.name}: {opt.py_type}\n') - dest.write('\n\n') - with ts_dest.open('wt', encoding='utf-8') as dest: - dest.write('// this file is auto-generated, do not edit!\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(f'export type OptionsContainerType = {{\n') - for opt in primary_options: - dest.write(f' {opt.name}: {opt.ts_type};\n') - dest.write('}\n') @classmethod def convert_cgi_options(cls, params: dict[str, str], @@ -295,8 +206,9 @@ def convert_options(cls, defaults=defaults) else: result = OptionsContainer(cls.get_parameter_map(), defaults) + param_map: dict[str, DashOption] if is_cgi: - param_map: dict[str, DashOption] = cls.get_cgi_map() + param_map = cls.get_cgi_map() else: param_map = cls.get_short_param_map() for key, value in params.items(): @@ -327,7 +239,3 @@ def convert_options(cls, for opt in OptionsRepository._all_options: if opt.prefix and opt.prefix not in OptionsContainer.OBJECT_FIELDS: OptionsContainer.OBJECT_FIELDS[opt.prefix] = OptionsContainer - -if __name__ == "__main__": - OptionsRepository.create_python_options_container_types( - Path("/tmp/options_types.py"), Path("/tmp/options_types.d.ts")) \ No newline at end of file diff --git a/dashlive/server/options/text_options.py b/dashlive/server/options/text_options.py index 33261a62..19125d5a 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 4d436620..6ea04346 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 2a6268c2..258d5846 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='