preliminary cleanup of Windows implementation

This commit is contained in:
Benoit Pierre 2021-05-26 19:45:25 +02:00
commit b39eeca9c4

View file

@ -225,33 +225,26 @@ class AbstractStenographMachine:
"""Send a StenoPacket to the machine and return the response or None""" """Send a StenoPacket to the machine and return the response or None"""
raise NotImplementedError('send_receive() is not implemented') raise NotImplementedError('send_receive() is not implemented')
if sys.platform.startswith('win32'): if sys.platform.startswith('win32'):
from ctypes import (
Structure, # For Windows we directly call Windows API functions.
POINTER,
c_ulonglong, from ctypes import windll, wintypes
windll, import ctypes
create_string_buffer,
sizeof,
byref,
pointer,
c_char,
)
from ctypes.wintypes import DWORD, HANDLE, BYTE
import uuid import uuid
# Class GUID for Stenograph USB Writer # Class GUID for Stenograph USB Writer
USB_WRITER_GUID = uuid.UUID('{c5682e20-8059-604a-b761-77c4de9d5dbf}') USB_WRITER_GUID = uuid.UUID('{c5682e20-8059-604a-b761-77c4de9d5dbf}')
class DeviceInterfaceData(Structure): class DeviceInterfaceData(ctypes.Structure):
_fields_ = [ _fields_ = [
('cbSize', DWORD), ('cbSize', wintypes.DWORD),
('InterfaceClassGuid', BYTE * 16), ('InterfaceClassGuid', wintypes.BYTE * 16),
('Flags', DWORD), ('Flags', wintypes.DWORD),
('Reserved', POINTER(c_ulonglong)) ('Reserved', ctypes.POINTER(ctypes.c_ulonglong))
] ]
# For Windows we directly call Windows API functions
SetupDiGetClassDevs = windll.setupapi.SetupDiGetClassDevsA SetupDiGetClassDevs = windll.setupapi.SetupDiGetClassDevsA
SetupDiEnumDeviceInterfaces = windll.setupapi.SetupDiEnumDeviceInterfaces SetupDiEnumDeviceInterfaces = windll.setupapi.SetupDiEnumDeviceInterfaces
SetupDiGetInterfaceDeviceDetail = ( SetupDiGetInterfaceDeviceDetail = (
@ -266,29 +259,30 @@ if sys.platform.startswith('win32'):
ERROR_INSUFFICIENT_BUFFER = 122 ERROR_INSUFFICIENT_BUFFER = 122
class StenographMachine: class StenographMachine:
def __init__(self): def __init__(self):
self._usb_device = HANDLE(0) self._usb_device = wintypes.HANDLE(0)
self._read_buffer = create_string_buffer(MAX_READ + StenoPacket.HEADER_SIZE) self._read_buffer = ctypes.create_string_buffer(MAX_READ + StenoPacket.HEADER_SIZE)
@staticmethod @staticmethod
def _open_device_instance(device_info, guid): def _open_device_instance(device_info, guid):
dev_interface_data = DeviceInterfaceData() dev_interface_data = DeviceInterfaceData()
dev_interface_data.cbSize = sizeof(dev_interface_data) dev_interface_data.cbSize = ctypes.sizeof(dev_interface_data)
status = SetupDiEnumDeviceInterfaces( status = SetupDiEnumDeviceInterfaces(
device_info, None, guid.bytes, 0, byref(dev_interface_data)) device_info, None, guid.bytes, 0, ctypes.byref(dev_interface_data))
if status == 0: if status == 0:
log.debug('status is zero') log.debug('status is zero')
return INVALID_HANDLE_VALUE return INVALID_HANDLE_VALUE
request_length = DWORD(0) request_length = wintypes.DWORD(0)
# Call with None to see how big a buffer we need for detail data. # Call with None to see how big a buffer we need for detail data.
SetupDiGetInterfaceDeviceDetail( SetupDiGetInterfaceDeviceDetail(
device_info, device_info,
byref(dev_interface_data), ctypes.byref(dev_interface_data),
None, None,
0, 0,
pointer(request_length), ctypes.pointer(request_length),
None None
) )
err = GetLastError() err = GetLastError()
@ -298,17 +292,17 @@ if sys.platform.startswith('win32'):
characters = request_length.value characters = request_length.value
class DeviceDetailData(Structure): class DeviceDetailData(ctypes.Structure):
_fields_ = [('cbSize', DWORD), _fields_ = [('cbSize', wintypes.DWORD),
('DevicePath', c_char * characters)] ('DevicePath', ctypes.c_char * characters)]
dev_detail_data = DeviceDetailData() dev_detail_data = DeviceDetailData()
dev_detail_data.cbSize = 5 dev_detail_data.cbSize = 5
# Now put the actual detail data into the buffer # Now put the actual detail data into the buffer
status = SetupDiGetInterfaceDeviceDetail( status = SetupDiGetInterfaceDeviceDetail(
device_info, byref(dev_interface_data), byref(dev_detail_data), device_info, ctypes.byref(dev_interface_data), ctypes.byref(dev_detail_data),
characters, pointer(request_length), None characters, ctypes.pointer(request_length), None
) )
if not status: if not status:
log.debug('not status') log.debug('not status')
@ -331,24 +325,24 @@ if sys.platform.startswith('win32'):
return usb_device return usb_device
def _usb_write_packet(self, request): def _usb_write_packet(self, request):
bytes_written = DWORD(0) bytes_written = wintypes.DWORD(0)
request_packet = request.pack() request_packet = request.pack()
WriteFile( WriteFile(
self._usb_device, self._usb_device,
request_packet, request_packet,
StenoPacket.HEADER_SIZE + request.data_length, StenoPacket.HEADER_SIZE + request.data_length,
byref(bytes_written), ctypes.byref(bytes_written),
None None
) )
return bytes_written.value return bytes_written.value
def _usb_read_packet(self): def _usb_read_packet(self):
bytes_read = DWORD(0) bytes_read = wintypes.DWORD(0)
ReadFile( ReadFile(
self._usb_device, self._usb_device,
byref(self._read_buffer), ctypes.byref(self._read_buffer),
MAX_READ + StenoPacket.HEADER_SIZE, MAX_READ + StenoPacket.HEADER_SIZE,
byref(bytes_read), ctypes.byref(bytes_read),
None None
) )
# Return None if not enough data was read. # Return None if not enough data was read.
@ -379,7 +373,9 @@ if sys.platform.startswith('win32'):
return None return None
writer_packet = self._usb_read_packet() writer_packet = self._usb_read_packet()
return writer_packet return writer_packet
else: else:
from usb import core, util from usb import core, util
class StenographMachine(AbstractStenographMachine): class StenographMachine(AbstractStenographMachine):