pycyphal.application.file module

pycyphal.application.file.Path

alias of Path_2_0

pycyphal.application.file.Error

alias of Error_1_0

pycyphal.application.file.Read

alias of Read_1_1

pycyphal.application.file.Write

alias of Write_1_1

pycyphal.application.file.List

alias of List_0_2

pycyphal.application.file.GetInfo

alias of GetInfo_0_2

pycyphal.application.file.Modify

alias of Modify_1_1

pycyphal.application.file.Unstructured

alias of Unstructured_1_0

class pycyphal.application.file.FileServer(node: Node, roots: Iterable[str | 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: Node, roots: Iterable[str | 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[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: Path | str | Path_2_0) → Tuple[Path, 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[Path, 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) → Error_1_0

__repr__() → str

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

Bases: object

This class is deprecated and should not be used in new applications; instead, consider using FileClient2.

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: Node, server_node_id: int, response_timeout: float = 3.0, priority: 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) → 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: int | None = None) → 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: 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

class pycyphal.application.file.FileClient2(local_node: Node, server_node_id: int, response_timeout: float = 3.0, priority: 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.

In contrast to FileClient, FileClient2 raises exceptions for errors reported over the network. The intent is to provide more pythonic error handling in the API. All possible exceptions are defined in this module; all of them are derived from OSError and also from a tag type RemoteFileError which can be used to easily distinguish file-related exceptions in exception handlers.

__init__(local_node: Node, server_node_id: int, response_timeout: float = 3.0, priority: 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) → Response

Proxy for uavcan.file.GetInfo.

Raises:

OSError – If the operation failed; see uavcan.file.Error

async remove(path: str) → None

Proxy for uavcan.file.Modify.

Raises:

OSError – If the operation failed; see uavcan.file.Error

async touch(path: str) → None

Proxy for uavcan.file.Modify.

Raises:

OSError – If the operation failed; see uavcan.file.Error

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

Proxy for uavcan.file.Modify.

Raises:

OSError – If the operation failed; see uavcan.file.Error

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

Proxy for uavcan.file.Modify.

Raises:

OSError – If the operation failed; see uavcan.file.Error

async read(path: str, offset: int = 0, size: int | None = None) → 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.

Raises:

OSError – If the read operation failed; see uavcan.file.Error

Returns:

data on success (empty if the offset is out of bounds or the file is empty).

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

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.

Raises:

OSError – If the write operation failed; see uavcan.file.Error

__repr__() → str

class pycyphal.application.file.RemoteFileError

Bases: object

This is a tag type used to differentiate Cyphal remote file errors.

exception pycyphal.application.file.FileTimeoutError

Bases: RemoteFileError, NetworkTimeoutError

The specialization of the network error for file access. It inherits from RemoteFileError and pycyphal.application.NetworkTimeoutError.

exception pycyphal.application.file.RemoteFileNotFoundError(filename: str)

Bases: RemoteFileError, FileNotFoundError

Exception type raised when a file server reports uavcan.file.Error.NOT_FOUND. This exception type inherits from RemoteFileError and FileNotFoundError.

__init__(filename: str) → None

Parameters:

filename (str) – File, which was not found on the remote end.

exception pycyphal.application.file.RemoteIOError(filename: str)

Bases: RemoteFileError, OSError

Exception type raised when a file server reports uavcan.file.Error.IO_ERROR. This exception type inherits from RemoteFileError and OSError.

__init__(filename: str) → None

Parameters:

filename (str) – File on which was operated on when the I/O error occured on the remote end.

exception pycyphal.application.file.RemoteAccessDeniedError(filename: str)

Bases: RemoteFileError, PermissionError

Exception type raised when a file server reports``uavcan.file.Error.ACCESS_DENIED``. This exception type inherits from RemoteFileError and exc:PermissionError.

__init__(filename: str) → None

Parameters:

filename (str) – File on which was operated on when the permission error occured on the remote end.

exception pycyphal.application.file.RemoteIsDirectoryError(filename: str)

Bases: RemoteFileError, IsADirectoryError

Exception type raised when a file server reports uavcan.file.Error.IS_DIRECTORY. This exception type inherits from RemoteFileError and IsADirectoryError .

__init__(filename: str) → None

Parameters:

filename (str) – File on which the I/O error occured on the remote end.

exception pycyphal.application.file.RemoteInvalidValueError(filename: str)

Bases: RemoteFileError, OSError

Exception type raised when a file server reports uavcan.file.Error.INVALID_VALUE. This exception type inherits from RemoteFileError and OSError.

__init__(filename: str) → None

Parameters:

filename (str) – File on which the invalid value error occured on the remote end.

exception pycyphal.application.file.RemoteFileTooLargeError(filename: str)

Bases: RemoteFileError, OSError

Exception type raised when a file server reports uavcan.file.Error.FILE_TOO_LARGE. This exception type inherits from RemoteFileError and OSError.

__init__(filename: str) → None

Parameters:

filename (str) – File for which the remote end reported it is too large.

exception pycyphal.application.file.RemoteOutOfSpaceError(filename: str)

Bases: RemoteFileError, OSError

Exception type raised when a file server reports uavcan.file.Error.OUT_OF_SPACE. This exception type inherits from RemoteFileError and OSError.

__init__(filename: str) → None

Parameters:

filename (str) – File on which was operated on when the remote end ran out of space.

exception pycyphal.application.file.RemoteNotSupportedError(filename: str)

Bases: RemoteFileError, OSError

Exception type raised when a file server reports uavcan.file.Error.NOT_SUPPORTED. This exception type inherits from RemoteFileError and OSError.

__init__(filename: str) → None

Parameters:

filename (str) – File on which an operation was requested which is not supported by the remote end

exception pycyphal.application.file.RemoteUnknownError(filename: str)

Bases: RemoteFileError, OSError

Exception type raised when a file server reports uavcan.file.Error.UNKNOWN_ERROR. This exception type inherits from RemoteFileError and OSError.

__init__(filename: str) → None

Parameters:

filename (str) – File on which was operated on when the remote end experienced an unknown error.