PTP Improvements (#150)

* Improve PTP section and update tests and examples

* Add cbor2 requirement and update parser to latest

* Oops - update piplock

* Cap PTP priorities at 255 and add tests

* Add enum for PTP profiles and tests

* Update PTP profile enum

Pipfile

@@ -8,6 +8,7 @@ pylint = "==2.6.0"
 coverage = "*"
 jsonschema = "*"
 jinja2 = "*"
+cbor2 = "*"
 
 [requires]
 python_version = "3.11"

src/main/python/camdkit/examples.py

@@ -136,11 +136,15 @@ def _get_complete_dynamic_clip():
     locked=True,
     source=SynchronizationSourceEnum.PTP,
     ptp=SynchronizationPTP(
-      offset=0.0,
-      domain=1,
-      leader="00:11:22:33:44:55"
-    ),
-    offsets=SynchronizationOffsets(1.0,2.0,3.0)
+      PTP_PROFILES[2],
+      1,
+      "00:11:22:33:44:55",
+      SynchronizationPTPPriorities(128, 128),
+      0.00000005,
+      0.000123,
+      100,
+      "GNSS"
+    )
   ),)
   #   transforms
   clip.global_stage = (GlobalPosition(100.0,200.0,300.0,100.0,200.0,300.0),)

src/main/python/camdkit/framework.py

@@ -15,6 +15,11 @@
 UINT48_MAX = 281474976710655 # 2^48 - 1
 
 DEFAULT_SUB_FRAME = 0
+PTP_PROFILES = [
+  "IEEE Std 1588-2019",
+  "IEEE Std 802.1AS-2020",
+  "SMPTE ST2059-2:2021",
+]
 
 class Sampling(Enum):
   STATIC = "Static"   # Data that does not change for a Clip or across many Frames
@@ -179,21 +184,103 @@ def to_json(value: typing.Any) -> typing.Any:
     }
 
 @dataclasses.dataclass
+class SynchronizationPTPPriorities:
+  """Data structure for PTP synchronization priorities"""
+
+  priority1: typing.Optional[int]
+  priority2: typing.Optional[int]
+
+  def validate(self):
+    return all([isinstance(self.priority1, int) and 
+                self.priority1 >= 0 and self.priority1 < 256,
+                isinstance(self.priority2, int) and
+                self.priority2 >= 0 and self.priority2 < 256])
+
+  @staticmethod
+  def to_json(value: typing.Any) -> typing.Any:
+    return {
+      "priority1": value.priority1,
+      "priority2": value.priority2,
+    }
+@dataclasses.dataclass
 class SynchronizationPTP:
   """Data structure for PTP synchronization"""
 
+  profile: typing.Optional[str] = None
   domain: typing.Optional[int] = None
-  leader: typing.Optional[str] = None
-  offset: typing.Optional[float] = None
+  leader_identity: typing.Optional[str] = None
+  leader_priorities: typing.Optional[SynchronizationPTPPriorities] = None
+  leader_accuracy: typing.Optional[float] = None
+  mean_path_delay: typing.Optional[float] = None
+  vlan: typing.Optional[int] = None
+  time_source: typing.Optional[str] = None
 
   def validate(self):
-    return all([isinstance(self.leader, str), 
-                isinstance(self.offset, float),
-                isinstance(self.domain, int)])
+    return all([isinstance(self.profile, str),
+                isinstance(self.domain, int),
+                isinstance(self.leader_priorities, SynchronizationPTPPriorities) and
+                self.leader_priorities.validate(),
+                isinstance(self.leader_accuracy, float),
+                isinstance(self.mean_path_delay, float)])
 
   @staticmethod
   def to_json(value: typing.Any) -> typing.Any:
-    return dataclasses.asdict(value)
+    ret = {
+      "profile": value.profile,
+      "domain": value.domain,
+      "leaderIdentity": value.leader_identity,
+      "leaderPriorities": SynchronizationPTPPriorities.to_json(value.leader_priorities),
+      "leaderAccuracy": value.leader_accuracy,
+      "meanPathDelay": value.mean_path_delay,
+      "meanPathDelay": value.mean_path_delay,
+    }
+    if value.vlan != None:
+      ret["vlan"] = value.vlan
+    if value.time_source != None:
+      ret["timeSource"] = value.time_source
+    return ret
+
+  @staticmethod
+  def from_json(value: typing.Any) -> typing.Any:
+    ptp = SynchronizationPTP(value["profile"],
+                             value["domain"],
+                             value["leaderIdentity"],
+                             SynchronizationPTPPriorities(value["leaderPriorities"]["priority1"],
+                                                          value["leaderPriorities"]["priority2"]),
+                             value["leaderAccuracy"],
+                             value["meanPathDelay"],
+                             )
+    if "vlan" in value:
+      ptp.vlan = value["vlan"]
+    if "timeSource" in value:
+      ptp.time_source = value["timeSource"]
+    return ptp
+
+  @staticmethod
+  def make_json_schema() -> dict:
+    return {
+      "type": "object",
+      "additionalProperties": False,
+      "properties": {
+        "profile": { "type": "string", "enum": PTP_PROFILES },
+        "domain": { "type": "integer", "minimum": 0, "maximum": 127 },
+        "leaderIdentity": { "type": "string", "pattern": r"(?:^[0-9a-f]{2}(?::[0-9a-f]{2}){5}$)|(?:^[0-9a-f]{2}(?:-[0-9a-f]{2}){5}$)"},
+        "leaderPriorities": {
+          "type": "object",
+          "additionalProperties": False,
+          "properties": {
+            "priority1": { "type": "integer", "minimum": 0, "maximum": 255 },
+            "priority2": { "type": "integer", "minimum": 0, "maximum": 255 },
+          },
+          "required": ["priority1", "priority2"]
+        },
+        "leaderAccuracy": { "type": "number", "minimum": 0 },
+        "meanPathDelay": { "type": "number", "minimum": 0 },
+        "vlan": { "type": "integer", "minimum": 0 },
+        "timeSource": { "type": "string", "minLength": 1 },
+      },
+      "required": ["profile", "domain", "leaderIdentity", "leaderPriorities", "leaderAccuracy", "meanPathDelay"]
+    }
 
 class TimingModeEnum(BaseEnum):
   """Enumeration for sample timing modes"""

src/main/python/camdkit/model.py

@@ -562,12 +562,28 @@ class TimingSynchronization(Parameter):
   example)
   ptp: If the synchronization source is a PTP leader, then this object
   contains:
-  - "leader": The MAC address of the PTP leader
-  - "offset": The timing offset in seconds from the sample timestamp to
-  the PTP timestamp
-  - "domain": The PTP domain number
+  - "profile": Specifies the PTP profile in use. This defines the operational
+  rules and parameters for synchronization. For example "SMPTE ST2059-2:2021"
+  for SMPTE 2110 based systems, or "IEEE Std 1588-2019" or
+  "IEEE Std 802.1AS-2020" for industrial applications
+  - "domain": Identifies the PTP domain the device belongs to. Devices in the
+  same domain can synchronize with each other
+  - "leaderIdentity": The unique identifier (usually MAC address) of the
+  current PTP leader (grandmaster)
+  - "leaderPriorities": The priority values of the leader used in the Best
+  Master Clock Algorithm (BMCA). Lower values indicate higher priority
+  - "priority1": Static priority set by the administrator
+  - "priority2": Dynamic priority based on the leader's role or clock quality
+  - "leaderAccuracy": The timing offset in seconds from the sample timestamp
+  to the PTP timestamp
+  - "meanPathDelay": The average round-trip delay between the device and the
+  PTP leader, measured in seconds
   source: The source of synchronization must be defined as one of the
   following:
+  - "vlan": Integer representing the VLAN ID for PTP traffic (e.g., 100 for
+  VLAN 100)
+  - "timeSource": Indicates the leader's source of time, such as GNSS, atomic
+  clock, or NTP
   - "genlock": The tracking device has an external black/burst or
   tri-level analog sync signal that is triggering the capture of
   tracking samples
@@ -595,16 +611,29 @@ def validate(value) -> bool:
       if not (isinstance(value.frequency, numbers.Rational) and value.frequency > 0):
         return False
     if value.ptp != None:
+      if not value.ptp.validate():
+        return False
+      # Required PTP fields
+      if any([value.ptp.profile == None,
+              value.ptp.domain == None,
+              value.ptp.leader_identity == None,
+              value.ptp.leader_priorities == None,
+              value.ptp.leader_accuracy == None,
+              value.ptp.mean_path_delay == None,
+          ]):
+        return False
+      if value.ptp.profile not in PTP_PROFILES:
+        return False
+      if not all([isinstance(value.ptp.domain, int), value.ptp.domain < 128, value.ptp.domain >= 0]):
+        return False
       # Validate MAC address
-      if value.ptp.leader != None and not (isinstance(value.ptp.leader,str) and 
-                                           re.match("[0-9a-f]{2}([-:]?)[0-9a-f]{2}(\\1[0-9a-f]{2}){4}$",
-                                           value.ptp.leader.lower())):
+      if not re.match("[0-9a-f]{2}([-:]?)[0-9a-f]{2}(\\1[0-9a-f]{2}){4}$", value.ptp.leader_identity.lower()):
+        return False
+      if value.ptp.leader_accuracy <= 0.0 or value.ptp.mean_path_delay <= 0.0:
         return False
-      if value.ptp.offset != None and not isinstance(value.ptp.offset, float):
+      if value.ptp.vlan != None and (not isinstance(value.ptp.vlan, int) or value.ptp.vlan <= 0):
         return False
-      if value.ptp.domain != None and not (isinstance(value.ptp.domain, int) \
-                                           and value.ptp.domain < 128 \
-                                           and value.ptp.domain >= 0):
+      if value.ptp.time_source != None and (not isinstance(value.ptp.time_source, str) or value.ptp.time_source == ""):
         return False
       if value.offsets != None and not value.offsets.validate():
         return False
@@ -620,7 +649,9 @@ def to_json(value: typing.Any) -> typing.Any:
     if "frequency" in d:
       d["frequency"] = { "num": d["frequency"].numerator, "denom": d["frequency"].denominator }
     if value.offsets is not None:
-        d["offsets"] = SynchronizationOffsets.to_json(value.offsets)
+      d["offsets"] = SynchronizationOffsets.to_json(value.offsets)
+    if value.ptp is not None:
+      d["ptp"] = SynchronizationPTP.to_json(value.ptp)
     return d
 
   @staticmethod
@@ -630,7 +661,7 @@ def from_json(value: typing.Any) -> typing.Any:
     sync.offsets = SynchronizationOffsets(translation=value["offsets"]["translation"],
                                           rotation=value["offsets"]["rotation"],
                                           lens_encoders=value["offsets"]["lensEncoders"])
-    sync.ptp = SynchronizationPTP(**value["ptp"])
+    sync.ptp = SynchronizationPTP.from_json(value["ptp"])
     sync.frequency = Fraction(value["frequency"]["num"], value["frequency"]["denom"])
     return sync
 
@@ -669,15 +700,7 @@ def make_json_schema() -> dict:
           }
         },
         "present": { "type": "boolean" },
-        "ptp": {
-          "type": "object",
-          "additionalProperties": False,
-          "properties": {
-            "leader": { "type": "string", "pattern": r"(?:^[0-9a-f]{2}(?::[0-9a-f]{2}){5}$)|(?:^[0-9a-f]{2}(?:-[0-9a-f]{2}){5}$)"},
-            "offset": { "type": "number" },
-            "domain": { "type": "integer", "minimum": 0, "maximum": 127 }
-          }
-        },
+        "ptp": SynchronizationPTP.make_json_schema(),
         "source": { "type": "string", "enum": [e.value for e in SynchronizationSourceEnum] },
       },
       "required": ["locked", "source"]

src/test/python/parser/opentrackio_lib.py

@@ -138,8 +138,8 @@ def import_schema(self):
         if self.schema_str:
             try:
                 self.sd = json.loads(self.schema_str)
-            except json.decoder.JSONDecodeError:
-                raise OpenTrackIOException(e.message)
+            except json.decoder.JSONDecodeError as e:
+                raise OpenTrackIOException(e.msg)
             if not self.sd:
                 raise OpenTrackIOException("Error: Failed to parse OpenTrackIO schema file.")
             else:                                   # we have a valid schema
@@ -286,9 +286,9 @@ def get_sample_time(self, format: Optional[TimeFormat] = None, part=None):
 
     def get_timecode_framerate(self):
         """Frame rate which the house timecode represents"""
-        if self.validate_dict_elements(self.pd,["timing","frameRate","num"]):
-            numerator = float(self.pd["timing"]["frameRate"]["num"])  
-            denominator = float(self.pd["timing"]["frameRate"]["denom"]) 
+        if self.validate_dict_elements(self.pd,["timing","sampleRate","num"]):
+            numerator = float(self.pd["timing"]["sampleRate"]["num"])  
+            denominator = float(self.pd["timing"]["sampleRate"]["denom"]) 
             return float(numerator / denominator)
         else:
             return None