mediagrains.grains.AudioGrain

  1from uuid import UUID
  2from mediatimestamp.immutable import Timestamp, SupportsMediaTimestamp, mediatimestamp, TimeOffset
  3from fractions import Fraction
  4
  5from typing import (
  6    Optional)
  7from ..typing import (
  8    AudioGrainMetadataDict,
  9    GrainDataParameterType)
 10
 11from ..cogenums import CogAudioFormat
 12from .Grain import Grain
 13
 14
 15def size_for_audio_format(cog_audio_format: CogAudioFormat, channels: int, samples: int) -> int:
 16    if (cog_audio_format & 0x200) == 0x200:  # compressed format, no idea of correct size
 17        return 0
 18
 19    if (cog_audio_format & 0x3) == 0x1:
 20        channels += 1
 21        channels //= 2
 22        channels *= 2
 23    if (cog_audio_format & 0xC) == 0xC:
 24        depth = 8
 25    elif (cog_audio_format & 0xf) == 0x04:
 26        depth = 4
 27    else:
 28        depth = ((cog_audio_format & 0xf) >> 2) + 2
 29    return channels * samples * depth
 30
 31
 32class AudioGrain(Grain):
 33    """\
 34A class representing a raw audio grain.
 35
 36Any grain can be freely cast to a tuple:
 37
 38  (meta, data)
 39
 40where meta is a dictionary containing the grain metadata, and data is the data element described below..
 41
 42The Grain class provides a number of properties which can be used to access
 43parts of the standard grain metadata, and this class inherits these:
 44
 45meta
 46    The meta dictionary object
 47
 48data
 49    Either None or an object which can be cast to bytes by passing it to the bytes
 50    constructor and will in of itself respond to the python-level portions of the bytes-like
 51    object protocol. It is not guaranteed that this object will always respond correctly to the
 52    C buffer-protocol, but it can always be converted into something that will by calling bytes on it.
 53
 54grain_type
 55    A string containing the type of the grain, always "audio"
 56
 57src_id
 58    A uuid.UUID object representing the source_id in the grain
 59
 60flow_id
 61    A uuid.UUID object representing the flow_id in the grain
 62
 63origin_timestamp
 64    An mediatimestamp.Timestamp object representing the origin timestamp
 65    of this grain.
 66
 67sync_timestamp
 68    An mediatimestamp.Timestamp object representing the sync timestamp
 69    of this grain.
 70
 71creation_timestamp
 72    An mediatimestamp.Timestamp object representing the creation timestamp
 73    of this grain.
 74
 75rate
 76    A fractions.Fraction object representing the grain rate in grains per second.
 77
 78duration
 79    A fractions.Fraction object representing the grain duration in seconds.
 80
 81timelabels
 82    A list object containing time label data
 83
 84length
 85    The length of the data element or 0 if that is None
 86
 87The AudioGrain class also provides additional properies
 88
 89format
 90    An enumerated value of type CogAudioFormat
 91
 92samples
 93    The number of audio samples per channel in this grain
 94
 95channels
 96    The number of channels in this grain
 97
 98sample_rate
 99    An integer indicating the number of samples per channel per second in this
100    audio flow.
101"""
102    def __init__(self,
103                 meta: Optional[AudioGrainMetadataDict] = None,
104                 data: Optional[GrainDataParameterType] = None,
105                 src_id: Optional[UUID] = None,
106                 flow_id: Optional[UUID] = None,
107                 origin_timestamp: Optional[SupportsMediaTimestamp] = None,
108                 sync_timestamp: Optional[SupportsMediaTimestamp] = None,
109                 creation_timestamp: Optional[SupportsMediaTimestamp] = None,
110                 rate: Fraction = Fraction(25, 1),
111                 duration: Fraction = Fraction(1, 25),
112                 cog_audio_format: CogAudioFormat = CogAudioFormat.INVALID,
113                 samples: int = 0,
114                 channels: int = 0,
115                 sample_rate: int = 48000):
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            if creation_timestamp is None:
124                creation_timestamp = Timestamp.get_time()
125            if origin_timestamp is None:
126                origin_timestamp = creation_timestamp
127            if sync_timestamp is None:
128                sync_timestamp = origin_timestamp
129            meta = {
130                "@_ns": "urn:x-ipstudio:ns:0.1",
131                "grain": {
132                    'grain_type': "audio",
133                    'source_id': str(src_id),
134                    'flow_id': str(flow_id),
135                    'origin_timestamp': str(mediatimestamp(origin_timestamp)),
136                    'sync_timestamp': str(mediatimestamp(sync_timestamp)),
137                    'creation_timestamp': str(mediatimestamp(creation_timestamp)),
138                    'rate': {
139                        'numerator': Fraction(rate).numerator,
140                        'denominator': Fraction(rate).denominator
141                    },
142                    'duration': {
143                        'numerator': Fraction(duration).numerator,
144                        'denominator': Fraction(duration).denominator
145                    },
146                    'cog_audio': {
147                        "format": cog_audio_format,
148                        "samples": samples,
149                        "channels": channels,
150                        "sample_rate": sample_rate
151                    }
152                }
153            }
154
155        if data is None:
156            size = size_for_audio_format(cog_audio_format, channels, samples)
157            data = bytearray(size)
158
159        super().__init__(meta, data)
160        self.meta: AudioGrainMetadataDict
161
162        self._factory = "AudioGrain"
163        self.meta['grain']['grain_type'] = 'audio'
164        if 'cog_audio' not in self.meta['grain']:
165            self.meta['grain']['cog_audio'] = {}  # type: ignore
166        if 'format' not in self.meta['grain']['cog_audio']:
167            self.meta['grain']['cog_audio']['format'] = int(CogAudioFormat.INVALID)
168        if 'samples' not in self.meta['grain']['cog_audio']:
169            self.meta['grain']['cog_audio']['samples'] = 0
170        if 'channels' not in self.meta['grain']['cog_audio']:
171            self.meta['grain']['cog_audio']['channels'] = 0
172        if 'sample_rate' not in self.meta['grain']['cog_audio']:
173            self.meta['grain']['cog_audio']['sample_rate'] = 0
174        self.meta['grain']['cog_audio']['format'] = int(self.meta['grain']['cog_audio']['format'])
175
176    def final_origin_timestamp(self) -> Timestamp:
177        return (self.origin_timestamp + TimeOffset.from_count(self.samples - 1, self.sample_rate, 1))
178
179    @property
180    def format(self) -> CogAudioFormat:
181        return CogAudioFormat(self.meta['grain']['cog_audio']['format'])
182
183    @format.setter
184    def format(self, value: CogAudioFormat) -> None:
185        self.meta['grain']['cog_audio']['format'] = int(value)
186
187    @property
188    def cog_audio_format(self) -> CogAudioFormat:
189        return CogAudioFormat(self.meta['grain']['cog_audio']['format'])
190
191    @cog_audio_format.setter
192    def cog_audio_format(self, value: CogAudioFormat) -> None:
193        self.meta['grain']['cog_audio']['format'] = int(value)
194
195    @property
196    def samples(self) -> int:
197        return self.meta['grain']['cog_audio']['samples']
198
199    @samples.setter
200    def samples(self, value: int) -> None:
201        self.meta['grain']['cog_audio']['samples'] = int(value)
202
203    @property
204    def channels(self) -> int:
205        return self.meta['grain']['cog_audio']['channels']
206
207    @channels.setter
208    def channels(self, value: int) -> None:
209        self.meta['grain']['cog_audio']['channels'] = int(value)
210
211    @property
212    def sample_rate(self) -> int:
213        return self.meta['grain']['cog_audio']['sample_rate']
214
215    @sample_rate.setter
216    def sample_rate(self, value: int) -> None:
217        self.meta['grain']['cog_audio']['sample_rate'] = int(value)
218
219    @property
220    def expected_length(self) -> int:
221        return size_for_audio_format(self.format, self.channels, self.samples)
222
223    @property
224    def media_rate(self) -> Optional[Fraction]:
225        if self.sample_rate:
226            return Fraction(self.sample_rate, 1)
227        else:
228            return None
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
class AudioGrain(mediagrains.grains.Grain.Grain):
 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'])
meta: mediagrains.typing.AudioGrainMetadataDict
def final_origin_timestamp(self) -> mediatimestamp.immutable.timestamp.Timestamp:
177    def final_origin_timestamp(self) -> Timestamp:
178        return (self.origin_timestamp + TimeOffset.from_count(self.samples - 1, self.sample_rate, 1))
format: mediagrains.cogenums.CogAudioFormat
180    @property
181    def format(self) -> CogAudioFormat:
182        return CogAudioFormat(self.meta['grain']['cog_audio']['format'])
cog_audio_format: mediagrains.cogenums.CogAudioFormat
188    @property
189    def cog_audio_format(self) -> CogAudioFormat:
190        return CogAudioFormat(self.meta['grain']['cog_audio']['format'])
samples: int
196    @property
197    def samples(self) -> int:
198        return self.meta['grain']['cog_audio']['samples']
channels: int
204    @property
205    def channels(self) -> int:
206        return self.meta['grain']['cog_audio']['channels']
sample_rate: int
212    @property
213    def sample_rate(self) -> int:
214        return self.meta['grain']['cog_audio']['sample_rate']
expected_length: int
220    @property
221    def expected_length(self) -> int:
222        return size_for_audio_format(self.format, self.channels, self.samples)
media_rate: Optional[fractions.Fraction]
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