Skip to content

resource 🔗

Interface to the struct vaccel_resource C object.

Resource 🔗

Resource(paths: list[Path] | list[str] | Path | str, type_: ResourceType)

Bases: CType

Wrapper for the struct vaccel_resource C object.

Manages the creation and initialization of a C struct vaccel_resource and provides access to it through Python properties.

Inherits

CType: Abstract base class for defining C data types.

Parameters:

Name Type Description Default

paths 🔗

 list[Path] | list[str] | Path | str

The path(s) to the file(s) that will be represented by the resource.

 required

type_ 🔗

 ResourceType

The type of the resource.

 required
Source code in vaccel/resource.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def __init__(
    self, paths: list[Path] | list[str] | Path | str, type_: ResourceType
):
    """Initializes a new `Resource` object.

    Args:
        paths: The path(s) to the file(s) that will be represented by the
            resource.
        type_: The type of the resource.
    """
    if isinstance(paths, list):
        self._paths = [str(path) for path in paths]
    else:
        self._paths = [str(paths)]
    self._c_paths = CList(self._paths)
    self._type = type_
    self._c_data = None
    self._c_obj_ptr = ffi.NULL
    super().__init__()

c_size property 🔗

c_size: int

Returns the size of the object in bytes.

id property 🔗

id: int

The resource identifier.

Returns:

Type Description
int

The resource's unique ID.

remote_id property 🔗

remote_id: int

The remote resource identifier.

Returns:

Type Description
int

The resource's remote ID.

value property 🔗

value: CData

Returns the value of the underlying C struct.

Returns:

Type Description
CData

The dereferenced 'struct vaccel_resource`

from_buffer classmethod 🔗

from_buffer(
    data: bytes | bytearray | memoryview, type_: ResourceType
) -> Resource

Initializes a new Resource object from byte-like data.

Parameters:

Name Type Description Default

data 🔗

 bytes | bytearray | memoryview

The data to be passed to the C struct.

 required

type_ 🔗

 ResourceType

The type of the resource.

 required

Returns:

Type Description
Resource

A new Resource object
Source code in vaccel/resource.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
@classmethod
def from_buffer(
    cls,
    data: bytes | bytearray | memoryview,
    type_: ResourceType,
) -> "Resource":
    """Initializes a new `Resource` object from byte-like data.

    Args:
        data: The data to be passed to the C struct.
        type_: The type of the resource.

    Returns:
        A new `Resource` object
    """
    inst = cls.__new__(cls)
    inst._data = data
    inst._type = type_
    inst._c_data = CBytes(inst._data)
    inst._c_paths = None
    inst._c_obj_ptr = ffi.NULL
    super().__init__(inst)
    return inst

from_numpy classmethod 🔗

from_numpy(data: ndarray) -> Resource

Initializes a new Resource object from a NumPy array.

Parameters:

Name Type Description Default

data 🔗

 ndarray

The NumPy array containing the resource data.

 required

Returns:

Type Description
Resource

A new Resource object

Raises:

Type Description
NotImplementedError

If NumPy is not installed.
Source code in vaccel/resource.py 
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
@classmethod
def from_numpy(cls, data: "np.ndarray") -> "Resource":
    """Initializes a new `Resource` object from a NumPy array.

    Args:
        data: The NumPy array containing the resource data.

    Returns:
        A new `Resource` object

    Raises:
        NotImplementedError: If NumPy is not installed.
    """
    if not HAS_NUMPY:
        msg = "NumPy is not available"
        raise NotImplementedError(msg)

    inst = cls.__new__(cls)
    inst._data = data
    inst._type = ResourceType.DATA
    inst._c_data = CNumpyArray(inst._data)
    inst._c_paths = None
    inst._c_obj_ptr = ffi.NULL
    super().__init__(inst)
    return inst

register 🔗

register(session: BaseSession) -> None

Registers the resource with a session.

Parameters:

Name Type Description Default

session 🔗

 BaseSession

The session to register the resource with.

 required

Raises:

Type Description
FFIError

If resource registration fails.
Source code in vaccel/resource.py 
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
def register(self, session: "Session") -> None:
    """Registers 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}",
        )

sync 🔗

sync(session: BaseSession) -> None

Synchronizes the resource data to reflect any remote changes.

Parameters:

Name Type Description Default

session 🔗

 BaseSession

The session the resource is registered with.

 required

Raises:

Type Description
FFIError

If resource synchronization fails.
Source code in vaccel/resource.py 
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
def sync(self, session: "Session") -> None:
    """Synchronizes the resource data to reflect any remote changes.

    Args:
        session: The session the resource is registered with.

    Raises:
        FFIError: If resource synchronization fails.
    """
    ret = lib.vaccel_resource_sync(
        self._c_ptr_or_raise,
        session._c_ptr_or_raise,
    )
    if ret != 0:
        raise FFIError(
            ret,
            f"Could not synchronize resource {self.id} "
            f"in session {session.id}",
        )

unregister 🔗

unregister(session: BaseSession) -> None

Unregisters the resource from a session.

Parameters:

Name Type Description Default

session 🔗

 BaseSession

The session to unregister the resource from.

 required

Raises:

Type Description
FFIError

If resource unregistration fails.
Source code in vaccel/resource.py 
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
def unregister(self, session: "Session") -> None:
    """Unregisters 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}",
        )