Merge pull request #43 from cta-wave/dpcat

Merge dpcat branch to main.

__init__.py

@@ -19,4 +19,4 @@
 License: Apache 2.0 https://www.apache.org/licenses/LICENSE-2.0.txt
 Licensor: Consumer Technology Association
 Contributor: Eurofins Digital Product Testing UK Limited
-"""
+"""

config.ini

@@ -53,8 +53,8 @@ no_qr_code_timeout = 5
 search_qr_area_to = 60
 # margin to add around the detected area to crop
 qr_area_margin = 50
-# system mode is for development purposes only
-#system_mode = Debug
+# qr code list check back count for duplicated qr code detection
+duplicated_qr_check_count = 3
 
 [TOLERANCES]
 start_frame_num_tolerance = 0

configuration_parser.py

@@ -25,16 +25,16 @@
 Licensor: Consumer Technology Association
 Contributor: Eurofins Digital Product Testing UK Limited
 """
-from exceptions import ConfigError
+import json
 import logging
 from typing import Dict, List
+from fractions import Fraction
 
-import requests
 import isodate
-import json
+import requests
+from exceptions import ConfigError
 from global_configurations import GlobalConfigurations
 
-
 logger = logging.getLogger(__name__)
 
 
@@ -55,18 +55,16 @@ class ConfigurationParser:
  def __init__(self, global_configurations: GlobalConfigurations):
  self.server_url = global_configurations.get_test_runner_url()
 
- # if conf.ini DEBUG mode is set then read test configuration settings from
+ # if debug mode is set then read test configuration settings from
  # a local file (used for development), instead of retrieving from Test Runner.
- if (global_configurations.get_system_mode()) == "Debug":
+ if (global_configurations.get_system_mode()) == "debug":
  self.test_config_json = self._get_json_from_local("test-config.json")
  self.tests_json = self._get_json_from_local("tests.json")
  else:
  self.test_config_json = self._get_json_from_tr("test-config.json")
  self.tests_json = self._get_json_from_tr("tests.json")
 
- def parse_config(
- self, content_type:str
- ) -> List[Dict[str, Dict[str, str]]]:
+ def parse_config(self, content_type: str) -> List[Dict[str, Dict[str, str]]]:
  if content_type == "audio":
  test_config = self.audio_config
  else:
@@ -81,22 +79,26 @@ def parse_tests_json(self, test_id: str):
  test_path = self.tests_json["tests"][test_id]["path"]
  test_code = self.tests_json["tests"][test_id]["code"]
 
- self.video_config = self.tests_json["tests"][test_id]["switchingSets"]["video"]
- self.audio_config = self.tests_json["tests"][test_id]["switchingSets"]["audio"]
+ self.video_config = self.tests_json["tests"][test_id]["switchingSets"][
+ "video"
+ ]
+ self.audio_config = self.tests_json["tests"][test_id]["switchingSets"][
+ "audio"
+ ]
 
  return test_path, test_code
  except KeyError as e:
  raise ConfigError(
  f"Unrecognised test id is detected. "
- f"Detected test id({test_id}) is not defined in \"tests.json\". "
+ f'Detected test id({test_id}) is not defined in "tests.json". '
  )
 
  def parse_fragment_duration(
  self,
  test_path: str,
  content_type: str,
  parameter: str,
- test_config: List[Dict[str, str]]
+ test_config: List[Dict[str, str]],
  ) -> dict:
  """parse fragment duration
  fragment_duration: single track playback
@@ -114,7 +116,9 @@ def parse_fragment_duration(
  raise TypeError
  # the multiplication happens so that we get the fragment duration in ms
  # we are interested just about the first video representation's fragment duration
- parameters_dict[parameter] = representation["fragment_duration"] * 1000
+ parameters_dict[parameter] = (
+ Fraction(str(representation["fragment_duration"])) * 1000
+ )
  break
  except (TypeError, KeyError) as e:
  raise ConfigError(
@@ -132,14 +136,16 @@ def parse_fragment_duration(
  if representation["fragment_duration"] == None:
  raise TypeError
  # the multiplication happens so that we get the fragment duration in ms
- parameters_dict[parameter][rep_index] = representation["fragment_duration"] * 1000
+ parameters_dict[parameter][rep_index] = (
+ representation["fragment_duration"] * 1000
+ )
  rep_index += 1
  except (TypeError, KeyError) as e:
  raise ConfigError(
  f"Failed to get a parameter:{e} for the test '{test_path}'"
  )
  elif parameter == "fragment_duration_multi_mpd":
- # fragment_duration_multi_mpd is used for tests when more then mpd is used
+ # fragment_duration_multi_mpd is used for tests when more then one mpd is used
  # for splicing set. This is an 2D array, this list is needed to identify
  # the switching points and splicing point, and to calculate durations
  parameters_dict[parameter] = {}
@@ -152,7 +158,9 @@ def parse_fragment_duration(
  if representation["fragment_duration"] == None:
  raise TypeError
  # the multiplication happens so that we get the fragment duration in ms
- parameters_dict[parameter][(content_index, rep_index)] = representation["fragment_duration"] * 1000
+ parameters_dict[parameter][(content_index, rep_index)] = (
+ representation["fragment_duration"] * 1000
+ )
  rep_index += 1
  except (TypeError, KeyError) as e:
  raise ConfigError(
@@ -181,14 +189,11 @@ def parse_cmaf_track_duration(
  return parameters_dict
 
  def parse_tests_json_content_config(
- self,
- parameters: list,
- test_path: str,
- content_type: str
+ self, parameters: list, test_path: str, content_type: str
  ) -> dict:
  """parse content related config parameters for current test"""
  parameters_dict = {}
- 
+
  # parse video/audio configuration
  # TODO: audio parsing, audio observation not implemented
  test_config = self.parse_config(content_type)
@@ -197,14 +202,16 @@ def parse_tests_json_content_config(
  for parameter in parameters:
  if parameter == "cmaf_track_duration":
  # cmaf_track_duration is only required in single mpd
- parameters_dict.update(self.parse_cmaf_track_duration(test_path,
- test_config[0]))
+ parameters_dict.update(
+ self.parse_cmaf_track_duration(test_path, test_config[0])
+ )
  else:
  # parse fragment duration handled differently for single mpd and mutiple
- parameters_dict.update(self.parse_fragment_duration(test_path,
- content_type,
- parameter,
- test_config))
+ parameters_dict.update(
+ self.parse_fragment_duration(
+ test_path, content_type, parameter, test_config
+ )
+ )
 
  return parameters_dict
 
@@ -227,9 +234,14 @@ def parse_test_config_json(
  value = self.test_config_json["all"][parameter]
  parameters_dict[parameter] = value
  except KeyError as e:
- raise ConfigError(
- f"Failed to get a parameter:{e} for the test '{test_path}'"
- )
+ # gap_duration by default is fragment_duration
+ # when undefined seek fragment_duration
+ if parameter == "gap_duration":
+ continue
+ else:
+ raise ConfigError(
+ f"Failed to get a parameter:{e} for the test '{test_path}'"
+ )
 
  return parameters_dict
 
@@ -268,12 +280,12 @@ class PlayoutParser:
  @staticmethod
  def get_switching_playout(playout: List[List[int]]) -> List[int]:
  """for switching set to extract track ID list
- switching set ID column 0 and the fragment ID column 2 
+ switching set ID column 0 and the fragment ID column 2
  are ignored for swithing set tests
  """
  switching_playout = [i[1] for i in playout]
  return switching_playout
- 
+
  @staticmethod
  def get_playout_sequence(switching_playout: List[int]):
  """for switching set return playout sequence
@@ -285,12 +297,12 @@ def get_playout_sequence(switching_playout: List[int]):
  if switching_playout[i] != switching_playout[i - 1]:
  playout_sequence.append(switching_playout[i])
  return playout_sequence
- 
+
  @staticmethod
  def get_splicing_period_list(
  playouts: List[List[int]], fragment_duration_multi_mpd: dict
  ) -> List[float]:
- """ return period duration list of splicing
+ """return period duration list of splicing
  e.g: [main duration, ad duration, main duration]
  """
  period_list = []
@@ -307,8 +319,7 @@ def get_splicing_period_list(
 
  @staticmethod
  def get_change_type_list(playouts: List[List[int]]) -> List[str]:
- """ save ecah change type in a list
- """
+ """save ecah change type in a list"""
  change_type_list = []
  switching_set = playouts[0][0]
  track_num = playouts[0][1]
@@ -323,28 +334,26 @@ def get_change_type_list(playouts: List[List[int]]) -> List[str]:
 
  @staticmethod
  def get_ending_playout_list(playouts: List[List[int]]) -> List[List[int]]:
- """ when content change save each previous ending playout
- """
+ """when content change save each previous ending playout"""
  ending_playout_list = []
  switching_set = playouts[0][0]
  track_num = playouts[0][1]
  for i, playout in enumerate(playouts):
- if (playout[0] != switching_set or playout[1] != track_num):
- ending_playout_list.append(playouts[i-1])
+ if playout[0] != switching_set or playout[1] != track_num:
+ ending_playout_list.append(playouts[i - 1])
  switching_set = playout[0]
  track_num = playout[1]
  return ending_playout_list
 
  @staticmethod
  def get_starting_playout_list(playouts: List[List[int]]) -> List[List[int]]:
- """ when content change save each current starting playout
- """
+ """when content change save each current starting playout"""
  starting_playout_list = []
  switching_set = playouts[0][0]
  track_num = playouts[0][1]
  for playout in playouts:
- if (playout[0] != switching_set or playout[1] != track_num):
+ if playout[0] != switching_set or playout[1] != track_num:
  starting_playout_list.append(playout)
  switching_set = playout[0]
  track_num = playout[1]
- return starting_playout_list
+ return starting_playout_list

dpctf_qr_decoder.py

@@ -25,10 +25,12 @@
 Licensor: Consumer Technology Association
 Contributor: Eurofins Digital Product Testing UK Limited
 """
-import re
-import logging
 import json
+import logging
+import re
 from datetime import datetime
+from fractions import Fraction
+
 from qr_recognition.qr_decoder import DecodedQr, QrDecoder
 
 logger = logging.getLogger(__name__)
@@ -53,7 +55,7 @@ class MezzanineDecodedQr(DecodedQr):
  """The media time encoded in this QR code"""
  frame_number: int
  """The media time encoded in this QR code"""
- frame_rate: float
+ frame_rate: Fraction
  """The frame rate encoded in this QR code"""
 
  first_camera_frame_num: int
@@ -68,7 +70,7 @@ def __init__(
  content_id: str,
  media_time: float,
  frame_number: int,
- frame_rate: float,
+ frame_rate: Fraction,
  camera_frame_num: int,
  ):
  super().__init__(data, location)
@@ -144,8 +146,9 @@ def __init__(self, data: str, location: list, session_token: str, test_id: str):
 
 
 class DPCTFQrDecoder(QrDecoder):
- def _translate_qr_test_runner(
- self, data: str, location: list, json_data, camera_frame_num: int
+ @staticmethod
+ def translate_qr_test_runner(
+ data: str, location: list, json_data, camera_frame_num: int
  ) -> DecodedQr:
  """translate different type of test runner qr code"""
  code = DecodedQr("", [])
@@ -195,14 +198,11 @@ def _translate_qr_test_runner(
 
  return code
 
- def _media_time_str_to_ms(self, media_time_str: str) -> float:
+ @staticmethod
+ def media_time_str_to_ms(media_time_str: str) -> float:
  """Change media time string to ms
  return media time from mezzanine QR code in milliseconds
  """
- # temp bug fix from content to be removed
- if media_time_str == "00:00:60.000":
- media_time_str = "00:01:00.000"
-
  media_time_datetime = datetime.strptime(media_time_str, "%H:%M:%S.%f")
 
  ms = media_time_datetime.microsecond / 1000
@@ -213,6 +213,20 @@ def _media_time_str_to_ms(self, media_time_str: str) -> float:
 
  return media_time
 
+ @staticmethod
+ def frame_rate_str_to_fraction(frame_rate_str: str) -> Fraction:
+ """Convert string frame rate to float
+ fractional frame rate fund match from map to get accurate number"""
+ frame_rate_map = {}
+ with open("frame_rate_map.json") as f:
+ frame_rate_map = json.load(f)
+ try:
+ res = frame_rate_map[frame_rate_str].split('/')
+ frame_rate = Fraction(int(res[0]), int(res[1]))
+ except Exception:
+ frame_rate = Fraction(float(frame_rate_str))
+ return frame_rate
+
  def translate_qr(
  self, data: str, location: list, camera_frame_num: int
  ) -> DecodedQr:
@@ -228,20 +242,21 @@ def translate_qr(
  match = _mezzanine_qr_data_re.match(data)
  if match:
  # matches a mezzanine signature so decode it as such
- media_time = self._media_time_str_to_ms(match.group(2))
+ media_time = DPCTFQrDecoder.media_time_str_to_ms(match.group(2))
+ frame_rate = DPCTFQrDecoder.frame_rate_str_to_fraction(match.group(4))
  code = MezzanineDecodedQr(
  data,
  location,
  match.group(1),
  media_time,
  int(match.group(3)),
- float(match.group(4)),
+ frame_rate,
  camera_frame_num,
  )
  else:
  try:
  json_data = json.loads(data)
- code = self._translate_qr_test_runner(
+ code = DPCTFQrDecoder.translate_qr_test_runner(
  data, location, json_data, camera_frame_num
  )
  except json.decoder.JSONDecodeError as e:

exceptions.py

@@ -23,6 +23,7 @@
 Contributor: Eurofins Digital Product Testing UK Limited
 """
 
+
 class ObsFrameError(Exception):
  """Base for specific exceptions thrown by Observation Framework.
 
@@ -43,9 +44,11 @@ class ObsFrameTerminate(ObsFrameError):
  """Specific exceptions thrown by Observation Framework.
  when this is raised the Observation Framework will be terminated.
  """
+
  pass
 
 
 class ConfigError(ObsFrameError):
  """Specific exception to indicate config errors"""
+
  pass