mmcv/video/optflow.py
def flowread(flow_or_path, quantize=False, concat_axis=0, *args, **kwargs):
    """Read an optical flow map.

    Args:
        flow_or_path (ndarray or str): A flow map or filepath.
        quantize (bool): whether to read quantized pair, if set to True,
            remaining args will be passed to :func:`dequantize_flow`.
        concat_axis (int): The axis that dx and dy are concatenated,
            can be either 0 or 1. Ignored if quantize is False.

    Returns:
        ndarray: Optical flow represented as a (h, w, 2) numpy array
    """
    if isinstance(flow_or_path, np.ndarray):
        if (flow_or_path.ndim != 3) or (flow_or_path.shape[-1] != 2):
            raise ValueError('Invalid flow with shape {}'.format(
                flow_or_path.shape))
        return flow_or_path
    elif not is_str(flow_or_path):
        raise TypeError(
            '"flow_or_path" must be a filename or numpy array, not {}'.format(
                type(flow_or_path)))

    if not quantize:
        with open(flow_or_path, 'rb') as f:
            try:
                header = f.read(4).decode('utf-8')
            except Exception:
                raise IOError('Invalid flow file: {}'.format(flow_or_path))
            else:
                if header != 'PIEH':
                    raise IOError(
                        'Invalid flow file: {}, header does not contain PIEH'.
                        format(flow_or_path))

            w = np.fromfile(f, np.int32, 1).squeeze()
            h = np.fromfile(f, np.int32, 1).squeeze()
            flow = np.fromfile(f, np.float32, w * h * 2).reshape((h, w, 2))
    else:
        assert concat_axis in [0, 1]
        cat_flow = imread(flow_or_path, flag='unchanged')
        if cat_flow.ndim != 2:
            raise IOError(
                '{} is not a valid quantized flow file, its dimension is {}.'.
                format(flow_or_path, cat_flow.ndim))
        assert cat_flow.shape[concat_axis] % 2 == 0
        dx, dy = np.split(cat_flow, 2, axis=concat_axis)
        flow = dequantize_flow(dx, dy, *args, **kwargs)

    return flow.astype(np.float32)
def flowread(flow_or_path, quantize=False, concat_axis=0, *args, **kwargs):
    if isinstance(flow_or_path, np.ndarray):
        if (flow_or_path.ndim != 3) or (flow_or_path.shape[-1] != 2):
            raise ValueError('Invalid flow with shape {}'.format(
                flow_or_path.shape))
        return flow_or_path
    elif not is_str(flow_or_path):
        raise TypeError(
            '"flow_or_path" must be a filename or numpy array, not {}'.format(
                type(flow_or_path)))

    if not quantize:
        with open(flow_or_path, 'rb') as f:
            try:
                header = f.read(4).decode('utf-8')
            except Exception:
                raise IOError('Invalid flow file: {}'.format(flow_or_path))
            else:
                if header != 'PIEH':
                    raise IOError(
                        'Invalid flow file: {}, header does not contain PIEH'.
                        format(flow_or_path))

            w = np.fromfile(f, np.int32, 1).squeeze()
            h = np.fromfile(f, np.int32, 1).squeeze()
            flow = np.fromfile(f, np.float32, w * h * 2).reshape((h, w, 2))
    else:
        assert concat_axis in [0, 1]
        cat_flow = imread(flow_or_path, flag='unchanged')
        if cat_flow.ndim != 2:
            raise IOError(
                '{} is not a valid quantized flow file, its dimension is {}.'.
                format(flow_or_path, cat_flow.ndim))
        assert cat_flow.shape[concat_axis] % 2 == 0
        dx, dy = np.split(cat_flow, 2, axis=concat_axis)
        flow = dequantize_flow(dx, dy, *args, **kwargs)

    return flow.astype(np.float32)
def flowread ( flow_or_path , quantize = False , concat_axis = 0 , * args , * * kwargs ) : if isinstance ( flow_or_path , np . ndarray ) : if ( flow_or_path . ndim != 3 ) or ( flow_or_path . shape [ - 1 ] != 2 ) : raise ValueError ( 'Invalid flow with shape {}' . format ( flow_or_path . shape ) ) return flow_or_path elif not is_str ( flow_or_path ) : raise TypeError ( '"flow_or_path" must be a filename or numpy array, not {}' . format ( type ( flow_or_path ) ) ) if not quantize : with open ( flow_or_path , 'rb' ) as f : try : header = f . read ( 4 ) . decode ( 'utf-8' ) except Exception : raise IOError ( 'Invalid flow file: {}' . format ( flow_or_path ) ) else : if header != 'PIEH' : raise IOError ( 'Invalid flow file: {}, header does not contain PIEH' . format ( flow_or_path ) ) w = np . fromfile ( f , np . int32 , 1 ) . squeeze ( ) h = np . fromfile ( f , np . int32 , 1 ) . squeeze ( ) flow = np . fromfile ( f , np . float32 , w * h * 2 ) . reshape ( ( h , w , 2 ) ) else : assert concat_axis in [ 0 , 1 ] cat_flow = imread ( flow_or_path , flag = 'unchanged' ) if cat_flow . ndim != 2 : raise IOError ( '{} is not a valid quantized flow file, its dimension is {}.' . format ( flow_or_path , cat_flow . ndim ) ) assert cat_flow . shape [ concat_axis ] % 2 == 0 dx , dy = np . split ( cat_flow , 2 , axis = concat_axis ) flow = dequantize_flow ( dx , dy , * args , * * kwargs ) return flow . astype ( np . float32 )
Read an optical flow map.

    Args:
        flow_or_path (ndarray or str): A flow map or filepath.
        quantize (bool): whether to read quantized pair, if set to True,
            remaining args will be passed to :func:`dequantize_flow`.
        concat_axis (int): The axis that dx and dy are concatenated,
            can be either 0 or 1. Ignored if quantize is False.

    Returns:
        ndarray: Optical flow represented as a (h, w, 2) numpy array
Read an optical flow map .