-
Notifications
You must be signed in to change notification settings - Fork 9
Expand file tree
/
Copy pathrawdata_bytes.py
More file actions
126 lines (100 loc) · 3.52 KB
/
rawdata_bytes.py
File metadata and controls
126 lines (100 loc) · 3.52 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
from .._utils import get_samplesize
from pluggy import HookimplMarker
from typing import Tuple
hookimpl = HookimplMarker("ffmpegio")
@hookimpl
def video_info(obj: dict) -> Tuple[Tuple[int, int, int], str]:
"""get video frame info
:param obj: dict containing video frame data with arbitrary number of frames
:type obj: object
:return: shape (height,width,components) and data type in numpy dtype str expression
:rtype: Tuple[Tuple[int, int, int], str]
"""
try:
return obj["shape"][-3:], obj["dtype"]
except:
return None
@hookimpl
def audio_info(obj: object) -> Tuple[int, str]:
"""get audio sample info
:param obj: dict containing audio data (with interleaving channels) with arbitrary number of samples
:type obj: dict
:return: number of channels and sample data type in numpy dtype str expression
:rtype: Tuple[Tuple[int], str]
"""
try:
return obj["shape"][-1:], obj["dtype"]
except:
return None
@hookimpl
def video_bytes(obj: object) -> memoryview:
"""return bytes-like object of packed video pixels, associated with `video_info()`
:param obj: dict containing video frame data with arbitrary number of frames
:type obj: dict
:return: packed bytes of video frames
:rtype: bytes-like object
"""
try:
return obj["buffer"]
except:
return None
@hookimpl
def audio_bytes(obj: object) -> memoryview:
"""return bytes-like object of packed audio samples
:param obj: dict containing audio data (with interleaving channels) with arbitrary number of samples
:type obj: dict
:return: packed bytes of audio samples
:rtype: bytes-like object
"""
try:
return obj["buffer"]
except:
return None
@hookimpl
def bytes_to_video(
b: bytes, dtype: str, shape: Tuple[int, int, int], squeeze: bool
) -> object:
"""convert bytes to rawvideo object
:param b: byte data of arbitrary number of video frames
:type b: bytes
:param dtype: data type numpy dtype string (e.g., '|u1', '<f4')
:type dtype: str
:param size: frame dimension in pixels and number of color components (height, width, components)
:type size: Tuple[int, int, int]
:param squeeze: True to remove all the singular dimensions
:type squeeze: bool
:return: dict holding the rawvideo frame data
:rtype: dict['buffer':bytes, 'dtype':str, 'shape': Tuple[int,int,int]]
"""
sh = (len(b) // get_samplesize(shape, dtype), *shape)
try:
return {
"buffer": b,
"dtype": dtype,
"shape": tuple(((i for i in sh if i != 1))) if squeeze else sh,
}
except:
return None
@hookimpl
def bytes_to_audio(b: bytes, dtype: str, shape: Tuple[int], squeeze: bool) -> object:
"""convert bytes to rawaudio object
:param b: byte data of arbitrary number of video frames
:type b: bytes
:param dtype: numpy dtype string of the bytes (e.g., '<s2', '<f4')
:type dtype: str
:param shape: number of interleaved audio channels (1-element tuple)
:type shape: Tuple[int]
:param squeeze: True to remove all the singular dimensions
:type squeeze: bool
:return: dict to hold the raw audio samples
:rtype: dict['buffer':bytes, 'dtype':str, 'shape': Tuple[int]]
"""
try:
sh = (len(b) // get_samplesize(shape, dtype), *shape)
return {
"buffer": b,
"dtype": dtype,
"shape": tuple(((i for i in sh if i != 1))) if squeeze else sh,
}
except:
return None