File valuefd.py#
File List > build > lib.linux-armv7l-cpython-311 > pypisoundmicro > valuefd.py
Go to the documentation of this file
import os
from .swig import pypisoundmicro as psm
from ._utils import copy_doc
from typing import Union
@copy_doc(psm.ValueFd)
class ValueFd:
"""A wrapper around a file descriptor for reading/writing element values."""
_fd_obj: psm.ValueFd = None
def __init__(self, fd: Union[psm.ValueFd, int]):
"""Create a ValueFd from an existing file descriptor.
Args:
fd (int): The file descriptor to wrap
"""
if isinstance(fd, psm.ValueFd):
self._fd_obj = fd
elif isinstance(fd, int):
self._fd_obj = psm.ValueFd(fd)
else:
raise TypeError('fd must be a psm.ValueFd or an int')
def __del__(self):
"""Close the file descriptor when the object is garbage collected."""
self.closeclose()
@property
def is_valid(self) -> bool:
""" Returns true if the object holds a valid fd."""
return self._fd_obj.isValid() if self._fd_obj else False
def take(self) -> int:
""" Returns the fd value, and won't close it in the destructor, transferring ownership."""
if self._fd_obj is None:
return -1
fd = self._fd_obj.take()
self._fd_obj = None
return fd
def get(self) -> int:
""" Returns the fd value for your use, but keeps ownership, and will close it upon destruction."""
return self._fd_obj.get() if self._fd_obj else -1
def close(self) -> None:
""" Closes the fd and forgets about it."""
if hasattr(self, '_fd_obj') and self._fd_obj is not None:
err = self._fd_obj.close()
if err != 0:
raise OSError(f'Failed to close file descriptor: {err}')
self._fd_obj = None
def write(self, value) -> int:
""" Outputs a decimal number to the fd."""
return self._fd_obj.write(value) if self._fd_obj else -1
def read(self) -> int:
"""
Reads a decimal number from the fd and returns it as integer.
:raises OSError: If the file descriptor is invalid or if an error occurs during reading.
:return: The read value.
"""
if self._fd_obj is None:
return (-1, os.EBADF)
result, err = self._fd_obj.read()
if err != 0:
raise OSError(f'Failed to read from file descriptor: {err}')
return result