mediagrains.grains
1from .AudioGrain import AudioGrain, size_for_audio_format 2from .VideoGrain import VideoGrain 3from .CodedAudioGrain import CodedAudioGrain 4from .CodedVideoGrain import CodedVideoGrain 5from .EventGrain import EventGrain 6from .Grain import Grain, attributes_for_grain_type, new_attributes_for_grain_type, GrainFactory 7 8__all__ = ["Grain", "VideoGrain", "CodedVideoGrain", "AudioGrain", "CodedAudioGrain", "EventGrain", 9 "attributes_for_grain_type", "new_attributes_for_grain_type", "size_for_audio_format", "GrainFactory"]
class
Grain(collections.abc.Sequence):
171class Grain(Sequence): 172 """\ 173A class representing a generic media grain. 174 175Any grain can be freely cast to a tuple: 176 177 (meta, data) 178 179where meta is a dictionary containing the grain metadata, and data is None or one of the following: 180* a bytes-like object 181* An object supporting the __bytes__ magic method 182* An awaitable returning a valid data element 183 184In addition the class provides a number of properties which can be used to 185access parts of the standard grain metadata, and all other grain classes 186inherit these: 187 188meta 189 The meta dictionary object 190 191data 192 One of the following: 193 * A byteslike object -- This becomes the grain's data element 194 * An object that has a method __bytes__ which returns a bytes-like object, which will be the grain's data 195 element 196 * None -- This grain has no data 197 198 If the data parameter passed on construction is an awaitable which will return a valid data element when awaited 199 then the grain's data element is 200 initially None, but the grain can be awaited to populate it 201 202 For convenience any grain can be awaited and will return the data element, regardless of whether the underlying 203 data is asynchronous or not 204 205 For additional convenience using a grain as an async context manager will ensure that the data element is populated 206 if it needs to be and can be. 207 208grain_type 209 A string containing the type of the grain, any value is possible 210 211source_id 212 A uuid.UUID object representing the source_id in the grain 213 214flow_id 215 A uuid.UUID object representing the flow_id in the grain 216 217origin_timestamp 218 An mediatimestamp.Timestamp object representing the origin timestamp 219 of this grain. 220 221sync_timestamp 222 An mediatimestamp.Timestamp object representing the sync timestamp 223 of this grain. 224 225creation_timestamp 226 An mediatimestamp.Timestamp object representing the creation timestamp 227 of this grain. 228 229rate 230 A fractions.Fraction object representing the grain rate in grains per second. 231 232duration 233 A fractions.Fraction object representing the grain duration in seconds. 234 235timelabels 236 A list object containing time label data 237 238length 239 The length of the data property, or 0 if it is None 240 241expected_length 242 How long the data would be expected to be based on what's listed in the metadata 243 244 245In addition these methods are provided for convenience: 246 247 248final_origin_timestamp() 249 The origin timestamp of the final sample in the grain. For most grain types this is the same as 250 origin_timestamp, but not for audio grains. 251 252origin_timerange() 253 The origin time range covered by the samples in the grain. 254 255presentation_origin_timestamp 256 The presentation timeline origin timestamp for the grain. 257 258final_presentation_origin_timestamp() 259 The presentation origin timestamp of the final sample in the grain. For most grain types this is the same as 260 presentation_origin_timestamp, but not for audio grains. 261 262presentation_origin_timerange() 263 The presentation timeline origin time range covered by the samples in the grain. 264 265normalise_time(value) 266 Returns a normalised Timestamp, TimeOffset or TimeRange using the media rate. 267 268media_rate 269 The video frame rate or audio sample rate as a Fraction or None. Returns None if there is no media 270 rate or the media rate == 0. 271 272 """ 273 def __init__(self, meta: GrainMetadataDict, data: GrainDataParameterType, **kwargs): 274 275 self.meta = meta 276 277 if meta is None: 278 pass 279 280 self._data_fetcher_coroutine: Optional[Awaitable[Optional[GrainDataType]]] 281 self._data_fetcher_length: int = 0 282 self._data: Optional[GrainDataType] 283 284 if isawaitable(data): 285 self._data_fetcher_coroutine = cast(Awaitable[Optional[GrainDataType]], data) 286 self._data = None 287 else: 288 self._data_fetcher_coroutine = None 289 self._data = cast(Optional[GrainDataType], data) 290 self._factory = "Grain" 291 if self.meta is not None: 292 # This code is here to deal with malformed inputs, and as such needs to cast away the type safety to operate 293 if "@_ns" not in self.meta: 294 cast(EmptyGrainMetadataDict, self.meta)['@_ns'] = "urn:x-ipstudio:ns:0.1" 295 if 'grain' not in self.meta: 296 cast(dict, self.meta)['grain'] = {} 297 if 'grain_type' not in self.meta['grain']: 298 cast(EmptyGrainMetadataDict, self.meta)['grain']['grain_type'] = "empty" 299 if 'creation_timestamp' not in self.meta['grain']: 300 cast(EmptyGrainMetadataDict, self.meta)['grain']['creation_timestamp'] = str(Timestamp.get_time()) 301 if 'origin_timestamp' not in self.meta['grain']: 302 cast(EmptyGrainMetadataDict, self.meta 303 )['grain']['origin_timestamp'] = self.meta['grain']['creation_timestamp'] 304 if 'sync_timestamp' not in self.meta['grain']: 305 cast(EmptyGrainMetadataDict, self.meta)['grain']['sync_timestamp'] = \ 306 self.meta['grain']['origin_timestamp'] 307 if 'rate' not in self.meta['grain']: 308 cast(EmptyGrainMetadataDict, self.meta)['grain']['rate'] = {'numerator': 0, 309 'denominator': 1} 310 if 'duration' not in self.meta['grain']: 311 cast(EmptyGrainMetadataDict, self.meta)['grain']['duration'] = {'numerator': 0, 312 'denominator': 1} 313 if 'source_id' not in self.meta['grain']: 314 cast(EmptyGrainMetadataDict, self.meta)['grain']['source_id'] = "00000000-0000-0000-0000-000000000000" 315 if 'flow_id' not in self.meta['grain']: 316 cast(EmptyGrainMetadataDict, self.meta)['grain']['flow_id'] = "00000000-0000-0000-0000-000000000000" 317 318 if isinstance(self.meta["grain"]["source_id"], UUID): 319 cast(EmptyGrainMetadataDict, self.meta)['grain']['source_id'] = str(self.meta['grain']['source_id']) 320 if isinstance(self.meta["grain"]["flow_id"], UUID): 321 cast(EmptyGrainMetadataDict, self.meta)['grain']['flow_id'] = str(self.meta['grain']['flow_id']) 322 if not isinstance(self.meta["grain"]["origin_timestamp"], str): 323 cast(EmptyGrainMetadataDict, self.meta)['grain']['origin_timestamp'] = _stringify_timestamp_input( 324 self.meta['grain']['origin_timestamp']) 325 if not isinstance(self.meta["grain"]["sync_timestamp"], str): 326 cast(EmptyGrainMetadataDict, self.meta)['grain']['sync_timestamp'] = _stringify_timestamp_input( 327 self.meta['grain']['sync_timestamp']) 328 if not isinstance(self.meta["grain"]["creation_timestamp"], str): 329 cast(EmptyGrainMetadataDict, self.meta)['grain']['creation_timestamp'] = _stringify_timestamp_input( 330 self.meta['grain']['creation_timestamp']) 331 if isinstance(self.meta['grain']['rate'], Fraction): 332 cast(EmptyGrainMetadataDict, self.meta)['grain']['rate'] = { 333 'numerator': self.meta['grain']['rate'].numerator, 334 'denominator': self.meta['grain']['rate'].denominator} 335 if isinstance(self.meta['grain']['duration'], Fraction): 336 cast(EmptyGrainMetadataDict, self.meta)['grain']['duration'] = { 337 'numerator': self.meta['grain']['duration'].numerator, 338 'denominator': self.meta['grain']['duration'].denominator} 339 else: 340 raise ValueError("Metadata dict passed to Grain was none!!") 341 342 def __len__(self) -> int: 343 return 2 344 345 @overload 346 def __getitem__(self, index: int) -> Union[GrainMetadataDict, Optional[GrainDataType]]: ... 347 348 @overload # noqa: F811 349 def __getitem__(self, index: slice) -> Union[Tuple[GrainMetadataDict], 350 Tuple[GrainMetadataDict, Optional[GrainDataType]], 351 Tuple[Optional[GrainDataType]], 352 Tuple[()]]: ... 353 354 def __getitem__(self, index): # noqa: F811 355 return (self.meta, self.data)[index] 356 357 def __repr__(self) -> str: 358 if not hasattr(self.data, "__len__"): 359 return "{}({!r})".format(self._factory, self.meta) 360 else: 361 return "{}({!r},< binary data of length {} >)".format(self._factory, self.meta, len(cast(Sized, self.data))) 362 363 def __eq__(self, other: object) -> bool: 364 return tuple(self) == other 365 366 def __ne__(self, other: object) -> bool: 367 return not (self == other) 368 369 def __copy__(self) -> "Grain": 370 return GrainFactory(copy(self.meta), self.data) 371 372 def __deepcopy__(self, memo) -> "Grain": 373 return GrainFactory(deepcopy(self.meta), deepcopy(self.data)) 374 375 def __bytes__(self) -> Optional[bytes]: 376 if isinstance(self._data, bytes): 377 return self._data 378 elif self._data is None: 379 return None 380 return bytes(self._data) 381 382 def has_data(self) -> bool: 383 return self._data is not None 384 385 def __await__(self) -> Generator[Any, None, Optional[GrainDataType]]: 386 async def __inner(): 387 if self._data is None and self._data_fetcher_coroutine is not None: 388 self._data = await self._data_fetcher_coroutine 389 return self._data 390 return __inner().__await__() 391 392 async def __aenter__(self): 393 await self 394 return self 395 396 async def __aexit__(self, *args, **kwargs): 397 pass 398 399 @property 400 def data(self) -> Optional[GrainDataType]: 401 return self._data 402 403 @data.setter 404 def data(self, value: GrainDataParameterType): 405 if isawaitable(value): 406 self._data = None 407 self._data_fetcher_coroutine = cast(Awaitable[Optional[GrainDataType]], value) 408 else: 409 self._data = cast(Optional[GrainDataType], value) 410 self._data_fetcher_coroutine = None 411 412 @property 413 def grain_type(self) -> str: 414 if not hasattr(self, 'meta'): 415 pass 416 return self.meta['grain']['grain_type'] 417 418 @grain_type.setter 419 def grain_type(self, value: str) -> None: 420 # We ignore the type safety rules for this assignment 421 self.meta['grain']['grain_type'] = value # type: ignore 422 423 @property 424 def source_id(self) -> UUID: 425 # Our code ensures that this will always be a string at runtime 426 return UUID(cast(str, self.meta['grain']['source_id'])) 427 428 @source_id.setter 429 def source_id(self, value: Union[UUID, str]) -> None: 430 cast(EmptyGrainMetadataDict, self.meta)['grain']['source_id'] = str(value) 431 432 @property 433 def src_id(self) -> UUID: 434 # Our code ensures that this will always be a string at runtime 435 return UUID(cast(str, self.meta['grain']['source_id'])) 436 437 @src_id.setter 438 def src_id(self, value: Union[UUID, str]) -> None: 439 cast(EmptyGrainMetadataDict, self.meta)['grain']['source_id'] = str(value) 440 441 @property 442 def flow_id(self) -> UUID: 443 return UUID(cast(str, self.meta['grain']['flow_id'])) 444 445 @flow_id.setter 446 def flow_id(self, value: Union[UUID, str]) -> None: 447 cast(EmptyGrainMetadataDict, self.meta)['grain']['flow_id'] = str(value) 448 449 @property 450 def origin_timestamp(self) -> Timestamp: 451 return Timestamp.from_tai_sec_nsec(cast(str, self.meta['grain']['origin_timestamp'])) 452 453 @origin_timestamp.setter 454 def origin_timestamp(self, value: Union[SupportsMediaTimestamp, SupportsMediaTimeOffset, str]): 455 cast(EmptyGrainMetadataDict, self.meta)['grain']['origin_timestamp'] = _stringify_timestamp_input(value) 456 457 def final_origin_timestamp(self) -> Timestamp: 458 return self.origin_timestamp 459 460 def origin_timerange(self) -> TimeRange: 461 return TimeRange(self.origin_timestamp, self.final_origin_timestamp(), TimeRange.INCLUSIVE) 462 463 @property 464 def presentation_origin_timestamp(self) -> Timestamp: 465 return self.origin_timestamp 466 467 def final_presentation_origin_timestamp(self) -> Timestamp: 468 return self.final_origin_timestamp() 469 470 def presentation_origin_timerange(self) -> TimeRange: 471 return self.origin_timerange() 472 473 @overload 474 def normalise_time(self, value: TimeOffset) -> TimeOffset: ... 475 476 @overload 477 def normalise_time(self, value: TimeRange) -> TimeRange: ... 478 479 def normalise_time(self, value): 480 if self.media_rate is not None: 481 return value.normalise(self.media_rate.numerator, self.media_rate.denominator) 482 else: 483 return value 484 485 @property 486 def media_rate(self) -> Optional[Fraction]: 487 return None 488 489 @property 490 def sync_timestamp(self) -> Timestamp: 491 return Timestamp.from_tai_sec_nsec(cast(str, self.meta['grain']['sync_timestamp'])) 492 493 @sync_timestamp.setter 494 def sync_timestamp(self, value: Union[SupportsMediaTimestamp, SupportsMediaTimeOffset, str]) -> None: 495 cast(EmptyGrainMetadataDict, self.meta)['grain']['sync_timestamp'] = _stringify_timestamp_input(value) 496 497 @property 498 def creation_timestamp(self) -> Timestamp: 499 return Timestamp.from_tai_sec_nsec(cast(str, self.meta['grain']['creation_timestamp'])) 500 501 @creation_timestamp.setter 502 def creation_timestamp(self, value: Union[SupportsMediaTimestamp, SupportsMediaTimeOffset, str]) -> None: 503 cast(EmptyGrainMetadataDict, self.meta)['grain']['creation_timestamp'] = _stringify_timestamp_input(value) 504 505 @property 506 def rate(self) -> Fraction: 507 return Fraction(cast(FractionDict, self.meta['grain']['rate'])['numerator'], 508 cast(FractionDict, self.meta['grain']['rate'])['denominator']) 509 510 @rate.setter 511 def rate(self, value: RationalTypes) -> None: 512 value = Fraction(value) 513 cast(EmptyGrainMetadataDict, self.meta)['grain']['rate'] = { 514 'numerator': value.numerator, 515 'denominator': value.denominator 516 } 517 518 @property 519 def duration(self) -> Fraction: 520 return Fraction(cast(FractionDict, self.meta['grain']['duration'])['numerator'], 521 cast(FractionDict, self.meta['grain']['duration'])['denominator']) 522 523 @duration.setter 524 def duration(self, value: RationalTypes) -> None: 525 value = Fraction(value) 526 cast(EmptyGrainMetadataDict, self.meta)['grain']['duration'] = { 527 'numerator': value.numerator, 528 'denominator': value.denominator 529 } 530 531 @property 532 def timelabels(self) -> "Grain.TIMELABELS": 533 return Grain.TIMELABELS(self) 534 535 @timelabels.setter 536 def timelabels(self, value: "Union[List[Grain.TIMELABEL], Grain.TIMELABELS]") -> None: 537 cast(EmptyGrainMetadataDict, self.meta)['grain']['timelabels'] = [] 538 for x in value: 539 self.timelabels.append(x) 540 541 def add_timelabel(self, tag: str, count: int, rate: Fraction, drop_frame: bool = False) -> None: 542 tl = Grain.TIMELABEL() 543 tl.tag = tag 544 tl.count = count 545 tl.rate = rate 546 tl.drop_frame = drop_frame 547 self.timelabels.append(tl) 548 549 class TIMELABEL(Mapping): 550 GrainMetadataDict = Dict[str, Any] 551 552 def __init__(self, meta: "Optional[Grain.TIMELABEL.GrainMetadataDict]" = None): 553 if meta is None: 554 meta = {} 555 self.meta = meta 556 if 'tag' not in self.meta: 557 self.meta['tag'] = '' 558 if 'timelabel' not in self.meta: 559 self.meta['timelabel'] = {} 560 if 'frames_since_midnight' not in self.meta['timelabel']: 561 self.meta['timelabel']['frames_since_midnight'] = 0 562 if 'frame_rate_numerator' not in self.meta['timelabel']: 563 self.meta['timelabel']['frame_rate_numerator'] = 0 564 if 'frame_rate_denominator' not in self.meta['timelabel']: 565 self.meta['timelabel']['frame_rate_denominator'] = 1 566 if 'drop_frame' not in self.meta['timelabel']: 567 self.meta['timelabel']['drop_frame'] = False 568 569 def __getitem__(self, key: str) -> Union[str, Dict[str, Union[int, bool]]]: 570 return self.meta[key] 571 572 def __setitem__(self, key: str, value: Union[str, Dict[str, Union[int, bool]]]) -> None: 573 if key not in ['tag', 'timelabel']: 574 raise KeyError 575 self.meta[key] = value 576 577 def __iter__(self) -> Iterator[str]: 578 return self.meta.__iter__() 579 580 def __len__(self) -> int: 581 return 2 582 583 def __eq__(self, other: object) -> bool: 584 return dict(self) == other 585 586 def __ne__(self, other: object) -> bool: 587 return not (self == other) 588 589 @property 590 def tag(self) -> str: 591 return self.meta['tag'] 592 593 @tag.setter 594 def tag(self, value: str) -> None: 595 self.meta['tag'] = value 596 597 @property 598 def count(self) -> int: 599 return self.meta['timelabel']['frames_since_midnight'] 600 601 @count.setter 602 def count(self, value: int) -> None: 603 self.meta['timelabel']['frames_since_midnight'] = int(value) 604 605 @property 606 def rate(self) -> Fraction: 607 return Fraction(self.meta['timelabel']['frame_rate_numerator'], 608 self.meta['timelabel']['frame_rate_denominator']) 609 610 @rate.setter 611 def rate(self, value: RationalTypes) -> None: 612 value = Fraction(value) 613 self.meta['timelabel']['frame_rate_numerator'] = value.numerator 614 self.meta['timelabel']['frame_rate_denominator'] = value.denominator 615 616 @property 617 def drop_frame(self) -> bool: 618 return self.meta['timelabel']['drop_frame'] 619 620 @drop_frame.setter 621 def drop_frame(self, value: bool) -> None: 622 self.meta['timelabel']['drop_frame'] = bool(value) 623 624 class TIMELABELS(MutableSequence): 625 def __init__(self, parent: "Grain"): 626 self.parent = parent 627 628 @overload 629 def __getitem__(self, key: int) -> "Grain.TIMELABEL": ... 630 631 @overload # noqa: F811 632 def __getitem__(self, key: slice) -> "List[Grain.TIMELABEL]": ... 633 634 def __getitem__(self, key): # noqa: F811 635 if 'timelabels' not in self.parent.meta['grain']: 636 raise IndexError("list index out of range") 637 if isinstance(key, int): 638 return Grain.TIMELABEL(self.parent.meta['grain']['timelabels'][key]) 639 else: 640 return [Grain.TIMELABEL(self.parent.meta['grain']['timelabels'][n]) for n in range(len(self))[key]] 641 642 @overload 643 def __setitem__(self, key: int, value: "Grain.TIMELABEL.GrainMetadataDict") -> None: ... 644 645 @overload # noqa: F811 646 def __setitem__(self, key: slice, value: "Iterable[Grain.TIMELABEL.GrainMetadataDict]") -> None: ... 647 648 def __setitem__(self, key, value): # noqa: F811 649 if 'timelabels' not in self.parent.meta['grain']: 650 raise IndexError("list assignment index out of range") 651 if isinstance(key, int): 652 self.parent.meta['grain']['timelabels'][key] = dict(Grain.TIMELABEL(value)) 653 else: 654 values = iter(value) 655 for n in key: 656 self.parent.meta['grain']['timelabels'][n] = dict(Grain.TIMELABEL(next(values))) 657 658 def __delitem__(self, key: Union[int, slice]) -> None: 659 if 'timelabels' not in self.parent.meta['grain']: 660 raise IndexError("list assignment index out of range") 661 662 del self.parent.meta['grain']['timelabels'][key] 663 if len(self.parent.meta['grain']['timelabels']) == 0: 664 del self.parent.meta['grain']['timelabels'] 665 666 def insert(self, key: int, value: "Grain.TIMELABEL.GrainMetadataDict") -> None: 667 if 'timelabels' not in self.parent.meta['grain']: 668 cast(EmptyGrainMetadataDict, self.parent.meta)['grain']['timelabels'] = [] 669 self.parent.meta['grain']['timelabels'].insert(key, cast(TimeLabel, dict(Grain.TIMELABEL(value)))) 670 671 def __len__(self) -> int: 672 if 'timelabels' not in self.parent.meta['grain']: 673 return 0 674 return len(self.parent.meta['grain']['timelabels']) 675 676 def __eq__(self, other: object) -> bool: 677 return list(self) == other 678 679 def __ne__(self, other: object) -> bool: 680 return not (self == other) 681 682 @property 683 def length(self) -> int: 684 if hasattr(self.data, "__len__"): 685 return len(cast(Sized, self.data)) 686 elif hasattr(self.data, "__bytes__"): 687 return len(bytes(cast(SupportsBytes, self.data))) 688 elif self.data is None and self._data_fetcher_coroutine is not None: 689 return self._data_fetcher_length 690 else: 691 return 0 692 693 @length.setter 694 def length(self, L: int) -> None: 695 if self.data is None and self._data_fetcher_coroutine is not None: 696 self._data_fetcher_length = L 697 else: 698 raise AttributeError 699 700 @property 701 def expected_length(self) -> int: 702 if 'length' in self.meta['grain']: 703 return cast(dict, self.meta['grain'])['length'] 704 else: 705 return self.length
A class representing a generic media grain.
Any grain can be freely cast to a tuple:
(meta, data)
where meta is a dictionary containing the grain metadata, and data is None or one of the following:
- a bytes-like object
- An object supporting the __bytes__ magic method
- An awaitable returning a valid data element
In addition the class provides a number of properties which can be used to access parts of the standard grain metadata, and all other grain classes inherit these:
meta The meta dictionary object
data One of the following: * A byteslike object -- This becomes the grain's data element * An object that has a method __bytes__ which returns a bytes-like object, which will be the grain's data element * None -- This grain has no data
If the data parameter passed on construction is an awaitable which will return a valid data element when awaited
then the grain's data element is
initially None, but the grain can be awaited to populate it
For convenience any grain can be awaited and will return the data element, regardless of whether the underlying
data is asynchronous or not
For additional convenience using a grain as an async context manager will ensure that the data element is populated
if it needs to be and can be.
grain_type A string containing the type of the grain, any value is possible
source_id A uuid.UUID object representing the source_id in the grain
flow_id A uuid.UUID object representing the flow_id in the grain
origin_timestamp An mediatimestamp.Timestamp object representing the origin timestamp of this grain.
sync_timestamp An mediatimestamp.Timestamp object representing the sync timestamp of this grain.
creation_timestamp An mediatimestamp.Timestamp object representing the creation timestamp of this grain.
rate A fractions.Fraction object representing the grain rate in grains per second.
duration A fractions.Fraction object representing the grain duration in seconds.
timelabels A list object containing time label data
length The length of the data property, or 0 if it is None
expected_length How long the data would be expected to be based on what's listed in the metadata
In addition these methods are provided for convenience:
final_origin_timestamp() The origin timestamp of the final sample in the grain. For most grain types this is the same as origin_timestamp, but not for audio grains.
origin_timerange() The origin time range covered by the samples in the grain.
presentation_origin_timestamp The presentation timeline origin timestamp for the grain.
final_presentation_origin_timestamp() The presentation origin timestamp of the final sample in the grain. For most grain types this is the same as presentation_origin_timestamp, but not for audio grains.
presentation_origin_timerange() The presentation timeline origin time range covered by the samples in the grain.
normalise_time(value) Returns a normalised Timestamp, TimeOffset or TimeRange using the media rate.
media_rate The video frame rate or audio sample rate as a Fraction or None. Returns None if there is no media rate or the media rate == 0.
Grain( meta: Union[mediagrains.typing.EmptyGrainMetadataDict, mediagrains.typing.AudioGrainMetadataDict, mediagrains.typing.CodedAudioGrainMetadataDict, mediagrains.typing.VideoGrainMetadataDict, mediagrains.typing.CodedVideoGrainMetadataDict, mediagrains.typing.EventGrainMetadataDict], data: Union[SupportsBytes, bytes, numpy.ndarray, Awaitable[Union[SupportsBytes, bytes, numpy.ndarray, NoneType]], NoneType], **kwargs)
273 def __init__(self, meta: GrainMetadataDict, data: GrainDataParameterType, **kwargs): 274 275 self.meta = meta 276 277 if meta is None: 278 pass 279 280 self._data_fetcher_coroutine: Optional[Awaitable[Optional[GrainDataType]]] 281 self._data_fetcher_length: int = 0 282 self._data: Optional[GrainDataType] 283 284 if isawaitable(data): 285 self._data_fetcher_coroutine = cast(Awaitable[Optional[GrainDataType]], data) 286 self._data = None 287 else: 288 self._data_fetcher_coroutine = None 289 self._data = cast(Optional[GrainDataType], data) 290 self._factory = "Grain" 291 if self.meta is not None: 292 # This code is here to deal with malformed inputs, and as such needs to cast away the type safety to operate 293 if "@_ns" not in self.meta: 294 cast(EmptyGrainMetadataDict, self.meta)['@_ns'] = "urn:x-ipstudio:ns:0.1" 295 if 'grain' not in self.meta: 296 cast(dict, self.meta)['grain'] = {} 297 if 'grain_type' not in self.meta['grain']: 298 cast(EmptyGrainMetadataDict, self.meta)['grain']['grain_type'] = "empty" 299 if 'creation_timestamp' not in self.meta['grain']: 300 cast(EmptyGrainMetadataDict, self.meta)['grain']['creation_timestamp'] = str(Timestamp.get_time()) 301 if 'origin_timestamp' not in self.meta['grain']: 302 cast(EmptyGrainMetadataDict, self.meta 303 )['grain']['origin_timestamp'] = self.meta['grain']['creation_timestamp'] 304 if 'sync_timestamp' not in self.meta['grain']: 305 cast(EmptyGrainMetadataDict, self.meta)['grain']['sync_timestamp'] = \ 306 self.meta['grain']['origin_timestamp'] 307 if 'rate' not in self.meta['grain']: 308 cast(EmptyGrainMetadataDict, self.meta)['grain']['rate'] = {'numerator': 0, 309 'denominator': 1} 310 if 'duration' not in self.meta['grain']: 311 cast(EmptyGrainMetadataDict, self.meta)['grain']['duration'] = {'numerator': 0, 312 'denominator': 1} 313 if 'source_id' not in self.meta['grain']: 314 cast(EmptyGrainMetadataDict, self.meta)['grain']['source_id'] = "00000000-0000-0000-0000-000000000000" 315 if 'flow_id' not in self.meta['grain']: 316 cast(EmptyGrainMetadataDict, self.meta)['grain']['flow_id'] = "00000000-0000-0000-0000-000000000000" 317 318 if isinstance(self.meta["grain"]["source_id"], UUID): 319 cast(EmptyGrainMetadataDict, self.meta)['grain']['source_id'] = str(self.meta['grain']['source_id']) 320 if isinstance(self.meta["grain"]["flow_id"], UUID): 321 cast(EmptyGrainMetadataDict, self.meta)['grain']['flow_id'] = str(self.meta['grain']['flow_id']) 322 if not isinstance(self.meta["grain"]["origin_timestamp"], str): 323 cast(EmptyGrainMetadataDict, self.meta)['grain']['origin_timestamp'] = _stringify_timestamp_input( 324 self.meta['grain']['origin_timestamp']) 325 if not isinstance(self.meta["grain"]["sync_timestamp"], str): 326 cast(EmptyGrainMetadataDict, self.meta)['grain']['sync_timestamp'] = _stringify_timestamp_input( 327 self.meta['grain']['sync_timestamp']) 328 if not isinstance(self.meta["grain"]["creation_timestamp"], str): 329 cast(EmptyGrainMetadataDict, self.meta)['grain']['creation_timestamp'] = _stringify_timestamp_input( 330 self.meta['grain']['creation_timestamp']) 331 if isinstance(self.meta['grain']['rate'], Fraction): 332 cast(EmptyGrainMetadataDict, self.meta)['grain']['rate'] = { 333 'numerator': self.meta['grain']['rate'].numerator, 334 'denominator': self.meta['grain']['rate'].denominator} 335 if isinstance(self.meta['grain']['duration'], Fraction): 336 cast(EmptyGrainMetadataDict, self.meta)['grain']['duration'] = { 337 'numerator': self.meta['grain']['duration'].numerator, 338 'denominator': self.meta['grain']['duration'].denominator} 339 else: 340 raise ValueError("Metadata dict passed to Grain was none!!")
timelabels: Grain.TIMELABELS
def
add_timelabel( self, tag: str, count: int, rate: fractions.Fraction, drop_frame: bool = False) -> None:
length: int
682 @property 683 def length(self) -> int: 684 if hasattr(self.data, "__len__"): 685 return len(cast(Sized, self.data)) 686 elif hasattr(self.data, "__bytes__"): 687 return len(bytes(cast(SupportsBytes, self.data))) 688 elif self.data is None and self._data_fetcher_coroutine is not None: 689 return self._data_fetcher_length 690 else: 691 return 0
Inherited Members
- collections.abc.Sequence
- index
- count
class
Grain.TIMELABEL(collections.abc.Mapping):
549 class TIMELABEL(Mapping): 550 GrainMetadataDict = Dict[str, Any] 551 552 def __init__(self, meta: "Optional[Grain.TIMELABEL.GrainMetadataDict]" = None): 553 if meta is None: 554 meta = {} 555 self.meta = meta 556 if 'tag' not in self.meta: 557 self.meta['tag'] = '' 558 if 'timelabel' not in self.meta: 559 self.meta['timelabel'] = {} 560 if 'frames_since_midnight' not in self.meta['timelabel']: 561 self.meta['timelabel']['frames_since_midnight'] = 0 562 if 'frame_rate_numerator' not in self.meta['timelabel']: 563 self.meta['timelabel']['frame_rate_numerator'] = 0 564 if 'frame_rate_denominator' not in self.meta['timelabel']: 565 self.meta['timelabel']['frame_rate_denominator'] = 1 566 if 'drop_frame' not in self.meta['timelabel']: 567 self.meta['timelabel']['drop_frame'] = False 568 569 def __getitem__(self, key: str) -> Union[str, Dict[str, Union[int, bool]]]: 570 return self.meta[key] 571 572 def __setitem__(self, key: str, value: Union[str, Dict[str, Union[int, bool]]]) -> None: 573 if key not in ['tag', 'timelabel']: 574 raise KeyError 575 self.meta[key] = value 576 577 def __iter__(self) -> Iterator[str]: 578 return self.meta.__iter__() 579 580 def __len__(self) -> int: 581 return 2 582 583 def __eq__(self, other: object) -> bool: 584 return dict(self) == other 585 586 def __ne__(self, other: object) -> bool: 587 return not (self == other) 588 589 @property 590 def tag(self) -> str: 591 return self.meta['tag'] 592 593 @tag.setter 594 def tag(self, value: str) -> None: 595 self.meta['tag'] = value 596 597 @property 598 def count(self) -> int: 599 return self.meta['timelabel']['frames_since_midnight'] 600 601 @count.setter 602 def count(self, value: int) -> None: 603 self.meta['timelabel']['frames_since_midnight'] = int(value) 604 605 @property 606 def rate(self) -> Fraction: 607 return Fraction(self.meta['timelabel']['frame_rate_numerator'], 608 self.meta['timelabel']['frame_rate_denominator']) 609 610 @rate.setter 611 def rate(self, value: RationalTypes) -> None: 612 value = Fraction(value) 613 self.meta['timelabel']['frame_rate_numerator'] = value.numerator 614 self.meta['timelabel']['frame_rate_denominator'] = value.denominator 615 616 @property 617 def drop_frame(self) -> bool: 618 return self.meta['timelabel']['drop_frame'] 619 620 @drop_frame.setter 621 def drop_frame(self, value: bool) -> None: 622 self.meta['timelabel']['drop_frame'] = bool(value)
A Mapping is a generic container for associating key/value pairs.
This class provides concrete generic implementations of all methods except for __getitem__, __iter__, and __len__.
Grain.TIMELABEL(meta: Optional[Dict[str, Any]] = None)
552 def __init__(self, meta: "Optional[Grain.TIMELABEL.GrainMetadataDict]" = None): 553 if meta is None: 554 meta = {} 555 self.meta = meta 556 if 'tag' not in self.meta: 557 self.meta['tag'] = '' 558 if 'timelabel' not in self.meta: 559 self.meta['timelabel'] = {} 560 if 'frames_since_midnight' not in self.meta['timelabel']: 561 self.meta['timelabel']['frames_since_midnight'] = 0 562 if 'frame_rate_numerator' not in self.meta['timelabel']: 563 self.meta['timelabel']['frame_rate_numerator'] = 0 564 if 'frame_rate_denominator' not in self.meta['timelabel']: 565 self.meta['timelabel']['frame_rate_denominator'] = 1 566 if 'drop_frame' not in self.meta['timelabel']: 567 self.meta['timelabel']['drop_frame'] = False
Inherited Members
- collections.abc.Mapping
- get
- keys
- items
- values
class
Grain.TIMELABELS(collections.abc.MutableSequence):
624 class TIMELABELS(MutableSequence): 625 def __init__(self, parent: "Grain"): 626 self.parent = parent 627 628 @overload 629 def __getitem__(self, key: int) -> "Grain.TIMELABEL": ... 630 631 @overload # noqa: F811 632 def __getitem__(self, key: slice) -> "List[Grain.TIMELABEL]": ... 633 634 def __getitem__(self, key): # noqa: F811 635 if 'timelabels' not in self.parent.meta['grain']: 636 raise IndexError("list index out of range") 637 if isinstance(key, int): 638 return Grain.TIMELABEL(self.parent.meta['grain']['timelabels'][key]) 639 else: 640 return [Grain.TIMELABEL(self.parent.meta['grain']['timelabels'][n]) for n in range(len(self))[key]] 641 642 @overload 643 def __setitem__(self, key: int, value: "Grain.TIMELABEL.GrainMetadataDict") -> None: ... 644 645 @overload # noqa: F811 646 def __setitem__(self, key: slice, value: "Iterable[Grain.TIMELABEL.GrainMetadataDict]") -> None: ... 647 648 def __setitem__(self, key, value): # noqa: F811 649 if 'timelabels' not in self.parent.meta['grain']: 650 raise IndexError("list assignment index out of range") 651 if isinstance(key, int): 652 self.parent.meta['grain']['timelabels'][key] = dict(Grain.TIMELABEL(value)) 653 else: 654 values = iter(value) 655 for n in key: 656 self.parent.meta['grain']['timelabels'][n] = dict(Grain.TIMELABEL(next(values))) 657 658 def __delitem__(self, key: Union[int, slice]) -> None: 659 if 'timelabels' not in self.parent.meta['grain']: 660 raise IndexError("list assignment index out of range") 661 662 del self.parent.meta['grain']['timelabels'][key] 663 if len(self.parent.meta['grain']['timelabels']) == 0: 664 del self.parent.meta['grain']['timelabels'] 665 666 def insert(self, key: int, value: "Grain.TIMELABEL.GrainMetadataDict") -> None: 667 if 'timelabels' not in self.parent.meta['grain']: 668 cast(EmptyGrainMetadataDict, self.parent.meta)['grain']['timelabels'] = [] 669 self.parent.meta['grain']['timelabels'].insert(key, cast(TimeLabel, dict(Grain.TIMELABEL(value)))) 670 671 def __len__(self) -> int: 672 if 'timelabels' not in self.parent.meta['grain']: 673 return 0 674 return len(self.parent.meta['grain']['timelabels']) 675 676 def __eq__(self, other: object) -> bool: 677 return list(self) == other 678 679 def __ne__(self, other: object) -> bool: 680 return not (self == other)
All the operations on a read-write sequence.
Concrete subclasses must provide __new__ or __init__, __getitem__, __setitem__, __delitem__, __len__, and insert().
def
insert(self, key: int, value: Dict[str, Any]) -> None:
666 def insert(self, key: int, value: "Grain.TIMELABEL.GrainMetadataDict") -> None: 667 if 'timelabels' not in self.parent.meta['grain']: 668 cast(EmptyGrainMetadataDict, self.parent.meta)['grain']['timelabels'] = [] 669 self.parent.meta['grain']['timelabels'].insert(key, cast(TimeLabel, dict(Grain.TIMELABEL(value))))
S.insert(index, value) -- insert value before index
Inherited Members
- collections.abc.MutableSequence
- append
- clear
- reverse
- extend
- pop
- remove
- collections.abc.Sequence
- index
- count
29class VideoGrain(Grain): 30 """\ 31A class representing a raw video grain. 32 33Any grain can be freely cast to a tuple: 34 35 (meta, data) 36 37where meta is a dictionary containing the grain metadata, and data is the data element described below. 38 39The Grain class provides a number of properties which can be used to access 40parts of the standard grain metadata, and this class inherits these: 41 42meta 43 The meta dictionary object 44 45data 46 Either None or an object which can be cast to bytes by passing it to the bytes 47 constructor and will in of itself respond to the python-level portions of the bytes-like 48 object protocol. It is not guaranteed that this object will always respond correctly to the 49 C buffer-protocol, but it can always be converted into something that will by calling bytes on it. 50 51grain_type 52 A string containing the type of the grain, always "video" 53 54source_id 55 A uuid.UUID object representing the source_id in the grain 56 57flow_id 58 A uuid.UUID object representing the flow_id in the grain 59 60origin_timestamp 61 An mediatimestamp.Timestamp object representing the origin timestamp 62 of this grain. 63 64sync_timestamp 65 An mediatimestamp.Timestamp object representing the sync timestamp 66 of this grain. 67 68creation_timestamp 69 An mediatimestamp.Timestamp object representing the creation timestamp 70 of this grain. 71 72rate 73 A fractions.Fraction object representing the grain rate in grains per second. 74 75duration 76 A fractions.Fraction object representing the grain duration in seconds. 77 78timelabels 79 A list object containing time label data 80 81length 82 The length of the data element or 0 if that is None 83 84The VideoGrain class also provides additional properies 85 86format 87 An enumerated value of type CogFrameFormat 88 89width 90 The video width in pixels 91 92height 93 The video height in pixels 94 95layout 96 An enumerated value of type CogFrameLayout 97 98extension 99 A numeric value indicating the offset from the start of the data array to 100 the start of the actual data, usually 0. 101 102source_aspect_ratio 103 A fractions.Fraction object indicating the video source aspect ratio, or None 104 105pixel_aspect_ratio 106 A fractions.Fraction object indicating the video pixel aspect ratio, or None 107 108components 109 A list-like sequence of VideoGrain.COMPONENT objects 110 """ 111 112 class COMPONENT(Mapping): 113 """ 114A class representing a video component, it may be treated as a dictionary of the form: 115 116 {"stride": <an integer>, 117 "offset": <an integer>, 118 "width": <an integer>, 119 "height": <an integer>, 120 "length": <an integer>} 121 122with additional properties allowing access to the members: 123 124stride 125 The offset in bytes between the first data byte of each line in the data 126 array and the first byte of the next. 127 128offset 129 The offset in bytes from the start of the data array to the first byte of 130 the first line of the data in this component. 131 132width 133 The number of samples per line in this component 134 135height 136 The number of lines in this component 137 138length 139 The total length of the data for this component in bytes 140""" 141 def __init__(self, meta: VideoGrainComponentDict): 142 self.meta = meta 143 144 def __getitem__(self, key: Literal['stride', 'offset', 'width', 'height', 'length']) -> int: 145 return self.meta[key] 146 147 def __setitem__(self, key: Literal['stride', 'offset', 'width', 'height', 'length'], value: int) -> None: 148 self.meta[key] = value 149 150 def __iter__(self) -> Iterator[str]: 151 return self.meta.__iter__() 152 153 def __len__(self) -> int: 154 return self.meta.__len__() 155 156 def __eq__(self, other: object) -> bool: 157 return dict(self) == other 158 159 def __ne__(self, other: object) -> bool: 160 return not (self == other) 161 162 @property 163 def stride(self) -> int: 164 return self.meta['stride'] 165 166 @stride.setter 167 def stride(self, value: int) -> None: 168 self.meta['stride'] = value 169 170 @property 171 def offset(self) -> int: 172 return self.meta['offset'] 173 174 @offset.setter 175 def offset(self, value: int) -> None: 176 self.meta['offset'] = value 177 178 @property 179 def width(self) -> int: 180 return self.meta['width'] 181 182 @width.setter 183 def width(self, value: int) -> None: 184 self.meta['width'] = value 185 186 @property 187 def height(self) -> int: 188 return self.meta['height'] 189 190 @height.setter 191 def height(self, value: int) -> None: 192 self.meta['height'] = value 193 194 @property 195 def length(self) -> int: 196 return self.meta['length'] 197 198 @length.setter 199 def length(self, value: int) -> None: 200 self.meta['length'] = value 201 202 class COMPONENT_LIST(MutableSequence): 203 def __init__(self, parent: "VideoGrain"): 204 self.parent = parent 205 206 @overload 207 def __getitem__(self, key: int) -> "VideoGrain.COMPONENT": ... 208 209 @overload # noqa: F811 210 def __getitem__(self, key: slice) -> "List[VideoGrain.COMPONENT]": ... 211 212 def __getitem__(self, key): # noqa: F811 213 if isinstance(key, int): 214 return type(self.parent).COMPONENT(self.parent.meta['grain']['cog_frame']['components'][key]) 215 else: 216 return [type(self.parent).COMPONENT( 217 self.parent.meta['grain']['cog_frame']['components'][k]) for k in range(len(self))[key]] 218 219 @overload 220 def __setitem__(self, key: int, value: VideoGrainComponentDict) -> None: ... 221 222 @overload # noqa: F811 223 def __setitem__(self, key: slice, value: Iterable[VideoGrainComponentDict]) -> None: ... 224 225 def __setitem__(self, key, value): # noqa: F811 226 if isinstance(key, int): 227 self.parent.meta['grain']['cog_frame']['components'][key] = type(self.parent).COMPONENT(value) 228 else: 229 values = iter(value) 230 for n in range(len(self))[key]: 231 self.parent.meta['grain']['cog_frame']['components'][n] = type(self.parent).COMPONENT(next(values)) 232 233 def __delitem__(self, key: Union[int, slice]) -> None: 234 del self.parent.meta['grain']['cog_frame']['components'][key] 235 236 def insert(self, key: int, value: VideoGrainComponentDict) -> None: 237 self.parent.meta['grain']['cog_frame']['components'].insert( 238 key, type(self.parent).COMPONENT(value)) # type: ignore 239 240 def __len__(self) -> int: 241 return len(self.parent.meta['grain']['cog_frame']['components']) 242 243 def __eq__(self, other: object) -> bool: 244 return list(self) == other 245 246 def __ne__(self, other: object) -> bool: 247 return not (self == other) 248 249 def __init__(self, 250 meta: Optional[VideoGrainMetadataDict] = None, 251 data: Optional[GrainDataParameterType] = None, 252 src_id: Optional[UUID] = None, 253 flow_id: Optional[UUID] = None, 254 origin_timestamp: Optional[SupportsMediaTimestamp] = None, 255 creation_timestamp: Optional[SupportsMediaTimestamp] = None, 256 sync_timestamp: Optional[SupportsMediaTimestamp] = None, 257 rate: Fraction = Fraction(25, 1), 258 duration: Fraction = Fraction(1, 25), 259 cog_frame_format: CogFrameFormat = CogFrameFormat.UNKNOWN, 260 width: int = 1920, 261 height: int = 1080, 262 cog_frame_layout: CogFrameLayout = CogFrameLayout.UNKNOWN): 263 264 if meta is None: 265 if not isinstance(src_id, UUID) and src_id is not None: 266 raise AttributeError(f"src_id: Seen type {type(src_id)}, expected UUID.") 267 if not isinstance(flow_id, UUID) and flow_id is not None: 268 raise AttributeError(f"flow_id: Seen type {type(flow_id)}, expected UUID.") 269 270 if creation_timestamp is None: 271 creation_timestamp = Timestamp.get_time() 272 if origin_timestamp is None: 273 origin_timestamp = creation_timestamp 274 if sync_timestamp is None: 275 sync_timestamp = origin_timestamp 276 meta = { 277 "@_ns": "urn:x-ipstudio:ns:0.1", 278 "grain": { 279 'grain_type': "video", 280 'source_id': str(src_id), 281 'flow_id': str(flow_id), 282 'origin_timestamp': str(mediatimestamp(origin_timestamp)), 283 'sync_timestamp': str(mediatimestamp(sync_timestamp)), 284 'creation_timestamp': str(mediatimestamp(creation_timestamp)), 285 'rate': { 286 'numerator': Fraction(rate).numerator, 287 'denominator': Fraction(rate).denominator, 288 }, 289 'duration': { 290 'numerator': Fraction(duration).numerator, 291 'denominator': Fraction(duration).denominator, 292 }, 293 'cog_frame': { 294 "format": cog_frame_format, 295 "width": width, 296 "height": height, 297 "layout": cog_frame_layout, 298 "extension": 0, 299 "components": [] 300 } 301 }, 302 } 303 304 def size_for_format(fmt: CogFrameFormat, w: int, h: int) -> int: 305 if ((fmt >> 8) & 0x1) == 0x00: # Cog frame is not packed 306 h_shift = (fmt & 0x01) 307 v_shift = ((fmt >> 1) & 0x01) 308 depth = (fmt & 0xc) 309 if depth == 0: 310 bpv = 1 311 elif depth == 4: 312 bpv = 2 313 else: 314 bpv = 4 315 return (w*h + 2*((w*h) >> (h_shift + v_shift)))*bpv 316 else: 317 if fmt in (CogFrameFormat.YUYV, CogFrameFormat.UYVY, CogFrameFormat.AYUV): 318 return w*h*2 319 elif fmt in (CogFrameFormat.RGBx, 320 CogFrameFormat.RGBA, 321 CogFrameFormat.xRGB, 322 CogFrameFormat.ARGB, 323 CogFrameFormat.BGRx, 324 CogFrameFormat.BGRA, 325 CogFrameFormat.xBGR, 326 CogFrameFormat.ABGR): 327 return w*h*4 328 elif fmt == CogFrameFormat.RGB: 329 return w*h*3 330 elif fmt == CogFrameFormat.v210: 331 return h*(((w + 47) // 48) * 128) 332 elif fmt == CogFrameFormat.v216: 333 return w*h*4 334 else: 335 return 0 336 if data is None: 337 size = size_for_format(cog_frame_format, width, height) 338 data = bytearray(size) 339 340 def components_for_format(fmt: CogFrameFormat, w: int, h: int) -> List[VideoGrainComponentDict]: 341 components: List[VideoGrainComponentDict] = [] 342 if ((fmt >> 8) & 0x1) == 0x00: # Cog frame is not packed 343 h_shift = (fmt & 0x01) 344 v_shift = ((fmt >> 1) & 0x01) 345 depth = (fmt & 0xc) 346 if depth == 0: 347 bpv = 1 348 elif depth == 4: 349 bpv = 2 350 else: 351 bpv = 4 352 offset = 0 353 components.append({ 354 'stride': w*bpv, 355 'offset': offset, 356 'width': w, 357 'height': h, 358 'length': w*h*bpv 359 }) 360 offset += w*h*bpv 361 components.append({ 362 'stride': (w >> h_shift)*bpv, 363 'offset': offset, 364 'width': w >> h_shift, 365 'height': h >> v_shift, 366 'length': ((w*h) >> (h_shift + v_shift))*bpv 367 }) 368 offset += ((w*h) >> (h_shift + v_shift))*bpv 369 components.append({ 370 'stride': (w >> h_shift)*bpv, 371 'offset': offset, 372 'width': w >> h_shift, 373 'height': h >> v_shift, 374 'length': ((w*h) >> (h_shift + v_shift))*bpv 375 }) 376 offset += ((w*h) >> (h_shift + v_shift))*bpv 377 else: 378 if fmt in (CogFrameFormat.YUYV, CogFrameFormat.UYVY, CogFrameFormat.AYUV): 379 components.append({ 380 'stride': w*2, 381 'offset': 0, 382 'width': w, 383 'height': h, 384 'length': h*w*2 385 }) 386 elif fmt in (CogFrameFormat.RGBx, 387 CogFrameFormat.RGBA, 388 CogFrameFormat.xRGB, 389 CogFrameFormat.ARGB, 390 CogFrameFormat.BGRx, 391 CogFrameFormat.BGRA, 392 CogFrameFormat.xBGR, 393 CogFrameFormat.ABGR): 394 components.append({ 395 'stride': w*4, 396 'offset': 0, 397 'width': w, 398 'height': h, 399 'length': h*w*4 400 }) 401 elif fmt == CogFrameFormat.RGB: 402 components.append({ 403 'stride': w*3, 404 'offset': 0, 405 'width': w, 406 'height': h, 407 'length': h*w*3 408 }) 409 elif fmt == CogFrameFormat.v210: 410 components.append({ 411 'stride': (((w + 47) // 48) * 128), 412 'offset': 0, 413 'width': w, 414 'height': h, 415 'length': h*(((w + 47) // 48) * 128) 416 }) 417 elif fmt == CogFrameFormat.v216: 418 components.append({ 419 'stride': w*4, 420 'offset': 0, 421 'width': w, 422 'height': h, 423 'length': h*w*4 424 }) 425 return components 426 427 if ("cog_frame" in meta['grain'] and 428 ("components" not in meta['grain']['cog_frame'] or 429 len(meta['grain']['cog_frame']['components']) == 0)): 430 meta['grain']['cog_frame']['components'] = components_for_format(cog_frame_format, width, height) 431 432 super().__init__(meta=meta, data=data) 433 self.meta: VideoGrainMetadataDict 434 435 self._factory = "VideoGrain" 436 self.meta['grain']['grain_type'] = 'video' 437 if 'cog_frame' not in self.meta['grain']: 438 self.meta['grain']['cog_frame'] = { 439 'format': int(CogFrameFormat.UNKNOWN), 440 'width': 0, 441 'height': 0, 442 'layout': int(CogFrameLayout.UNKNOWN), 443 'extension': 0, 444 'components': [] 445 } 446 self.meta['grain']['cog_frame']['format'] = int(self.meta['grain']['cog_frame']['format']) 447 self.meta['grain']['cog_frame']['layout'] = int(self.meta['grain']['cog_frame']['layout']) 448 self.components = VideoGrain.COMPONENT_LIST(self) 449 450 @property 451 @deprecated(version="4.0.0", reason="Referencing `format` directly is deprecated in favour of `cog_frame_format`") 452 def format(self) -> CogFrameFormat: 453 return CogFrameFormat(self.meta['grain']['cog_frame']['format']) 454 455 @format.setter 456 @deprecated(version="4.0.0", reason="Referencing `format` directly is deprecated in favour of `cog_frame_format`") 457 def format(self, value: CogFrameFormat) -> None: 458 self.meta['grain']['cog_frame']['format'] = int(value) 459 460 @property 461 def cog_frame_format(self) -> CogFrameFormat: 462 return CogFrameFormat(self.meta['grain']['cog_frame']['format']) 463 464 @cog_frame_format.setter 465 def cog_frame_format(self, value: CogFrameFormat) -> None: 466 self.meta['grain']['cog_frame']['format'] = int(value) 467 468 @property 469 def width(self) -> int: 470 return self.meta['grain']['cog_frame']['width'] 471 472 @width.setter 473 def width(self, value: int) -> None: 474 self.meta['grain']['cog_frame']['width'] = value 475 476 @property 477 def height(self) -> int: 478 return self.meta['grain']['cog_frame']['height'] 479 480 @height.setter 481 def height(self, value: int) -> None: 482 self.meta['grain']['cog_frame']['height'] = value 483 484 @property 485 @deprecated(version="4.0.0", reason="Referencing `layout` directly is deprecated in favour of `cog_frame_layout`") 486 def layout(self) -> CogFrameLayout: 487 return CogFrameLayout(self.meta['grain']['cog_frame']['layout']) 488 489 @layout.setter 490 @deprecated(version="4.0.0", reason="Referencing `layout` directly is deprecated in favour of `cog_frame_layout`") 491 def layout(self, value: CogFrameLayout) -> None: 492 self.meta['grain']['cog_frame']['layout'] = int(value) 493 494 @property 495 def cog_frame_layout(self) -> CogFrameLayout: 496 return CogFrameLayout(self.meta['grain']['cog_frame']['layout']) 497 498 @cog_frame_layout.setter 499 def cog_frame_layout(self, value: CogFrameLayout) -> None: 500 self.meta['grain']['cog_frame']['layout'] = int(value) 501 502 @property 503 def extension(self) -> int: 504 return self.meta['grain']['cog_frame']['extension'] 505 506 @extension.setter 507 def extension(self, value: int) -> None: 508 self.meta['grain']['cog_frame']['extension'] = value 509 510 @property 511 def source_aspect_ratio(self) -> Optional[Fraction]: 512 if 'source_aspect_ratio' in self.meta['grain']['cog_frame']: 513 return Fraction(cast(FractionDict, self.meta['grain']['cog_frame']['source_aspect_ratio'])['numerator'], 514 cast(FractionDict, self.meta['grain']['cog_frame']['source_aspect_ratio'])['denominator']) 515 else: 516 return None 517 518 @source_aspect_ratio.setter 519 def source_aspect_ratio(self, value: RationalTypes) -> None: 520 value = Fraction(value) 521 self.meta['grain']['cog_frame']['source_aspect_ratio'] = {'numerator': value.numerator, 522 'denominator': value.denominator} 523 524 @property 525 def pixel_aspect_ratio(self) -> Optional[Fraction]: 526 if 'pixel_aspect_ratio' in self.meta['grain']['cog_frame']: 527 return Fraction(cast(FractionDict, self.meta['grain']['cog_frame']['pixel_aspect_ratio'])['numerator'], 528 cast(FractionDict, self.meta['grain']['cog_frame']['pixel_aspect_ratio'])['denominator']) 529 else: 530 return None 531 532 @pixel_aspect_ratio.setter 533 def pixel_aspect_ratio(self, value: RationalTypes) -> None: 534 value = Fraction(value) 535 self.meta['grain']['cog_frame']['pixel_aspect_ratio'] = {'numerator': value.numerator, 536 'denominator': value.denominator} 537 538 @property 539 def expected_length(self) -> int: 540 length = 0 541 for component in self.components: 542 if component.offset + component.length > length: 543 length = component.offset + component.length 544 return length 545 546 @property 547 def media_rate(self) -> Optional[Fraction]: 548 if self.rate: 549 return self.rate 550 else: 551 return None
A class representing a raw video grain.
Any grain can be freely cast to a tuple:
(meta, data)
where meta is a dictionary containing the grain metadata, and data is the data element described below.
The Grain class provides a number of properties which can be used to access parts of the standard grain metadata, and this class inherits these:
meta The meta dictionary object
data Either None or an object which can be cast to bytes by passing it to the bytes constructor and will in of itself respond to the python-level portions of the bytes-like object protocol. It is not guaranteed that this object will always respond correctly to the C buffer-protocol, but it can always be converted into something that will by calling bytes on it.
grain_type A string containing the type of the grain, always "video"
source_id A uuid.UUID object representing the source_id in the grain
flow_id A uuid.UUID object representing the flow_id in the grain
origin_timestamp An mediatimestamp.Timestamp object representing the origin timestamp of this grain.
sync_timestamp An mediatimestamp.Timestamp object representing the sync timestamp of this grain.
creation_timestamp An mediatimestamp.Timestamp object representing the creation timestamp of this grain.
rate A fractions.Fraction object representing the grain rate in grains per second.
duration A fractions.Fraction object representing the grain duration in seconds.
timelabels A list object containing time label data
length The length of the data element or 0 if that is None
The VideoGrain class also provides additional properies
format An enumerated value of type CogFrameFormat
width The video width in pixels
height The video height in pixels
layout An enumerated value of type CogFrameLayout
extension A numeric value indicating the offset from the start of the data array to the start of the actual data, usually 0.
source_aspect_ratio A fractions.Fraction object indicating the video source aspect ratio, or None
pixel_aspect_ratio A fractions.Fraction object indicating the video pixel aspect ratio, or None
components A list-like sequence of VideoGrain.COMPONENT objects
VideoGrain( meta: Optional[mediagrains.typing.VideoGrainMetadataDict] = None, data: Union[SupportsBytes, bytes, numpy.ndarray, Awaitable[Union[SupportsBytes, bytes, numpy.ndarray, NoneType]], NoneType] = None, src_id: Optional[uuid.UUID] = None, flow_id: Optional[uuid.UUID] = None, origin_timestamp: Optional[mediatimestamp.immutable.timestamp.SupportsMediaTimestamp] = None, creation_timestamp: Optional[mediatimestamp.immutable.timestamp.SupportsMediaTimestamp] = None, sync_timestamp: Optional[mediatimestamp.immutable.timestamp.SupportsMediaTimestamp] = None, rate: fractions.Fraction = Fraction(25, 1), duration: fractions.Fraction = Fraction(1, 25), cog_frame_format: mediagrains.cogenums.CogFrameFormat = <CogFrameFormat.UNKNOWN: 4294967294>, width: int = 1920, height: int = 1080, cog_frame_layout: mediagrains.cogenums.CogFrameLayout = <CogFrameLayout.UNKNOWN: 4294967294>)
249 def __init__(self, 250 meta: Optional[VideoGrainMetadataDict] = None, 251 data: Optional[GrainDataParameterType] = None, 252 src_id: Optional[UUID] = None, 253 flow_id: Optional[UUID] = None, 254 origin_timestamp: Optional[SupportsMediaTimestamp] = None, 255 creation_timestamp: Optional[SupportsMediaTimestamp] = None, 256 sync_timestamp: Optional[SupportsMediaTimestamp] = None, 257 rate: Fraction = Fraction(25, 1), 258 duration: Fraction = Fraction(1, 25), 259 cog_frame_format: CogFrameFormat = CogFrameFormat.UNKNOWN, 260 width: int = 1920, 261 height: int = 1080, 262 cog_frame_layout: CogFrameLayout = CogFrameLayout.UNKNOWN): 263 264 if meta is None: 265 if not isinstance(src_id, UUID) and src_id is not None: 266 raise AttributeError(f"src_id: Seen type {type(src_id)}, expected UUID.") 267 if not isinstance(flow_id, UUID) and flow_id is not None: 268 raise AttributeError(f"flow_id: Seen type {type(flow_id)}, expected UUID.") 269 270 if creation_timestamp is None: 271 creation_timestamp = Timestamp.get_time() 272 if origin_timestamp is None: 273 origin_timestamp = creation_timestamp 274 if sync_timestamp is None: 275 sync_timestamp = origin_timestamp 276 meta = { 277 "@_ns": "urn:x-ipstudio:ns:0.1", 278 "grain": { 279 'grain_type': "video", 280 'source_id': str(src_id), 281 'flow_id': str(flow_id), 282 'origin_timestamp': str(mediatimestamp(origin_timestamp)), 283 'sync_timestamp': str(mediatimestamp(sync_timestamp)), 284 'creation_timestamp': str(mediatimestamp(creation_timestamp)), 285 'rate': { 286 'numerator': Fraction(rate).numerator, 287 'denominator': Fraction(rate).denominator, 288 }, 289 'duration': { 290 'numerator': Fraction(duration).numerator, 291 'denominator': Fraction(duration).denominator, 292 }, 293 'cog_frame': { 294 "format": cog_frame_format, 295 "width": width, 296 "height": height, 297 "layout": cog_frame_layout, 298 "extension": 0, 299 "components": [] 300 } 301 }, 302 } 303 304 def size_for_format(fmt: CogFrameFormat, w: int, h: int) -> int: 305 if ((fmt >> 8) & 0x1) == 0x00: # Cog frame is not packed 306 h_shift = (fmt & 0x01) 307 v_shift = ((fmt >> 1) & 0x01) 308 depth = (fmt & 0xc) 309 if depth == 0: 310 bpv = 1 311 elif depth == 4: 312 bpv = 2 313 else: 314 bpv = 4 315 return (w*h + 2*((w*h) >> (h_shift + v_shift)))*bpv 316 else: 317 if fmt in (CogFrameFormat.YUYV, CogFrameFormat.UYVY, CogFrameFormat.AYUV): 318 return w*h*2 319 elif fmt in (CogFrameFormat.RGBx, 320 CogFrameFormat.RGBA, 321 CogFrameFormat.xRGB, 322 CogFrameFormat.ARGB, 323 CogFrameFormat.BGRx, 324 CogFrameFormat.BGRA, 325 CogFrameFormat.xBGR, 326 CogFrameFormat.ABGR): 327 return w*h*4 328 elif fmt == CogFrameFormat.RGB: 329 return w*h*3 330 elif fmt == CogFrameFormat.v210: 331 return h*(((w + 47) // 48) * 128) 332 elif fmt == CogFrameFormat.v216: 333 return w*h*4 334 else: 335 return 0 336 if data is None: 337 size = size_for_format(cog_frame_format, width, height) 338 data = bytearray(size) 339 340 def components_for_format(fmt: CogFrameFormat, w: int, h: int) -> List[VideoGrainComponentDict]: 341 components: List[VideoGrainComponentDict] = [] 342 if ((fmt >> 8) & 0x1) == 0x00: # Cog frame is not packed 343 h_shift = (fmt & 0x01) 344 v_shift = ((fmt >> 1) & 0x01) 345 depth = (fmt & 0xc) 346 if depth == 0: 347 bpv = 1 348 elif depth == 4: 349 bpv = 2 350 else: 351 bpv = 4 352 offset = 0 353 components.append({ 354 'stride': w*bpv, 355 'offset': offset, 356 'width': w, 357 'height': h, 358 'length': w*h*bpv 359 }) 360 offset += w*h*bpv 361 components.append({ 362 'stride': (w >> h_shift)*bpv, 363 'offset': offset, 364 'width': w >> h_shift, 365 'height': h >> v_shift, 366 'length': ((w*h) >> (h_shift + v_shift))*bpv 367 }) 368 offset += ((w*h) >> (h_shift + v_shift))*bpv 369 components.append({ 370 'stride': (w >> h_shift)*bpv, 371 'offset': offset, 372 'width': w >> h_shift, 373 'height': h >> v_shift, 374 'length': ((w*h) >> (h_shift + v_shift))*bpv 375 }) 376 offset += ((w*h) >> (h_shift + v_shift))*bpv 377 else: 378 if fmt in (CogFrameFormat.YUYV, CogFrameFormat.UYVY, CogFrameFormat.AYUV): 379 components.append({ 380 'stride': w*2, 381 'offset': 0, 382 'width': w, 383 'height': h, 384 'length': h*w*2 385 }) 386 elif fmt in (CogFrameFormat.RGBx, 387 CogFrameFormat.RGBA, 388 CogFrameFormat.xRGB, 389 CogFrameFormat.ARGB, 390 CogFrameFormat.BGRx, 391 CogFrameFormat.BGRA, 392 CogFrameFormat.xBGR, 393 CogFrameFormat.ABGR): 394 components.append({ 395 'stride': w*4, 396 'offset': 0, 397 'width': w, 398 'height': h, 399 'length': h*w*4 400 }) 401 elif fmt == CogFrameFormat.RGB: 402 components.append({ 403 'stride': w*3, 404 'offset': 0, 405 'width': w, 406 'height': h, 407 'length': h*w*3 408 }) 409 elif fmt == CogFrameFormat.v210: 410 components.append({ 411 'stride': (((w + 47) // 48) * 128), 412 'offset': 0, 413 'width': w, 414 'height': h, 415 'length': h*(((w + 47) // 48) * 128) 416 }) 417 elif fmt == CogFrameFormat.v216: 418 components.append({ 419 'stride': w*4, 420 'offset': 0, 421 'width': w, 422 'height': h, 423 'length': h*w*4 424 }) 425 return components 426 427 if ("cog_frame" in meta['grain'] and 428 ("components" not in meta['grain']['cog_frame'] or 429 len(meta['grain']['cog_frame']['components']) == 0)): 430 meta['grain']['cog_frame']['components'] = components_for_format(cog_frame_format, width, height) 431 432 super().__init__(meta=meta, data=data) 433 self.meta: VideoGrainMetadataDict 434 435 self._factory = "VideoGrain" 436 self.meta['grain']['grain_type'] = 'video' 437 if 'cog_frame' not in self.meta['grain']: 438 self.meta['grain']['cog_frame'] = { 439 'format': int(CogFrameFormat.UNKNOWN), 440 'width': 0, 441 'height': 0, 442 'layout': int(CogFrameLayout.UNKNOWN), 443 'extension': 0, 444 'components': [] 445 } 446 self.meta['grain']['cog_frame']['format'] = int(self.meta['grain']['cog_frame']['format']) 447 self.meta['grain']['cog_frame']['layout'] = int(self.meta['grain']['cog_frame']['layout']) 448 self.components = VideoGrain.COMPONENT_LIST(self)
source_aspect_ratio: Optional[fractions.Fraction]
510 @property 511 def source_aspect_ratio(self) -> Optional[Fraction]: 512 if 'source_aspect_ratio' in self.meta['grain']['cog_frame']: 513 return Fraction(cast(FractionDict, self.meta['grain']['cog_frame']['source_aspect_ratio'])['numerator'], 514 cast(FractionDict, self.meta['grain']['cog_frame']['source_aspect_ratio'])['denominator']) 515 else: 516 return None
pixel_aspect_ratio: Optional[fractions.Fraction]
524 @property 525 def pixel_aspect_ratio(self) -> Optional[Fraction]: 526 if 'pixel_aspect_ratio' in self.meta['grain']['cog_frame']: 527 return Fraction(cast(FractionDict, self.meta['grain']['cog_frame']['pixel_aspect_ratio'])['numerator'], 528 cast(FractionDict, self.meta['grain']['cog_frame']['pixel_aspect_ratio'])['denominator']) 529 else: 530 return None
Inherited Members
- Grain
- has_data
- data
- grain_type
- source_id
- src_id
- flow_id
- origin_timestamp
- final_origin_timestamp
- origin_timerange
- presentation_origin_timestamp
- final_presentation_origin_timestamp
- presentation_origin_timerange
- normalise_time
- sync_timestamp
- creation_timestamp
- rate
- duration
- timelabels
- add_timelabel
- TIMELABEL
- TIMELABELS
- length
- collections.abc.Sequence
- index
- count
class
VideoGrain.COMPONENT(collections.abc.Mapping):
112 class COMPONENT(Mapping): 113 """ 114A class representing a video component, it may be treated as a dictionary of the form: 115 116 {"stride": <an integer>, 117 "offset": <an integer>, 118 "width": <an integer>, 119 "height": <an integer>, 120 "length": <an integer>} 121 122with additional properties allowing access to the members: 123 124stride 125 The offset in bytes between the first data byte of each line in the data 126 array and the first byte of the next. 127 128offset 129 The offset in bytes from the start of the data array to the first byte of 130 the first line of the data in this component. 131 132width 133 The number of samples per line in this component 134 135height 136 The number of lines in this component 137 138length 139 The total length of the data for this component in bytes 140""" 141 def __init__(self, meta: VideoGrainComponentDict): 142 self.meta = meta 143 144 def __getitem__(self, key: Literal['stride', 'offset', 'width', 'height', 'length']) -> int: 145 return self.meta[key] 146 147 def __setitem__(self, key: Literal['stride', 'offset', 'width', 'height', 'length'], value: int) -> None: 148 self.meta[key] = value 149 150 def __iter__(self) -> Iterator[str]: 151 return self.meta.__iter__() 152 153 def __len__(self) -> int: 154 return self.meta.__len__() 155 156 def __eq__(self, other: object) -> bool: 157 return dict(self) == other 158 159 def __ne__(self, other: object) -> bool: 160 return not (self == other) 161 162 @property 163 def stride(self) -> int: 164 return self.meta['stride'] 165 166 @stride.setter 167 def stride(self, value: int) -> None: 168 self.meta['stride'] = value 169 170 @property 171 def offset(self) -> int: 172 return self.meta['offset'] 173 174 @offset.setter 175 def offset(self, value: int) -> None: 176 self.meta['offset'] = value 177 178 @property 179 def width(self) -> int: 180 return self.meta['width'] 181 182 @width.setter 183 def width(self, value: int) -> None: 184 self.meta['width'] = value 185 186 @property 187 def height(self) -> int: 188 return self.meta['height'] 189 190 @height.setter 191 def height(self, value: int) -> None: 192 self.meta['height'] = value 193 194 @property 195 def length(self) -> int: 196 return self.meta['length'] 197 198 @length.setter 199 def length(self, value: int) -> None: 200 self.meta['length'] = value
A class representing a video component, it may be treated as a dictionary of the form:
{"stride": <an integer>,
"offset": <an integer>,
"width": <an integer>,
"height": <an integer>,
"length": <an integer>}
with additional properties allowing access to the members:
stride The offset in bytes between the first data byte of each line in the data array and the first byte of the next.
offset The offset in bytes from the start of the data array to the first byte of the first line of the data in this component.
width The number of samples per line in this component
height The number of lines in this component
length The total length of the data for this component in bytes
Inherited Members
- collections.abc.Mapping
- get
- keys
- items
- values
class
VideoGrain.COMPONENT_LIST(collections.abc.MutableSequence):
202 class COMPONENT_LIST(MutableSequence): 203 def __init__(self, parent: "VideoGrain"): 204 self.parent = parent 205 206 @overload 207 def __getitem__(self, key: int) -> "VideoGrain.COMPONENT": ... 208 209 @overload # noqa: F811 210 def __getitem__(self, key: slice) -> "List[VideoGrain.COMPONENT]": ... 211 212 def __getitem__(self, key): # noqa: F811 213 if isinstance(key, int): 214 return type(self.parent).COMPONENT(self.parent.meta['grain']['cog_frame']['components'][key]) 215 else: 216 return [type(self.parent).COMPONENT( 217 self.parent.meta['grain']['cog_frame']['components'][k]) for k in range(len(self))[key]] 218 219 @overload 220 def __setitem__(self, key: int, value: VideoGrainComponentDict) -> None: ... 221 222 @overload # noqa: F811 223 def __setitem__(self, key: slice, value: Iterable[VideoGrainComponentDict]) -> None: ... 224 225 def __setitem__(self, key, value): # noqa: F811 226 if isinstance(key, int): 227 self.parent.meta['grain']['cog_frame']['components'][key] = type(self.parent).COMPONENT(value) 228 else: 229 values = iter(value) 230 for n in range(len(self))[key]: 231 self.parent.meta['grain']['cog_frame']['components'][n] = type(self.parent).COMPONENT(next(values)) 232 233 def __delitem__(self, key: Union[int, slice]) -> None: 234 del self.parent.meta['grain']['cog_frame']['components'][key] 235 236 def insert(self, key: int, value: VideoGrainComponentDict) -> None: 237 self.parent.meta['grain']['cog_frame']['components'].insert( 238 key, type(self.parent).COMPONENT(value)) # type: ignore 239 240 def __len__(self) -> int: 241 return len(self.parent.meta['grain']['cog_frame']['components']) 242 243 def __eq__(self, other: object) -> bool: 244 return list(self) == other 245 246 def __ne__(self, other: object) -> bool: 247 return not (self == other)
All the operations on a read-write sequence.
Concrete subclasses must provide __new__ or __init__, __getitem__, __setitem__, __delitem__, __len__, and insert().
VideoGrain.COMPONENT_LIST(parent: VideoGrain)
def
insert( self, key: int, value: mediagrains.typing.VideoGrainComponentDict) -> None:
236 def insert(self, key: int, value: VideoGrainComponentDict) -> None: 237 self.parent.meta['grain']['cog_frame']['components'].insert( 238 key, type(self.parent).COMPONENT(value)) # type: ignore
S.insert(index, value) -- insert value before index
Inherited Members
- collections.abc.MutableSequence
- append
- clear
- reverse
- extend
- pop
- remove
- collections.abc.Sequence
- index
- count
25class CodedVideoGrain(Grain): 26 """\ 27A class representing a coded video grain. 28 29Any grain can be freely cast to a tuple: 30 31 (meta, data) 32 33where meta is a dictionary containing the grain metadata, and data is the data element described below. 34 35The Grain class provides a number of properties which can be used to access 36parts of the standard grain metadata, and this class inherits these: 37 38meta 39 The meta dictionary object 40 41data 42 Either None or an object which can be cast to bytes by passing it to the bytes 43 constructor and will in of itself respond to the python-level portions of the bytes-like 44 object protocol. It is not guaranteed that this object will always respond correctly to the 45 C buffer-protocol, but it can always be converted into something that will by calling bytes on it. 46 47grain_type 48 A string containing the type of the grain, always "coded_video" 49 50source_id 51 A uuid.UUID object representing the source_id in the grain 52 53flow_id 54 A uuid.UUID object representing the flow_id in the grain 55 56origin_timestamp 57 An mediatimestamp.Timestamp object representing the origin timestamp 58 of this grain. 59 60sync_timestamp 61 An mediatimestamp.Timestamp object representing the sync timestamp 62 of this grain. 63 64creation_timestamp 65 An mediatimestamp.Timestamp object representing the creation timestamp 66 of this grain. 67 68rate 69 A fractions.Fraction object representing the grain rate in grains per second. 70 71duration 72 A fractions.Fraction object representing the grain duration in seconds. 73 74timelabels 75 A list object containing time label data 76 77length 78 The length of the data element or 0 if that is None 79 80The CodedVideoGrain class also provides additional properies 81 82format 83 An enumerated value of type CogFrameFormat 84 85layout 86 An enumerated value of type CogFrameLayout 87 88origin_width 89 The original video width in pixels 90 91origin_height 92 The original video height in pixels 93 94coded_width 95 The coded video width in pixels 96 97coded_height 98 The coded video height in pixels 99 100temporal_offset 101 An optional signed integer value indicating the offset from the origin timestamp of 102 this grain to the expected presentation time of the picture in frames. 103 104unit_offsets 105 A list-like object containing integer offsets of coded units within the 106 data array. 107""" 108 def __init__(self, 109 meta: Optional[CodedVideoGrainMetadataDict] = None, 110 data: Optional[GrainDataParameterType] = None, 111 src_id: Optional[UUID] = None, 112 flow_id: Optional[UUID] = None, 113 origin_timestamp: Optional[SupportsMediaTimestamp] = None, 114 creation_timestamp: Optional[SupportsMediaTimestamp] = None, 115 sync_timestamp: Optional[SupportsMediaTimestamp] = None, 116 rate: Fraction = Fraction(25, 1), 117 duration: Fraction = Fraction(1, 25), 118 cog_frame_format: CogFrameFormat = CogFrameFormat.UNKNOWN, 119 origin_width: int = 1920, 120 origin_height: int = 1080, 121 coded_width: Optional[int] = None, 122 coded_height: Optional[int] = None, 123 is_key_frame: Optional[bool] = None, 124 temporal_offset: Optional[int] = None, 125 length: Optional[int] = None, 126 cog_frame_layout: CogFrameLayout = CogFrameLayout.UNKNOWN, 127 unit_offsets: Optional[List[int]] = None): 128 if coded_width is None: 129 coded_width = origin_width 130 if coded_height is None: 131 coded_height = origin_height 132 133 if length is None: 134 if data is not None and hasattr(data, "__len__"): 135 length = len(cast(Sized, data)) 136 else: 137 length = 0 138 139 if meta is None: 140 if not isinstance(src_id, UUID) and src_id is not None: 141 raise AttributeError(f"src_id: Seen type {type(src_id)}, expected UUID.") 142 if not isinstance(flow_id, UUID) and flow_id is not None: 143 raise AttributeError(f"flow_id: Seen type {type(flow_id)}, expected UUID.") 144 145 if creation_timestamp is None: 146 creation_timestamp = Timestamp.get_time() 147 if origin_timestamp is None: 148 origin_timestamp = creation_timestamp 149 if sync_timestamp is None: 150 sync_timestamp = origin_timestamp 151 meta = { 152 "@_ns": "urn:x-ipstudio:ns:0.1", 153 "grain": { 154 "grain_type": "coded_video", 155 "source_id": str(src_id), 156 "flow_id": str(flow_id), 157 "origin_timestamp": str(mediatimestamp(origin_timestamp)), 158 "sync_timestamp": str(mediatimestamp(sync_timestamp)), 159 "creation_timestamp": str(mediatimestamp(creation_timestamp)), 160 "rate": { 161 "numerator": Fraction(rate).numerator, 162 "denominator": Fraction(rate).denominator, 163 }, 164 "duration": { 165 "numerator": Fraction(duration).numerator, 166 "denominator": Fraction(duration).denominator, 167 }, 168 "cog_coded_frame": { 169 "format": cog_frame_format, 170 "origin_width": origin_width, 171 "origin_height": origin_height, 172 "coded_width": coded_width, 173 "coded_height": coded_height, 174 "layout": cog_frame_layout 175 } 176 }, 177 } 178 if is_key_frame is not None: 179 meta["grain"]["cog_coded_frame"]["is_key_frame"] = is_key_frame 180 if temporal_offset is not None: 181 meta["grain"]["cog_coded_frame"]["temporal_offset"] = temporal_offset 182 183 if data is None: 184 data = bytearray(length) 185 186 if "grain" in meta and "cog_coded_frame" in meta['grain'] and unit_offsets is not None: 187 meta['grain']['cog_coded_frame']['unit_offsets'] = unit_offsets 188 189 super().__init__(meta, data) 190 self.meta: CodedVideoGrainMetadataDict 191 192 self._factory = "CodedVideoGrain" 193 self.meta['grain']['grain_type'] = 'coded_video' 194 if 'cog_coded_frame' not in self.meta['grain']: 195 self.meta['grain']['cog_coded_frame'] = {} # type: ignore 196 if 'format' not in self.meta['grain']['cog_coded_frame']: 197 self.meta['grain']['cog_coded_frame']['format'] = int(CogFrameFormat.UNKNOWN) 198 if 'layout' not in self.meta['grain']['cog_coded_frame']: 199 self.meta['grain']['cog_coded_frame']['layout'] = int(CogFrameLayout.UNKNOWN) 200 if 'origin_width' not in self.meta['grain']['cog_coded_frame']: 201 self.meta['grain']['cog_coded_frame']['origin_width'] = 0 202 if 'origin_height' not in self.meta['grain']['cog_coded_frame']: 203 self.meta['grain']['cog_coded_frame']['origin_height'] = 0 204 if 'coded_width' not in self.meta['grain']['cog_coded_frame']: 205 self.meta['grain']['cog_coded_frame']['coded_width'] = 0 206 if 'coded_height' not in self.meta['grain']['cog_coded_frame']: 207 self.meta['grain']['cog_coded_frame']['coded_height'] = 0 208 if 'length' not in self.meta['grain']['cog_coded_frame']: 209 self.meta['grain']['cog_coded_frame']['length'] = 0 210 self.meta['grain']['cog_coded_frame']['format'] = int(self.meta['grain']['cog_coded_frame']['format']) 211 self.meta['grain']['cog_coded_frame']['layout'] = int(self.meta['grain']['cog_coded_frame']['layout']) 212 213 @property 214 def format(self) -> CogFrameFormat: 215 return CogFrameFormat(self.meta['grain']['cog_coded_frame']['format']) 216 217 @format.setter 218 def format(self, value: CogFrameFormat) -> None: 219 self.meta['grain']['cog_coded_frame']['format'] = int(value) 220 221 @property 222 def cog_frame_format(self) -> CogFrameFormat: 223 return CogFrameFormat(self.meta['grain']['cog_coded_frame']['format']) 224 225 @cog_frame_format.setter 226 def cog_frame_format(self, value: CogFrameFormat) -> None: 227 self.meta['grain']['cog_coded_frame']['format'] = int(value) 228 229 @property 230 def layout(self) -> CogFrameLayout: 231 return CogFrameLayout(self.meta['grain']['cog_coded_frame']['layout']) 232 233 @layout.setter 234 def layout(self, value: CogFrameLayout) -> None: 235 self.meta['grain']['cog_coded_frame']['layout'] = int(value) 236 237 @property 238 def cog_frame_layout(self) -> CogFrameLayout: 239 return CogFrameLayout(self.meta['grain']['cog_coded_frame']['layout']) 240 241 @cog_frame_layout.setter 242 def cog_frame_layout(self, value: CogFrameLayout) -> None: 243 self.meta['grain']['cog_coded_frame']['layout'] = int(value) 244 245 @property 246 def origin_width(self) -> int: 247 return self.meta['grain']['cog_coded_frame']['origin_width'] 248 249 @origin_width.setter 250 def origin_width(self, value: int) -> None: 251 self.meta['grain']['cog_coded_frame']['origin_width'] = value 252 253 @property 254 def origin_height(self) -> int: 255 return self.meta['grain']['cog_coded_frame']['origin_height'] 256 257 @origin_height.setter 258 def origin_height(self, value: int) -> None: 259 self.meta['grain']['cog_coded_frame']['origin_height'] = value 260 261 @property 262 def coded_width(self) -> int: 263 return self.meta['grain']['cog_coded_frame']['coded_width'] 264 265 @coded_width.setter 266 def coded_width(self, value: int) -> None: 267 self.meta['grain']['cog_coded_frame']['coded_width'] = value 268 269 @property 270 def coded_height(self) -> int: 271 return self.meta['grain']['cog_coded_frame']['coded_height'] 272 273 @coded_height.setter 274 def coded_height(self, value: int) -> None: 275 self.meta['grain']['cog_coded_frame']['coded_height'] = value 276 277 @property 278 def is_key_frame(self) -> bool | None: 279 return self.meta['grain']['cog_coded_frame'].get('is_key_frame') 280 281 @is_key_frame.setter 282 def is_key_frame(self, value: bool | None) -> None: 283 if value is not None: 284 self.meta['grain']['cog_coded_frame']['is_key_frame'] = bool(value) 285 else: 286 try: 287 del self.meta['grain']['cog_coded_frame']['is_key_frame'] 288 except KeyError: 289 pass 290 291 @property 292 def temporal_offset(self) -> int | None: 293 return self.meta['grain']['cog_coded_frame'].get('temporal_offset') 294 295 @temporal_offset.setter 296 def temporal_offset(self, value: int | None) -> None: 297 if value is not None: 298 self.meta['grain']['cog_coded_frame']['temporal_offset'] = value 299 else: 300 try: 301 del self.meta['grain']['cog_coded_frame']['temporal_offset'] 302 except KeyError: 303 pass 304 305 @property 306 def source_aspect_ratio(self) -> Optional[Fraction]: 307 if 'source_aspect_ratio' in self.meta['grain']['cog_coded_frame']: 308 return Fraction(cast(FractionDict, 309 self.meta['grain']['cog_coded_frame']['source_aspect_ratio'])['numerator'], 310 cast(FractionDict, 311 self.meta['grain']['cog_coded_frame']['source_aspect_ratio'])['denominator']) 312 else: 313 return None 314 315 @source_aspect_ratio.setter 316 def source_aspect_ratio(self, value: RationalTypes) -> None: 317 value = Fraction(value) 318 self.meta['grain']['cog_coded_frame']['source_aspect_ratio'] = {'numerator': value.numerator, 319 'denominator': value.denominator} 320 321 @property 322 def pixel_aspect_ratio(self) -> Optional[Fraction]: 323 if 'pixel_aspect_ratio' in self.meta['grain']['cog_coded_frame']: 324 return Fraction(cast(FractionDict, 325 self.meta['grain']['cog_coded_frame']['pixel_aspect_ratio'])['numerator'], 326 cast(FractionDict, 327 self.meta['grain']['cog_coded_frame']['pixel_aspect_ratio'])['denominator']) 328 else: 329 return None 330 331 @pixel_aspect_ratio.setter 332 def pixel_aspect_ratio(self, value: RationalTypes) -> None: 333 value = Fraction(value) 334 self.meta['grain']['cog_coded_frame']['pixel_aspect_ratio'] = {'numerator': value.numerator, 335 'denominator': value.denominator} 336 337 class UNITOFFSETS(MutableSequence): 338 def __init__(self, parent: "CodedVideoGrain"): 339 self.parent = parent 340 341 @overload 342 def __getitem__(self, key: int) -> int: ... 343 344 @overload # noqa: F811 345 def __getitem__(self, key: slice) -> List[int]: ... 346 347 def __getitem__(self, key): # noqa: F811 348 if 'unit_offsets' in self.parent.meta['grain']['cog_coded_frame']: 349 return self.parent.meta['grain']['cog_coded_frame']['unit_offsets'][key] 350 else: 351 raise IndexError("list index out of range") 352 353 @overload 354 def __setitem__(self, key: int, value: int) -> None: ... 355 356 @overload # noqa: F811 357 def __setitem__(self, key: slice, value: Iterable[int]) -> None: ... 358 359 def __setitem__(self, key, value): # noqa: F811 360 if 'unit_offsets' in self.parent.meta['grain']['cog_coded_frame']: 361 self.parent.meta['grain']['cog_coded_frame']['unit_offsets'][key] = value 362 else: 363 raise IndexError("list assignment index out of range") 364 365 def __delitem__(self, key: Union[int, slice]) -> None: 366 if 'unit_offsets' in self.parent.meta['grain']['cog_coded_frame']: 367 del cast(List[int], self.parent.meta['grain']['cog_coded_frame']['unit_offsets'])[key] 368 if len(self.parent.meta['grain']['cog_coded_frame']['unit_offsets']) == 0: 369 del self.parent.meta['grain']['cog_coded_frame']['unit_offsets'] 370 else: 371 raise IndexError("list assignment index out of range") 372 373 def insert(self, key: int, value: int) -> None: 374 if 'unit_offsets' not in self.parent.meta['grain']['cog_coded_frame']: 375 d: List[int] = [] 376 d.insert(key, value) 377 self.parent.meta['grain']['cog_coded_frame']['unit_offsets'] = d 378 else: 379 cast(List[int], self.parent.meta['grain']['cog_coded_frame']['unit_offsets']).insert(key, value) 380 381 def __len__(self) -> int: 382 if 'unit_offsets' in self.parent.meta['grain']['cog_coded_frame']: 383 return len(self.parent.meta['grain']['cog_coded_frame']['unit_offsets']) 384 else: 385 return 0 386 387 def __eq__(self, other: object) -> bool: 388 return list(self) == other 389 390 def __ne__(self, other: object) -> bool: 391 return not (self == other) 392 393 def __repr__(self) -> str: 394 if 'unit_offsets' not in self.parent.meta['grain']['cog_coded_frame']: 395 return repr([]) 396 else: 397 return repr(self.parent.meta['grain']['cog_coded_frame']['unit_offsets']) 398 399 @property 400 def unit_offsets(self) -> "CodedVideoGrain.UNITOFFSETS": 401 return CodedVideoGrain.UNITOFFSETS(self) 402 403 @unit_offsets.setter 404 def unit_offsets(self, value: Iterable[int]) -> None: 405 if value is not None and not (hasattr(value, "__len__") and len(cast(Sized, value)) == 0): 406 self.meta['grain']['cog_coded_frame']['unit_offsets'] = list(value) 407 elif 'unit_offsets' in self.meta['grain']['cog_coded_frame']: 408 del self.meta['grain']['cog_coded_frame']['unit_offsets'] 409 410 @property 411 def media_rate(self) -> Optional[Fraction]: 412 if self.rate: 413 return self.rate 414 else: 415 return None 416 417 @property 418 def presentation_origin_timestamp(self) -> Timestamp: 419 if self.rate is not None and self.temporal_offset is not None: 420 return self.origin_timestamp + Timestamp.from_count(self.temporal_offset, self.rate) 421 else: 422 return self.origin_timestamp 423 424 def final_presentation_origin_timestamp(self) -> Timestamp: 425 if self.rate is not None and self.temporal_offset is not None: 426 return self.final_origin_timestamp() + Timestamp.from_count(self.temporal_offset, self.rate) 427 else: 428 return self.final_origin_timestamp() 429 430 def presentation_origin_timerange(self) -> TimeRange: 431 origin_tr = self.origin_timerange() 432 if self.rate is not None and self.temporal_offset is not None: 433 if origin_tr.start is not None: 434 start = origin_tr.start + Timestamp.from_count(self.temporal_offset, self.rate) 435 else: 436 start = None 437 438 if origin_tr.end is not None: 439 end = origin_tr.end + Timestamp.from_count(self.temporal_offset, self.rate) 440 else: 441 end = None 442 443 return TimeRange(start, end, origin_tr.inclusivity) 444 else: 445 return origin_tr
A class representing a coded video grain.
Any grain can be freely cast to a tuple:
(meta, data)
where meta is a dictionary containing the grain metadata, and data is the data element described below.
The Grain class provides a number of properties which can be used to access parts of the standard grain metadata, and this class inherits these:
meta The meta dictionary object
data Either None or an object which can be cast to bytes by passing it to the bytes constructor and will in of itself respond to the python-level portions of the bytes-like object protocol. It is not guaranteed that this object will always respond correctly to the C buffer-protocol, but it can always be converted into something that will by calling bytes on it.
grain_type A string containing the type of the grain, always "coded_video"
source_id A uuid.UUID object representing the source_id in the grain
flow_id A uuid.UUID object representing the flow_id in the grain
origin_timestamp An mediatimestamp.Timestamp object representing the origin timestamp of this grain.
sync_timestamp An mediatimestamp.Timestamp object representing the sync timestamp of this grain.
creation_timestamp An mediatimestamp.Timestamp object representing the creation timestamp of this grain.
rate A fractions.Fraction object representing the grain rate in grains per second.
duration A fractions.Fraction object representing the grain duration in seconds.
timelabels A list object containing time label data
length The length of the data element or 0 if that is None
The CodedVideoGrain class also provides additional properies
format An enumerated value of type CogFrameFormat
layout An enumerated value of type CogFrameLayout
origin_width The original video width in pixels
origin_height The original video height in pixels
coded_width The coded video width in pixels
coded_height The coded video height in pixels
temporal_offset An optional signed integer value indicating the offset from the origin timestamp of this grain to the expected presentation time of the picture in frames.
unit_offsets A list-like object containing integer offsets of coded units within the data array.
CodedVideoGrain( meta: Optional[mediagrains.typing.CodedVideoGrainMetadataDict] = None, data: Union[SupportsBytes, bytes, numpy.ndarray, Awaitable[Union[SupportsBytes, bytes, numpy.ndarray, NoneType]], NoneType] = None, src_id: Optional[uuid.UUID] = None, flow_id: Optional[uuid.UUID] = None, origin_timestamp: Optional[mediatimestamp.immutable.timestamp.SupportsMediaTimestamp] = None, creation_timestamp: Optional[mediatimestamp.immutable.timestamp.SupportsMediaTimestamp] = None, sync_timestamp: Optional[mediatimestamp.immutable.timestamp.SupportsMediaTimestamp] = None, rate: fractions.Fraction = Fraction(25, 1), duration: fractions.Fraction = Fraction(1, 25), cog_frame_format: mediagrains.cogenums.CogFrameFormat = <CogFrameFormat.UNKNOWN: 4294967294>, origin_width: int = 1920, origin_height: int = 1080, coded_width: Optional[int] = None, coded_height: Optional[int] = None, is_key_frame: Optional[bool] = None, temporal_offset: Optional[int] = None, length: Optional[int] = None, cog_frame_layout: mediagrains.cogenums.CogFrameLayout = <CogFrameLayout.UNKNOWN: 4294967294>, unit_offsets: Optional[List[int]] = None)
108 def __init__(self, 109 meta: Optional[CodedVideoGrainMetadataDict] = None, 110 data: Optional[GrainDataParameterType] = None, 111 src_id: Optional[UUID] = None, 112 flow_id: Optional[UUID] = None, 113 origin_timestamp: Optional[SupportsMediaTimestamp] = None, 114 creation_timestamp: Optional[SupportsMediaTimestamp] = None, 115 sync_timestamp: Optional[SupportsMediaTimestamp] = None, 116 rate: Fraction = Fraction(25, 1), 117 duration: Fraction = Fraction(1, 25), 118 cog_frame_format: CogFrameFormat = CogFrameFormat.UNKNOWN, 119 origin_width: int = 1920, 120 origin_height: int = 1080, 121 coded_width: Optional[int] = None, 122 coded_height: Optional[int] = None, 123 is_key_frame: Optional[bool] = None, 124 temporal_offset: Optional[int] = None, 125 length: Optional[int] = None, 126 cog_frame_layout: CogFrameLayout = CogFrameLayout.UNKNOWN, 127 unit_offsets: Optional[List[int]] = None): 128 if coded_width is None: 129 coded_width = origin_width 130 if coded_height is None: 131 coded_height = origin_height 132 133 if length is None: 134 if data is not None and hasattr(data, "__len__"): 135 length = len(cast(Sized, data)) 136 else: 137 length = 0 138 139 if meta is None: 140 if not isinstance(src_id, UUID) and src_id is not None: 141 raise AttributeError(f"src_id: Seen type {type(src_id)}, expected UUID.") 142 if not isinstance(flow_id, UUID) and flow_id is not None: 143 raise AttributeError(f"flow_id: Seen type {type(flow_id)}, expected UUID.") 144 145 if creation_timestamp is None: 146 creation_timestamp = Timestamp.get_time() 147 if origin_timestamp is None: 148 origin_timestamp = creation_timestamp 149 if sync_timestamp is None: 150 sync_timestamp = origin_timestamp 151 meta = { 152 "@_ns": "urn:x-ipstudio:ns:0.1", 153 "grain": { 154 "grain_type": "coded_video", 155 "source_id": str(src_id), 156 "flow_id": str(flow_id), 157 "origin_timestamp": str(mediatimestamp(origin_timestamp)), 158 "sync_timestamp": str(mediatimestamp(sync_timestamp)), 159 "creation_timestamp": str(mediatimestamp(creation_timestamp)), 160 "rate": { 161 "numerator": Fraction(rate).numerator, 162 "denominator": Fraction(rate).denominator, 163 }, 164 "duration": { 165 "numerator": Fraction(duration).numerator, 166 "denominator": Fraction(duration).denominator, 167 }, 168 "cog_coded_frame": { 169 "format": cog_frame_format, 170 "origin_width": origin_width, 171 "origin_height": origin_height, 172 "coded_width": coded_width, 173 "coded_height": coded_height, 174 "layout": cog_frame_layout 175 } 176 }, 177 } 178 if is_key_frame is not None: 179 meta["grain"]["cog_coded_frame"]["is_key_frame"] = is_key_frame 180 if temporal_offset is not None: 181 meta["grain"]["cog_coded_frame"]["temporal_offset"] = temporal_offset 182 183 if data is None: 184 data = bytearray(length) 185 186 if "grain" in meta and "cog_coded_frame" in meta['grain'] and unit_offsets is not None: 187 meta['grain']['cog_coded_frame']['unit_offsets'] = unit_offsets 188 189 super().__init__(meta, data) 190 self.meta: CodedVideoGrainMetadataDict 191 192 self._factory = "CodedVideoGrain" 193 self.meta['grain']['grain_type'] = 'coded_video' 194 if 'cog_coded_frame' not in self.meta['grain']: 195 self.meta['grain']['cog_coded_frame'] = {} # type: ignore 196 if 'format' not in self.meta['grain']['cog_coded_frame']: 197 self.meta['grain']['cog_coded_frame']['format'] = int(CogFrameFormat.UNKNOWN) 198 if 'layout' not in self.meta['grain']['cog_coded_frame']: 199 self.meta['grain']['cog_coded_frame']['layout'] = int(CogFrameLayout.UNKNOWN) 200 if 'origin_width' not in self.meta['grain']['cog_coded_frame']: 201 self.meta['grain']['cog_coded_frame']['origin_width'] = 0 202 if 'origin_height' not in self.meta['grain']['cog_coded_frame']: 203 self.meta['grain']['cog_coded_frame']['origin_height'] = 0 204 if 'coded_width' not in self.meta['grain']['cog_coded_frame']: 205 self.meta['grain']['cog_coded_frame']['coded_width'] = 0 206 if 'coded_height' not in self.meta['grain']['cog_coded_frame']: 207 self.meta['grain']['cog_coded_frame']['coded_height'] = 0 208 if 'length' not in self.meta['grain']['cog_coded_frame']: 209 self.meta['grain']['cog_coded_frame']['length'] = 0 210 self.meta['grain']['cog_coded_frame']['format'] = int(self.meta['grain']['cog_coded_frame']['format']) 211 self.meta['grain']['cog_coded_frame']['layout'] = int(self.meta['grain']['cog_coded_frame']['layout'])
source_aspect_ratio: Optional[fractions.Fraction]
305 @property 306 def source_aspect_ratio(self) -> Optional[Fraction]: 307 if 'source_aspect_ratio' in self.meta['grain']['cog_coded_frame']: 308 return Fraction(cast(FractionDict, 309 self.meta['grain']['cog_coded_frame']['source_aspect_ratio'])['numerator'], 310 cast(FractionDict, 311 self.meta['grain']['cog_coded_frame']['source_aspect_ratio'])['denominator']) 312 else: 313 return None
pixel_aspect_ratio: Optional[fractions.Fraction]
321 @property 322 def pixel_aspect_ratio(self) -> Optional[Fraction]: 323 if 'pixel_aspect_ratio' in self.meta['grain']['cog_coded_frame']: 324 return Fraction(cast(FractionDict, 325 self.meta['grain']['cog_coded_frame']['pixel_aspect_ratio'])['numerator'], 326 cast(FractionDict, 327 self.meta['grain']['cog_coded_frame']['pixel_aspect_ratio'])['denominator']) 328 else: 329 return None
unit_offsets: CodedVideoGrain.UNITOFFSETS
def
presentation_origin_timerange(self) -> mediatimestamp.immutable.timerange.TimeRange:
430 def presentation_origin_timerange(self) -> TimeRange: 431 origin_tr = self.origin_timerange() 432 if self.rate is not None and self.temporal_offset is not None: 433 if origin_tr.start is not None: 434 start = origin_tr.start + Timestamp.from_count(self.temporal_offset, self.rate) 435 else: 436 start = None 437 438 if origin_tr.end is not None: 439 end = origin_tr.end + Timestamp.from_count(self.temporal_offset, self.rate) 440 else: 441 end = None 442 443 return TimeRange(start, end, origin_tr.inclusivity) 444 else: 445 return origin_tr
Inherited Members
- Grain
- has_data
- data
- grain_type
- source_id
- src_id
- flow_id
- origin_timestamp
- final_origin_timestamp
- origin_timerange
- normalise_time
- sync_timestamp
- creation_timestamp
- rate
- duration
- timelabels
- add_timelabel
- TIMELABEL
- TIMELABELS
- length
- expected_length
- collections.abc.Sequence
- index
- count
class
CodedVideoGrain.UNITOFFSETS(collections.abc.MutableSequence):
337 class UNITOFFSETS(MutableSequence): 338 def __init__(self, parent: "CodedVideoGrain"): 339 self.parent = parent 340 341 @overload 342 def __getitem__(self, key: int) -> int: ... 343 344 @overload # noqa: F811 345 def __getitem__(self, key: slice) -> List[int]: ... 346 347 def __getitem__(self, key): # noqa: F811 348 if 'unit_offsets' in self.parent.meta['grain']['cog_coded_frame']: 349 return self.parent.meta['grain']['cog_coded_frame']['unit_offsets'][key] 350 else: 351 raise IndexError("list index out of range") 352 353 @overload 354 def __setitem__(self, key: int, value: int) -> None: ... 355 356 @overload # noqa: F811 357 def __setitem__(self, key: slice, value: Iterable[int]) -> None: ... 358 359 def __setitem__(self, key, value): # noqa: F811 360 if 'unit_offsets' in self.parent.meta['grain']['cog_coded_frame']: 361 self.parent.meta['grain']['cog_coded_frame']['unit_offsets'][key] = value 362 else: 363 raise IndexError("list assignment index out of range") 364 365 def __delitem__(self, key: Union[int, slice]) -> None: 366 if 'unit_offsets' in self.parent.meta['grain']['cog_coded_frame']: 367 del cast(List[int], self.parent.meta['grain']['cog_coded_frame']['unit_offsets'])[key] 368 if len(self.parent.meta['grain']['cog_coded_frame']['unit_offsets']) == 0: 369 del self.parent.meta['grain']['cog_coded_frame']['unit_offsets'] 370 else: 371 raise IndexError("list assignment index out of range") 372 373 def insert(self, key: int, value: int) -> None: 374 if 'unit_offsets' not in self.parent.meta['grain']['cog_coded_frame']: 375 d: List[int] = [] 376 d.insert(key, value) 377 self.parent.meta['grain']['cog_coded_frame']['unit_offsets'] = d 378 else: 379 cast(List[int], self.parent.meta['grain']['cog_coded_frame']['unit_offsets']).insert(key, value) 380 381 def __len__(self) -> int: 382 if 'unit_offsets' in self.parent.meta['grain']['cog_coded_frame']: 383 return len(self.parent.meta['grain']['cog_coded_frame']['unit_offsets']) 384 else: 385 return 0 386 387 def __eq__(self, other: object) -> bool: 388 return list(self) == other 389 390 def __ne__(self, other: object) -> bool: 391 return not (self == other) 392 393 def __repr__(self) -> str: 394 if 'unit_offsets' not in self.parent.meta['grain']['cog_coded_frame']: 395 return repr([]) 396 else: 397 return repr(self.parent.meta['grain']['cog_coded_frame']['unit_offsets'])
All the operations on a read-write sequence.
Concrete subclasses must provide __new__ or __init__, __getitem__, __setitem__, __delitem__, __len__, and insert().
CodedVideoGrain.UNITOFFSETS(parent: CodedVideoGrain)
def
insert(self, key: int, value: int) -> None:
373 def insert(self, key: int, value: int) -> None: 374 if 'unit_offsets' not in self.parent.meta['grain']['cog_coded_frame']: 375 d: List[int] = [] 376 d.insert(key, value) 377 self.parent.meta['grain']['cog_coded_frame']['unit_offsets'] = d 378 else: 379 cast(List[int], self.parent.meta['grain']['cog_coded_frame']['unit_offsets']).insert(key, value)
S.insert(index, value) -- insert value before index
Inherited Members
- collections.abc.MutableSequence
- append
- clear
- reverse
- extend
- pop
- remove
- collections.abc.Sequence
- index
- count
33class AudioGrain(Grain): 34 """\ 35A class representing a raw audio grain. 36 37Any grain can be freely cast to a tuple: 38 39 (meta, data) 40 41where meta is a dictionary containing the grain metadata, and data is the data element described below.. 42 43The Grain class provides a number of properties which can be used to access 44parts of the standard grain metadata, and this class inherits these: 45 46meta 47 The meta dictionary object 48 49data 50 Either None or an object which can be cast to bytes by passing it to the bytes 51 constructor and will in of itself respond to the python-level portions of the bytes-like 52 object protocol. It is not guaranteed that this object will always respond correctly to the 53 C buffer-protocol, but it can always be converted into something that will by calling bytes on it. 54 55grain_type 56 A string containing the type of the grain, always "audio" 57 58src_id 59 A uuid.UUID object representing the source_id in the grain 60 61flow_id 62 A uuid.UUID object representing the flow_id in the grain 63 64origin_timestamp 65 An mediatimestamp.Timestamp object representing the origin timestamp 66 of this grain. 67 68sync_timestamp 69 An mediatimestamp.Timestamp object representing the sync timestamp 70 of this grain. 71 72creation_timestamp 73 An mediatimestamp.Timestamp object representing the creation timestamp 74 of this grain. 75 76rate 77 A fractions.Fraction object representing the grain rate in grains per second. 78 79duration 80 A fractions.Fraction object representing the grain duration in seconds. 81 82timelabels 83 A list object containing time label data 84 85length 86 The length of the data element or 0 if that is None 87 88The AudioGrain class also provides additional properies 89 90format 91 An enumerated value of type CogAudioFormat 92 93samples 94 The number of audio samples per channel in this grain 95 96channels 97 The number of channels in this grain 98 99sample_rate 100 An integer indicating the number of samples per channel per second in this 101 audio flow. 102""" 103 def __init__(self, 104 meta: Optional[AudioGrainMetadataDict] = None, 105 data: Optional[GrainDataParameterType] = None, 106 src_id: Optional[UUID] = None, 107 flow_id: Optional[UUID] = None, 108 origin_timestamp: Optional[SupportsMediaTimestamp] = None, 109 sync_timestamp: Optional[SupportsMediaTimestamp] = None, 110 creation_timestamp: Optional[SupportsMediaTimestamp] = None, 111 rate: Fraction = Fraction(25, 1), 112 duration: Fraction = Fraction(1, 25), 113 cog_audio_format: CogAudioFormat = CogAudioFormat.INVALID, 114 samples: int = 0, 115 channels: int = 0, 116 sample_rate: int = 48000): 117 118 if meta is None: 119 if not isinstance(src_id, UUID) and src_id is not None: 120 raise AttributeError(f"src_id: Seen type {type(src_id)}, expected UUID.") 121 if not isinstance(flow_id, UUID) and flow_id is not None: 122 raise AttributeError(f"flow_id: Seen type {type(flow_id)}, expected UUID.") 123 124 if creation_timestamp is None: 125 creation_timestamp = Timestamp.get_time() 126 if origin_timestamp is None: 127 origin_timestamp = creation_timestamp 128 if sync_timestamp is None: 129 sync_timestamp = origin_timestamp 130 meta = { 131 "@_ns": "urn:x-ipstudio:ns:0.1", 132 "grain": { 133 'grain_type': "audio", 134 'source_id': str(src_id), 135 'flow_id': str(flow_id), 136 'origin_timestamp': str(mediatimestamp(origin_timestamp)), 137 'sync_timestamp': str(mediatimestamp(sync_timestamp)), 138 'creation_timestamp': str(mediatimestamp(creation_timestamp)), 139 'rate': { 140 'numerator': Fraction(rate).numerator, 141 'denominator': Fraction(rate).denominator 142 }, 143 'duration': { 144 'numerator': Fraction(duration).numerator, 145 'denominator': Fraction(duration).denominator 146 }, 147 'cog_audio': { 148 "format": cog_audio_format, 149 "samples": samples, 150 "channels": channels, 151 "sample_rate": sample_rate 152 } 153 } 154 } 155 156 if data is None: 157 size = size_for_audio_format(cog_audio_format, channels, samples) 158 data = bytearray(size) 159 160 super().__init__(meta, data) 161 self.meta: AudioGrainMetadataDict 162 163 self._factory = "AudioGrain" 164 self.meta['grain']['grain_type'] = 'audio' 165 if 'cog_audio' not in self.meta['grain']: 166 self.meta['grain']['cog_audio'] = {} # type: ignore 167 if 'format' not in self.meta['grain']['cog_audio']: 168 self.meta['grain']['cog_audio']['format'] = int(CogAudioFormat.INVALID) 169 if 'samples' not in self.meta['grain']['cog_audio']: 170 self.meta['grain']['cog_audio']['samples'] = 0 171 if 'channels' not in self.meta['grain']['cog_audio']: 172 self.meta['grain']['cog_audio']['channels'] = 0 173 if 'sample_rate' not in self.meta['grain']['cog_audio']: 174 self.meta['grain']['cog_audio']['sample_rate'] = 0 175 self.meta['grain']['cog_audio']['format'] = int(self.meta['grain']['cog_audio']['format']) 176 177 def final_origin_timestamp(self) -> Timestamp: 178 return (self.origin_timestamp + TimeOffset.from_count(self.samples - 1, self.sample_rate, 1)) 179 180 @property 181 def format(self) -> CogAudioFormat: 182 return CogAudioFormat(self.meta['grain']['cog_audio']['format']) 183 184 @format.setter 185 def format(self, value: CogAudioFormat) -> None: 186 self.meta['grain']['cog_audio']['format'] = int(value) 187 188 @property 189 def cog_audio_format(self) -> CogAudioFormat: 190 return CogAudioFormat(self.meta['grain']['cog_audio']['format']) 191 192 @cog_audio_format.setter 193 def cog_audio_format(self, value: CogAudioFormat) -> None: 194 self.meta['grain']['cog_audio']['format'] = int(value) 195 196 @property 197 def samples(self) -> int: 198 return self.meta['grain']['cog_audio']['samples'] 199 200 @samples.setter 201 def samples(self, value: int) -> None: 202 self.meta['grain']['cog_audio']['samples'] = int(value) 203 204 @property 205 def channels(self) -> int: 206 return self.meta['grain']['cog_audio']['channels'] 207 208 @channels.setter 209 def channels(self, value: int) -> None: 210 self.meta['grain']['cog_audio']['channels'] = int(value) 211 212 @property 213 def sample_rate(self) -> int: 214 return self.meta['grain']['cog_audio']['sample_rate'] 215 216 @sample_rate.setter 217 def sample_rate(self, value: int) -> None: 218 self.meta['grain']['cog_audio']['sample_rate'] = int(value) 219 220 @property 221 def expected_length(self) -> int: 222 return size_for_audio_format(self.format, self.channels, self.samples) 223 224 @property 225 def media_rate(self) -> Optional[Fraction]: 226 if self.sample_rate: 227 return Fraction(self.sample_rate, 1) 228 else: 229 return None
A class representing a raw audio grain.
Any grain can be freely cast to a tuple:
(meta, data)
where meta is a dictionary containing the grain metadata, and data is the data element described below..
The Grain class provides a number of properties which can be used to access parts of the standard grain metadata, and this class inherits these:
meta The meta dictionary object
data Either None or an object which can be cast to bytes by passing it to the bytes constructor and will in of itself respond to the python-level portions of the bytes-like object protocol. It is not guaranteed that this object will always respond correctly to the C buffer-protocol, but it can always be converted into something that will by calling bytes on it.
grain_type A string containing the type of the grain, always "audio"
src_id A uuid.UUID object representing the source_id in the grain
flow_id A uuid.UUID object representing the flow_id in the grain
origin_timestamp An mediatimestamp.Timestamp object representing the origin timestamp of this grain.
sync_timestamp An mediatimestamp.Timestamp object representing the sync timestamp of this grain.
creation_timestamp An mediatimestamp.Timestamp object representing the creation timestamp of this grain.
rate A fractions.Fraction object representing the grain rate in grains per second.
duration A fractions.Fraction object representing the grain duration in seconds.
timelabels A list object containing time label data
length The length of the data element or 0 if that is None
The AudioGrain class also provides additional properies
format An enumerated value of type CogAudioFormat
samples The number of audio samples per channel in this grain
channels The number of channels in this grain
sample_rate An integer indicating the number of samples per channel per second in this audio flow.
AudioGrain( meta: Optional[mediagrains.typing.AudioGrainMetadataDict] = None, data: Union[SupportsBytes, bytes, numpy.ndarray, Awaitable[Union[SupportsBytes, bytes, numpy.ndarray, NoneType]], NoneType] = None, src_id: Optional[uuid.UUID] = None, flow_id: Optional[uuid.UUID] = None, origin_timestamp: Optional[mediatimestamp.immutable.timestamp.SupportsMediaTimestamp] = None, sync_timestamp: Optional[mediatimestamp.immutable.timestamp.SupportsMediaTimestamp] = None, creation_timestamp: Optional[mediatimestamp.immutable.timestamp.SupportsMediaTimestamp] = None, rate: fractions.Fraction = Fraction(25, 1), duration: fractions.Fraction = Fraction(1, 25), cog_audio_format: mediagrains.cogenums.CogAudioFormat = <CogAudioFormat.INVALID: 4294967295>, samples: int = 0, channels: int = 0, sample_rate: int = 48000)
103 def __init__(self, 104 meta: Optional[AudioGrainMetadataDict] = None, 105 data: Optional[GrainDataParameterType] = None, 106 src_id: Optional[UUID] = None, 107 flow_id: Optional[UUID] = None, 108 origin_timestamp: Optional[SupportsMediaTimestamp] = None, 109 sync_timestamp: Optional[SupportsMediaTimestamp] = None, 110 creation_timestamp: Optional[SupportsMediaTimestamp] = None, 111 rate: Fraction = Fraction(25, 1), 112 duration: Fraction = Fraction(1, 25), 113 cog_audio_format: CogAudioFormat = CogAudioFormat.INVALID, 114 samples: int = 0, 115 channels: int = 0, 116 sample_rate: int = 48000): 117 118 if meta is None: 119 if not isinstance(src_id, UUID) and src_id is not None: 120 raise AttributeError(f"src_id: Seen type {type(src_id)}, expected UUID.") 121 if not isinstance(flow_id, UUID) and flow_id is not None: 122 raise AttributeError(f"flow_id: Seen type {type(flow_id)}, expected UUID.") 123 124 if creation_timestamp is None: 125 creation_timestamp = Timestamp.get_time() 126 if origin_timestamp is None: 127 origin_timestamp = creation_timestamp 128 if sync_timestamp is None: 129 sync_timestamp = origin_timestamp 130 meta = { 131 "@_ns": "urn:x-ipstudio:ns:0.1", 132 "grain": { 133 'grain_type': "audio", 134 'source_id': str(src_id), 135 'flow_id': str(flow_id), 136 'origin_timestamp': str(mediatimestamp(origin_timestamp)), 137 'sync_timestamp': str(mediatimestamp(sync_timestamp)), 138 'creation_timestamp': str(mediatimestamp(creation_timestamp)), 139 'rate': { 140 'numerator': Fraction(rate).numerator, 141 'denominator': Fraction(rate).denominator 142 }, 143 'duration': { 144 'numerator': Fraction(duration).numerator, 145 'denominator': Fraction(duration).denominator 146 }, 147 'cog_audio': { 148 "format": cog_audio_format, 149 "samples": samples, 150 "channels": channels, 151 "sample_rate": sample_rate 152 } 153 } 154 } 155 156 if data is None: 157 size = size_for_audio_format(cog_audio_format, channels, samples) 158 data = bytearray(size) 159 160 super().__init__(meta, data) 161 self.meta: AudioGrainMetadataDict 162 163 self._factory = "AudioGrain" 164 self.meta['grain']['grain_type'] = 'audio' 165 if 'cog_audio' not in self.meta['grain']: 166 self.meta['grain']['cog_audio'] = {} # type: ignore 167 if 'format' not in self.meta['grain']['cog_audio']: 168 self.meta['grain']['cog_audio']['format'] = int(CogAudioFormat.INVALID) 169 if 'samples' not in self.meta['grain']['cog_audio']: 170 self.meta['grain']['cog_audio']['samples'] = 0 171 if 'channels' not in self.meta['grain']['cog_audio']: 172 self.meta['grain']['cog_audio']['channels'] = 0 173 if 'sample_rate' not in self.meta['grain']['cog_audio']: 174 self.meta['grain']['cog_audio']['sample_rate'] = 0 175 self.meta['grain']['cog_audio']['format'] = int(self.meta['grain']['cog_audio']['format'])
Inherited Members
- Grain
- has_data
- data
- grain_type
- source_id
- src_id
- flow_id
- origin_timestamp
- origin_timerange
- presentation_origin_timestamp
- final_presentation_origin_timestamp
- presentation_origin_timerange
- normalise_time
- sync_timestamp
- creation_timestamp
- rate
- duration
- timelabels
- add_timelabel
- TIMELABEL
- TIMELABELS
- length
- collections.abc.Sequence
- index
- count
18class CodedAudioGrain(Grain): 19 """\ 20A class representing a coded audio grain. 21 22Any grain can be freely cast to a tuple: 23 24 (meta, data) 25 26where meta is a dictionary containing the grain metadata, and data is a 27bytes-like object containing the coded audio data. 28 29The Grain class provides a number of properties which can be used to access 30parts of the standard grain metadata, and this class inherits these: 31 32meta 33 The meta dictionary object 34 35data 36 Either None or an object which can be cast to bytes by passing it to the bytes 37 constructor and will in of itself respond to the python-level portions of the bytes-like 38 object protocol. It is not guaranteed that this object will always respond correctly to the 39 C buffer-protocol, but it can always be converted into something that will by calling bytes on it. 40 41grain_type 42 A string containing the type of the grain, always "coded_audio" 43 44src_id 45 A uuid.UUID object representing the source_id in the grain 46 47flow_id 48 A uuid.UUID object representing the flow_id in the grain 49 50origin_timestamp 51 An mediatimestamp.Timestamp object representing the origin timestamp 52 of this grain. 53 54sync_timestamp 55 An mediatimestamp.Timestamp object representing the sync timestamp 56 of this grain. 57 58creation_timestamp 59 An mediatimestamp.Timestamp object representing the creation timestamp 60 of this grain. 61 62rate 63 A fractions.Fraction object representing the grain rate in grains per second. 64 65duration 66 A fractions.Fraction object representing the grain duration in seconds. 67 68timelabels 69 A list object containing time label data 70 71length 72 The length of the data element or 0 if that is None 73 74The AudioGrain class also provides additional properies 75 76format 77 An enumerated value of type CogAudioFormat 78 79samples 80 The number of audio samples per channel in this grain 81 82channels 83 The number of channels in this grain 84 85sample_rate 86 An integer indicating the number of samples per channel per second in this 87 audio flow. 88 89priming 90 An integer 91 92remainder 93 An integer 94""" 95 def __init__(self, 96 meta: Optional[CodedAudioGrainMetadataDict] = None, 97 data: Optional[GrainDataParameterType] = None, 98 src_id: Optional[UUID] = None, 99 flow_id: Optional[UUID] = None, 100 origin_timestamp: Optional[SupportsMediaTimestamp] = None, 101 creation_timestamp: Optional[SupportsMediaTimestamp] = None, 102 sync_timestamp: Optional[SupportsMediaTimestamp] = None, 103 rate: Fraction = Fraction(25, 1), 104 duration: Fraction = Fraction(1, 25), 105 cog_audio_format: CogAudioFormat = CogAudioFormat.INVALID, 106 samples: int = 0, 107 channels: int = 0, 108 priming: int = 0, 109 remainder: int = 0, 110 sample_rate: int = 48000, 111 length: Optional[int] = None): 112 113 if length is None: 114 if data is not None and hasattr(data, "__len__"): 115 length = len(cast(Sized, data)) 116 else: 117 length = 0 118 119 if meta is None: 120 if not isinstance(src_id, UUID) and src_id is not None: 121 raise AttributeError(f"src_id: Seen type {type(src_id)}, expected UUID.") 122 if not isinstance(flow_id, UUID) and flow_id is not None: 123 raise AttributeError(f"flow_id: Seen type {type(flow_id)}, expected UUID.") 124 125 creation_timestamp = Timestamp.get_time() if not creation_timestamp else creation_timestamp 126 if origin_timestamp is None: 127 origin_timestamp = creation_timestamp 128 if sync_timestamp is None: 129 sync_timestamp = origin_timestamp 130 meta = { 131 "@_ns": "urn:x-ipstudio:ns:0.1", 132 "grain": { 133 'grain_type': "coded_audio", 134 'source_id': str(src_id), 135 'flow_id': str(flow_id), 136 'origin_timestamp': str(mediatimestamp(origin_timestamp)), 137 'sync_timestamp': str(mediatimestamp(sync_timestamp)), 138 'creation_timestamp': str(mediatimestamp(creation_timestamp)), 139 'rate': { 140 'numerator': Fraction(rate).numerator, 141 'denominator': Fraction(rate).denominator 142 }, 143 'duration': { 144 'numerator': Fraction(duration).numerator, 145 'denominator': Fraction(duration).denominator 146 }, 147 'cog_coded_audio': { 148 "format": cog_audio_format, 149 "samples": samples, 150 "channels": channels, 151 "priming": priming, 152 "remainder": remainder, 153 "sample_rate": sample_rate 154 } 155 } 156 } 157 158 if data is None: 159 data = bytearray(length) 160 161 super().__init__(meta, data) 162 self.meta: CodedAudioGrainMetadataDict 163 164 self._factory = "CodedAudioGrain" 165 self.meta['grain']['grain_type'] = 'coded_audio' 166 if 'cog_coded_audio' not in self.meta['grain']: 167 self.meta['grain']['cog_coded_audio'] = {} # type: ignore 168 if 'format' not in self.meta['grain']['cog_coded_audio']: 169 self.meta['grain']['cog_coded_audio']['format'] = int(CogAudioFormat.INVALID) 170 if 'channels' not in self.meta['grain']['cog_coded_audio']: 171 self.meta['grain']['cog_coded_audio']['channels'] = 0 172 if 'samples' not in self.meta['grain']['cog_coded_audio']: 173 self.meta['grain']['cog_coded_audio']['samples'] = 0 174 if 'priming' not in self.meta['grain']['cog_coded_audio']: 175 self.meta['grain']['cog_coded_audio']['priming'] = 0 176 if 'remainder' not in self.meta['grain']['cog_coded_audio']: 177 self.meta['grain']['cog_coded_audio']['remainder'] = 0 178 if 'sample_rate' not in self.meta['grain']['cog_coded_audio']: 179 self.meta['grain']['cog_coded_audio']['sample_rate'] = 48000 180 self.meta['grain']['cog_coded_audio']['format'] = int(self.meta['grain']['cog_coded_audio']['format']) 181 182 def final_origin_timestamp(self) -> Timestamp: 183 return (self.origin_timestamp + TimeOffset.from_count(self.samples - 1, self.sample_rate, 1)) 184 185 @property 186 def format(self) -> CogAudioFormat: 187 return CogAudioFormat(self.meta['grain']['cog_coded_audio']['format']) 188 189 @format.setter 190 def format(self, value: int) -> None: 191 self.meta['grain']['cog_coded_audio']['format'] = int(value) 192 193 @property 194 def cog_audio_format(self) -> CogAudioFormat: 195 return CogAudioFormat(self.meta['grain']['cog_coded_audio']['format']) 196 197 @cog_audio_format.setter 198 def cog_audio_format(self, value: int) -> None: 199 self.meta['grain']['cog_coded_audio']['format'] = int(value) 200 201 @property 202 def channels(self) -> int: 203 return self.meta['grain']['cog_coded_audio']['channels'] 204 205 @channels.setter 206 def channels(self, value: int) -> None: 207 self.meta['grain']['cog_coded_audio']['channels'] = value 208 209 @property 210 def samples(self) -> int: 211 return self.meta['grain']['cog_coded_audio']['samples'] 212 213 @samples.setter 214 def samples(self, value: int) -> None: 215 self.meta['grain']['cog_coded_audio']['samples'] = value 216 217 @property 218 def priming(self) -> int: 219 return self.meta['grain']['cog_coded_audio']['priming'] 220 221 @priming.setter 222 def priming(self, value: int) -> None: 223 self.meta['grain']['cog_coded_audio']['priming'] = value 224 225 @property 226 def remainder(self) -> int: 227 return self.meta['grain']['cog_coded_audio']['remainder'] 228 229 @remainder.setter 230 def remainder(self, value: int) -> None: 231 self.meta['grain']['cog_coded_audio']['remainder'] = value 232 233 @property 234 def sample_rate(self) -> int: 235 return self.meta['grain']['cog_coded_audio']['sample_rate'] 236 237 @sample_rate.setter 238 def sample_rate(self, value: int) -> None: 239 self.meta['grain']['cog_coded_audio']['sample_rate'] = value 240 241 @property 242 def media_rate(self) -> Optional[Fraction]: 243 if self.sample_rate: 244 return Fraction(self.sample_rate, 1) 245 else: 246 return None
A class representing a coded audio grain.
Any grain can be freely cast to a tuple:
(meta, data)
where meta is a dictionary containing the grain metadata, and data is a bytes-like object containing the coded audio data.
The Grain class provides a number of properties which can be used to access parts of the standard grain metadata, and this class inherits these:
meta The meta dictionary object
data Either None or an object which can be cast to bytes by passing it to the bytes constructor and will in of itself respond to the python-level portions of the bytes-like object protocol. It is not guaranteed that this object will always respond correctly to the C buffer-protocol, but it can always be converted into something that will by calling bytes on it.
grain_type A string containing the type of the grain, always "coded_audio"
src_id A uuid.UUID object representing the source_id in the grain
flow_id A uuid.UUID object representing the flow_id in the grain
origin_timestamp An mediatimestamp.Timestamp object representing the origin timestamp of this grain.
sync_timestamp An mediatimestamp.Timestamp object representing the sync timestamp of this grain.
creation_timestamp An mediatimestamp.Timestamp object representing the creation timestamp of this grain.
rate A fractions.Fraction object representing the grain rate in grains per second.
duration A fractions.Fraction object representing the grain duration in seconds.
timelabels A list object containing time label data
length The length of the data element or 0 if that is None
The AudioGrain class also provides additional properies
format An enumerated value of type CogAudioFormat
samples The number of audio samples per channel in this grain
channels The number of channels in this grain
sample_rate An integer indicating the number of samples per channel per second in this audio flow.
priming An integer
remainder An integer
CodedAudioGrain( meta: Optional[mediagrains.typing.CodedAudioGrainMetadataDict] = None, data: Union[SupportsBytes, bytes, numpy.ndarray, Awaitable[Union[SupportsBytes, bytes, numpy.ndarray, NoneType]], NoneType] = None, src_id: Optional[uuid.UUID] = None, flow_id: Optional[uuid.UUID] = None, origin_timestamp: Optional[mediatimestamp.immutable.timestamp.SupportsMediaTimestamp] = None, creation_timestamp: Optional[mediatimestamp.immutable.timestamp.SupportsMediaTimestamp] = None, sync_timestamp: Optional[mediatimestamp.immutable.timestamp.SupportsMediaTimestamp] = None, rate: fractions.Fraction = Fraction(25, 1), duration: fractions.Fraction = Fraction(1, 25), cog_audio_format: mediagrains.cogenums.CogAudioFormat = <CogAudioFormat.INVALID: 4294967295>, samples: int = 0, channels: int = 0, priming: int = 0, remainder: int = 0, sample_rate: int = 48000, length: Optional[int] = None)
95 def __init__(self, 96 meta: Optional[CodedAudioGrainMetadataDict] = None, 97 data: Optional[GrainDataParameterType] = None, 98 src_id: Optional[UUID] = None, 99 flow_id: Optional[UUID] = None, 100 origin_timestamp: Optional[SupportsMediaTimestamp] = None, 101 creation_timestamp: Optional[SupportsMediaTimestamp] = None, 102 sync_timestamp: Optional[SupportsMediaTimestamp] = None, 103 rate: Fraction = Fraction(25, 1), 104 duration: Fraction = Fraction(1, 25), 105 cog_audio_format: CogAudioFormat = CogAudioFormat.INVALID, 106 samples: int = 0, 107 channels: int = 0, 108 priming: int = 0, 109 remainder: int = 0, 110 sample_rate: int = 48000, 111 length: Optional[int] = None): 112 113 if length is None: 114 if data is not None and hasattr(data, "__len__"): 115 length = len(cast(Sized, data)) 116 else: 117 length = 0 118 119 if meta is None: 120 if not isinstance(src_id, UUID) and src_id is not None: 121 raise AttributeError(f"src_id: Seen type {type(src_id)}, expected UUID.") 122 if not isinstance(flow_id, UUID) and flow_id is not None: 123 raise AttributeError(f"flow_id: Seen type {type(flow_id)}, expected UUID.") 124 125 creation_timestamp = Timestamp.get_time() if not creation_timestamp else creation_timestamp 126 if origin_timestamp is None: 127 origin_timestamp = creation_timestamp 128 if sync_timestamp is None: 129 sync_timestamp = origin_timestamp 130 meta = { 131 "@_ns": "urn:x-ipstudio:ns:0.1", 132 "grain": { 133 'grain_type': "coded_audio", 134 'source_id': str(src_id), 135 'flow_id': str(flow_id), 136 'origin_timestamp': str(mediatimestamp(origin_timestamp)), 137 'sync_timestamp': str(mediatimestamp(sync_timestamp)), 138 'creation_timestamp': str(mediatimestamp(creation_timestamp)), 139 'rate': { 140 'numerator': Fraction(rate).numerator, 141 'denominator': Fraction(rate).denominator 142 }, 143 'duration': { 144 'numerator': Fraction(duration).numerator, 145 'denominator': Fraction(duration).denominator 146 }, 147 'cog_coded_audio': { 148 "format": cog_audio_format, 149 "samples": samples, 150 "channels": channels, 151 "priming": priming, 152 "remainder": remainder, 153 "sample_rate": sample_rate 154 } 155 } 156 } 157 158 if data is None: 159 data = bytearray(length) 160 161 super().__init__(meta, data) 162 self.meta: CodedAudioGrainMetadataDict 163 164 self._factory = "CodedAudioGrain" 165 self.meta['grain']['grain_type'] = 'coded_audio' 166 if 'cog_coded_audio' not in self.meta['grain']: 167 self.meta['grain']['cog_coded_audio'] = {} # type: ignore 168 if 'format' not in self.meta['grain']['cog_coded_audio']: 169 self.meta['grain']['cog_coded_audio']['format'] = int(CogAudioFormat.INVALID) 170 if 'channels' not in self.meta['grain']['cog_coded_audio']: 171 self.meta['grain']['cog_coded_audio']['channels'] = 0 172 if 'samples' not in self.meta['grain']['cog_coded_audio']: 173 self.meta['grain']['cog_coded_audio']['samples'] = 0 174 if 'priming' not in self.meta['grain']['cog_coded_audio']: 175 self.meta['grain']['cog_coded_audio']['priming'] = 0 176 if 'remainder' not in self.meta['grain']['cog_coded_audio']: 177 self.meta['grain']['cog_coded_audio']['remainder'] = 0 178 if 'sample_rate' not in self.meta['grain']['cog_coded_audio']: 179 self.meta['grain']['cog_coded_audio']['sample_rate'] = 48000 180 self.meta['grain']['cog_coded_audio']['format'] = int(self.meta['grain']['cog_coded_audio']['format'])
Inherited Members
- Grain
- has_data
- data
- grain_type
- source_id
- src_id
- flow_id
- origin_timestamp
- origin_timerange
- presentation_origin_timestamp
- final_presentation_origin_timestamp
- presentation_origin_timerange
- normalise_time
- sync_timestamp
- creation_timestamp
- rate
- duration
- timelabels
- add_timelabel
- TIMELABEL
- TIMELABELS
- length
- expected_length
- collections.abc.Sequence
- index
- count
30class EventGrain(Grain): 31 """\ 32A class representing an event grain. 33 34Any grain can be freely cast to a tuple: 35 36 (meta, None) 37 38where meta is a dictionary containing the grain metadata. 39 40The Grain class provides a number of properties which can be used to access 41parts of the standard grain metadata, and this class inherits these: 42 43meta 44 The meta dictionary object 45 46data 47 Either None or an object which can be cast to bytes by passing it to the bytes 48 constructor and will in of itself respond to the python-level portions of the bytes-like 49 object protocol. It is not guaranteed that this object will always respond correctly to the 50 C buffer-protocol, but it can always be converted into something that will by calling bytes on it. 51 52grain_type 53 A string containing the type of the grain, always "event" 54 55source_id 56 A uuid.UUID object representing the source_id in the grain 57 58flow_id 59 A uuid.UUID object representing the flow_id in the grain 60 61origin_timestamp 62 An mediatimestamp.Timestamp object representing the origin timestamp 63 of this grain. 64 65sync_timestamp 66 An mediatimestamp.Timestamp object representing the sync timestamp 67 of this grain. 68 69creation_timestamp 70 An mediatimestamp.Timestamp object representing the creation timestamp 71 of this grain. 72 73rate 74 A fractions.Fraction object representing the grain rate in grains per second. 75 76duration 77 A fractions.Fraction object representing the grain duration in seconds. 78 79timelabels 80 A list object containing time label data 81 82length 83 the length of the json representation in data 84 85The EventGrain class also provides additional properies 86 87event_type 88 A urn representing the type of the event 89 90topic 91 A string which should be an identifier for the topic of the event 92 93event_data 94 A list-like sequence object of EVENTGRAIN.DATA objects representing the 95 data in the event 96 97And the class provides one additional method: 98 99append(path, pre=None, post=None) 100 Adds a new data element to the event_data property with path set to the 101 provided string, and pre and post set optionally. All calls should use 102 only json serialisable objects for the values of pre and post. 103 """ 104 def __init__(self, 105 meta: Optional[EventGrainMetadataDict] = None, 106 data: Optional[GrainDataParameterType] = None, 107 src_id: Optional[UUID] = None, 108 flow_id: Optional[UUID] = None, 109 origin_timestamp: Optional[SupportsMediaTimestamp] = None, 110 creation_timestamp: Optional[SupportsMediaTimestamp] = None, 111 sync_timestamp: Optional[SupportsMediaTimestamp] = None, 112 rate: Fraction = Fraction(25, 1), 113 duration: Fraction = Fraction(1, 25), 114 event_type: str = '', 115 topic: str = ''): 116 117 if meta is None: 118 if not isinstance(src_id, UUID) and src_id is not None: 119 raise AttributeError(f"src_id: Seen type {type(src_id)}, expected UUID.") 120 if not isinstance(flow_id, UUID) and flow_id is not None: 121 raise AttributeError(f"flow_id: Seen type {type(flow_id)}, expected UUID.") 122 123 cts = creation_timestamp 124 if cts is None: 125 cts = Timestamp.get_time() 126 if origin_timestamp is None: 127 origin_timestamp = cts 128 if sync_timestamp is None: 129 sync_timestamp = origin_timestamp 130 meta = EventGrainMetadataDict({ 131 "@_ns": "urn:x-ipstudio:ns:0.1", 132 "grain": { 133 "grain_type": "event", 134 "source_id": str(src_id), 135 "flow_id": str(flow_id), 136 "origin_timestamp": str(mediatimestamp(origin_timestamp)), 137 "sync_timestamp": str(mediatimestamp(sync_timestamp)), 138 "creation_timestamp": str(mediatimestamp(cts)), 139 "rate": { 140 "numerator": Fraction(rate).numerator, 141 "denominator": Fraction(rate).denominator, 142 }, 143 "duration": { 144 "numerator": Fraction(duration).numerator, 145 "denominator": Fraction(duration).denominator, 146 }, 147 "event_payload": { 148 "type": event_type, 149 "topic": topic, 150 "data": [] 151 } 152 }, 153 }) 154 155 super().__init__(meta, None) 156 self.meta: EventGrainMetadataDict 157 158 self._factory = "EventGrain" 159 self.meta['grain']['grain_type'] = 'event' 160 if 'event_payload' not in self.meta['grain']: 161 self.meta['grain']['event_payload'] = { 162 'type': "", 163 'topic': "", 164 'data': []} 165 if isawaitable(data): 166 self._data_fetcher_coroutine = cast(Awaitable[Optional[GrainDataType]], data) 167 elif data is not None: 168 if isinstance(data, bytes): 169 self.data = data 170 else: 171 self.data = bytes(cast(SupportsBytes, data)) 172 if 'type' not in self.meta['grain']['event_payload']: 173 self.meta['grain']['event_payload']['type'] = "" 174 if 'topic' not in self.meta['grain']['event_payload']: 175 self.meta['grain']['event_payload']['topic'] = "" 176 if 'data' not in self.meta['grain']['event_payload']: 177 self.meta['grain']['event_payload']['data'] = [] 178 179 # ignoring typing here because mypy does not accept narrowing of the type here 180 # (bytes is in the GrainDataType union) and it doesn't accept @overload 181 # of Grain.data @property 182 @property # type: ignore 183 def data(self) -> bytes: 184 return json.dumps({'type': self.event_type, 185 'topic': self.topic, 186 'data': [dict(datum) for datum in self.event_data]}).encode('utf-8') 187 188 @data.setter 189 def data(self, value: Union[str, bytes]): 190 if not isinstance(value, str): 191 payload = json.loads(value.decode('utf-8')) 192 else: 193 payload = json.loads(value) 194 195 if 'type' not in payload or 'topic' not in payload or 'data' not in payload: 196 raise ValueError("incorrectly formated event payload") 197 self.event_type = payload['type'] 198 self.topic = payload['topic'] 199 self.meta['grain']['event_payload']['data'] = [] 200 for datum in payload['data']: 201 d: EventGrainDatumDict = {'path': datum['path']} 202 if 'pre' in datum: 203 d['pre'] = datum['pre'] 204 if 'post' in datum: 205 d['post'] = datum['post'] 206 self.meta['grain']['event_payload']['data'].append(d) 207 208 def __repr__(self) -> str: 209 return "EventGrain({!r})".format(self.meta) 210 211 @property 212 def event_type(self) -> str: 213 return self.meta['grain']['event_payload']['type'] 214 215 @event_type.setter 216 def event_type(self, value: str) -> None: 217 self.meta['grain']['event_payload']['type'] = value 218 219 @property 220 def topic(self) -> str: 221 return self.meta['grain']['event_payload']['topic'] 222 223 @topic.setter 224 def topic(self, value: str) -> None: 225 self.meta['grain']['event_payload']['topic'] = value 226 227 class DATA(Mapping): 228 """\ 229A class representing a data element within an event grain. 230 231It can be trated as a dictionary: 232 233 {"path": "/a/path", 234 "pre": <json serialisable object>, 235 "post": <json serialisable object>} 236 237But also provides additional properties: 238 239path 240 The path 241 242pre 243 The pre value, or None if none is present. If set to None will remove "pre" 244 key from dictionary. 245 246post 247 The post value, or None if none is present. If set to None will remove 248 "post" key from dictionary. 249""" 250 def __init__(self, meta: EventGrainDatumDict): 251 self.meta = meta 252 253 def __getitem__(self, key: Literal['path', 'pre', 'post']) -> MediaJSONSerialisable: 254 return self.meta[key] 255 256 def __setitem__(self, key: Literal['path', 'pre', 'post'], value: MediaJSONSerialisable) -> None: 257 self.meta[key] = value 258 259 def __delitem__(self, key: Literal['pre', 'post']) -> None: 260 del self.meta[key] 261 262 def __iter__(self) -> Iterator[str]: 263 return self.meta.__iter__() 264 265 def __len__(self) -> int: 266 return self.meta.__len__() 267 268 def __eq__(self, other: object) -> bool: 269 return dict(self) == other 270 271 def __ne__(self, other: object) -> bool: 272 return not (self == other) 273 274 @property 275 def path(self) -> str: 276 return self.meta['path'] 277 278 @path.setter 279 def path(self, value: str) -> None: 280 self.meta['path'] = value 281 282 @property 283 def pre(self) -> Optional[MediaJSONSerialisable]: 284 if 'pre' in self.meta: 285 return self.meta['pre'] 286 else: 287 return None 288 289 @pre.setter 290 def pre(self, value: Optional[MediaJSONSerialisable]) -> None: 291 if value is not None: 292 self.meta['pre'] = value 293 else: 294 del self.meta['pre'] 295 296 @property 297 def post(self) -> Optional[MediaJSONSerialisable]: 298 if 'post' in self.meta: 299 return self.meta['post'] 300 else: 301 return None 302 303 @post.setter 304 def post(self, value: Optional[MediaJSONSerialisable]) -> None: 305 if value is not None: 306 self.meta['post'] = value 307 elif 'post' in self.meta: 308 del self.meta['post'] 309 310 @property 311 def event_data(self) -> List["EventGrain.DATA"]: 312 return [EventGrain.DATA(datum) for datum in self.meta['grain']['event_payload']['data']] 313 314 @event_data.setter 315 def event_data(self, value: List[EventGrainDatumDict]) -> None: 316 self.meta['grain']['event_payload']['data'] = [cast(EventGrainDatumDict, dict(datum)) for datum in value] 317 318 def append(self, path: str, pre: Optional[MediaJSONSerialisable] = None, 319 post: Optional[MediaJSONSerialisable] = None) -> None: 320 datum = EventGrainDatumDict(path=path) 321 if pre is not None: 322 datum['pre'] = pre 323 if post is not None: 324 datum['post'] = post 325 self.meta['grain']['event_payload']['data'].append(datum)
A class representing an event grain.
Any grain can be freely cast to a tuple:
(meta, None)
where meta is a dictionary containing the grain metadata.
The Grain class provides a number of properties which can be used to access parts of the standard grain metadata, and this class inherits these:
meta The meta dictionary object
data Either None or an object which can be cast to bytes by passing it to the bytes constructor and will in of itself respond to the python-level portions of the bytes-like object protocol. It is not guaranteed that this object will always respond correctly to the C buffer-protocol, but it can always be converted into something that will by calling bytes on it.
grain_type A string containing the type of the grain, always "event"
source_id A uuid.UUID object representing the source_id in the grain
flow_id A uuid.UUID object representing the flow_id in the grain
origin_timestamp An mediatimestamp.Timestamp object representing the origin timestamp of this grain.
sync_timestamp An mediatimestamp.Timestamp object representing the sync timestamp of this grain.
creation_timestamp An mediatimestamp.Timestamp object representing the creation timestamp of this grain.
rate A fractions.Fraction object representing the grain rate in grains per second.
duration A fractions.Fraction object representing the grain duration in seconds.
timelabels A list object containing time label data
length the length of the json representation in data
The EventGrain class also provides additional properies
event_type A urn representing the type of the event
topic A string which should be an identifier for the topic of the event
event_data A list-like sequence object of EVENTGRAIN.DATA objects representing the data in the event
And the class provides one additional method:
append(path, pre=None, post=None) Adds a new data element to the event_data property with path set to the provided string, and pre and post set optionally. All calls should use only json serialisable objects for the values of pre and post.
EventGrain( meta: Optional[mediagrains.typing.EventGrainMetadataDict] = None, data: Union[SupportsBytes, bytes, numpy.ndarray, Awaitable[Union[SupportsBytes, bytes, numpy.ndarray, NoneType]], NoneType] = None, src_id: Optional[uuid.UUID] = None, flow_id: Optional[uuid.UUID] = None, origin_timestamp: Optional[mediatimestamp.immutable.timestamp.SupportsMediaTimestamp] = None, creation_timestamp: Optional[mediatimestamp.immutable.timestamp.SupportsMediaTimestamp] = None, sync_timestamp: Optional[mediatimestamp.immutable.timestamp.SupportsMediaTimestamp] = None, rate: fractions.Fraction = Fraction(25, 1), duration: fractions.Fraction = Fraction(1, 25), event_type: str = '', topic: str = '')
104 def __init__(self, 105 meta: Optional[EventGrainMetadataDict] = None, 106 data: Optional[GrainDataParameterType] = None, 107 src_id: Optional[UUID] = None, 108 flow_id: Optional[UUID] = None, 109 origin_timestamp: Optional[SupportsMediaTimestamp] = None, 110 creation_timestamp: Optional[SupportsMediaTimestamp] = None, 111 sync_timestamp: Optional[SupportsMediaTimestamp] = None, 112 rate: Fraction = Fraction(25, 1), 113 duration: Fraction = Fraction(1, 25), 114 event_type: str = '', 115 topic: str = ''): 116 117 if meta is None: 118 if not isinstance(src_id, UUID) and src_id is not None: 119 raise AttributeError(f"src_id: Seen type {type(src_id)}, expected UUID.") 120 if not isinstance(flow_id, UUID) and flow_id is not None: 121 raise AttributeError(f"flow_id: Seen type {type(flow_id)}, expected UUID.") 122 123 cts = creation_timestamp 124 if cts is None: 125 cts = Timestamp.get_time() 126 if origin_timestamp is None: 127 origin_timestamp = cts 128 if sync_timestamp is None: 129 sync_timestamp = origin_timestamp 130 meta = EventGrainMetadataDict({ 131 "@_ns": "urn:x-ipstudio:ns:0.1", 132 "grain": { 133 "grain_type": "event", 134 "source_id": str(src_id), 135 "flow_id": str(flow_id), 136 "origin_timestamp": str(mediatimestamp(origin_timestamp)), 137 "sync_timestamp": str(mediatimestamp(sync_timestamp)), 138 "creation_timestamp": str(mediatimestamp(cts)), 139 "rate": { 140 "numerator": Fraction(rate).numerator, 141 "denominator": Fraction(rate).denominator, 142 }, 143 "duration": { 144 "numerator": Fraction(duration).numerator, 145 "denominator": Fraction(duration).denominator, 146 }, 147 "event_payload": { 148 "type": event_type, 149 "topic": topic, 150 "data": [] 151 } 152 }, 153 }) 154 155 super().__init__(meta, None) 156 self.meta: EventGrainMetadataDict 157 158 self._factory = "EventGrain" 159 self.meta['grain']['grain_type'] = 'event' 160 if 'event_payload' not in self.meta['grain']: 161 self.meta['grain']['event_payload'] = { 162 'type': "", 163 'topic': "", 164 'data': []} 165 if isawaitable(data): 166 self._data_fetcher_coroutine = cast(Awaitable[Optional[GrainDataType]], data) 167 elif data is not None: 168 if isinstance(data, bytes): 169 self.data = data 170 else: 171 self.data = bytes(cast(SupportsBytes, data)) 172 if 'type' not in self.meta['grain']['event_payload']: 173 self.meta['grain']['event_payload']['type'] = "" 174 if 'topic' not in self.meta['grain']['event_payload']: 175 self.meta['grain']['event_payload']['topic'] = "" 176 if 'data' not in self.meta['grain']['event_payload']: 177 self.meta['grain']['event_payload']['data'] = []
event_data: List[EventGrain.DATA]
def
append( self, path: str, pre: Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]]]]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]]]]]]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]]]]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]]]]]]]], NoneType] = None, post: Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]]]]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]]]]]]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]]]]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]]]]]]]], NoneType] = None) -> None:
318 def append(self, path: str, pre: Optional[MediaJSONSerialisable] = None, 319 post: Optional[MediaJSONSerialisable] = None) -> None: 320 datum = EventGrainDatumDict(path=path) 321 if pre is not None: 322 datum['pre'] = pre 323 if post is not None: 324 datum['post'] = post 325 self.meta['grain']['event_payload']['data'].append(datum)
Inherited Members
- Grain
- has_data
- grain_type
- source_id
- src_id
- flow_id
- origin_timestamp
- final_origin_timestamp
- origin_timerange
- presentation_origin_timestamp
- final_presentation_origin_timestamp
- presentation_origin_timerange
- normalise_time
- media_rate
- sync_timestamp
- creation_timestamp
- rate
- duration
- timelabels
- add_timelabel
- TIMELABEL
- TIMELABELS
- length
- expected_length
- collections.abc.Sequence
- index
- count
class
EventGrain.DATA(collections.abc.Mapping):
227 class DATA(Mapping): 228 """\ 229A class representing a data element within an event grain. 230 231It can be trated as a dictionary: 232 233 {"path": "/a/path", 234 "pre": <json serialisable object>, 235 "post": <json serialisable object>} 236 237But also provides additional properties: 238 239path 240 The path 241 242pre 243 The pre value, or None if none is present. If set to None will remove "pre" 244 key from dictionary. 245 246post 247 The post value, or None if none is present. If set to None will remove 248 "post" key from dictionary. 249""" 250 def __init__(self, meta: EventGrainDatumDict): 251 self.meta = meta 252 253 def __getitem__(self, key: Literal['path', 'pre', 'post']) -> MediaJSONSerialisable: 254 return self.meta[key] 255 256 def __setitem__(self, key: Literal['path', 'pre', 'post'], value: MediaJSONSerialisable) -> None: 257 self.meta[key] = value 258 259 def __delitem__(self, key: Literal['pre', 'post']) -> None: 260 del self.meta[key] 261 262 def __iter__(self) -> Iterator[str]: 263 return self.meta.__iter__() 264 265 def __len__(self) -> int: 266 return self.meta.__len__() 267 268 def __eq__(self, other: object) -> bool: 269 return dict(self) == other 270 271 def __ne__(self, other: object) -> bool: 272 return not (self == other) 273 274 @property 275 def path(self) -> str: 276 return self.meta['path'] 277 278 @path.setter 279 def path(self, value: str) -> None: 280 self.meta['path'] = value 281 282 @property 283 def pre(self) -> Optional[MediaJSONSerialisable]: 284 if 'pre' in self.meta: 285 return self.meta['pre'] 286 else: 287 return None 288 289 @pre.setter 290 def pre(self, value: Optional[MediaJSONSerialisable]) -> None: 291 if value is not None: 292 self.meta['pre'] = value 293 else: 294 del self.meta['pre'] 295 296 @property 297 def post(self) -> Optional[MediaJSONSerialisable]: 298 if 'post' in self.meta: 299 return self.meta['post'] 300 else: 301 return None 302 303 @post.setter 304 def post(self, value: Optional[MediaJSONSerialisable]) -> None: 305 if value is not None: 306 self.meta['post'] = value 307 elif 'post' in self.meta: 308 del self.meta['post']
A class representing a data element within an event grain.
It can be trated as a dictionary:
{"path": "/a/path",
"pre": <json serialisable object>,
"post": <json serialisable object>}
But also provides additional properties:
path The path
pre The pre value, or None if none is present. If set to None will remove "pre" key from dictionary.
post The post value, or None if none is present. If set to None will remove "post" key from dictionary.
pre: Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]]]]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]]]]]]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]]]]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]]]]]]]], NoneType]
post: Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]]]]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]]]]]]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]]]]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]], Mapping[str, Union[str, int, uuid.UUID, mediatimestamp.immutable.timestamp.Timestamp, mediatimestamp.immutable.timerange.TimeRange, fractions.Fraction, Sequence[Any], Mapping[str, Any]]]]]]]]], NoneType]
Inherited Members
- collections.abc.Mapping
- get
- keys
- items
- values
def
attributes_for_grain_type( grain_type: str | mediagrains.grains.Grain.GrainType, new: bool = False) -> List[str]:
145def attributes_for_grain_type(grain_type: str | GrainType, new: bool = False) -> List[str]: 146 """Returns a list of attributes for a particular grain type. Useful for testing.""" 147 # cast to the correct string 148 if type(grain_type) is GrainType: 149 grain_type = ["grain", "video", "audio", "coded_video", "coded_audio", "event"][grain_type] 150 151 COMMON_ATTRS = ['source_id' if not new else "src_id", 'flow_id', 'origin_timestamp', 'sync_timestamp', 152 'creation_timestamp', 'rate', 'duration'] 153 if grain_type == "event": 154 return COMMON_ATTRS + ["event_type", "topic", "event_data"] 155 elif grain_type == "audio": 156 return COMMON_ATTRS + ["format" if not new else "cog_audio_format", "samples", "channels", "sample_rate"] 157 elif grain_type == "coded_audio": 158 return COMMON_ATTRS + ["format" if not new else "cog_audio_format", "samples", "channels", "sample_rate", 159 "priming", "remainder"] 160 elif grain_type == "coded_video": 161 return COMMON_ATTRS + ["format" if not new else "cog_frame_format", "coded_width", "coded_height", 162 "layout" if not new else "cog_frame_layout", "origin_width", "origin_height", 163 "is_key_frame", "temporal_offset", "unit_offsets"] 164 elif grain_type == "video": 165 return COMMON_ATTRS + ["format" if not new else "cog_frame_format", "width", "height", 166 "layout" if not new else "cog_frame_layout"] 167 else: 168 return COMMON_ATTRS
Returns a list of attributes for a particular grain type. Useful for testing.
def
size_for_audio_format( cog_audio_format: mediagrains.cogenums.CogAudioFormat, channels: int, samples: int) -> int:
16def size_for_audio_format(cog_audio_format: CogAudioFormat, channels: int, samples: int) -> int: 17 if (cog_audio_format & 0x200) == 0x200: # compressed format, no idea of correct size 18 return 0 19 20 if (cog_audio_format & 0x3) == 0x1: 21 channels += 1 22 channels //= 2 23 channels *= 2 24 if (cog_audio_format & 0xC) == 0xC: 25 depth = 8 26 elif (cog_audio_format & 0xf) == 0x04: 27 depth = 4 28 else: 29 depth = ((cog_audio_format & 0xf) >> 2) + 2 30 return channels * samples * depth
def
GrainFactory( meta: Union[mediagrains.typing.EmptyGrainMetadataDict, mediagrains.typing.AudioGrainMetadataDict, mediagrains.typing.CodedAudioGrainMetadataDict, mediagrains.typing.VideoGrainMetadataDict, mediagrains.typing.CodedVideoGrainMetadataDict, mediagrains.typing.EventGrainMetadataDict, NoneType] = None, data: Union[SupportsBytes, bytes, numpy.ndarray, Awaitable[Union[SupportsBytes, bytes, numpy.ndarray, NoneType]], NoneType] = None, src_id: Optional[uuid.UUID] = None, flow_id: Optional[uuid.UUID] = None, origin_timestamp: Optional[mediatimestamp.immutable.timestamp.SupportsMediaTimestamp] = None, sync_timestamp: Optional[mediatimestamp.immutable.timestamp.SupportsMediaTimestamp] = None, creation_timestamp: Optional[mediatimestamp.immutable.timestamp.SupportsMediaTimestamp] = None, rate: fractions.Fraction = Fraction(0, 1), duration: fractions.Fraction = Fraction(0, 1), **kwargs):
59def GrainFactory(meta: Optional[GrainMetadataDict] = None, 60 data: GrainDataParameterType = None, 61 src_id: Optional[UUID] = None, 62 flow_id: Optional[UUID] = None, 63 origin_timestamp: Optional[SupportsMediaTimestamp] = None, 64 sync_timestamp: Optional[SupportsMediaTimestamp] = None, 65 creation_timestamp: Optional[SupportsMediaTimestamp] = None, 66 rate: Fraction = Fraction(0, 1), 67 duration: Fraction = Fraction(0, 1), 68 **kwargs): 69 from .VideoGrain import VideoGrain 70 from .AudioGrain import AudioGrain 71 from .CodedVideoGrain import CodedVideoGrain 72 from .CodedAudioGrain import CodedAudioGrain 73 from .EventGrain import EventGrain 74 if meta is None: 75 if creation_timestamp is None: 76 creation_timestamp = Timestamp.get_time() 77 if origin_timestamp is None: 78 origin_timestamp = creation_timestamp 79 if sync_timestamp is None: 80 sync_timestamp = origin_timestamp 81 82 if src_id is None: 83 raise AttributeError("src_id is None. Meta is None so src_id must not be None.") 84 if flow_id is None: 85 raise AttributeError("flow_id is None. Meta is None so flow_id must not be None.") 86 87 if not isinstance(src_id, UUID): 88 raise AttributeError(f"src_id: Seen type {type(src_id)}, expected UUID.") 89 if not isinstance(flow_id, UUID): 90 raise AttributeError(f"flow_id: Seen type {type(flow_id)}, expected UUID.") 91 92 meta = EmptyGrainMetadataDict({ 93 "@_ns": "urn:x-ipstudio:ns:0.1", 94 "grain": { 95 'grain_type': "empty", 96 'source_id': str(src_id), 97 'flow_id': str(flow_id), 98 'origin_timestamp': str(mediatimestamp(origin_timestamp)), 99 'sync_timestamp': str(mediatimestamp(sync_timestamp)), 100 'creation_timestamp': str(mediatimestamp(creation_timestamp)), 101 'rate': { 102 'numerator': Fraction(rate).numerator, 103 'denominator': Fraction(rate).denominator 104 }, 105 'duration': { 106 'numerator': Fraction(duration).numerator, 107 'denominator': Fraction(duration).denominator 108 } 109 } 110 }) 111 data = None 112 if 'grain' in meta and 'grain_type' in meta['grain'] and meta['grain']['grain_type'] == 'video': 113 return VideoGrain(cast(VideoGrainMetadataDict, meta), data) 114 elif 'grain' in meta and 'grain_type' in meta['grain'] and meta['grain']['grain_type'] == 'audio': 115 return AudioGrain(cast(AudioGrainMetadataDict, meta), data) 116 elif 'grain' in meta and 'grain_type' in meta['grain'] and meta['grain']['grain_type'] == 'coded_video': 117 return CodedVideoGrain(cast(CodedVideoGrainMetadataDict, meta), data) 118 elif 'grain' in meta and 'grain_type' in meta['grain'] and meta['grain']['grain_type'] == 'coded_audio': 119 return CodedAudioGrain(cast(CodedAudioGrainMetadataDict, meta), data) 120 elif 'grain' in meta and 'grain_type' in meta['grain'] and meta['grain']['grain_type'] in ['event', 'data']: 121 return EventGrain(cast(EventGrainMetadataDict, meta), data) 122 else: 123 return Grain(cast(GrainMetadataDict, meta), data)