hwIo package

Raw input/output for braille displays via serial and HID. See the L{Serial} and L{Hid} classes. Braille display drivers must be thread-safe to use this, as it utilises a background thread. See L{braille.BrailleDisplayDriver.isThreadSafe}.

hwIo.bgThread: IoThread
hwIo.initialize()
hwIo.terminate()

Submodules

hwIo.base module

Raw input/output for braille displays via serial and HID. See the L{Serial} and L{Hid} classes. Braille display drivers must be thread-safe to use this, as it utilises a background thread. See L{braille.BrailleDisplayDriver.isThreadSafe}.

hwIo.base._isDebug()
class hwIo.base.IoBase(fileHandle: c_void_p, onReceive: Callable[[bytes], None], writeFileHandle: c_void_p | None = None, onReceiveSize: int = 1, onReadError: Callable[[int], bool] | None = None, ioThread: IoThread | None = None)

Bases: object

Base class for raw I/O. This watches for data of a specified size and calls a callback when it is received.

Constructor. @param fileHandle: A handle to an open I/O device opened for overlapped I/O.

If L{writeFileHandle} is specified, this is only for input. The serial implementation uses a _port_handle member for this argument.

@param onReceive: A callable taking the received data as its only argument. @param writeFileHandle: A handle to an open output device opened for overlapped I/O. @param onReceiveSize: The size (in bytes) of the data with which to call C{onReceive}. @param onReadError: If provided, a callback that takes the error code for a failed read

and returns True if the I/O loop should exit cleanly or False if an exception should be thrown

@param ioThread: If provided, the I/O thread used for background reads.

if C{None}, defaults to L{hwIo.bgThread}

_ioThreadRef: ReferenceType[IoThread]
_initialRead()

Performs the initial background read by queuing it as an APC to the IO background thread provided at initialization time.

waitForRead(timeout: int | float) bool

Wait for a chunk of data to be received and processed. This will return after L{onReceive} has been called or when the timeout elapses. @param timeout: The maximum time to wait in seconds. @return: C{True} if received data was processed before the timeout,

C{False} if not.

_prepareWriteBuffer(data: bytes) Tuple[int, c_char_p]

Private helper method to allow derived classes to prepare buffers in different ways

write(data: bytes)
close()
_asyncRead(param: int | None = None)
_ioDone(error, numberOfBytes: int, overlapped)
_notifyReceive(data: bytes)

Called when data is received. The base implementation just calls the onReceive callback provided to the constructor. This can be extended to perform tasks before/after the callback. @type data: bytes

class hwIo.base.Serial(*args, onReceive: Callable[[bytes], None], onReadError: Callable[[int], bool] | None = None, ioThread: IoThread | None = None, **kwargs)

Bases: IoBase

Raw I/O for serial devices. This extends pyserial to call a callback when data is received.

Constructor. Pass the arguments you would normally pass to L{serial.Serial}. There are also some additional keyword arguments (the first, onReceive, is required).

@param onReceive: A callable taking a byte of received data as its only argument.

This callable can then call C{read} to get additional data if desired.

@param onReadError: If provided, a callback that takes the error code for a failed read

and returns True if the I/O loop should exit cleanly or False if an exception should be thrown

@param ioThread: If provided, the I/O thread used for background reads.

if C{None}, defaults to L{hwIo.bgThread}

read(size=1) bytes
write(data: bytes)
close()
_notifyReceive(data: bytes)

Called when data is received. The base implementation just calls the onReceive callback provided to the constructor. This can be extended to perform tasks before/after the callback. @type data: bytes

_setTimeout(timeout: int | None)
class hwIo.base.Bulk(path: str, epIn: int, epOut: int, onReceive: Callable[[bytes], None], onReceiveSize: int = 1, onReadError: Callable[[int], bool] | None = None, ioThread: IoThread | None = None)

Bases: IoBase

Raw I/O for bulk USB devices. This implementation assumes that the used Bulk device has two separate end points for input and output.

Constructor. @param path: The device path. @param epIn: The endpoint to read data from. @param epOut: The endpoint to write data to. @param onReceive: A callable taking a received input report as its only argument. @param onReadError: An optional callable that handles read errors.

It takes an error code and returns True if the error has been handled, allowing the read loop to exit cleanly, or False if an exception should be thrown.

@param ioThread: If provided, the I/O thread used for background reads.

if C{None}, defaults to L{hwIo.bgThread}

close()
hwIo.base.boolToByte(arg: bool) bytes
hwIo.base.intToByte(arg: int) bytes

Convert an int (value < 256) to a single byte bytes object

hwIo.base.getByte(arg: bytes, index: int) bytes

Return the single byte at index

hwIo.hid module

Raw input/output for braille displays via HID Braille display drivers must be thread-safe to use this, as it utilises a background thread. See L{braille.BrailleDisplayDriver.isThreadSafe}.

exception hwIo.hid.HidPError

Bases: RuntimeError

hwIo.hid.check_HidP_status(func, *args)
class hwIo.hid.HidReport(device)

Bases: object

_reportType: HIDP_REPORT_TYPE
_reportSize: int
_reportBuf: Array
class hwIo.hid.HidInputReport(device, data)

Bases: HidReport

_reportType: HIDP_REPORT_TYPE = 0
getUsages(usagePage, linkCollection=0)
getDataItems()
class hwIo.hid.HidOutputReport(device, reportID=0)

Bases: HidReport

_reportType: HIDP_REPORT_TYPE = 1
property data
setUsageValueArray(usagePage, linkCollection, usage, data)
class hwIo.hid.Hid(path: str, onReceive: Callable[[bytes], None], exclusive: bool = True, onReadError: Callable[[int], bool] | None = None, ioThread: IoThread | None = None)

Bases: IoBase

Raw I/O for HID devices.

Constructor. @param path: The device path.

This can be retrieved using L{hwPortUtils.listHidDevices}.

@param onReceive: A callable taking a received input report as its only argument. @param exclusive: Whether to block other application’s access to this device. @param onReadError: An optional callable that handles read errors.

It takes an error code and returns True if the error has been handled, allowing the read loop to exit cleanly, or False if an exception should be thrown.

@param ioThread: If provided, the I/O thread used for background reads.

if C{None}, defaults to L{hwIo.bgThread}

_featureSize: int
property caps
property inputButtonCaps: Array[HIDP_VALUE_CAPS]
property inputValueCaps: Array[HIDP_VALUE_CAPS]
property outputValueCaps: Array[HIDP_VALUE_CAPS]
_prepareWriteBuffer(data: bytes) Tuple[int, c_char_p]

For HID devices, the buffer to be written must match the OutputReportByteLength fetched from HIDP_CAPS, to ensure this is the case we create a buffer of that size. We also check that data is not bigger than the write size, which we do not currently support. If it becomes necessary to support this, we could split the data and send it several chunks.

getFeature(reportId: bytes) bytes

Get a feature report from this device. @param reportId: The report id. @return: The report, including the report id.

setFeature(report: bytes) None

Send a feature report to this device. @param report: The report, including its id.

setOutputReport(report: bytes) None

Write the given report to the device using HidD_SetOutputReport. This is instead of using the standard WriteFile which may freeze with some USB HID implementations. @param report: The report, including its id.

close()

hwIo.ioThread module

hwIo.ioThread._generateApcParams() Generator[int, None, None]

Generator of APC params for internal use. Params generated using this generator are passed to our internal APC to lookup Python functions. A parameter passed to an APC is of type ULONG_PTR, which has a size of 4 bytes. Therefore, we use a counter which starts at 0, counts up to 0xffffffff, wraps back to 0 and continues cycling.

class hwIo.ioThread.IoThread

Bases: Thread

A thread used for background writes and raw I/O, e.g. for braille displays.

This constructor should always be called with keyword arguments. Arguments are:

group should be None; reserved for future extension when a ThreadGroup class is implemented.

target is the callable object to be invoked by the run() method. Defaults to None, meaning nothing is called.

name is the thread name. By default, a unique name is constructed of the form “Thread-N” where N is a small decimal number.

args is a list or tuple of arguments for the target invocation. Defaults to ().

kwargs is a dictionary of keyword arguments for the target invocation. Defaults to {}.

If a subclass overrides the constructor, it must make sure to invoke the base class constructor (Thread.__init__()) before doing anything else to the thread.

exit: bool = False
_apcParamCounter = <generator object _generateApcParams>
_apcStore: Dict[int, Tuple[Callable[[int], None] | BoundMethodWeakref[Callable[[int], None]] | AnnotatableWeakref[Callable[[int], None]], int]] = {}

Store of Python functions to be called as APC. This allows us to have a single APC function in the class rather than on each instance, which prevents reference cycles.

_completionRoutineStore: Dict[int, Tuple[BoundMethodWeakref[Callable[[int, int, LP__OVERLAPPED], None]] | AnnotatableWeakref[Callable[[int, int, LP__OVERLAPPED], None]], _OVERLAPPED]] = {}

Store of Python functions to be called as Overlapped Completion Routine. This allows us to have a single completion routine in the class rather than on each instance, which prevents reference cycles. Note that we use the address of the OVERLAPPED structure as key in this store, eventhough the structure is also stored in the value. The OVERLAPPED structure can’t be used as key because ctypes does not have OOR (original object return), it constructs a new, equivalent object each time you retrieve the contents of a LPOVERLAPPED.

_internalApc = <WinFunctionType object>
_internalCompletionRoutine = <WinFunctionType object>
start()

Start the thread’s activity.

It must be called at most once per thread object. It arranges for the object’s run() method to be invoked in a separate thread of control.

This method will raise a RuntimeError if called more than once on the same thread object.

_registerToCallAsApc(func: Callable[[int], None], param: int = 0) int

Internal method to store a python function to be called in an Asynchronous Procedure Call (APC). The function and param are saved in a store on the IoThread instance. When our internal APC executes the function, the entry be popped from the store. This method does not queue the APC itself. The saved python function will be weakly referenced, therefore the caller should keep a reference to the python function. @param func: The function to be called in an APC. @param param: The parameter passed to the APC when called. @returns: The internal param to pass to the internal APC.

queueAsApc(func: Callable[[int], None], param: int = 0)

safely queues a Python function call as an Asynchronous Procedure Call (APC). The function and param are saved in a store on the IoThread instance. When our internal APC executes the function, the entry will be popped from the store. The queued python function will be weakly referenced, therefore the caller should keep a reference to the python function. @param func: The function to be called in an APC. @param param: The parameter passed to the APC when called.

setWaitableTimer(handle: int | c_void_p, dueTime: int, func: Callable[[int], None], param: int = 0)

“Safe wrapper around winKernel.setWaitableTimer that uses an internal APC. A weak reference to the function and its param are saved in a store on the IoThread instance. When our internal APC executes the function, the entry will be popped from the store. Note that as the python function is weakly referenced, the caller should keep a reference to the python function. @param handle: A handle to the timer object. @param dueTime: Relative time (in miliseconds). @param func: The function to be executed when the timer elapses. @param param: The parameter passed to the APC when called.

queueAsCompletionRoutine(func: Callable[[int, int, LP__OVERLAPPED], None], overlapped: _OVERLAPPED)

safely queues a Python function call as an overlapped completion routine. A weak reference to the Python function is saved in a store on the IoThread instance When our internal completion routine executes the function, it will be popped from the store. The wrapped python function is weakly referenced, therefore the caller should keep a reference to the python function (not the completion routine itself). @param func: The function to be wrapped in a completion routine. @param overlapped: The overlapped structure @returns: The completion routine.

stop(timeout: float | None = None)
run()

Method representing the thread’s activity.

You may override this method in a subclass. The standard run() method invokes the callable object passed to the object’s constructor as the target argument, if any, with sequential and keyword arguments taken from the args and kwargs arguments, respectively.