pycyphal.application.file module

pycyphal.application.file.Path

alias of uavcan.file.Path_2_0.Path_2_0

pycyphal.application.file.Error

alias of uavcan.file.Error_1_0.Error_1_0

pycyphal.application.file.Read

alias of uavcan.file.Read_1_1.Read_1_1

pycyphal.application.file.Write

alias of uavcan.file.Write_1_1.Write_1_1

pycyphal.application.file.List

alias of uavcan.file.List_0_2.List_0_2

pycyphal.application.file.GetInfo

alias of uavcan.file.GetInfo_0_2.GetInfo_0_2

pycyphal.application.file.Modify

alias of uavcan.file.Modify_1_1.Modify_1_1

pycyphal.application.file.Unstructured

alias of uavcan.primitive.Unstructured_1_0.Unstructured_1_0

class pycyphal.application.file.FileServer(node: pycyphal.application._node.Node, roots: Iterable[Union[str, pathlib.Path]])

Bases: object

Exposes local filesystems via the standard RPC-services defined in uavcan.file. The lifetime of this instance matches the lifetime of its node.

__init__(node: pycyphal.application._node.Node, roots: Iterable[Union[str, pathlib.Path]]) → None

Parameters

• node – The node instance to initialize the file server on. It shall not be anonymous, otherwise it’s a pycyphal.transport.OperationNotDefinedForAnonymousNodeError.

• roots – All file operations will be performed in the specified directories. The first directory to match takes precedence. New files are created in the first directory.

property roots: List[pathlib.Path]

File operations will be performed within these root directories. The first directory to match takes precedence. New files are created in the first directory in the list. The list can be modified.

locate(p: Union[pathlib.Path, str, uavcan.file.Path_2_0.Path_2_0]) → Tuple[pathlib.Path, pathlib.Path]

Iterate through roots until a root r is found such that r/p exists and return (r, p). Otherwise, return nonexistent (roots[0], p). The leading slash makes no difference because we only search through the specified roots.

Raises

FileNotFoundError if roots is empty.

glob(pat: str) → Iterable[Tuple[pathlib.Path, pathlib.Path]]

Search for entries matching the pattern across roots, in order. Return tuple of (root, match), where match is relative to its root. Ordering not enforced.

static convert_error(ex: Exception) → uavcan.file.Error_1_0.Error_1_0

__repr__() → str

class pycyphal.application.file.FileClient(local_node: pycyphal.application._node.Node, server_node_id: int, response_timeout: float = 3.0, priority: pycyphal.transport._transfer.Priority = Priority.SLOW)

Bases: object

A trivial proxy that provides a higher-level and more pythonic API on top of the standard RPC-services from uavcan.file. Client instances are created lazily at first request and then kept alive until this instance is closed. All remote operations raise FileTimeoutError on timeout.

__init__(local_node: pycyphal.application._node.Node, server_node_id: int, response_timeout: float = 3.0, priority: pycyphal.transport._transfer.Priority = Priority.SLOW) → None

Parameters

• local_node – RPC-service clients will be created on this node.

• server_node_id – All requests will be sent to this node-ID.

• response_timeout – Raise FileTimeoutError if the server does not respond in this time.

• priority – Transfer priority for requests (and, therefore, responses).

property data_transfer_capacity: int

A convenience constant derived from DSDL: the maximum number of bytes per read/write transfer. Larger reads/writes are non-atomic.

property server_node_id: int

The node-ID of the remote file server.

close() → None

Close all RPC-service client instances created up to this point.

async list(path: str) → AsyncIterable[str]

Proxy for uavcan.file.List. Invokes requests in series until all elements are listed.

async get_info(path: str) → uavcan.file.GetInfo_0_2.GetInfo_0_2.Response

Proxy for uavcan.file.GetInfo. Be sure to check the error code in the returned object.

async remove(path: str) → int

Proxy for uavcan.file.Modify.

Returns

See uavcan.file.Error

async touch(path: str) → int

Proxy for uavcan.file.Modify.

Returns

See uavcan.file.Error

async copy(src: str, dst: str, overwrite: bool = False) → int

Proxy for uavcan.file.Modify.

Returns

See uavcan.file.Error

async move(src: str, dst: str, overwrite: bool = False) → int

Proxy for uavcan.file.Modify.

Returns

See uavcan.file.Error

async read(path: str, offset: int = 0, size: Optional[int] = None) → Union[int, bytes]

Proxy for uavcan.file.Read.

Parameters

• path – The file to read.

• offset – Read offset from the beginning of the file. Currently, it must be positive; negative offsets from the end of the file may be supported later.

• size – Read requests will be stopped after the end of the file is reached or at least this many bytes are read. If None (default), the entire file will be read (this may exhaust local memory). If zero, this call is a no-op.

Returns

uavcan.file.Error.value on error (e.g., no file), data on success (empty if the offset is out of bounds or the file is empty).

async write(path: str, data: Union[memoryview, bytes], offset: int = 0, *, truncate: bool = True) → int

Proxy for uavcan.file.Write.

Parameters

• path – The file to write.

• data – The data to write at the specified offset. The number of write requests depends on the size of data.

• offset – Write offset from the beginning of the file. Currently, it must be positive; negative offsets from the end of the file may be supported later.

• truncate – If True, the rest of the file after offset + len(data) will be truncated. This is done by sending an empty write request, as prescribed by the Specification.

Returns

See uavcan.file.Error

__repr__() → str

exception pycyphal.application.file.FileTimeoutError

Bases: pycyphal.application.NetworkTimeoutError

The specialization of the network error for file access.