Utilities

linuxpy.util

Utility functions used by the library. Mostly for internal usage

aclosing(thing)

Bases: AbstractAsyncContextManager

Async context manager for safely finalizing an asynchronously cleaned-up resource such as an async generator, calling its aclose() method.

This is a copy of contextlib.aclosing needed for python 3.9 compatibility

Version(major: int, minor: int, patch: int)

Simple version supporting only a.b.c format

tuple property

Returns a 3 element tuple representing the version so Version(3,2,1).tuple() yields the tuple (3, 2, 1)

from_tuple(sequence: Iterable[Union[str, int]]) classmethod

Create a Version from any sequence/iterable of 3 elements

Examples that create a Version "3.2.1" object:

Version.from_tuple([3, 2, 1])
Version.from_tuple(["3", 2, "1"])
Version.from_tuple(range(3, 0, -1))

# This will work although not recommended since it will not work
# when any of the members is bigger than 9
Version.from_tuple("321")

from_str(text) classmethod

Create a Version from text

Example that create a Version "3.2.1" object:

Version.from_str("3.2.1")

from_number(number: int) classmethod

Create a Version from an integer where each member corresponds to one byte in the integer so that 0xFFFEFD corresponds to 255.254.253

Example that create a Version "3.2.1" object:

Version.from_number((3<<16) + (2<<8) + 1)

iter_chunks(lst: Sequence, size: int) -> Iterable

Batch data from the sequence into groups of length n. The last batch may be shorter than size.

chunks(lst: Sequence, size: int) -> tuple

Batch data from the sequence into groups of length n. The last batch may be shorter than size.

bcd_version_tuple(bcd: int) -> tuple[int, ...]

Returns the tuple version of a BCD (binary-coded decimal) number

bcd_version(bcd: int) -> str

Returns the text version of a BCD (binary-coded decimal) number

to_fd(fd: FDLike)

Return a file descriptor from a file object.

Parameters: fd -- file object or file descriptor

Returns: corresponding file descriptor

Raises: ValueError if the object is invalid

try_numeric(text: str)

Try to translate given text into int, int base 16 or float. Returns the orig and return the original text if it fails.

PARAMETER	DESCRIPTION
text	text to be translated TYPE: str

RETURNS	DESCRIPTION
	int, float or str: The converted text

add_reader_asyncio(fd: FDLike, callback: Callable, *args, loop: Optional[asyncio.AbstractEventLoop] = None)

Add reader during the context and remove it after

Selector(fds: Collection[FDLike], events=selectors.EVENT_READ) -> selectors.DefaultSelector

A selectors.DefaultSelector with given fds registered

selector_stream(selector: selectors.BaseSelector, timeout: Optional[float] = None) -> Iterable[SelectorEvent]

A stream of selector read events

selector_file_stream(fds: Collection[FDLike], timeout: Optional[float] = None) -> Iterable[SelectorEvent]

An inifinte stream of selector read events

file_stream(fds: Collection[FDLike], timeout: Optional[float] = None) -> Iterable[FDLike]

An infinite stream of read ready file descriptors

event_stream(fds: Collection[FDLike], read: Callable[[FDLike], T], timeout: Optional[float] = None) -> Iterable[T]

An infinite stream of events. The given read callable is called for each file that is reported as ready

async_selector_stream(selector: selectors.BaseSelector) -> AsyncIterator[SelectorEvent] async

An asyncronous infinite stream of selector read events

async_selector_file_stream(fds: Collection[FDLike]) -> AsyncIterator[SelectorEvent] async

An asyncronous infinite stream of selector read events

async_file_stream(fds: Collection[FDLike]) -> AsyncIterator[FDLike] async

An asyncronous infinite stream of read ready files

async_event_stream(fds: Collection[FDLike], read: Callable[[FDLike], T]) async

An asyncronous stream of events. The given read callable is called for each file that is reported as ready

make_find(iter_devices: Callable[[], Iterator], needs_open=True) -> Callable

Create a find function for the given callable. The callable should return an iterable where each element has the context manager capability (ie, it can be used in a with statement)

bit_indexes(number: int) -> list[int]

Return the list of indexes that have an active bit on the number.

Example bit_indexes(74) gives [1, 3, 6] (74 == 0b1001010)

sequence_indexes(it: Iterable[T]) -> dict[T, int]

Return a map with key being the value in the given iterable and value the index where it appears in the map. If the same element appears more than once then the last occurence is returned. Elements must be non mutable so they can appear in the returned map Example:

sequence_indexes([10, "bla", 1]) == {10: 0, "bla": 1, 1: 2}

PARAMETER	DESCRIPTION
lines	The iterable to be returned TYPE: Iterable[int]

RETURNS	DESCRIPTION
dict[T, int]	dict[int, int]: map of values and their indexes

index_mask(indexes: dict[T, int], elements: Iterable[T]) -> int

Return a bit mask with active bits from elements. The bit index for each element is given by the indexes

PARAMETER	DESCRIPTION
indexes	bit index map TYPE: dict[T, int]
elements	list of active elements TYPE: Iterable[T]

RETURNS	DESCRIPTION
int	bit mask TYPE: int

random_name(min_length=32, max_length=32)

Generates a random name like text of ascii characters (letters or digits).

The first character is always a letter