Skip to content

Commit

Permalink
Add metadata section and tests
Browse files Browse the repository at this point in the history
  • Loading branch information
jamesmosys committed Jun 24, 2024
1 parent 05b74b6 commit ad3cd35
Show file tree
Hide file tree
Showing 5 changed files with 170 additions and 3 deletions.
51 changes: 51 additions & 0 deletions src/main/python/camdkit/framework.py
Original file line number Diff line number Diff line change
Expand Up @@ -215,6 +215,27 @@ def make_json_schema() -> dict:
}
}

class BooleanParameter(Parameter):

@staticmethod
def validate(value) -> bool:
"""The parameter shall be a boolean."""
return isinstance(value, bool)

@staticmethod
def to_json(value: typing.Any) -> typing.Any:
return value

@staticmethod
def from_json(value: typing.Any) -> typing.Any:
return bool(value)

@staticmethod
def make_json_schema() -> dict:
return {
"type": "boolean"
}

class StringParameter(Parameter):

@staticmethod
Expand All @@ -237,6 +258,36 @@ def make_json_schema() -> dict:
"minLength": 1,
"maxLength": 1023
}

class ArrayParameter(Parameter):

item_class = None

def validate(self, value) -> bool:
"""The parameter shall be a tuple of items of the class itemClass. The tuple can be empty"""

if self.item_class == None:
return False
if not isinstance(value, tuple):
return False
for item in value:
if not isinstance(item, self.item_class) and not self.item_class.validate(item):
return False
return True

@staticmethod
def to_json(value: typing.Any) -> typing.Any:
return list(value)

@staticmethod
def from_json(value: typing.Any) -> typing.Any:
return str(value)

def make_json_schema(self) -> dict:
return {
"type": "array",
"items": self.item_class.make_json_schema()
}

class UUIDURNParameter(Parameter):

Expand Down
46 changes: 44 additions & 2 deletions src/main/python/camdkit/model.py
Original file line number Diff line number Diff line change
Expand Up @@ -190,6 +190,43 @@ def make_json_schema() -> dict:
"minimum": 1,
"maximum": 360000
}

class MetadataStatus(StringParameter):
"""Free string that describes the status of the system - e.g. 'Optical Good' in a tracking system"""

canonical_name = "status"
sampling = Sampling.REGULAR
units = None
section = "metadata"

class MetadataRecording(BooleanParameter):
"""True if the system is recording data - e.g. tracking data"""

canonical_name = "recording"
sampling = Sampling.REGULAR
section = "metadata"

class MetadataSlate(StringParameter):
"""Free string that describes the recording slate - e.g. 'A101_A_4'"""

canonical_name = "slate"
sampling = Sampling.REGULAR
section = "metadata"

class MetadataNotes(StringParameter):
"""Free string for notes"""

canonical_name = "notes"
sampling = Sampling.REGULAR
section = "metadata"

class MetadataRelatedPackets(ArrayParameter):
"""List of packet unique IDs that are related to this packet."""

canonical_name = "relatedPackets"
sampling = Sampling.REGULAR
units = None
item_class = UUIDURNParameter

class PacketId(UUIDURNParameter):
"""Unique identifier of the packet in which data is being traansported."""
Expand Down Expand Up @@ -670,8 +707,13 @@ class Clip(ParameterContainer):
fdl_link: typing.Optional[str] = FDLLink()
shutter_angle: typing.Optional[numbers.Integral] = ShutterAngle()
# TODO JU rest of the tracking model! Also re-order sensibly
packet_id: typing.Optional[str] = PacketId()
protocol: typing.Optional[str] = Protocol()
packet_id: typing.Optional[typing.Tuple[str]] = PacketId()
protocol: typing.Optional[typing.Tuple[str]] = Protocol()
metadata_status: typing.Optional[typing.Tuple[str]] = MetadataStatus()
metadata_recording: typing.Optional[typing.Tuple[bool]] = MetadataRecording()
metadata_slate: typing.Optional[typing.Tuple[str]] = MetadataSlate()
metadata_notes: typing.Optional[typing.Tuple[str]] = MetadataNotes()
metadata_related_packets: typing.Optional[typing.Tuple[tuple]] = MetadataRelatedPackets()
timing_mode: typing.Optional[typing.Tuple[TimingMode]] = TimingMode()
timing_timestamp: typing.Optional[typing.Tuple[TimingTimestamp]] = TimingTimestamp()
timing_sequence_number: typing.Optional[typing.Tuple[NonNegativeIntegerParameter]] = TimingSequenceNumber()
Expand Down
31 changes: 30 additions & 1 deletion src/main/python/camdkit/mosys/f4.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,9 +34,30 @@ class F4:
FIELD_ID_APERTURE = 0x60
COMMAND_BYTE = 0xf4
FIELD_ID_TIMECODE = 0xF8
TRACKING_STATUS = 0xF9

ANGLE_FACTOR = 1000
LINEAR_FACTOR = 1000

TRACKING_STATUS_STRINGS = [
"Undefined",
"Tracking",
"Optical Good",
"Optical Acceptable",
"Optical Unreliable",
"Optical Unstable",
"Optical Lost",
"Lost Too Few Stars",
"Location Searching",
"Busy or Waiting",
"Busy Loading Map",
"No Map Loaded",
"Test Signal",
"Mechanical Encoders Only",
"I/O Error",
"Internal Error"
]

class F4AxisBlock:
axis_id: int = 0
axis_status: int = 0
Expand Down Expand Up @@ -139,6 +160,10 @@ def _axis_block_to_lens_type(self, axis_block: F4AxisBlock) -> int:
def _axis_block_to_lens_param(self, axis_block: F4AxisBlock) -> float:
return self._four_bytes_to_float(axis_block.axis_status, axis_block.data_bits1, axis_block.data_bits2, axis_block.data_bits3)

def _axis_block_to_status_string(self, axis_block: F4AxisBlock) -> str:
detail = ((axis_block.data_bits2 >> 4) & 0xF)
return F4.TRACKING_STATUS_STRINGS[detail]

def _compute_f4_checksum(self, buffer: bytes) -> int:
total: int = 0x40
for i in range(0, self._packet.size - 1):
Expand Down Expand Up @@ -171,7 +196,7 @@ def get_tracking_frame(self) -> Clip:
frame.packet_id = (uuid.uuid1().urn,)
frame.timing_mode = ("internal",)
frame.timing_sequence_number = (self._frame_number,)
#frame.metadata.recording = (self._packet.status & (1 << 4)) != 0
frame.metadata_recording = ((self._packet.status & (1 << 4)) != 0,)
for i in range(0, self._packet.axis_count):
axis_block = self._packet.axis_block_list[i]
match axis_block.axis_id:
Expand Down Expand Up @@ -231,6 +256,10 @@ def get_tracking_frame(self) -> Clip:
frame.timing_frame_rate = (frame_rate,)
frequency = frame_rate
pass
case F4.TRACKING_STATUS:
frame.metadata_status = (self._axis_block_to_status_string(axis_block),)
pass

sync = Synchronization(
locked=(self._packet.status & (1 << 5)) != 0,
source=SynchronizationSourceEnum.GENLOCK,
Expand Down
40 changes: 40 additions & 0 deletions src/test/python/test_model.py
Original file line number Diff line number Diff line change
Expand Up @@ -286,6 +286,46 @@ def test_protocol(self):
clip.protocol = value
self.assertTupleEqual(clip.protocol, value)

def test_metadata(self):
clip = camdkit.model.Clip()

self.assertIsNone(clip.metadata_status)
self.assertIsNone(clip.metadata_recording)
self.assertIsNone(clip.metadata_slate)
self.assertIsNone(clip.metadata_notes)
self.assertIsNone(clip.metadata_related_packets)

with self.assertRaises(ValueError):
clip.metadata_status = ""
with self.assertRaises(ValueError):
clip.metadata_recording = 0
with self.assertRaises(ValueError):
clip.metadata_recording = "True"
with self.assertRaises(ValueError):
clip.metadata_slate = ""
with self.assertRaises(ValueError):
clip.metadata_notes = ""
with self.assertRaises(ValueError):
clip.metadata_related_packets = ("urn:uuid:f81d4fae-7dec-11d0-a765-00a0c91e6bf6",
"urn:uuid:f81d4fae-7dec-11d0-a765-00a0c91e6bf6")

value = ("Optical Good",)
clip.metadata_status = value
self.assertTupleEqual(clip.metadata_status, value)
value = (True,False)
clip.metadata_recording = value
self.assertTupleEqual(clip.metadata_recording, value)
value = ("A104_A_4",)
clip.metadata_slate = value
self.assertTupleEqual(clip.metadata_slate, value)
value = ("Test notes",)
clip.metadata_notes = value
self.assertTupleEqual(clip.metadata_notes, value)
value = (("urn:uuid:f81d4fae-7dec-11d0-a765-00a0c91e6bf6",
"urn:uuid:f81d4fae-7dec-11d0-a765-00a0c91e6bf6"),)
clip.metadata_related_packets = value
self.assertTupleEqual(clip.metadata_related_packets, value)

def test_timing_mode_model(self):
clip = camdkit.model.Clip()

Expand Down
5 changes: 5 additions & 0 deletions src/tools/python/generate_example.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,11 @@ def main():
clip = Clip()
clip.packet_id = (uuid.uuid1().urn,)
clip.protocol = ("OpenTrackIO_0.1.0",)
clip.metadata_status = ("Optical Good",)
clip.metadata_recording = (False,)
clip.metadata_slate = ("A101_A_4",)
clip.metadata_notes = ("Example generated packet.",)
clip.metadata_related_packets = ((uuid.uuid1().urn,uuid.uuid1().urn),)
clip.timing_mode = (TimingModeEnum.INTERNAL,)
clip.timing_timestamp = (Timestamp(1718806554, 0),)
clip.timing_sequence_number = (0,)
Expand Down

0 comments on commit ad3cd35

Please sign in to comment.