from abc import abstractmethod, abstractproperty
from ebu_tt_live.utils import AutoRegisteringABCMeta, AbstractStaticMember, validate_types_only
# Interfaces
# ==========
[docs]class ICarriageMechanism(object):
"""
Basic interface for the carrige mechanisms
"""
__metaclass__ = AutoRegisteringABCMeta
[docs]class IProducerCarriage(ICarriageMechanism):
"""
Carriage mechanism interface for producer nodes.
"""
_expects = AbstractStaticMember(validate_types_only)
[docs] @classmethod
def expects(cls):
"""
Data type expected
:return:
"""
if isinstance(cls._expects, AbstractStaticMember):
raise TypeError('Classmethod relies on abstract property: \'_expects\'')
return cls._expects
[docs] @abstractmethod
def register_producer_node(self, node):
"""
Register the producer node in the carriage mechanism
:param node: The node to connect to.
"""
@abstractproperty
def producer_node(self):
"""
Node accessor
:return:
"""
[docs] @abstractmethod
def emit_data(self, data, **kwargs):
"""
Implement protocol specific postprocessing here.
:param kwargs: Extra parameters to send down
:param data:
:return:
"""
[docs] @abstractmethod
def resume_producing(self):
"""
This makes sure that the producers can be pulled. This is good for timer or manual triggering
:return:
"""
[docs]class IConsumerCarriage(ICarriageMechanism):
"""
Carriage mechanism interface for consumer nodes.
"""
_provides = AbstractStaticMember(validate_types_only)
[docs] @classmethod
def provides(cls):
"""
Data type provided
:return:
"""
if isinstance(cls._provides, AbstractStaticMember):
raise TypeError('Classmethod relies on abstract property: \'_provides\'')
return cls._provides
[docs] @abstractmethod
def register_consumer_node(self, node):
"""
Register the consumer node in the carriage mechanism
:param node:
:return:
"""
@abstractproperty
def consumer_node(self):
"""
Node accessor
:return:
"""
[docs] @abstractmethod
def on_new_data(self, data, **kwargs):
"""
Implement protocol specific preprocessing here.
:param kwargs: Extra parameters to send down
"""