).I--- 55 ---I I 48 112 I I I I 16 80 I I--- 21 ---I
. Ich mach erstmal die Taster fertig und versuche alles was man brauchen könnte schonmal in Variablen zu packen.
[eno]
[A0]
type = bool
enforce_update = true
eno_id = 00:11:22:33:44
eno_rorg = F6_02
eno_button = A0
[A1]
type = bool
enforce_update = true
eno_id = 00:11:22:33:44
eno_rorg = F6_02
eno_button = A1
[B0]
type = bool
enforce_update = true
eno_id = 00:11:22:33:44
eno_rorg = F6_02
eno_button = B1
[B0]
type = bool
enforce_update = true
eno_id = 00:11:22:33:44
eno_rorg = F6_02
eno_button = B0
x=21 a = bitstring.BitArray(uint=x,length=8) print(a[:3])
print((0x15 & 0xE0) >> 5)
.
0x00 --- Reserved 0x01 RADIO Radio telegram 0x02 RESPONSE Response to any packet 0x03 RADIO_SUB_TEL Radio subtelegram 0x04 EVENT Event message 0x05 COMMON_COMMAND Common command 0x06 SMART_ACK_COMMAND Smart Ack command 0x07 REMOTE_MAN_COMMAND Remote management command 0x08 --- Reserved for EnOcean 0x09 RADIO_MESSAGE Radio message 0x0A RADIO_ADVANCED Advanced protocol radio telegram
0XF6 RPS 0XD5 1BS 0xA5 4BS 0XD2 VLD u.s.w.
[A1] type = bool enforce_updates = true enocean_id = 00:22:60:37 enocean_rorg = F6_02_02 enocean_rorg_id = A1 #Button-Name in Plugin [A0] type = bool enforce_updates = true enocean_id = 00:22:60:37 enocean_rorg = F6_02_02 enocean_rorg_id = A0 #Button-Name in Plugin
#!/usr/bin/python
import serial
import os
import sys
import logging
import struct
import time
#CO_RD_VERSION 3
FCSTAB = [
0x00, 0x07, 0x0e, 0x09, 0x1c, 0x1b, 0x12, 0x15,
0x38, 0x3f, 0x36, 0x31, 0x24, 0x23, 0x2a, 0x2d,
0x70, 0x77, 0x7e, 0x79, 0x6c, 0x6b, 0x62, 0x65,
0x48, 0x4f, 0x46, 0x41, 0x54, 0x53, 0x5a, 0x5d,
0xe0, 0xe7, 0xee, 0xe9, 0xfc, 0xfb, 0xf2, 0xf5,
0xd8, 0xdf, 0xd6, 0xd1, 0xc4, 0xc3, 0xca, 0xcd,
0x90, 0x97, 0x9e, 0x99, 0x8c, 0x8b, 0x82, 0x85,
0xa8, 0xaf, 0xa6, 0xa1, 0xb4, 0xb3, 0xba, 0xbd,
0xc7, 0xc0, 0xc9, 0xce, 0xdb, 0xdc, 0xd5, 0xd2,
0xff, 0xf8, 0xf1, 0xf6, 0xe3, 0xe4, 0xed, 0xea,
0xb7, 0xb0, 0xb9, 0xbe, 0xab, 0xac, 0xa5, 0xa2,
0x8f, 0x88, 0x81, 0x86, 0x93, 0x94, 0x9d, 0x9a,
0x27, 0x20, 0x29, 0x2e, 0x3b, 0x3c, 0x35, 0x32,
0x1f, 0x18, 0x11, 0x16, 0x03, 0x04, 0x0d, 0x0a,
0x57, 0x50, 0x59, 0x5e, 0x4b, 0x4c, 0x45, 0x42,
0x6f, 0x68, 0x61, 0x66, 0x73, 0x74, 0x7d, 0x7a,
0x89, 0x8e, 0x87, 0x80, 0x95, 0x92, 0x9b, 0x9c,
0xb1, 0xb6, 0xbf, 0xb8, 0xad, 0xaa, 0xa3, 0xa4,
0xf9, 0xfe, 0xf7, 0xf0, 0xe5, 0xe2, 0xeb, 0xec,
0xc1, 0xc6, 0xcf, 0xc8, 0xdd, 0xda, 0xd3, 0xd4,
0x69, 0x6e, 0x67, 0x60, 0x75, 0x72, 0x7b, 0x7c,
0x51, 0x56, 0x5f, 0x58, 0x4d, 0x4a, 0x43, 0x44,
0x19, 0x1e, 0x17, 0x10, 0x05, 0x02, 0x0b, 0x0c,
0x21, 0x26, 0x2f, 0x28, 0x3d, 0x3a, 0x33, 0x34,
0x4e, 0x49, 0x40, 0x47, 0x52, 0x55, 0x5c, 0x5b,
0x76, 0x71, 0x78, 0x7f, 0x6A, 0x6d, 0x64, 0x63,
0x3e, 0x39, 0x30, 0x37, 0x22, 0x25, 0x2c, 0x2b,
0x06, 0x01, 0x08, 0x0f, 0x1a, 0x1d, 0x14, 0x13,
0xae, 0xa9, 0xa0, 0xa7, 0xb2, 0xb5, 0xbc, 0xbb,
0x96, 0x91, 0x98, 0x9f, 0x8a, 0x8D, 0x84, 0x83,
0xde, 0xd9, 0xd0, 0xd7, 0xc2, 0xc5, 0xcc, 0xcb,
0xe6, 0xe1, 0xe8, 0xef, 0xfa, 0xfd, 0xf4, 0xf3
]
DEVICE = dict(
switch = 246, # 0xf6 RPS, org 05
contact = 213, # 0xd5 1BS, org 06
sensor = 165, # 0xa5 4BS, org 07
msc = 209, # 0xd1 MSC
vld = 210, # 0xd2 VLD
ute = 212 # 0xd4 UTE
)
VALID_ID = dict()
logger = logging.getLogger('EnOcean')
class EnOceanType():
def switch():
pass
class EnOcean():
_items = []
def __init__(self, smarthome, serialport):
self._sh = smarthome
self.port = serialport
self._tcm = serial.Serial(serialport, 57600, timeout=0.5)
def run(self):
self.alive = True
global from_shell
from_shell = False
self.connect()
msg = []
while True:
readin = self._tcm.read(1000)
if readin:
msg += readin
# check if header is complete (6bytes including sync)
# 0x55 (SYNC) + 4bytes (HEADER) + 1byte(CRC)
while (len(msg) >= 5):
#check header for CRC
if (msg[0] == 0x55) and (self._calc_crc8(msg[1:5]) == msg[5]):
# header bytes: length of data; length of data; optional length; packet type
### HEADER RECEIVED ###
print ('header: {}'.format(msg[:6]))
### HEADER_DATA_LENTGH ###
data_length = (msg[1] << 8) + msg[2] #data
print ("data_length:{}".format(data_length))
### HEADER_OPT_DATA_LENGTH ###
opt_length = msg[3] #opt
print ("opt_length:{}".format(opt_length))
### HEADER_PACKET_TYPE ###
packet_type = msg[4]
print ("packet_type:{}".format(packet_type))
### CALCULATED MESSAGE_LENTGH FROM HEADER ###
msg_length = data_length + opt_length + 7
print ("msg_length:{}".format(msg_length))
# check if data is complete (and ok):
if (len(msg) < msg_length):
break
if (self._calc_crc8(msg[6:msg_length-1]) == msg[msg_length-1]):
self._complete = True
### RECEIVED ###
print ('RX: {}'.format(msg[0:msg_length])) #print data
# check complete check if packet_type = RADIO (0x01):
if (self._complete) and (packet_type == 0x01):
### DATA ###
data=msg[6:msg_length-(opt_length+1)]
print ("data:{}".format(data))
### RORG ###
rorg = data[0]
print ("device:{}".format(rorg))
### PAYLOAD ###
payload = data[1:-5] #offset 5 bytes to ignore at end - ignore rorg too
print ("payload:{}".format(payload))
### SENDER_ID ###
#print (self.build_mac(data[2:6]))
sender_id = self.build_mac(data[2:6])
print ("SenderID: {}".format(sender_id))
### OPT_DATA ###
opt=msg[(6+data_length):msg_length-1]
if len(opt) > 0:
print ("opt:{}".format(opt))
### SUB_TEL_NUMBER ###
print ("SubTelNumber: {}".format(opt[0]))
### DESTINATION ###
print ("Destination: {}".format(self.build_mac(opt[1:5])))
### LINK_QUALITY ###
print ("dBm: -{}".format(opt[5]))
### SECURITY_LEVEL ###
print ("SecurityLevel: {}".format(opt[6]))
### CRC ###
print ("crc:{}".format(msg[msg_length-1]))
print ("VALIDATE")
self.validate(rorg, sender_id, payload, opt)
msg = msg[msg_length:]
else:
msg.pop(0)
# self.radio([0xA5, 0x11, 0x22, 0x33, 0x44, 0x00, 0x00, 0x00, 0x00, 0x30]) # 4BS + 9bytes Data
# self.radio([0xF6, 0x66, 0x00, 0x00, 0x00, 0x00, 0x20]) # RPS
def validate(self, rorg, sender_id, payload, opt):
#think about this example:
#x = {'F6_02':[('B0',<nimm Bit 2 als Funktion aber nur wenn nicht x>), ('A1',<nimm Bit 3 wenn...>)], 'F6_xx ...........}
valid = False
for item in self._items:
# validate id for item id:
if item.conf['enocean_id'] == sender_id:
print ("validated {0} for {1}".format(sender_id,item))
valid = True
if valid:
#### 0xF6 SPECIFIC RPS ####
if rorg == 0xF6: #0xf6
print ("F6-02-02 Command-Set RPS")
spec_payload = payload[0]
print ("payload: {}".format(spec_payload))
#a = bitstring.BitArray(uint=payload,length=8)
x = spec_payload
#print(a[:3]) # last 3 bits
button_1 = (ord(chr(x)) & 0xE0) >> 5
print ("First Button-ID: {}".format((ord(chr(x)) & 0xE0) >> 5)) #sixth to eigth
#print(a[3:4]) # 4th bit
pressed = (ord(chr(x)) & 0x10) >> 4
print ("Pressed: {}".format((ord(chr(x)) & 0x10) >> 4)) #fifth
#print(a[-4:-1]) # second to fourth bit
button_2 = (ord(chr(x)) & 0x0E) >> 1
print ("Second Button-ID: {}".format((ord(chr(x)) & 0x0E) >> 1)) #second to fourth
#print(a[:-1]) # first bit
valid_2 = (ord(chr(x)) & 0x01)
print ("Second Button is valid: {}".format((ord(chr(x)) & 0x01))) #first
button_codes=['A1','A0','B1','B0']
if pressed == 1:
if valid_2 == 1:
button_result = button_codes[button_1]+button_codes[button_2]
else:
button_result = button_codes[button_1]
print ("Resulting Button: {}".format(button_result))
#### END OF 0xF6 SPECIFIC ####
#### 0xD5 SPECIFIC 1BS ####
if rorg == 0xD5: #0xD5
print ("D5 Command-Set 1BS")
#### 0xA5 SPECIFIC 4BS ####
if rorg == 0xA5: #0x00
print ("A5 Command-Set 4BS")
#### 0x00 SPECIFIC Extended Telegram ####
if rorg == 0x00: #0x00
print ("00 Identification-Set")
def parse_item(self, item):
if 'enocean_id' in item.conf:
enocean_id = item.conf['enocean_id']
if 'enocean_rorg' in item.conf:
enocean_rorg = item.conf['enocean_rorg']
self._items.append(item)
#return self.update_item
def update_item(self, item, caller=None, source=None, dest=None):
if caller != 'EnOcean':
if item():
pass
#if 'roomba_cmd' in item.conf:
# cmd_string = item.conf['roomba_cmd']
# logger.debug("Roomba: item = true")
# self.drive(cmd_string)
def build_mac(self,ints):
hex_string = ''
for i in ints:
hex_string += "%0.2X" % i
hex_string = ':'.join(hex_string[i:i+2] for i in range(0,8,2))
return hex_string
def stop(self):
self.alive = False
def connect(self):
logger.debug("EnOcean: connect")
# request version information / firmware
self._send([0x55, 0x00, 0x01, 0x00, 0x05, 0x00, 0x03, 0x00])
#0x03 = CO_RD_VERSION = Read the device (SW) version / (HW) version, chip ID etc.
def radio(self, msg):
self._send([0x55, (len(msg) >> 8), (len(msg) & 0xFF), 0x07, 0x01, 0x00] + msg + [0x03, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0x00, 0x00])
# SYNC,DataLength,DataLength,OptLength,PacketType,CRC + DATA + SubTelNum,DestinationID,DestinationID,DestinationID,DestinationID,dBm,SecurityLevel
# 0xFF 0xFF 0xFF 0xFF ---> Broadcast
def _calc_crc8(self, msg, crc=0):
for i in msg:
crc = FCSTAB[crc ^ i]
return crc
def _send(self, msg):
msg[5] = self._calc_crc8(msg[1:5])
msg[-1] = self._calc_crc8(msg[6:-1])
send = msg
print ('TX: {}'.format(msg))
self._tcm.write(bytearray(msg))
def main():
try:
if sys.argv[1]:
serialport = sys.argv[1]
print ("TCM310 on {0}".format(serialport))
smarthome = ''
global from_shell
from_shell = True
tcm = EnOcean(smarthome,serialport)
tcm.run()
except Exception as e:
print ("[main]: {0}".format(e))
print ("usage: __init__.py <serial_port>")
return 1
if __name__ == "__main__":
sys.exit(main())
[enocean]
class_name = EnOcean
class_path = plugins.enocean
serialport = /dev/ttyUSB0
print ("F6-02-02 Command-Set RPS")
spec_payload = payload[0]
print ("payload: {}".format(spec_payload))
x = spec_payload
button_1 = (ord(chr(x)) & 0xE0) >> 5
print ("First Button-ID: {}".format((ord(chr(x)) & 0xE0) >> 5)) #sixth to eigth
pressed = (ord(chr(x)) & 0x10) >> 4
print ("Pressed: {}".format((ord(chr(x)) & 0x10) >> 4)) #fifth
button_2 = (ord(chr(x)) & 0x0E) >> 1
print ("Second Button-ID: {}".format((ord(chr(x)) & 0x0E) >> 1)) #second to fourth
valid_2 = (ord(chr(x)) & 0x01)
print ("Second Button is valid: {}".format((ord(chr(x)) & 0x01))) #first
button_codes=['A1','A0','B1','B0']
if pressed == 1:
if valid_2 == 1:
button_result = button_codes[button_1]+button_codes[button_2]
else:
button_result = button_codes[button_1]
print ("Resulting Button: {}".format(button_result)) #possible reults: A1,A0,B1,B0,A1B1,A0B0
my_items = {'mac_as_str_or_bytes':[item-ref],
'2nd mac':[item2-ref, item3-ref]}
my_dict = {'F6_02_02':{'A1':'pl & (1<<4) == (1<<4)','A0':'pl & (1<<5) == (1<<5)'},
'F6_02_03':{'foo':'foo_transform'}}
# in my_dict können natürlich auch komplexere Operation zum finden des Wertes gemacht werden! auch ein 'foo if bla else None' ginge...
# example frame received - yes, mac is too short i know
frame = bytes([0xDE, 0xAD, 0xBE, 0xEF, 0xF6, (1<<5)])
#iterate over all items having this MAC
if not frame[1:4] in my_items:
print('unknown MAC')
return
for item in my_items[frame[1:4]]:
# match rorg with item:
if not item.conf['enocean_rorg'][3] == frame[5]:
print('wrong RORG')
continue
if (item.conf['enocean_rorg'] in my_dict) and (item.conf['enocean_id'] in my_dict[item.conf['enocean_rorg']]):
pl = frame[4:]
# for id = 'A0' this will be 'True' but for 'A1' this will be evaluated as 'False'
value = eval(my_dict[item.conf['enocean_rorg']][item.conf['enocean_id']])
# allow "None" to enable "eval" to decide not to update item!
if value is not None:
item(value,'cooles Plugin')
Kommentar