Auto-generate type hints for OptionsContainer

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:

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(

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:

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:

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:

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),

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:

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 = '<int>'
                 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 = '<iso-datetime>'
                 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

dashlive/server/events/factory.py

@@ -1,50 +1,38 @@
 #############################################################################
 #
-#   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,
         Scte35Events.PREFIX: Scte35Events,
     }
 
     @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:
                 EventClazz = cls.EVENT_TYPES[name]
             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