gryf/python-linak-desk-control
1
0
Fork 0
mirror of https://github.com/gryf/python-linak-desk-control.git synced 2025-12-17 11:30:29 +01:00
Code Issues Projects Releases Wiki Activity

Removing trailng spaces, removing tabs.

Browse Source
This commit is contained in:
gryf
2021-02-23 20:01:16 +01:00
parent 9b984fb8de
commit bd9455766b
1 changed files with 259 additions and 259 deletions
Download Patch File Download Diff File Expand all files Collapse all files

518
linak-desk-control.py
View File

@@ -1,16 +1,16 @@
# Application to let your desk dance. # Application to let your desk dance.
# Copyright (C) 2018 Lukas Schreiner <dev@lschreiner.de> # Copyright (C) 2018 Lukas Schreiner <dev@lschreiner.de>
# #
# This program is free software: you can redistribute it and/or modify it under # This program is free software: you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free Software # the terms of the GNU General Public License as published by the Free Software
# Foundation, either version 3 of the License, or (at your option) any later version. # Foundation, either version 3 of the License, or (at your option) any later version.
# #
# This program is distributed in the hope that it will be useful, but WITHOUT ANY # This program is distributed in the hope that it will be useful, but WITHOUT ANY
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A # WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A
# PARTICULAR PURPOSE. See the GNU General Public License for more details. # PARTICULAR PURPOSE. See the GNU General Public License for more details.
# #
# You should have received a copy of the GNU General Public License along with this # You should have received a copy of the GNU General Public License along with this
# program. If not, see <https://www.gnu.org/licenses/>. # program. If not, see <https://www.gnu.org/licenses/>.
from ctypes import sizeof from ctypes import sizeof
import sys import sys
@@ -48,307 +48,307 @@ HEIGHT_MOVE_UPWARDS = 32768
HEIGHT_MOVE_END = 32769 HEIGHT_MOVE_END = 32769
class Status(object): class Status(object):
positionLost = True positionLost = True
antiColision = True antiColision = True
overloadDown = True overloadDown = True
overloadUp = True overloadUp = True
unknown = 4 unknown = 4
@classmethod @classmethod
def fromBuf(sr, buf): def fromBuf(sr, buf):
self = sr() self = sr()
attr = ['positionLost', 'antiColision', 'overloadDown', 'overloadUp'] attr = ['positionLost', 'antiColision', 'overloadDown', 'overloadUp']
bitlist = '{:0>8s}'.format(bin(int(buf, base=16)).lstrip('0b')) bitlist = '{:0>8s}'.format(bin(int(buf, base=16)).lstrip('0b'))
for i in range(0, 4): for i in range(0, 4):
setattr(self, attr[i], True if bitlist[i] == '1' else False) setattr(self, attr[i], True if bitlist[i] == '1' else False)
# set unknown # set unknown
self.unkown = int(buf[1:], 16) self.unkown = int(buf[1:], 16)
return self return self
class StatusPositionSpeed(object): class StatusPositionSpeed(object):
pos = None pos = None
status = None status = None
speed = 0 speed = 0
@classmethod @classmethod
def fromBuf(sr, buf): def fromBuf(sr, buf):
self = sr() self = sr()
self.pos = int(buf[2:4] + buf[:2], 16) self.pos = int(buf[2:4] + buf[:2], 16)
self.status = Status.fromBuf(buf[4:6]) self.status = Status.fromBuf(buf[4:6])
self.speed = int(buf[6:8], 16) self.speed = int(buf[6:8], 16)
return self return self
class ValidFlags(object): class ValidFlags(object):
ID00_Ref1_pos_stat_speed = True ID00_Ref1_pos_stat_speed = True
ID01_Ref2_pos_stat_speed = True ID01_Ref2_pos_stat_speed = True
ID02_Ref3_pos_stat_speed = True ID02_Ref3_pos_stat_speed = True
ID03_Ref4_pos_stat_speed = True ID03_Ref4_pos_stat_speed = True
ID10_Ref1_controlInput = True ID10_Ref1_controlInput = True
ID11_Ref2_controlInput = True ID11_Ref2_controlInput = True
ID12_Ref3_controlInput = True ID12_Ref3_controlInput = True
ID13_Ref4_controlInput = True ID13_Ref4_controlInput = True
ID04_Ref5_pos_stat_speed = True ID04_Ref5_pos_stat_speed = True
ID28_Diagnostic = True ID28_Diagnostic = True
ID05_Ref6_pos_stat_speed = True ID05_Ref6_pos_stat_speed = True
ID37_Handset1command = True ID37_Handset1command = True
ID38_Handset2command = True ID38_Handset2command = True
ID06_Ref7_pos_stat_speed = True ID06_Ref7_pos_stat_speed = True
ID07_Ref8_pos_stat_speed = True ID07_Ref8_pos_stat_speed = True
unknown = True unknown = True
@classmethod @classmethod
def fromBuf(sr, buf): def fromBuf(sr, buf):
self = sr() self = sr()
attr = ['ID00_Ref1_pos_stat_speed', 'ID01_Ref2_pos_stat_speed', 'ID02_Ref3_pos_stat_speed', 'ID03_Ref4_pos_stat_speed', 'ID10_Ref1_controlInput', 'ID11_Ref2_controlInput', 'ID12_Ref3_controlInput', 'ID13_Ref4_controlInput', 'ID04_Ref5_pos_stat_speed', 'ID28_Diagnostic', 'ID05_Ref6_pos_stat_speed', 'ID37_Handset1command', 'ID38_Handset2command', 'ID06_Ref7_pos_stat_speed', 'ID07_Ref8_pos_stat_speed', 'unknown'] attr = ['ID00_Ref1_pos_stat_speed', 'ID01_Ref2_pos_stat_speed', 'ID02_Ref3_pos_stat_speed', 'ID03_Ref4_pos_stat_speed', 'ID10_Ref1_controlInput', 'ID11_Ref2_controlInput', 'ID12_Ref3_controlInput', 'ID13_Ref4_controlInput', 'ID04_Ref5_pos_stat_speed', 'ID28_Diagnostic', 'ID05_Ref6_pos_stat_speed', 'ID37_Handset1command', 'ID38_Handset2command', 'ID06_Ref7_pos_stat_speed', 'ID07_Ref8_pos_stat_speed', 'unknown']
bitlist = '{:0>16s}'.format(bin(int(buf, base=16)).lstrip('0b')) bitlist = '{:0>16s}'.format(bin(int(buf, base=16)).lstrip('0b'))
for i in range(0, len(bitlist)): for i in range(0, len(bitlist)):
setattr(self, attr[i], True if bitlist[i] == '1' else False) setattr(self, attr[i], True if bitlist[i] == '1' else False)
return self return self
class StatusReport(object): class StatusReport(object):
featureRaportID = 0 featureRaportID = 0
numberOfBytes = 0 numberOfBytes = 0
validFlag = None validFlag = None
ref1 = None ref1 = None
ref2 = None ref2 = None
ref3 = None ref3 = None
ref4 = None ref4 = None
ref1cnt = 0 ref1cnt = 0
ref2cnt = 0 ref2cnt = 0
ref3cnt = 0 ref3cnt = 0
ref4cnt = 0 ref4cnt = 0
ref5 = None ref5 = None
diagnostic = None diagnostic = None
undefined1 = None undefined1 = None
handset1 = 0 handset1 = 0
handset2 = 0 handset2 = 0
ref6 = None ref6 = None
ref7 = None ref7 = None
ref8 = None ref8 = None
undefined2 = None undefined2 = None
@classmethod @classmethod
def fromBuf(sr, buf): def fromBuf(sr, buf):
self = sr() self = sr()
raw = buf.hex() raw = buf.hex()
self.featureRaportID = buf[0] self.featureRaportID = buf[0]
self.numberOfBytes = buf[1] self.numberOfBytes = buf[1]
self.validFlag = ValidFlags.fromBuf(raw[4:8]) self.validFlag = ValidFlags.fromBuf(raw[4:8])
self.ref1 = StatusPositionSpeed.fromBuf(raw[8:8+8]) self.ref1 = StatusPositionSpeed.fromBuf(raw[8:8+8])
self.ref2 = StatusPositionSpeed.fromBuf(raw[16:16+8]) self.ref2 = StatusPositionSpeed.fromBuf(raw[16:16+8])
self.ref3 = StatusPositionSpeed.fromBuf(raw[24:24+8]) self.ref3 = StatusPositionSpeed.fromBuf(raw[24:24+8])
self.ref4 = StatusPositionSpeed.fromBuf(raw[32:32+8]) self.ref4 = StatusPositionSpeed.fromBuf(raw[32:32+8])
self.ref1cnt = int(raw[42:44] + raw[40:42], 16) self.ref1cnt = int(raw[42:44] + raw[40:42], 16)
self.ref2cnt = int(raw[46:48] + raw[44:46], 16) self.ref2cnt = int(raw[46:48] + raw[44:46], 16)
self.ref3cnt = int(raw[50:52] + raw[48:50], 16) self.ref3cnt = int(raw[50:52] + raw[48:50], 16)
self.ref4cnt = int(raw[54:56] + raw[52:54], 16) self.ref4cnt = int(raw[54:56] + raw[52:54], 16)
self.ref5 = StatusPositionSpeed.fromBuf(raw[56:56+8]) self.ref5 = StatusPositionSpeed.fromBuf(raw[56:56+8])
self.diagnostic = raw[64:64+16] self.diagnostic = raw[64:64+16]
self.undefined1 = raw[80:84] self.undefined1 = raw[80:84]
self.handset1 = int(raw[86:88] + raw[84:86], 16) self.handset1 = int(raw[86:88] + raw[84:86], 16)
self.handset2 = int(raw[88:90] + raw[86:88], 16) self.handset2 = int(raw[88:90] + raw[86:88], 16)
self.ref6 = StatusPositionSpeed.fromBuf(raw[90:90+8]) self.ref6 = StatusPositionSpeed.fromBuf(raw[90:90+8])
self.ref7 = StatusPositionSpeed.fromBuf(raw[98:98+8]) self.ref7 = StatusPositionSpeed.fromBuf(raw[98:98+8])
self.ref8 = StatusPositionSpeed.fromBuf(raw[106:106+8]) self.ref8 = StatusPositionSpeed.fromBuf(raw[106:106+8])
self.undefined2 = raw[114:] self.undefined2 = raw[114:]
return self return self
class LinakController(object): class LinakController(object):
_handle = None _handle = None
_ctx = None _ctx = None
def __init__(self, vendor_id=0x12d3, product_id=0x0002): def __init__(self, vendor_id=0x12d3, product_id=0x0002):
self._ctx =usb1.USBContext() self._ctx =usb1.USBContext()
#self._ctx.setDebug(4) #self._ctx.setDebug(4)
self._handle = self._ctx.openByVendorIDAndProductID( self._handle = self._ctx.openByVendorIDAndProductID(
vendor_id, vendor_id,
product_id, product_id,
skip_on_error=True, skip_on_error=True,
) )
if not self._handle: if not self._handle:
raise Exception('Could not connect to usb device') raise Exception('Could not connect to usb device')
self._handle.claimInterface(0) self._handle.claimInterface(0)
self._initDevice() self._initDevice()
def close(self): def close(self):
if self._handle: if self._handle:
self._handle.releaseInterface(0) self._handle.releaseInterface(0)
del(self._handle) del(self._handle)
del(self._ctx) del(self._ctx)
def _controlWriteRead(self, request_type, request, value, index, data, timeout=0): def _controlWriteRead(self, request_type, request, value, index, data, timeout=0):
data, data_buffer = usb1.create_initialised_buffer(data) data, data_buffer = usb1.create_initialised_buffer(data)
transferred = self._handle._controlTransfer(request_type, request, value, index, data, transferred = self._handle._controlTransfer(request_type, request, value, index, data,
sizeof(data), timeout) sizeof(data), timeout)
return transferred, data_buffer[:transferred] return transferred, data_buffer[:transferred]
def _getStatusReport(self): def _getStatusReport(self):
buf = bytearray(b'\x00'*LEN_STATUS_REPORT) buf = bytearray(b'\x00'*LEN_STATUS_REPORT)
buf[0] = CMD_STATUS_REPORT buf[0] = CMD_STATUS_REPORT
#print('> {:s}'.format(buf.hex())) #print('> {:s}'.format(buf.hex()))
x, buf = self._controlWriteRead( x, buf = self._controlWriteRead(
TYPE_GET_CI, TYPE_GET_CI,
HID_REPORT_GET, HID_REPORT_GET,
REQ_GET_STATUS, REQ_GET_STATUS,
0, 0,
buf, buf,
LINAK_TIMEOUT LINAK_TIMEOUT
) )
# check if the response match to request! # check if the response match to request!
if buf[0] != CMD_STATUS_REPORT: if buf[0] != CMD_STATUS_REPORT:
raise Exception('Invalid status report received!') raise Exception('Invalid status report received!')
return buf return buf
def _setStatusReport(self): def _setStatusReport(self):
buf = bytearray(b'\x00'*LEN_STATUS_REPORT) buf = bytearray(b'\x00'*LEN_STATUS_REPORT)
buf[0] = CMD_MODE_OF_OPERATION buf[0] = CMD_MODE_OF_OPERATION
buf[1] = DEF_MODE_OF_OPERATION buf[1] = DEF_MODE_OF_OPERATION
buf[2] = 0 buf[2] = 0
buf[3] = 251 buf[3] = 251
x, buf = self._controlWriteRead( x, buf = self._controlWriteRead(
TYPE_SET_CI, TYPE_SET_CI,
HID_REPORT_SET, HID_REPORT_SET,
REQ_INIT, REQ_INIT,
0, 0,
buf, buf,
LINAK_TIMEOUT LINAK_TIMEOUT
) )
if x != LEN_STATUS_REPORT: if x != LEN_STATUS_REPORT:
raise Exception('Device is not ready yet. Initialization failed in step 1.') raise Exception('Device is not ready yet. Initialization failed in step 1.')
def _move(self, height): def _move(self, height):
buf = bytearray(b'\x00' * LEN_STATUS_REPORT) buf = bytearray(b'\x00' * LEN_STATUS_REPORT)
buf[0] = CMD_CONTROL_CBC buf[0] = CMD_CONTROL_CBC
hHex = '{:04x}'.format(height) hHex = '{:04x}'.format(height)
hHigh = int(hHex[2:], 16) hHigh = int(hHex[2:], 16)
hLow = int(hHex[:2], 16) hLow = int(hHex[:2], 16)
buf[1] = hHigh buf[1] = hHigh
buf[2] = hLow buf[2] = hLow
buf[3] = hHigh buf[3] = hHigh
buf[4] = hLow buf[4] = hLow
buf[5] = hHigh buf[5] = hHigh
buf[6] = hLow buf[6] = hLow
buf[7] = hHigh buf[7] = hHigh
buf[8] = hLow buf[8] = hLow
x, buf = self._controlWriteRead( x, buf = self._controlWriteRead(
TYPE_SET_CI, TYPE_SET_CI,
HID_REPORT_SET, HID_REPORT_SET,
REQ_MOVE, REQ_MOVE,
0, 0,
buf, buf,
LINAK_TIMEOUT LINAK_TIMEOUT
) )
return x == LEN_STATUS_REPORT return x == LEN_STATUS_REPORT
def _moveDown(self): def _moveDown(self):
return self._move(HEIGHT_MOVE_DOWNWARDS) return self._move(HEIGHT_MOVE_DOWNWARDS)
def _moveUp(self): def _moveUp(self):
return self._move(HEIGHT_MOVE_UPWARDS) return self._move(HEIGHT_MOVE_UPWARDS)
def _moveEnd(self): def _moveEnd(self):
return self._move(HEIGHT_MOVE_END) return self._move(HEIGHT_MOVE_END)
def _isStatusReportNotReady(self, buf): def _isStatusReportNotReady(self, buf):
if buf[0] != CMD_STATUS_REPORT or buf[1] != NRB_STATUS_REPORT: if buf[0] != CMD_STATUS_REPORT or buf[1] != NRB_STATUS_REPORT:
return False return False
for i in range(2, LEN_STATUS_REPORT - 5): for i in range(2, LEN_STATUS_REPORT - 5):
if buf[i] != 0: if buf[i] != 0:
return False return False
return True return True
def _initDevice(self): def _initDevice(self):
buf = self._getStatusReport() buf = self._getStatusReport()
if not self._isStatusReportNotReady(buf): if not self._isStatusReportNotReady(buf):
return return
else: else:
print('Device not ready!') print('Device not ready!')
self._setStatusReport() self._setStatusReport()
time.sleep(1000/1000000.0) time.sleep(1000/1000000.0)
if not _moveEnd(): if not _moveEnd():
raise Exception('Device not ready - initialization failed on step 2 (moveEnd)') raise Exception('Device not ready - initialization failed on step 2 (moveEnd)')
time.sleep(100000/1000000.0) time.sleep(100000/1000000.0)
def move(self, target): def move(self, target):
a = max_a = 3 a = max_a = 3
epsilon = 13 epsilon = 13
oldH = 0 oldH = 0
while True: while True:
self._move(target) self._move(target)
time.sleep(200000/1000000.0) time.sleep(200000/1000000.0)
buf = self._getStatusReport() buf = self._getStatusReport()
r = StatusReport.fromBuf(buf) r = StatusReport.fromBuf(buf)
distance = r.ref1cnt - r.ref1.pos distance = r.ref1cnt - r.ref1.pos
delta = oldH-r.ref1.pos delta = oldH-r.ref1.pos
if abs(distance) <= epsilon or abs(delta) <= epsilon or oldH == r.ref1.pos: if abs(distance) <= epsilon or abs(delta) <= epsilon or oldH == r.ref1.pos:
a -= 1 a -= 1
else: else:
a = max_a a = max_a
print( print(
'Current height: {:d}; target height: {:d}; distance: {:d}'.format( 'Current height: {:d}; target height: {:d}; distance: {:d}'.format(
r.ref1.pos, r.ref1.pos,
target, target,
distance distance
) )
) )
if a == 0: if a == 0:
break break
oldH = r.ref1.pos oldH = r.ref1.pos
return abs(r.ref1.pos - target) <= epsilon return abs(r.ref1.pos - target) <= epsilon
def getHeight(self): def getHeight(self):
buf = self._getStatusReport() buf = self._getStatusReport()
r = StatusReport.fromBuf(buf) r = StatusReport.fromBuf(buf)
return r.ref1.pos, r.ref1.pos/98.0 return r.ref1.pos, r.ref1.pos/98.0
if __name__ == '__main__': if __name__ == '__main__':
import argparse import argparse
parser = argparse.ArgumentParser(description='Get the control on your desk!') parser = argparse.ArgumentParser(description='Get the control on your desk!')
parser.add_argument('command', choices=['move', 'height'], help='Command to execute.') parser.add_argument('command', choices=['move', 'height'], help='Command to execute.')
parser.add_argument('height', type=int, nargs='?', help='For command "move", give the destination height.') parser.add_argument('height', type=int, nargs='?', help='For command "move", give the destination height.')
args = parser.parse_args() args = parser.parse_args()
if args.command == 'move' and not args.height: if args.command == 'move' and not args.height:
sys.stderr.write('Height missing in case of move!\n') sys.stderr.write('Height missing in case of move!\n')
parser.print_help() parser.print_help()
sys.exit(1) sys.exit(1)
co = LinakController() co = LinakController()
try: try:
if args.command == 'move': if args.command == 'move':
r = co.move(args.height) r = co.move(args.height)
if r: if r:
print('Command executed successfuly') print('Command executed successfuly')
else: else:
print('Command failed') print('Command failed')
elif args.command == 'height': elif args.command == 'height':
h, hcm = co.getHeight() h, hcm = co.getHeight()
print('Current height is: {:d} / {:.2f} cm'.format(h, hcm)) print('Current height is: {:d} / {:.2f} cm'.format(h, hcm))
except Exception as e: except Exception as e:
co.close() co.close()
raise e raise e
finally: finally:
co.close() co.close()