tf 🔗

def tf_model_delete(self, resource: Resource) -> Status:
    """Performs the Tensorflow model deletion operation.

    Wraps the `vaccel_tf_session_delete()` C operation.

    Args:
        resource: A resource with the model to unload.

    Returns:
        The status of the operation execution.

    Raises:
        RuntimeError: If the `Session` is uninitialized.
        FFIError: If the C operation fails.
    """
    if not self._c_ptr:
        msg = "Uninitialized session"
        raise RuntimeError(msg)

    status = Status()
    ret = lib.vaccel_tf_session_delete(
        self._c_ptr, resource._c_ptr, status._c_ptr
    )
    if ret != 0:
        raise FFIError(ret, "Tensorflow model delete failed")
    return status

def tf_model_load(self, resource: Resource) -> Status:
    """Performs the Tensorflow model loading operation.

    Wraps the `vaccel_tf_session_load()` C operation.

    Args:
        resource: A resource with the model to load.

    Returns:
        The status of the operation execution.

    Raises:
        RuntimeError: If the `Session` is uninitialized.
        FFIError: If the C operation fails.
    """
    if not self._c_ptr:
        msg = "Uninitialized session"
        raise RuntimeError(msg)

    status = Status()
    ret = lib.vaccel_tf_session_load(
        self._c_ptr, resource._c_ptr, status._c_ptr
    )
    if ret != 0:
        raise FFIError(ret, "Tensorflow model loading failed")
    return status

def tf_model_run(
    self,
    resource: Resource,
    in_nodes: list[Node],
    in_tensors: list[Tensor],
    out_nodes: list[Node],
    run_options: Buffer | None = None,
) -> (list[Tensor], Status):
    """Performs the Tensorflow model run operation.

    Wraps the `vaccel_tf_session_run()` C operation.

    Args:
        resource: A resource with the model to run.
        in_nodes: The input nodes for the inference.
        in_tensors: The input tensors for the inference.
        out_nodes: The output nodes for the inference.
        run_options: The inference options.

    Returns:
        A tuple containing:
            - The output tensors
            - The status of the operation execution.

    Raises:
        RuntimeError: If the `Session` is uninitialized.
        FFIError: If the C operation fails.
    """
    if not self._c_ptr:
        msg = "Uninitialized session"
        raise RuntimeError(msg)

    run_options_ptr = (
        ffi.NULL if run_options is None else run_options._c_ptr
    )
    c_in_nodes = CList(in_nodes)
    c_in_tensors = CList.from_ptrs(in_tensors)
    c_out_nodes = CList(out_nodes)
    c_out_tensors = CList.from_ptrs([Tensor.empty()] * len(c_out_nodes))
    status = Status()

    ret = lib.vaccel_tf_session_run(
        self._c_ptr,
        resource._c_ptr,
        run_options_ptr,
        c_in_nodes._c_ptr,
        c_in_tensors._c_ptr,
        len(c_in_nodes),
        c_out_nodes._c_ptr,
        c_out_tensors._c_ptr,
        len(c_out_nodes),
        status._c_ptr,
    )
    if ret != 0:
        raise FFIError(ret, "Tensorflow model run failed")

    out_tensors = [Tensor.from_c_obj(t) for t in c_out_tensors.value]
    return (out_tensors, status)

@classmethod
def empty(cls) -> "Tensor":
    """Initializes a new empty `Tensor` object.

    The object a NULL pointer in place of the C struct.

    Returns:
        A new `Tensor` object
    """
    inst = cls.__new__(cls)
    inst._c_obj_ptr = ffi.new("struct vaccel_tf_tensor **")
    inst._c_obj = inst._c_obj_ptr[0]
    inst._c_obj_ptr[0] = ffi.NULL
    inst._c_size = ffi.sizeof("struct vaccel_tf_tensor *")
    return inst

@classmethod
def from_c_obj(cls, c_obj: ffi.CData) -> "Tensor":
    """Initializes a new `Tensor` object from an existing C struct.

    Args:
        c_obj: A pointer to a `struct vaccel_tf_tensor` C object.

    Returns:
        A new `Tensor` object
    """
    type_str = ffi.getctype(ffi.typeof(c_obj))
    if type_str != "struct vaccel_tf_tensor *":
        msg = f"Expected 'struct vaccel_tf_tensor *', got '{type_str}'"
        raise TypeError(msg)

    inst = cls.__new__(cls)
    inst._c_obj = c_obj
    inst._c_size = ffi.sizeof(inst._c_obj)
    inst._dims = None
    inst._data = None
    inst._data_type = None
    inst._c_obj_ptr = None
    return inst