summaryrefslogtreecommitdiff
path: root/src/py/cam/helpers.py
blob: 2d906667eba0333fe44fc95e5f89248d56e7a271 (plain)
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
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
# SPDX-License-Identifier: GPL-2.0-or-later
# Copyright (C) 2022, Tomi Valkeinen <tomi.valkeinen@ideasonboard.com>
#
# Debayering code from PiCamera documentation

from numpy.lib.stride_tricks import as_strided
import libcamera as libcam
import libcamera.utils
import numpy as np


def demosaic(data, r0, g0, g1, b0):
    # Separate the components from the Bayer data to RGB planes

    rgb = np.zeros(data.shape + (3,), dtype=data.dtype)
    rgb[r0[1]::2, r0[0]::2, 0] = data[r0[1]::2, r0[0]::2]  # Red
    rgb[g0[1]::2, g0[0]::2, 1] = data[g0[1]::2, g0[0]::2]  # Green
    rgb[g1[1]::2, g1[0]::2, 1] = data[g1[1]::2, g1[0]::2]  # Green
    rgb[b0[1]::2, b0[0]::2, 2] = data[b0[1]::2, b0[0]::2]  # Blue

    # Below we present a fairly naive de-mosaic method that simply
    # calculates the weighted average of a pixel based on the pixels
    # surrounding it. The weighting is provided by a byte representation of
    # the Bayer filter which we construct first:

    bayer = np.zeros(rgb.shape, dtype=np.uint8)
    bayer[r0[1]::2, r0[0]::2, 0] = 1  # Red
    bayer[g0[1]::2, g0[0]::2, 1] = 1  # Green
    bayer[g1[1]::2, g1[0]::2, 1] = 1  # Green
    bayer[b0[1]::2, b0[0]::2, 2] = 1  # Blue

    # Allocate an array to hold our output with the same shape as the input
    # data. After this we define the size of window that will be used to
    # calculate each weighted average (3x3). Then we pad out the rgb and
    # bayer arrays, adding blank pixels at their edges to compensate for the
    # size of the window when calculating averages for edge pixels.

    output = np.empty(rgb.shape, dtype=rgb.dtype)
    window = (3, 3)
    borders = (window[0] - 1, window[1] - 1)
    border = (borders[0] // 2, borders[1] // 2)

    rgb = np.pad(rgb, [
        (border[0], border[0]),
        (border[1], border[1]),
        (0, 0),
    ], 'constant')
    bayer = np.pad(bayer, [
        (border[0], border[0]),
        (border[1], border[1]),
        (0, 0),
    ], 'constant')

    # For each plane in the RGB data, we use a nifty numpy trick
    # (as_strided) to construct a view over the plane of 3x3 matrices. We do
    # the same for the bayer array, then use Einstein summation on each
    # (np.sum is simpler, but copies the data so it's slower), and divide
    # the results to get our weighted average:

    for plane in range(3):
        p = rgb[..., plane]
        b = bayer[..., plane]
        pview = as_strided(p, shape=(
            p.shape[0] - borders[0],
            p.shape[1] - borders[1]) + window, strides=p.strides * 2)
        bview = as_strided(b, shape=(
            b.shape[0] - borders[0],
            b.shape[1] - borders[1]) + window, strides=b.strides * 2)
        psum = np.einsum('ijkl->ij', pview)
        bsum = np.einsum('ijkl->ij', bview)
        output[..., plane] = psum // bsum

    return output


def to_rgb(fmt, size, data):
    w = size.width
    h = size.height

    if fmt == libcam.formats.YUYV:
        # YUV422
        yuyv = data.reshape((h, w // 2 * 4))

        # YUV444
        yuv = np.empty((h, w, 3), dtype=np.uint8)
        yuv[:, :, 0] = yuyv[:, 0::2]                    # Y
        yuv[:, :, 1] = yuyv[:, 1::4].repeat(2, axis=1)  # U
        yuv[:, :, 2] = yuyv[:, 3::4].repeat(2, axis=1)  # V

        m = np.array([
            [1.0, 1.0, 1.0],
            [-0.000007154783816076815, -0.3441331386566162, 1.7720025777816772],
            [1.4019975662231445, -0.7141380310058594, 0.00001542569043522235]
        ])

        rgb = np.dot(yuv, m)
        rgb[:, :, 0] -= 179.45477266423404
        rgb[:, :, 1] += 135.45870971679688
        rgb[:, :, 2] -= 226.8183044444304
        rgb = rgb.astype(np.uint8)

    elif fmt == libcam.formats.RGB888:
        rgb = data.reshape((h, w, 3))
        rgb[:, :, [0, 1, 2]] = rgb[:, :, [2, 1, 0]]

    elif fmt == libcam.formats.BGR888:
        rgb = data.reshape((h, w, 3))

    elif fmt in [libcam.formats.ARGB8888, libcam.formats.XRGB8888]:
        rgb = data.reshape((h, w, 4))
        rgb = np.flip(rgb, axis=2)
        # drop alpha component
        rgb = np.delete(rgb, np.s_[0::4], axis=2)

    elif str(fmt).startswith('S'):
        fmt = str(fmt)
        bayer_pattern = fmt[1:5]
        bitspp = int(fmt[5:])

        if bitspp == 8:
            data = data.reshape((h, w))
            data = data.astype(np.uint16)
        elif bitspp in [10, 12]:
            data = data.view(np.uint16)
            data = data.reshape((h, w))
        else:
            raise Exception('Bad bitspp:' + str(bitspp))

        idx = bayer_pattern.find('R')
        assert(idx != -1)
        r0 = (idx % 2, idx // 2)

        idx = bayer_pattern.find('G')
        assert(idx != -1)
        g0 = (idx % 2, idx // 2)

        idx = bayer_pattern.find('G', idx + 1)
        assert(idx != -1)
        g1 = (idx % 2, idx // 2)

        idx = bayer_pattern.find('B')
        assert(idx != -1)
        b0 = (idx % 2, idx // 2)

        rgb = demosaic(data, r0, g0, g1, b0)
        rgb = (rgb >> (bitspp - 8)).astype(np.uint8)

    else:
        rgb = None

    return rgb


# A naive format conversion to 24-bit RGB
def mfb_to_rgb(mfb: libcamera.utils.MappedFrameBuffer, cfg: libcam.StreamConfiguration):
    data = np.array(mfb.planes[0], dtype=np.uint8)
    rgb = to_rgb(cfg.pixel_format, cfg.size, data)
    return rgb
pan class="hl opt">): if mojom.IsArrayKind(element): return GetAllAttrs(element.kind) if mojom.IsMapKind(element): return {**GetAllAttrs(element.key_kind), **GetAllAttrs(element.value_kind)} if isinstance(element, mojom.Parameter): return GetAllAttrs(element.kind) if mojom.IsEnumKind(element): return element.attributes if element.attributes is not None else {} if mojom.IsStructKind(element) and len(element.fields) == 0: return element.attributes if element.attributes is not None else {} if not mojom.IsStructKind(element): if hasattr(element, 'attributes'): return element.attributes or {} return {} attrs = [(x.attributes) for x in element.fields] ret = {} for d in attrs: ret.update(d or {}) if hasattr(element, 'attributes'): ret.update(element.attributes or {}) return ret def NeedsControlSerializer(element): types = GetAllTypes(element) for type in ['ControlList', 'ControlInfoMap']: if f'x:{type}' in types: raise Exception(f'Unknown type "{type}" in {element.mojom_name}, did you mean "libcamera.{type}"?') return "ControlList" in types or "ControlInfoMap" in types def HasFd(element): attrs = GetAllAttrs(element) if isinstance(element, mojom.Kind): types = GetAllTypes(element) else: types = GetAllTypes(element.kind) return "SharedFD" in types or (attrs is not None and "hasFd" in attrs) def WithDefaultValues(element): return [x for x in element if HasDefaultValue(x)] def WithFds(element): return [x for x in element if HasFd(x)] def MethodParamInputs(method): return method.parameters def MethodParamOutputs(method): if method.response_parameters is None: return [] if MethodReturnValue(method) == 'void': return method.response_parameters if len(method.response_parameters) <= 1: return [] return method.response_parameters[1:] def MethodParamsHaveFd(parameters): return len([x for x in parameters if HasFd(x)]) > 0 def MethodInputHasFd(method): return MethodParamsHaveFd(method.parameters) def MethodOutputHasFd(method): return MethodParamsHaveFd(MethodParamOutputs(method)) def MethodParamNames(method): params = [] for param in method.parameters: params.append(param.mojom_name) for param in MethodParamOutputs(method): params.append(param.mojom_name) return params def MethodParameters(method): params = [] for param in method.parameters: params.append('const %s %s%s' % (GetNameForElement(param), '' if IsPod(param) or IsEnum(param) else '&', param.mojom_name)) for param in MethodParamOutputs(method): params.append(f'{GetNameForElement(param)} *{param.mojom_name}') return params def MethodReturnValue(method): if method.response_parameters is None or len(method.response_parameters) == 0: return 'void' first_output = method.response_parameters[0] if ((len(method.response_parameters) == 1 and IsPod(first_output)) or first_output.kind == mojom.INT32): return GetNameForElement(first_output) return 'void' def IsAsync(method): # Events are always async if re.match("^IPA.*EventInterface$", method.interface.mojom_name): return True elif re.match("^IPA.*Interface$", method.interface.mojom_name): if method.attributes is None: return False elif 'async' in method.attributes and method.attributes['async']: return True return False def IsArray(element): return mojom.IsArrayKind(element.kind) def IsControls(element): return mojom.IsStructKind(element.kind) and (element.kind.mojom_name == "ControlList" or element.kind.mojom_name == "ControlInfoMap") def IsEnum(element): return mojom.IsEnumKind(element.kind) # Only works the enum definition, not types def IsScoped(element): attributes = getattr(element, 'attributes', None) if not attributes: return False return 'scopedEnum' in attributes def IsEnumScoped(element): if not IsEnum(element): return False return IsScoped(element.kind) def IsFd(element): return mojom.IsStructKind(element.kind) and element.kind.mojom_name == "SharedFD" def IsFlags(element): attributes = getattr(element, 'attributes', None) if not attributes: return False return 'flags' in attributes def IsMap(element): return mojom.IsMapKind(element.kind) def IsPlainStruct(element): return mojom.IsStructKind(element.kind) and not IsControls(element) and not IsFd(element) def IsPod(element): return element.kind in _kind_to_cpp_type def IsStr(element): return element.kind.spec == 's' def BitWidth(element): if element.kind in _bit_widths: return _bit_widths[element.kind] if mojom.IsEnumKind(element.kind): return '32' return '' def ByteWidthFromCppType(t): key = None for mojo_type, cpp_type in _kind_to_cpp_type.items(): if t == cpp_type: key = mojo_type if key is None: raise Exception('invalid type') return str(int(_bit_widths[key]) // 8) # Get the type name for a given element def GetNameForElement(element): # Flags if IsFlags(element): return f'Flags<{GetFullNameForElement(element.kind)}>' # structs if (mojom.IsEnumKind(element) or mojom.IsInterfaceKind(element) or mojom.IsStructKind(element)): return element.mojom_name # vectors if (mojom.IsArrayKind(element)): elem_name = GetFullNameForElement(element.kind) return f'std::vector<{elem_name}>' # maps if (mojom.IsMapKind(element)): key_name = GetFullNameForElement(element.key_kind) value_name = GetFullNameForElement(element.value_kind) return f'std::map<{key_name}, {value_name}>' # struct fields and function parameters if isinstance(element, (mojom.Field, mojom.Method, mojom.Parameter)): # maps and vectors if (mojom.IsArrayKind(element.kind) or mojom.IsMapKind(element.kind)): return GetNameForElement(element.kind) # strings if (mojom.IsReferenceKind(element.kind) and element.kind.spec == 's'): return 'std::string' # PODs if element.kind in _kind_to_cpp_type: return _kind_to_cpp_type[element.kind] # structs and enums return element.kind.mojom_name # PODs that are members of vectors/maps if (hasattr(element, '__hash__') and element in _kind_to_cpp_type): return _kind_to_cpp_type[element] if (hasattr(element, 'spec')): # strings that are members of vectors/maps if (element.spec == 's'): return 'std::string' # structs that aren't defined in mojom that are members of vectors/maps if (element.spec[0] == 'x'): return element.spec.replace('x:', '').replace('.', '::') if (mojom.IsInterfaceRequestKind(element) or mojom.IsAssociatedKind(element) or mojom.IsPendingRemoteKind(element) or mojom.IsPendingReceiverKind(element) or mojom.IsUnionKind(element)): raise Exception('Unsupported element: %s' % element) raise Exception('Unexpected element: %s' % element) def GetFullNameForElement(element): name = GetNameForElement(element) namespace_str = '' if (mojom.IsStructKind(element) or mojom.IsEnumKind(element)): namespace_str = element.module.mojom_namespace.replace('.', '::') elif (hasattr(element, 'kind') and (mojom.IsStructKind(element.kind) or mojom.IsEnumKind(element.kind))): namespace_str = element.kind.module.mojom_namespace.replace('.', '::') if namespace_str == '': return name if IsFlags(element): return GetNameForElement(element) return f'{namespace_str}::{name}' def ValidateZeroLength(l, s, cap=True): if l is None: return if len(l) > 0: raise Exception(f'{s.capitalize() if cap else s} should be empty') def ValidateSingleLength(l, s, cap=True): if len(l) > 1: raise Exception(f'Only one {s} allowed') if len(l) < 1: raise Exception(f'{s.capitalize() if cap else s} is required') def GetMainInterface(interfaces): intf = [x for x in interfaces if re.match("^IPA.*Interface", x.mojom_name) and not re.match("^IPA.*EventInterface", x.mojom_name)] ValidateSingleLength(intf, 'main interface') return None if len(intf) == 0 else intf[0] def GetEventInterface(interfaces): event = [x for x in interfaces if re.match("^IPA.*EventInterface", x.mojom_name)] ValidateSingleLength(event, 'event interface') return None if len(event) == 0 else event[0] def ValidateNamespace(namespace): if namespace == '': raise Exception('Must have a namespace') if not re.match(r'^ipa\.[0-9A-Za-z_]+', namespace): raise Exception('Namespace must be of the form "ipa.{pipeline_name}"') def ValidateInterfaces(interfaces): # Validate presence of main interface intf = GetMainInterface(interfaces) if intf is None: raise Exception('Must have main IPA interface') # Validate presence of event interface event = GetEventInterface(interfaces) if intf is None: raise Exception('Must have event IPA interface') # Validate required main interface functions f_init = [x for x in intf.methods if x.mojom_name == 'init'] f_start = [x for x in intf.methods if x.mojom_name == 'start'] f_stop = [x for x in intf.methods if x.mojom_name == 'stop'] ValidateSingleLength(f_init, 'init()', False) ValidateSingleLength(f_start, 'start()', False) ValidateSingleLength(f_stop, 'stop()', False) f_stop = f_stop[0] # No need to validate init() and start() as they are customizable # Validate parameters to stop() ValidateZeroLength(f_stop.parameters, 'input parameter to stop()') ValidateZeroLength(f_stop.parameters, 'output parameter from stop()') # Validate that event interface has at least one event if len(event.methods) < 1: raise Exception('Event interface must have at least one event') # Validate that all async methods don't have return values intf_methods_async = [x for x in intf.methods if IsAsync(x)] for method in intf_methods_async: ValidateZeroLength(method.response_parameters, f'{method.mojom_name} response parameters', False) event_methods_async = [x for x in event.methods if IsAsync(x)] for method in event_methods_async: