danvics/python-validity
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

A couple of small tweaks

Browse source 
- added a [lame] way to cancel an active capture
- detect when usb device is disconnected (seems to work fine with suspend/resume)
- try to gracefully end dbus service when device is disconnected
This commit is contained in:
Viktor Dragomiretskyy 2019-05-27 20:01:17 +12:00
parent b690db7498
commit ae7f5f9a3b
3 changed files with 132 additions and 30 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

52
dbus-service.py
View file

@ -8,10 +8,12 @@ from time import sleep
from prototype import *
import pwd
load97()
print("Starting up")
loop = GLib.MainLoop()
usb.quit = lambda e: loop.quit()
def uname2identity(uname):
if uname == '':
# For some reason Gnome enrollment UI does not send the user name
@ -22,6 +24,8 @@ def uname2identity(uname):
sidstr='S-1-5-21-111111111-1111111111-1111111111-%d' % pw.pw_uid
return sid_from_string(sidstr)
class AlreadyInUse(Exception):
__name__ = 'net.reactivated.Fprint.Error.AlreadyInUse'
class Device():
name = 'Validity sensor'
@ -29,40 +33,55 @@ class Device():
def __init__(self):
setattr(self, 'num-enroll-stages', 7)
setattr(self, 'scan-type', 'press')
self.capturing = False
def Claim(self, usr):
print('In Claim %s' % usr)
def Release(self):
print('In Release')
self.caimed = False
def ListEnrolledFingers(self, usr):
print('In ListEnrolledFingers %s' % usr)
try:
print('In ListEnrolledFingers %s' % usr)
usr=db.lookup_user(uname2identity(usr))
usr=db.lookup_user(uname2identity(usr))
if usr == None:
print('User not found on this device')
return []
rc = [subtype_to_string(f['subtype']) for f in usr.fingers]
print(repr(rc))
return rc
if usr == None:
print('User not found on this device')
return []
rc = [subtype_to_string(f['subtype']) for f in usr.fingers]
print(repr(rc))
return rc
except Exception as e:
loop.quit()
raise e
def do_scan(self):
if self.capturing:
return
try:
self.capturing = True
z=identify()
self.VerifyStatus('verify-match', True)
except Exception as e:
print('Opps: %s' % repr(e))
self.VerifyStatus('verify-no-match', True)
#loop.quit();
raise e
finally:
self.capturing = False
def VerifyStart(self, finger_name):
print('In VerifyStart %s' % finger_name)
Thread(target=lambda: self.do_scan()).start()
#self.do_scan()
def VerifyStop(self):
print('In VerifyStop')
cancel_capture()
def DeleteEnrolledFingers(self, user):
print('In DeleteEnrolledFingers %s' % user)
@ -82,8 +101,9 @@ class Device():
print('Enroll was successfull')
self.EnrollStatus('enroll-completed', True)
except Exception as e:
print('Enroll failed %s' % repr(e))
self.EnrollStatus('enroll-failed', True)
#loop.quit();
raise e
def EnrollStart(self, finger_name):
print('In EnrollStart %s' % finger_name)
@ -129,4 +149,12 @@ bus.publish('net.reactivated.Fprint',
('/net/reactivated/Fprint/Manager', Manager()),
('/net/reactivated/Fprint/Device/0', Device())
)
open97()
usb.trace_enabled = True
tls.trace_enabled = True
loop.run()
print("Normal exit")

62
prototype.py
View file

@ -102,9 +102,20 @@ c1e1e1c221f201d21201e1c1f34242221201f20221f201e241e241d2020221e2420231d221e211e1
def start_scan(cmd):
assert_status(tls.app(cmd))
def cancel_capture():
usb.queue.put(b'')
#sleep(0.2)
#rsp=tls.app(b'\x04')
#assert_status(rsp)
usb.read_82()
def wait_for_finger():
while True:
b=usb.wait_int()
if len(b) == 0:
raise Exception('Cancelled')
if b[0] == 2:
break
@ -124,21 +135,28 @@ def stop_prg():
return tls.app(unhexlify('5100200000'))
def capture(prg):
usb.purge_int_queue()
start_scan(prg)
b=usb.wait_int()
if b[0] != 0:
raise Exception('Unexpected interrupt type' % hexlify(b).decode())
wait_for_finger()
wait_till_finished()
res=stop_prg()
b=usb.wait_int()
if b[0] != 3:
raise Exception('Unexpected interrupt type %s' % hexlify(b).decode())
try:
wait_for_finger()
wait_till_finished()
finally:
res=stop_prg()
while True:
b=usb.wait_int()
if b[0] != 3:
raise Exception('Unexpected interrupt type %s' % hexlify(b).decode())
if b[1] == 0x43:
break
assert_status(res)
res = res[2:]
@ -257,11 +275,11 @@ def parse_dict(x):
def identify():
glow_start_scan()
err = capture(identify_prg)
if err != 0:
raise Exception('Capture failed: %08x' % err)
try:
err = capture(identify_prg)
if err != 0:
raise Exception('Capture failed: %08x' % err)
# which finger?
stg_id=0 # match against any storage
usr_id=0 # match against any user
@ -327,3 +345,21 @@ def identify():
return rsp
def read_flash(which, size):
block_size = 0x1000
blocks = [tls.read_flash(which, addr, 0x1000) for addr in range(0, size, 0x1000)]
return b''.join(blocks)
def dump_flash():
with open('db_flash.bin', 'wb') as f:
f.write(read_flash(4, 0x30000))
with open('cert_flash.bin', 'wb') as f:
f.write(read_flash(1, 0x1000))
with open('calib_flash.bin', 'wb') as f:
f.write(read_flash(6, 0x8000))
with open('5_flash.bin', 'wb') as f:
f.write(read_flash(5, 0x8000))

48
usb97.py
View file

@ -5,6 +5,9 @@ from binascii import *
from util import assert_status
from struct import unpack
from usb.util import claim_interface, release_interface
from queue import Queue
from threading import Thread
from usb.core import USBError
def unhex(x):
return unhexlify(re.sub('\W', '', x))
@ -40,13 +43,25 @@ aa41249faeef4d838e280cb5d6fc19cfe86c75f
class Usb():
def __init__(self):
self.trace_enabled = False
self.queue = Queue(maxsize=10)
self.quit = None
def purge_int_queue(self):
try:
while True:
self.queue.get_nowait()
except:
pass
def open(self):
self.dev = usb.core.find(idVendor=0x138a, idProduct=0x0097)
self.dev.default_timeout = 5000
self.thread = Thread(target=lambda: self.int_thread())
self.thread.daemon = True
self.thread.start()
def send_init(self):
self.dev.set_configuration()
#self.dev.set_configuration()
# TODO analyse responses, detect hardware type
assert_status(self.cmd(unhexlify('01')))
@ -59,7 +74,7 @@ class Usb():
(err,), rsp = unpack('<H', rsp[:2]), rsp[2:]
if err != 0:
print('Clean slate')
self.cmd(init_hardcoded_clean_slate)
self.cmd(init_hardcoded_clean_slate)
self.cmd(unhexlify('3e'))
# why twice?
@ -73,10 +88,33 @@ class Usb():
self.trace('<cmd< %s' % hexlify(resp).decode())
return resp
def read_82(self):
try:
resp = self.dev.read(130, 100*1024, 100)
resp = bytes(resp)
self.trace('<130< %s' % hexlify(resp).decode())
return resp
except Exception as e:
self.trace('<130< Error: %s' % repr(e))
return None
def int_thread(self):
try:
while True:
resp = self.dev.read(131, 1024, timeout=0)
resp = bytes(resp)
self.trace('<int< %s' % hexlify(resp).decode())
self.queue.put(resp)
except USBError as e:
self.trace('<int< Exception on interrupt thread: %s' % repr(e))
if self.quit != None:
self.quit(e)
finally:
self.trace('<int< Interrupt thread is dead')
def wait_int(self):
resp = self.dev.read(131, 1024, timeout=0)
resp = bytes(resp)
self.trace('<int< %s' % hexlify(resp).decode())
resp = self.queue.get()
return resp
def trace(self, s):