vaccel 🔗

def is_registered(self, session: "Session") -> bool:
    """Checks if the resource is registered with the session.

    Args:
        session: The session to check for registration.

    Returns:
        True if the resource is registered with the session.
    """
    return session in self.__sessions

def register(self, session: "Session") -> None:
    """Register the resource with a session.

    Args:
        session: The session to register the resource with.

    Raises:
        FFIError: If resource registration fails.
    """
    ret = lib.vaccel_resource_register(
        self._c_ptr_or_raise,
        session._c_ptr_or_raise,
    )
    if ret != 0:
        raise FFIError(
            ret,
            f"Could not register resource {self.id} "
            f"with session {session.id}",
        )
    self.__sessions.append(session)

def unregister(self, session: "Session") -> None:
    """Unregister the resource from a session.

    Args:
        session: The session to unregister the resource from.

    Raises:
        FFIError: If resource unregistration fails.
    """
    ret = lib.vaccel_resource_unregister(
        self._c_ptr_or_raise,
        session._c_ptr_or_raise,
    )
    if ret != 0:
        raise FFIError(
            ret,
            f"Could not unregister resource {self.id} "
            f"from session {session.id}",
        )
    self.__sessions.remove(session)

def classify(self, image: bytes) -> (str, str):
    """Performs the image classification operation.

    Wraps the `vaccel_image_classification()` C operation.

    Args:
        image: The image data as a `bytes` object.

    Returns:
        A tuple containing:
            - The classification tag.
            - The resulting image filename.

    Raises:
        FFIError: If the C operation fails.
    """
    img = CBytes(image)
    out_text = CBytes(bytearray(self._out_len))
    out_imgname = CBytes(bytearray(self._out_len))

    ret = lib.vaccel_image_classification(
        self._c_ptr_or_raise,
        img._c_ptr,
        out_text._c_ptr,
        out_imgname._c_ptr,
        len(img),
        len(out_text),
        len(out_imgname),
    )
    if ret:
        raise FFIError(ret, "Image classification failed")

    return out_text.to_str(), out_imgname.to_str()

def depth(self, image: bytes) -> str:
    """Performs the image depth estimation operation.

    Wraps the `vaccel_image_depth()` C operation.

    Args:
        image: The image data as a `bytes` object.

    Returns:
        The resulting image filename.

    Raises:
        FFIError: If the C operation fails.
    """
    img = CBytes(image)
    out_imgname = CBytes(bytearray(self._out_len))

    ret = lib.vaccel_image_depth(
        self._c_ptr_or_raise,
        img._c_ptr,
        out_imgname._c_ptr,
        len(img),
        len(out_imgname),
    )
    if ret:
        raise FFIError(ret, "Image depth estimation failed")

    return out_imgname.to_str()

def detect(self, image: bytes) -> str:
    """Performs the image detection operation.

    Wraps the `vaccel_image_detection()` C operation.

    Args:
        image: The image data as a `bytes` object.

    Returns:
        The resulting image filename.

    Raises:
        FFIError: If the C operation fails.
    """
    img = CBytes(image)
    out_imgname = CBytes(bytearray(self._out_len))

    ret = lib.vaccel_image_detection(
        self._c_ptr_or_raise,
        img._c_ptr,
        out_imgname._c_ptr,
        len(img),
        len(out_imgname),
    )
    if ret:
        raise FFIError(ret, "Image detection failed")

    return out_imgname.to_str()

def exec(
    self,
    library: str | Path,
    symbol: str,
    arg_read: list[Any] | None = None,
    arg_write: list[Any] | None = None,
) -> list[Any] | None:
    """Performs the Exec operation.

    Wraps the `vaccel_exec()` C operation.

    Args:
        library: The path to the shared object containing the function that
            will be called.
        symbol: The name of the function contained in the above shared
            object.
        arg_read: The input arguments that will be passed to the
            called function.
        arg_write: The output arguments that will be passed to the
            called function.

    Returns:
        The resulting outputs.

    Raises:
        FFIError: If the C operation fails.
    """
    if arg_read is not None:
        c_arg_read = CList([Arg(arg) for arg in arg_read])
        c_arg_read_ptr = c_arg_read._c_ptr
        c_arg_read_len = len(c_arg_read)
    else:
        c_arg_read = None
        c_arg_read_ptr = ffi.NULL
        c_arg_read_len = 0

    if arg_write is not None:
        c_arg_write = CList([Arg(arg) for arg in arg_write])
        c_arg_write_ptr = c_arg_write._c_ptr
        c_arg_write_len = len(c_arg_write)
    else:
        c_arg_write = None
        c_arg_write_ptr = ffi.NULL
        c_arg_write_len = 0

    ret = lib.vaccel_exec(
        self._c_ptr_or_raise,
        str(library).encode(),
        symbol.encode(),
        c_arg_read_ptr,
        c_arg_read_len,
        c_arg_write_ptr,
        c_arg_write_len,
    )
    if ret != 0:
        raise FFIError(ret, "Exec operation failed")

    if c_arg_write is not None:
        return [c_arg_write[i].buf for i in range(len(c_arg_write))]
    return None

def exec_with_resource(
    self,
    resource: Resource,
    symbol: str,
    arg_read: list[Any] | None = None,
    arg_write: list[Any] | None = None,
) -> list[Any] | None:
    """Performs the Exec with resource operation.

    Wraps the `vaccel_exec_with_resource()` C operation.

    Args:
        resource: The resource of the shared object containing the function
            that will be called.
        symbol: The name of the function contained in the above shared
            object.
        arg_read: The input arguments that will be passed to the
            called function.
        arg_write: The output arguments that will be passed to the
            called function.

    Returns:
        The resulting outputs.

    Raises:
        FFIError: If the C operation fails.
    """
    if arg_read is not None:
        c_arg_read = CList([Arg(arg) for arg in arg_read])
        c_arg_read_ptr = c_arg_read._c_ptr
        c_arg_read_len = len(c_arg_read)
    else:
        c_arg_read = None
        c_arg_read_ptr = ffi.NULL
        c_arg_read_len = 0

    if arg_write is not None:
        c_arg_write = CList([Arg(arg) for arg in arg_write])
        c_arg_write_ptr = c_arg_write._c_ptr
        c_arg_write_len = len(c_arg_write)
    else:
        c_arg_write = None
        c_arg_write_ptr = ffi.NULL
        c_arg_write_len = 0

    ret = lib.vaccel_exec_with_resource(
        self._c_ptr_or_raise,
        resource._c_ptr,
        symbol.encode(),
        c_arg_read_ptr,
        c_arg_read_len,
        c_arg_write_ptr,
        c_arg_write_len,
    )
    if ret != 0:
        raise FFIError(ret, "Exec with resource operation failed")

    if c_arg_write is not None:
        return [c_arg_write[i].buf for i in range(len(c_arg_write))]
    return None

def fpga_arraycopy(self, a: list[int]) -> list[int]:
    """Performs the matrix copying operation.

    Wraps the `vaccel_fpga_arraycopy()` C operation.

    Args:
        a: The matrix A to be copied.

    Returns:
        A copy of the matrix A.

    Raises:
        FFIError: If the C operation fails.
    """
    c_a = CList(a)
    c_out_a = CList([0] * len(a))

    ret = lib.vaccel_fpga_arraycopy(
        self._c_ptr_or_raise, c_a._c_ptr, c_out_a._c_ptr, len(c_a)
    )
    if ret != 0:
        raise FFIError(ret, "FPGA array copy failed")

    return [int(item) for item in c_out_a.value]

def fpga_mmult(self, a: list[float], b: list[float]) -> list[float]:
    """Performs the matrix multiplication operation.

    Wraps the `vaccel_fpga_mmult()` C operation.

    Args:
        a: A matrix A.
        b: A matrix B.

    Returns:
        The multiplication result of matrices A and B.

    Raises:
        FFIError: If the C operation fails.
    """
    c_a = CList(a)
    c_b = CList(b)
    c_c = CList([float(0)] * len(a))

    ret = lib.vaccel_fpga_mmult(
        self._c_ptr_or_raise, c_a._c_ptr, c_b._c_ptr, c_c._c_ptr, len(c_a)
    )
    if ret != 0:
        raise FFIError(ret, "FPGA matrix multiplication failed")

    return [float(item) for item in c_c.value]

def fpga_parallel(
    self, a: list[float], b: list[float]
) -> (list[float], list[float]):
    """Performs the parallel matrix addition and multiplication operation.

    Wraps the `vaccel_fpga_parallel()` C operation.

    Args:
        a: A matrix A.
        b: A matrix B.

    Returns:
        A tuple containing:
            - The result of the addition of matrices A and B.
            - The result of the multiplication of matrices A and B.

    Raises:
        FFIError: If the C operation fails.
    """
    c_a = CList(a)
    c_b = CList(b)
    c_add_output = CList([float(0)] * len(a))
    c_mult_output = CList([float(0)] * len(a))

    ret = lib.vaccel_fpga_parallel(
        self._c_ptr_or_raise,
        c_a._c_ptr,
        c_b._c_ptr,
        c_add_output._c_ptr,
        c_mult_output._c_ptr,
        len(c_a),
    )
    if ret != 0:
        raise FFIError(
            ret, "FPGA parallel matrix addition and multiplication failed"
        )

    return (
        [float(item) for item in c_add_output.value],
        [float(item) for item in c_mult_output.value],
    )

def fpga_vadd(self, a: list[float], b: list[float]) -> list[float]:
    """Performs the matrix addition operation.

    Wraps the `vaccel_fpga_vadd()` C operation.

    Args:
        a: A matrix A.
        b: A matrix B.

    Returns:
        The addition result of matrices A and B.

    Raises:
        FFIError: If the C operation fails.
    """
    c_a = CList(a)
    c_b = CList(b)
    c_c = CList([float(0)] * len(a))

    ret = lib.vaccel_fpga_vadd(
        self._c_ptr_or_raise,
        c_a._c_ptr,
        c_b._c_ptr,
        c_c._c_ptr,
        len(c_a),
        len(c_b),
    )
    if ret != 0:
        raise FFIError(ret, "FPGA vector addition failed")

    return [float(item) for item in c_c.value]

def genop(self, arg_read: list[Arg], arg_write: list[Arg]) -> None:
    """Performs the Generic operation.

    Wraps the `vaccel_genop()` C operation.

    Args:
        arg_read: The input arguments of the operation.
        arg_write: The output arguments of the operation. Modified in place.

    Raises:
        FFIError: If the C operation fails.
    """
    c_args_read = CList(arg_read)
    c_args_write = CList(arg_write)

    ret = lib.vaccel_genop(
        self._c_ptr_or_raise,
        c_args_read._c_ptr,
        len(c_args_read),
        c_args_write._c_ptr,
        len(c_args_write),
    )
    if ret:
        raise FFIError(ret, "Generic operation failed")

def has_resource(self, resource: Resource) -> bool:
    """Check if a resource is registered with the session.

    Args:
        resource: The resource to check for registration.

    Returns:
        True if the resource is registered.
    """
    return (
        lib.vaccel_session_has_resource(
            self._c_ptr_or_raise, resource._c_ptr
        )
        != 0
    )

def minmax(
    self, indata: bytes, ndata: int, low_threshold: int, high_threshold: int
) -> (bytes, float, float):
    """Performs the minmax operation.

    Wraps the `vaccel_minmax()` C operation.

    Args:
        indata: The input data as a `bytes` object.
        ndata: The number of data to be read provided data object.
        low_threshold: The threshold for the min value.
        high_threshold: The threshold for the max value.

    Returns:
        A tuple containing:
            - The resulting output data.
            - The detected min value of the data.
            - The detected max value of the data.

    Raises:
        FFIError: If the C operation fails.
    """
    c_indata = CBytes(indata)
    c_outdata = CBytes(bytearray(ndata * ffi.sizeof("double")))
    c_min = CFloat(float(0), "double")
    c_max = CFloat(float(0), "double")

    ret = lib.vaccel_minmax(
        self._c_ptr_or_raise,
        c_indata._as_c_array("double"),
        ndata,
        low_threshold,
        high_threshold,
        c_outdata._as_c_array("double"),
        c_min._c_ptr,
        c_max._c_ptr,
    )
    if ret != 0:
        raise FFIError(ret, "Minmax operation failed")

    return (c_outdata.value, c_min.value, c_max.value)

def noop(self) -> None:
    """Performs the NoOp operation.

    Wraps the `vaccel_noop()` C operation.

    Raises:
        FFIError: If the C operation fails.
    """
    ret = lib.vaccel_noop(self._c_ptr_or_raise)
    if ret != 0:
        raise FFIError(ret, "NoOp operation failed")

def pose(self, image: bytes) -> str:
    """Performs the image pose estimation operation.

    Wraps the `vaccel_image_pose()` C operation.

    Args:
        image: The image data as a `bytes` object.

    Returns:
        The resulting image filename.

    Raises:
        FFIError: If the C operation fails.
    """
    img = CBytes(image)
    out_imgname = CBytes(bytearray(self._out_len))

    ret = lib.vaccel_image_pose(
        self._c_ptr_or_raise,
        img._c_ptr,
        out_imgname._c_ptr,
        len(img),
        len(out_imgname),
    )
    if ret:
        raise FFIError(ret, "Image pose estimation failed")

    return out_imgname.to_str()

def segment(self, image: bytes) -> str:
    """Performs the image segmentation operations.

    Wraps the `vaccel_image_segmentation()` C operation.

    Args:
        image: The image data as a `bytes` object.

    Returns:
        The resulting image filename.

    Raises:
        FFIError: If the C operation fails.
    """
    img = CBytes(image)
    out_imgname = CBytes(bytearray(self._out_len))

    ret = lib.vaccel_image_segmentation(
        self._c_ptr_or_raise,
        img._c_ptr,
        out_imgname._c_ptr,
        len(img),
        len(out_imgname),
    )
    if ret:
        raise FFIError(ret, "Image segmentation failed")

    return out_imgname.to_str()

def sgemm(
    self,
    m: int,
    n: int,
    k: int,
    alpha: float,
    a: list[float],
    lda: int,
    b: list[float],
    ldb: int,
    beta: float,
    ldc: int,
) -> list[float]:
    """Performs the SGEMM operation.

    Wraps the `vaccel_sgemm()` C operation.

    Args:
        m: The number of rows in matrix A and matrix C.
        n: The number of columns in matrix B and matrix C.
        k: The number of columns in matrix A and rows in matrix B.
        alpha: Scalar multiplier for the matrix product A * B.
        a: The matrix A in row-major order with shape (m, k).
        lda: The leading dimension of matrix A (usually m).
        b: The matrix B in row-major order with shape (k, n).
        ldb: The leading dimension of matrix B (usually k).
        beta: Scalar multiplier for matrix C.
        ldc: The leading dimension of matrix C (usually m).

    Returns:
        The resulting matrix C in row-major order with shape (m, n).

    Raises:
        FFIError: If the C operation fails.
    """
    c_a = CList(a)
    c_b = CList(b)
    c_c = CList([float(0)] * m * n)

    ret = lib.vaccel_sgemm(
        self._c_ptr_or_raise,
        m,
        n,
        k,
        alpha,
        c_a._c_ptr,
        lda,
        c_b._c_ptr,
        ldb,
        beta,
        c_c._c_ptr,
        ldc,
    )
    if ret != 0:
        raise FFIError(ret, "SGEMM failed")

    return c_c.value