storm/fenrir
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Code cleanup and bug fixes.

Browse Source
This commit is contained in:
Storm Dragon 2025-06-20 02:19:57 -04:00
commit a394ea0222
6 changed files with 158 additions and 452 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/fenrirscreenreader/core/debugManager.py
View File

@ -32,6 +32,8 @@ class debugManager():
pathlib.Path(directory).mkdir(parents=True, exist_ok=True) pathlib.Path(directory).mkdir(parents=True, exist_ok=True)
try: try:
self._file = open(self._fileName,'a') self._file = open(self._fileName,'a')
# Set readable permissions for all users (644)
os.chmod(self._fileName, 0o644)
self._fileOpened = True self._fileOpened = True
except Exception as e: except Exception as e:
print(e) print(e)

56
src/fenrirscreenreader/core/settingsManager.py
View File

@ -49,33 +49,35 @@ class settingsManager():
def getBindingBackup(self): def getBindingBackup(self):
return self.bindingsBackup.copy() return self.bindingsBackup.copy()
def loadSoundIcons(self, soundIconPath): def loadSoundIcons(self, soundIconPath):
siConfig = open(soundIconPath + '/soundicons.conf',"r") try:
while(True): with open(soundIconPath + '/soundicons.conf', "r") as siConfig:
line = siConfig.readline() while(True):
if not line: line = siConfig.readline()
break if not line:
line = line.replace('\n','') break
if line.replace(" ","") == '': line = line.replace('\n','')
continue if line.replace(" ","") == '':
if line.replace(" ","").startswith("#"): continue
continue if line.replace(" ","").startswith("#"):
if line.count("=") != 1: continue
continue if line.count("=") != 1:
Values = line.split('=') continue
soundIcon = Values[0].upper() Values = line.split('=')
Values[1] = Values[1].replace("'","") soundIcon = Values[0].upper()
Values[1] = Values[1].replace('"',"") Values[1] = Values[1].replace("'","")
soundIconFile = '' Values[1] = Values[1].replace('"',"")
if os.path.exists(Values[1]): soundIconFile = ''
soundIconFile = Values[1] if os.path.exists(Values[1]):
else: soundIconFile = Values[1]
if not soundIconPath.endswith("/"): else:
soundIconPath += '/' if not soundIconPath.endswith("/"):
if os.path.exists(soundIconPath + Values[1]): soundIconPath += '/'
soundIconFile = soundIconPath + Values[1] if os.path.exists(soundIconPath + Values[1]):
self.env['soundIcons'][soundIcon] = soundIconFile soundIconFile = soundIconPath + Values[1]
self.env['runtime']['debug'].writeDebugOut("SoundIcon: " + soundIcon + '.' + soundIconFile, debug.debugLevel.INFO, onAnyLevel=True) self.env['soundIcons'][soundIcon] = soundIconFile
siConfig.close() self.env['runtime']['debug'].writeDebugOut("SoundIcon: " + soundIcon + '.' + soundIconFile, debug.debugLevel.INFO, onAnyLevel=True)
except (IOError, OSError) as e:
self.env['runtime']['debug'].writeDebugOut('loadSoundIcons: failed to load sound icons from ' + soundIconPath + '. Error: ' + str(e), debug.debugLevel.ERROR)
def getSettingsFile(self): def getSettingsFile(self):
return self.settingsFile return self.settingsFile
def setSettingsFile(self, settingsFile): def setSettingsFile(self, settingsFile):

2
src/fenrirscreenreader/fenrirVersion.py
View File

@ -4,5 +4,5 @@
# Fenrir TTY screen reader # Fenrir TTY screen reader
# By Chrys, Storm Dragon, and contributers. # By Chrys, Storm Dragon, and contributers.
version = "2025.06.17" version = "2025.06.20"
codeName = "master" codeName = "master"

351
src/fenrirscreenreader/inputDriver/atspiDriver.py
View File

@ -1,351 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Fenrir TTY screen reader
# By Chrys, Storm Dragon, and contributers.
_evdevAvailable = False
_udevAvailable = False
_evdevAvailableError = ''
_udevAvailableError = ''
try:
import gi
from gi.repository import GLib
gi.require_version('Gst', '1.0')
from gi.repository import Gst
gi.require_version('Atspi', '2.0')
import pyatspi
available = True
except Exception as e:
_availableError = str(e)
import time
from fenrirscreenreader.core.eventData import fenrirEventType
from fenrirscreenreader.core import inputData
from fenrirscreenreader.core import debug
from fenrirscreenreader.core.inputDriver import inputDriver
'''
def on_key_input(event):
print(event)
mainloop = GLib.MainLoop()
thread = threading.Thread(target=mainloop.run)
thread.start()
#pyatspi.Registry.registerKeystrokeListener(on_key_input, kind=(pyatspi.KEY_PRESSED_EVENT, pyatspi.KEY_RELEASED_EVENT))
pyatspi.Registry.registerKeystrokeListener(on_key_input,mask=pyatspi.allModifiers(), kind=(pyatspi.KEY_PRESS,pyatspi.KEY_RELEASE,pyatspi.KEY_PRESSRELEASE), synchronous=True, preemptive=True)
pyatspi.Registry.start()
'''
class driver(inputDriver):
def __init__(self):
inputDriver.__init__(self)
def initialize(self, environment):
self.env = environment
self.env['runtime']['inputManager'].setShortcutType('KEY')
global _available
self._initialized = _available
if not self._initialized:
global _availableError
currError = ' '
if not _evdevAvailable:
currError += _availableError
self.env['runtime']['debug'].writeDebugOut('InputDriver:' + currError, debug.debugLevel.ERROR)
return
self.env['runtime']['processManager'].addCustomEventThread(self.inputWatchdog)
def inputWatchdog(self,active , eventQueue):
try:
while active.value:
r, w, x = select(self.iDevices, [], [], 0.7)
for fd in r:
event = None
foreward = False
eventFired = False
try:
event = self.iDevices[fd].read_one()
except:
self.removeDevice(fd)
while(event):
self.env['input']['eventBuffer'].append( [self.iDevices[fd], self.uDevices[fd], event])
if event.type == evdev.events.EV_KEY:
if event.code != 0:
currMapEvent = self.mapEvent(event)
if not currMapEvent:
foreward = True
if not isinstance(currMapEvent['EventName'], str):
foreward = True
if not foreward or eventFired:
if currMapEvent['EventState'] in [0,1,2]:
eventQueue.put({"Type":fenrirEventType.KeyboardInput,"Data":currMapEvent.copy()})
eventFired = True
else:
if not event.type in [0,4]:
foreward = True
event = self.iDevices[fd].read_one()
if foreward and not eventFired:
self.writeEventBuffer()
self.clearEventBuffer()
except Exception as e:
self.env['runtime']['debug'].writeDebugOut("INPUT WATCHDOG CRASH: "+str(e),debug.debugLevel.ERROR)
def writeEventBuffer(self):
if not self._initialized:
return
for iDevice, uDevice, event in self.env['input']['eventBuffer']:
try:
if uDevice:
if self.gDevices[iDevice.fd]:
self.writeUInput(uDevice, event)
except Exception as e:
pass
def writeUInput(self, uDevice, event):
if not self._initialized:
return
uDevice.write_event(event)
uDevice.syn()
def updateInputDevices(self, newDevices = None, init = False):
if init:
self.removeAllDevices()
deviceFileList = None
if newDevices and not init:
deviceFileList = newDevices
else:
deviceFileList = evdev.list_devices()
if len(deviceFileList) == self.iDeviceNo:
return
if not deviceFileList:
return
mode = self.env['runtime']['settingsManager'].getSetting('keyboard', 'device').upper()
iDevicesFiles = []
for device in self.iDevices:
iDevicesFiles.append(self.iDevices[device].fn)
eventType = evdev.events
for deviceFile in deviceFileList:
try:
if not deviceFile:
continue
if deviceFile == '':
continue
if deviceFile in iDevicesFiles:
continue
try:
with open(deviceFile) as f:
pass
except Exception as e:
continue
# 3 pos absolute
# 2 pos relative
# 1 Keys
try:
currDevice = evdev.InputDevice(deviceFile)
except:
continue
try:
if currDevice.name.upper() in ['','SPEAKUP','PY-EVDEV-UINPUT']:
continue
if currDevice.phys.upper() in ['','SPEAKUP','PY-EVDEV-UINPUT']:
continue
if 'BRLTTY' in currDevice.name.upper():
continue
except:
pass
cap = currDevice.capabilities()
if mode in ['ALL','NOMICE']:
if eventType.EV_KEY in cap:
if 116 in cap[eventType.EV_KEY] and len(cap[eventType.EV_KEY]) < 10:
self.env['runtime']['debug'].writeDebugOut('Device Skipped (has 116):' + currDevice.name,debug.debugLevel.INFO)
continue
if len(cap[eventType.EV_KEY]) < 60:
self.env['runtime']['debug'].writeDebugOut('Device Skipped (< 60 keys):' + currDevice.name,debug.debugLevel.INFO)
continue
if mode == 'ALL':
self.addDevice(currDevice)
self.env['runtime']['debug'].writeDebugOut('Device added (ALL):' + self.iDevices[currDevice.fd].name, debug.debugLevel.INFO)
elif mode == 'NOMICE':
if not ((eventType.EV_REL in cap) or (eventType.EV_ABS in cap)):
self.addDevice(currDevice)
self.env['runtime']['debug'].writeDebugOut('Device added (NOMICE):' + self.iDevices[currDevice.fd].name,debug.debugLevel.INFO)
else:
self.env['runtime']['debug'].writeDebugOut('Device Skipped (NOMICE):' + currDevice.name,debug.debugLevel.INFO)
else:
self.env['runtime']['debug'].writeDebugOut('Device Skipped (no EV_KEY):' + currDevice.name,debug.debugLevel.INFO)
elif currDevice.name.upper() in mode.split(','):
self.addDevice(currDevice)
self.env['runtime']['debug'].writeDebugOut('Device added (Name):' + self.iDevices[currDevice.fd].name,debug.debugLevel.INFO)
except Exception as e:
self.env['runtime']['debug'].writeDebugOut("Device Skipped (Exception): " + deviceFile +' ' + currDevice.name +' '+ str(e),debug.debugLevel.INFO)
self.iDeviceNo = len(evdev.list_devices())
self.updateMPiDevicesFD()
def updateMPiDevicesFD(self):
try:
for fd in self.iDevices:
if not fd in self.iDevicesFD:
self.iDevicesFD.append(fd)
for fd in self.iDevicesFD:
if not fd in self.iDevices:
self.iDevicesFD.remove(fd)
except:
pass
def mapEvent(self, event):
if not self._initialized:
return None
if not event:
return None
mEvent = inputData.inputEvent
try:
mEvent['EventName'] = evdev.ecodes.keys[event.code]
mEvent['EventValue'] = event.code
mEvent['EventSec'] = event.sec
mEvent['EventUsec'] = event.usec
mEvent['EventState'] = event.value
mEvent['EventType'] = event.type
return mEvent
except Exception as e:
return None
def getLedState(self, led = 0):
if not self.hasIDevices():
return False
# 0 = Numlock
# 1 = Capslock
# 2 = Rollen
for fd, dev in self.iDevices.items():
if led in dev.leds():
return True
return False
def toggleLedState(self, led = 0):
if not self.hasIDevices():
return False
ledState = self.getLedState(led)
for i in self.iDevices:
# 17 LEDs
if 17 in self.iDevices[i].capabilities():
if ledState == 1:
self.iDevices[i].set_led(led , 0)
else:
self.iDevices[i].set_led(led , 1)
def grabAllDevices(self):
if not self._initialized:
return
for fd in self.iDevices:
self.grabDevice(fd)
def ungrabAllDevices(self):
if not self._initialized:
return
for fd in self.iDevices:
self.ungrabDevice(fd)
def createUInputDev(self, fd):
if not self.env['runtime']['settingsManager'].getSettingAsBool('keyboard', 'grabDevices'):
self.uDevices[fd] = None
return
try:
test = self.uDevices[fd]
return
except KeyError:
self.uDevices[fd] = None
if self.uDevices[fd] != None:
return
try:
self.uDevices[fd] = UInput.from_device(self.iDevices[fd])
except Exception as e:
try:
self.env['runtime']['debug'].writeDebugOut('InputDriver evdev: compat fallback: ' + str(e),debug.debugLevel.WARNING)
dev = self.iDevices[fd]
cap = dev.capabilities()
del cap[0]
self.uDevices[fd] = UInput(
cap,
dev.name,
)
except Exception as e:
self.env['runtime']['debug'].writeDebugOut('InputDriver evdev: init Uinput not possible: ' + str(e),debug.debugLevel.ERROR)
return
def addDevice(self, newDevice):
self.env['runtime']['debug'].writeDebugOut('InputDriver evdev: device added: ' + str(newDevice.fd) + ' ' +str(newDevice),debug.debugLevel.INFO)
self.iDevices[newDevice.fd] = newDevice
self.gDevices[newDevice.fd] = False
self.createUInputDev(newDevice.fd)
def grabDevice(self, fd):
if not self.env['runtime']['settingsManager'].getSettingAsBool('keyboard', 'grabDevices'):
return
try:
self.iDevices[fd].grab()
self.gDevices[fd] = True
self.env['runtime']['debug'].writeDebugOut('InputDriver evdev: grab device ('+ str(self.iDevices[fd].name) + ')',debug.debugLevel.INFO)
except IOError:
self.gDevices[fd] = True
except Exception as e:
self.env['runtime']['debug'].writeDebugOut('InputDriver evdev: grabing not possible: ' + str(e),debug.debugLevel.ERROR)
def ungrabDevice(self,fd):
if not self.env['runtime']['settingsManager'].getSettingAsBool('keyboard', 'grabDevices'):
return
try:
self.gDevices[fd] = False
self.iDevices[fd].ungrab()
self.env['runtime']['debug'].writeDebugOut('InputDriver evdev: ungrab device ('+ str(self.iDevices[fd].name) + ')',debug.debugLevel.INFO)
except:
pass
def removeDevice(self,fd):
self.env['runtime']['debug'].writeDebugOut('InputDriver evdev: device removed: ' + str(fd) + ' ' +str(self.iDevices[fd]),debug.debugLevel.INFO)
self.clearEventBuffer()
try:
self.ungrabDevice(fd)
except:
pass
try:
self.iDevices[fd].close()
except:
pass
try:
self.uDevices[fd].close()
except:
pass
try:
del(self.iDevices[fd])
except:
pass
try:
del(self.uDevices[fd])
except:
pass
try:
del(self.gDevices[fd])
except:
pass
self.updateMPiDevicesFD()
def hasIDevices(self):
if not self._initialized:
return False
if not self.iDevices:
return False
if len(self.iDevices) == 0:
return False
return True
def removeAllDevices(self):
if not self.hasIDevices():
return
devices = self.iDevices.copy()
for fd in devices:
self.removeDevice(fd)
self.iDevices.clear()
self.uDevices.clear()
self.gDevices.clear()
self.iDeviceNo = 0

124
src/fenrirscreenreader/inputDriver/evdevDriver.py
View File

@ -24,6 +24,7 @@ from select import select
import multiprocessing import multiprocessing
from multiprocessing.sharedctypes import Value from multiprocessing.sharedctypes import Value
from ctypes import c_bool from ctypes import c_bool
import threading
from fenrirscreenreader.core.eventData import fenrirEventType from fenrirscreenreader.core.eventData import fenrirEventType
from fenrirscreenreader.core import inputData from fenrirscreenreader.core import inputData
@ -41,6 +42,7 @@ class driver(inputDriver):
self.iDeviceNo = 0 self.iDeviceNo = 0
self.watchDog = Value(c_bool, True) self.watchDog = Value(c_bool, True)
self.UInputinject = UInput() self.UInputinject = UInput()
self._deviceLock = threading.Lock()
def initialize(self, environment): def initialize(self, environment):
self.env = environment self.env = environment
self.env['runtime']['inputManager'].setShortcutType('KEY') self.env['runtime']['inputManager'].setShortcutType('KEY')
@ -101,29 +103,45 @@ class driver(inputDriver):
def inputWatchdog(self, active, eventQueue): def inputWatchdog(self, active, eventQueue):
try: try:
while active.value: while active.value:
r, w, x = select(self.iDevices, [], [], 0.8) # Get a snapshot of devices for select() to avoid lock contention
with self._deviceLock:
devices_snapshot = self.iDevices.copy()
if not devices_snapshot:
time.sleep(0.1)
continue
r, w, x = select(devices_snapshot, [], [], 0.8)
event = None event = None
foundKeyInSequence = False foundKeyInSequence = False
foreward = False foreward = False
eventFired = False eventFired = False
for fd in r: for fd in r:
# Check if device still exists before accessing
with self._deviceLock:
if fd not in self.iDevices:
continue
device = self.iDevices[fd]
udevice = self.uDevices.get(fd)
try: try:
event = self.iDevices[fd].read_one() event = device.read_one()
except: except:
self.removeDevice(fd) self.removeDevice(fd)
continue
while(event): while(event):
self.env['runtime']['debug'].writeDebugOut('inputWatchdog: EVENT:' + str(event), debug.debugLevel.INFO) self.env['runtime']['debug'].writeDebugOut('inputWatchdog: EVENT:' + str(event), debug.debugLevel.INFO)
self.env['input']['eventBuffer'].append([self.iDevices[fd], self.uDevices[fd], event]) self.env['input']['eventBuffer'].append([device, udevice, event])
if event.type == evdev.events.EV_KEY: if event.type == evdev.events.EV_KEY:
if not foundKeyInSequence: if not foundKeyInSequence:
foundKeyInSequence = True foundKeyInSequence = True
if event.code != 0: if event.code != 0:
currMapEvent = self.mapEvent(event) currMapEvent = self.mapEvent(event)
if not currMapEvent: if not currMapEvent:
event = self.iDevices[fd].read_one() event = device.read_one()
continue continue
if not isinstance(currMapEvent['EventName'], str): if not isinstance(currMapEvent['EventName'], str):
event = self.iDevices[fd].read_one() event = device.read_one()
continue continue
if currMapEvent['EventState'] in [0, 1, 2]: if currMapEvent['EventState'] in [0, 1, 2]:
eventQueue.put({"Type": fenrirEventType.KeyboardInput, "Data": currMapEvent.copy()}) eventQueue.put({"Type": fenrirEventType.KeyboardInput, "Data": currMapEvent.copy()})
@ -132,7 +150,7 @@ class driver(inputDriver):
if event.type in [2, 3]: if event.type in [2, 3]:
foreward = True foreward = True
event = self.iDevices[fd].read_one() event = device.read_one()
if not foundKeyInSequence: if not foundKeyInSequence:
if foreward and not eventFired: if foreward and not eventFired:
self.writeEventBuffer() self.writeEventBuffer()
@ -351,24 +369,26 @@ class driver(inputDriver):
def addDevice(self, newDevice): def addDevice(self, newDevice):
self.env['runtime']['debug'].writeDebugOut('InputDriver evdev: device added: ' + str(newDevice.fd) + ' ' + str(newDevice), debug.debugLevel.INFO) self.env['runtime']['debug'].writeDebugOut('InputDriver evdev: device added: ' + str(newDevice.fd) + ' ' + str(newDevice), debug.debugLevel.INFO)
try: with self._deviceLock:
self.iDevices[newDevice.fd] = newDevice
self.createUInputDev(newDevice.fd)
self.gDevices[newDevice.fd] = False
except:
# if it doesnt work clean up
try: try:
del(self.iDevices[newDevice.fd]) self.iDevices[newDevice.fd] = newDevice
except: self.createUInputDev(newDevice.fd)
pass self.gDevices[newDevice.fd] = False
try: except Exception as e:
del(self.uDevices[newDevice.fd]) self.env['runtime']['debug'].writeDebugOut('InputDriver evdev: error adding device: ' + str(e), debug.debugLevel.ERROR)
except: # if it doesnt work clean up
pass try:
try: del(self.iDevices[newDevice.fd])
del(self.gDevices[newDevice.fd]) except:
except: pass
pass try:
del(self.uDevices[newDevice.fd])
except:
pass
try:
del(self.gDevices[newDevice.fd])
except:
pass
def grabDevice(self, fd): def grabDevice(self, fd):
if not self.env['runtime']['settingsManager'].getSettingAsBool('keyboard', 'grabDevices'): if not self.env['runtime']['settingsManager'].getSettingAsBool('keyboard', 'grabDevices'):
@ -424,33 +444,37 @@ class driver(inputDriver):
return True return True
def removeDevice(self, fd): def removeDevice(self, fd):
self.env['runtime']['debug'].writeDebugOut('InputDriver evdev: device removed: ' + str(fd) + ' ' + str(self.iDevices[fd]), debug.debugLevel.INFO) with self._deviceLock:
self.clearEventBuffer() try:
try: self.env['runtime']['debug'].writeDebugOut('InputDriver evdev: device removed: ' + str(fd) + ' ' + str(self.iDevices[fd]), debug.debugLevel.INFO)
self.ungrabDevice(fd) except:
except: self.env['runtime']['debug'].writeDebugOut('InputDriver evdev: device removed: ' + str(fd), debug.debugLevel.INFO)
pass self.clearEventBuffer()
try: try:
self.iDevices[fd].close() self.ungrabDevice(fd)
except: except:
pass pass
try: try:
self.uDevices[fd].close() self.iDevices[fd].close()
except: except:
pass pass
try: try:
del(self.iDevices[fd]) self.uDevices[fd].close()
except: except:
pass pass
try: try:
del(self.uDevices[fd]) del(self.iDevices[fd])
except: except:
pass pass
try: try:
del(self.gDevices[fd]) del(self.uDevices[fd])
except: except:
pass pass
self.updateMPiDevicesFD() try:
del(self.gDevices[fd])
except:
pass
self.updateMPiDevicesFD()
def hasIDevices(self): def hasIDevices(self):
if not self._initialized: if not self._initialized:

75
src/fenrirscreenreader/screenDriver/vcsaDriver.py
View File

@ -58,9 +58,8 @@ class driver(screenDriver):
def getCurrScreen(self): def getCurrScreen(self):
self.env['screen']['oldTTY'] = self.env['screen']['newTTY'] self.env['screen']['oldTTY'] = self.env['screen']['newTTY']
try: try:
currScreenFile = open('/sys/devices/virtual/tty/tty0/active','r') with open('/sys/devices/virtual/tty/tty0/active','r') as currScreenFile:
self.env['screen']['newTTY'] = str(currScreenFile.read()[3:-1]) self.env['screen']['newTTY'] = str(currScreenFile.read()[3:-1])
currScreenFile.close()
except Exception as e: except Exception as e:
self.env['runtime']['debug'].writeDebugOut(str(e),debug.debugLevel.ERROR) self.env['runtime']['debug'].writeDebugOut(str(e),debug.debugLevel.ERROR)
def injectTextToScreen(self, text, screen = None): def injectTextToScreen(self, text, screen = None):
@ -118,26 +117,35 @@ class driver(screenDriver):
break break
return d return d
def updateWatchdog(self, active , eventQueue): def updateWatchdog(self, active , eventQueue):
vcsa = {}
vcsu = {}
tty = None
watchdog = None
try: try:
useVCSU = os.access('/dev/vcsu', os.R_OK) useVCSU = os.access('/dev/vcsu', os.R_OK)
vcsa = {}
vcsaDevices = glob.glob('/dev/vcsa*') vcsaDevices = glob.glob('/dev/vcsa*')
vcsu = {}
vcsuDevices = None vcsuDevices = None
lastScreenContent = b'' lastScreenContent = b''
# Open TTY file with proper cleanup
tty = open('/sys/devices/virtual/tty/tty0/active','r') tty = open('/sys/devices/virtual/tty/tty0/active','r')
currScreen = str(tty.read()[3:-1]) currScreen = str(tty.read()[3:-1])
oldScreen = currScreen oldScreen = currScreen
# Open VCSA devices with proper cleanup tracking
for vcsaDev in vcsaDevices: for vcsaDev in vcsaDevices:
index = str(vcsaDev[9:]) index = str(vcsaDev[9:])
vcsa[index] = open(vcsaDev,'rb') vcsa[index] = open(vcsaDev,'rb')
if index == currScreen: if index == currScreen:
lastScreenContent = self.readFile(vcsa[index]) lastScreenContent = self.readFile(vcsa[index])
# Open VCSU devices if available
if useVCSU: if useVCSU:
vcsuDevices = glob.glob('/dev/vcsu*') vcsuDevices = glob.glob('/dev/vcsu*')
for vcsuDev in vcsuDevices: for vcsuDev in vcsuDevices:
index = str(vcsuDev[9:]) index = str(vcsuDev[9:])
vcsu[index] = open(vcsuDev,'rb') vcsu[index] = open(vcsuDev,'rb')
self.updateCharMap(currScreen) self.updateCharMap(currScreen)
watchdog = select.epoll() watchdog = select.epoll()
watchdog.register(vcsa[currScreen], select.POLLPRI | select.POLLERR) watchdog.register(vcsa[currScreen], select.POLLPRI | select.POLLERR)
@ -220,6 +228,28 @@ class driver(screenDriver):
except Exception as e: except Exception as e:
self.env['runtime']['debug'].writeDebugOut('VCSA:updateWatchdog:' + str(e),debug.debugLevel.ERROR) self.env['runtime']['debug'].writeDebugOut('VCSA:updateWatchdog:' + str(e),debug.debugLevel.ERROR)
time.sleep(0.2) time.sleep(0.2)
finally:
# Clean up all file handles
try:
if watchdog:
watchdog.close()
except:
pass
try:
if tty:
tty.close()
except:
pass
for handle in vcsa.values():
try:
handle.close()
except:
pass
for handle in vcsu.values():
try:
handle.close()
except:
pass
def createScreenEventData(self, screen, vcsaContent, vcsuContent = None): def createScreenEventData(self, screen, vcsaContent, vcsuContent = None):
eventData = { eventData = {
@ -252,27 +282,26 @@ class driver(screenDriver):
def updateCharMap(self, screen): def updateCharMap(self, screen):
self.charmap = {} self.charmap = {}
try: try:
tty = open('/dev/tty' + screen, 'rb') with open('/dev/tty' + screen, 'rb') as tty:
GIO_UNIMAP = 0x4B66
VT_GETHIFONTMASK = 0x560D
himask = array("H", (0,))
ioctl(tty, VT_GETHIFONTMASK, himask)
self.hichar, = unpack_from("@H", himask)
sz = 512
line = ''
while True:
try:
unipairs = array("H", [0]*(2*sz))
unimapdesc = array("B", pack("@HP", sz, unipairs.buffer_info()[0]))
ioctl(tty.fileno(), GIO_UNIMAP, unimapdesc)
break
except Exception as e:
self.env['runtime']['debug'].writeDebugOut('VCSA:updateCharMap:scaling up sz=' + str(sz) + ' ' + str(e),debug.debugLevel.WARNING)
sz *= 2
except Exception as e: except Exception as e:
self.env['runtime']['debug'].writeDebugOut('VCSA:updateCharMap:' + str(e),debug.debugLevel.ERROR) self.env['runtime']['debug'].writeDebugOut('VCSA:updateCharMap:' + str(e),debug.debugLevel.ERROR)
return return
GIO_UNIMAP = 0x4B66
VT_GETHIFONTMASK = 0x560D
himask = array("H", (0,))
ioctl(tty, VT_GETHIFONTMASK, himask)
self.hichar, = unpack_from("@H", himask)
sz = 512
line = ''
while True:
try:
unipairs = array("H", [0]*(2*sz))
unimapdesc = array("B", pack("@HP", sz, unipairs.buffer_info()[0]))
ioctl(tty.fileno(), GIO_UNIMAP, unimapdesc)
break
except Exception as e:
self.env['runtime']['debug'].writeDebugOut('VCSA:updateCharMap:scaling up sz=' + str(sz) + ' ' + str(e),debug.debugLevel.WARNING)
sz *= 2
tty.close()
ncodes, = unpack_from("@H", unimapdesc) ncodes, = unpack_from("@H", unimapdesc)
utable = unpack_from("@%dH" % (2*ncodes), unipairs) utable = unpack_from("@%dH" % (2*ncodes), unipairs)
for u, b in zip(utable[::2], utable[1::2]): for u, b in zip(utable[::2], utable[1::2]):