INNER CODE UNIT · Python

_decode_shared_response

unrealcv/unrealcv · client/python/unrealcv/__init__.py:610

    def _decode_shared_response(response, output_format):
        if isinstance(response, bytes):
            response = response.decode('utf-8', errors='replace')
        if not isinstance(response, str) or response.lstrip().lower().startswith('error'):
            raise ValueError(f'Shared-memory capture failed: {response!r}')
        metadata = json.loads(response)
        name = metadata['name']
        num_bytes = int(metadata['num_bytes'])
        shape = tuple(int(value) for value in metadata['shape'])
        dtype = np.dtype(metadata['dtype'])
        with mmap.mmap(-1, num_bytes, tagname=name, access=mmap.ACCESS_READ) as mapping:
            data = np.frombuffer(mapping, dtype=dtype, count=num_bytes // dtype.itemsize).copy()
        frame = data.reshape(shape)
        if output_format == 'npy':
            output = BytesIO()
            np.save(output, frame, allow_pickle=False)
            return output.getvalue()
        if frame.ndim != 3 or frame.shape[-1] != 4:

View source record →