zbdev

#!/usr/bin/env python3
# zbdev pretends to be a Zigbee device
# the local interface should be running the NIC.py firmware so that
# it is hexdumping the raw packets that are being received

import os
import sys
import serial
import threading
import readline
import time
from select import select
from binascii import unhexlify, hexlify
from struct import pack, unpack
from ZbPy import Device
from ZbPy import ZigbeeNetwork
from ZbPy import ZigbeeApplication

# re-open stdout as binary
#stdout = os.fdopen(sys.stdout.fileno(), "wb")
#stderr = os.fdopen(sys.stderr.fileno(), "wb")
from sys import stdout, stderr

device = "/dev/ttyACM0"
speed = 115200
do_reset = False
serial_dev = serial.Serial(device, speed, timeout=0.1)

def process_line(line):	
	try:
		data = bytearray(unhexlify(line))
		return data
	except Exception as e:
		#print(str(e) + ": " + str(line))
		print("Read '" + str(line) + "'")
		return None

line = b''
def process_serial():
	global line
	c = serial_dev.read(1)
	if c is None:
		return
	if c == b'\r':
		return

	#print(str(c,"UTF-8"), end='')
	#stderr.write(c)

#	if (b'0' <= c and c <= b'9') \
#	or (b'a' <= c and c <= b'f'):
	if b'\n' != c:
		line += c
		return

	if c == b'\n' and line != b'':
		pkt = process_line(line)
		line = b''
		return pkt

	# either a line was processed, or a non-hex char
	# was received, in either case restart the line
	line = b''
	return

def wait_for(s):
	l = ''
	while True:
		c = serial_dev.read(1)
		if c is None:
			continue

		c = str(c, "UTF-8")
		if '\r' == c:
			continue

		if '\n' == c:
			print("Read '", l, "'")
			l = ''
			continue

		l += c

		if l == s:
			print(l)
			return

# send the commands to reboot and load the NIC firmware
serial_dev.write(b"\x03\x03\x03")
serial_dev.flushOutput()
time.sleep(0.2)
wait_for(">>> ")
if do_reset:
	serial_dev.write(b"reset()\r") # why \r?
	serial_dev.flushOutput()
	wait_for("MicroPython v")
	wait_for(">>> ")
serial_dev.write(b"import ZbPy.NIC\r")
serial_dev.flushOutput()
wait_for(">>> ")
serial_dev.write(b"ZbPy.NIC.loop()\r")
serial_dev.flushOutput()

#stderr.write(b"reset\n")


last_status = 0

addr = [
	b'\x8co\xdb\xfe\xff\xd7k\x08', # mac
	None,		# nwk
	None,		# pan
]

# Mock the Radio with the serial NIC
class Radio:
	def __init__(self, serial_dev):
		self.serial_dev = serial_dev
	def tx(self, b):
		s = hexlify(b)
		print("TX: ", s)
		s += b"\n"
		#self.serial_dev.write(s)
		#self.serial_dev.flushOutput()
		#return

		# slow down the transmission a bit
		chunk = 32
		for i in range(0,len(s),chunk):
			p = s[i:i+chunk]
			print(p, end=',')
			self.serial_dev.write(p)
			self.serial_dev.flushOutput()

			if i + chunk < len(s):
				time.sleep(0.05)
	def rx(self):
		return process_serial()
	def mac(self):
		return addr[0]
	def address(self,new_nwk=None):
		if new_nwk is not None:
			addr[1] = new_nwk
		return addr[1]
	def pan(self,new_pan=None):
		if new_pan is not None:
			addr[2] = new_pan
		return addr[2]
			

zbdev = Device.IEEEDevice(
	radio = Radio(serial_dev),
)

nwkdev = Device.NetworkDevice(
	dev = zbdev,
	seq = int(time.time())
)


#
# Handle Zigbee Device profile cluster messages
# These are important for joining the network, etc.
#
aps_counter = 0

def zcl_device(nwkdev, pkt):
	global aps_counter

	aps = pkt.payload
	if aps.frame_type != ZigbeeApplication.FRAME_TYPE_DATA:
		return

	if aps.cluster == ZigbeeApplication.CLUSTER_DEVICE_NODE_DESCRIPTOR_RESPONSE:
		print("Node Descriptor:", pkt)
	elif aps.cluster == ZigbeeApplication.CLUSTER_DEVICE_NODE_DESCRIPTOR_REQUEST:
		print("--- Node Descriptor Request ---")
		aps_counter = (aps_counter + 1) & 0x7F
		nwkdev.tx(dst=pkt.src, ack_req=True, payload=ZigbeeApplication.ZigbeeApplication(
			profile = ZigbeeApplication.PROFILE_DEVICE,
			cluster = ZigbeeApplication.CLUSTER_DEVICE_NODE_DESCRIPTOR_RESPONSE,
			seq = aps_counter,
			payload = pack("<BBHHBHBHHHB",
				0x12, # seq
				0x00, # success
				addr[1], # our nwk
				0x4002, # type == end device, simple descriptor, 2.4 GHz OQPSK band
				0x80, # capabilities
				0x117c, # manufacturer code ikea
				0x52, # max buffer size
				0x0052, # max inomcing transfer size
				0x2c00, # server flags/stack compliance
				0x0052, # max outgoing transfer size
				0x00, # capability field (no extended data)
			),
		))
	elif aps.cluster == ZigbeeApplication.CLUSTER_DEVICE_ENDPOINT_REQUEST:
		print("--- Endpoint Request ---")
		aps_counter = (aps_counter + 1) & 0x7F
		nwkdev.tx(dst=pkt.src, ack_req=True, payload=ZigbeeApplication.ZigbeeApplication(
			profile = ZigbeeApplication.PROFILE_DEVICE,
			cluster = ZigbeeApplication.CLUSTER_DEVICE_ENDPOINT_RESPONSE,
			seq = aps_counter,
			payload = pack("<BBHBB",
				0x13, # seq
				0x00, # success
				addr[1], # our nwk
				1, # ony one endpoint
				1, # and that endpoint is #1
			),
		))
	elif aps.cluster == ZigbeeApplication.CLUSTER_DEVICE_ENDPOINT_DESCR_REQUEST:
		print("--- Endpoint Descriptor Request ---")
		aps_counter = (aps_counter + 1) & 0x7F
		nwkdev.tx(dst=pkt.src, ack_req=True, payload=ZigbeeApplication.ZigbeeApplication(
			profile = ZigbeeApplication.PROFILE_DEVICE,
			cluster = ZigbeeApplication.CLUSTER_DEVICE_ENDPOINT_DESCR_RESPONSE,
			seq = aps_counter,
			payload = pack("<BBH"+"BBHHB"+"BHHHHHHHHH"+"BHHHH",
				0x14, # seq
				0x00, # success
				addr[1], # our nwk
				# size and descriptor header
				1 + 3*2 + 1 + 9*2 + 1 + 6*2, # size
				1, # endpoint
				0xc05e, # zll?  0x0104, # profile home automation
				0x0100, # dimable lamp
				0x02, # application version
				# input clusters
				9, # input cluster count
0,
3,
4,
5,
6,
8,
768,
2821,
4096,

				#0x0000, # basic cluster
				#0x0001, # power configuration
				#0x0003, # identity cluster
				#0x0004, # ? cluster
				#0x0005, # ? cluster
				#0x0006, # on/off cluster
				#0x0008, # level control cluster
				#0x0300, # color control
				# output clusers
				4, # output cluster count
5, 25, 32, 4096,
				#0x0003, # identity cluster
				#0x0004, # ? cluster
				#0x0005, # ? cluster
				#0x0006, # on/off cluster
				#0x0008, # level control cluster
				#0x0300, # color control
			),
		))


manufacturer = b"IKEA of Sweden"
model = b"FLOALT panel WS 30x30"
datecode = b"20200830"
build = b"0.0.001"

attributes = {
0x0000: pack("<HBBB", 0x0000, 0, 0x20, 0x03), # ZCL version
0x0001: pack("<HBBB", 0x0001, 0, 0x20, 0x33), # application version
0x0002: pack("<HBBB", 0x0002, 0, 0x20, 0x62), # stack version
0x0003: pack("<HBBB", 0x0003, 0, 0x20, 0x01), # HW version
0x0004: pack("<HBBB", 0x0004, 0, 0x42, len(manufacturer)) + manufacturer,
0x0005: pack("<HBBB", 0x0005, 0, 0x42, len(model)) + model,
0x0006: pack("<HBBB", 0x0006, 0, 0x42, len(datecode)) + datecode,
#0x0007: pack("<HBBB", 0x0007, 0, 0x30, 0x03), # power source == battery
0x0007: pack("<HBBB", 0x0007, 0, 0x30, 0x01), # power source == mains
0x4000: pack("<HBBB", 0x4000, 0, 0x42, len(build)) + build,
}

#
# Home automation commands
#
def zcl_homeautomation(nwkdev, pkt):
	global aps_counter

	aps = pkt.payload
	if aps.frame_type == ZigbeeApplication.FRAME_TYPE_DATA:
		print("--- Read home automation data ---")
		aps_counter = (aps_counter + 1) & 0x7F
		fcf = aps.payload[0]
		seq = aps.payload[1]
		cmd = aps.payload[2]

		reply = b''
		for off in range(3,len(aps.payload),2):
			(key,) = unpack("<H", aps.payload[off:off+2])
			print("attr %d:" % (off), key)
			if key in attributes:
				reply += attributes[key]
			else:
				reply += pack("<HB", key, 0x01)

		nwkdev.tx(dst=pkt.src, ack_req=True, payload=ZigbeeApplication.ZigbeeApplication(
			profile = aps.profile,
			cluster = aps.cluster,
			ack_req = True,
			seq = aps_counter,
			src = aps.dst, # swap the src/dst for the reply
			dst = aps.src,
			payload = pack("<BBB",
				0x18, # profile-wide
				aps_counter,
				1, # command Read Attribute Response
			) + reply
		))

def aps_handler(nwkdev, pkt):
	if not pkt.valid:
		print("ENCRYPTION FAILED:", pkt)
		return

	aps = ZigbeeApplication.ZigbeeApplication(data=pkt.payload)
	if aps.security:
		print("APS security not yet supported")
		return

	pkt.payload = aps

	if aps.profile == ZigbeeApplication.PROFILE_DEVICE:
		zcl_device(nwkdev, pkt)
	elif aps.profile == ZigbeeApplication.PROFILE_HOME:
		zcl_homeautomation(nwkdev, pkt)
	else:
		print("UNKNOWN PROFILE %04x" % (aps.profile), pkt)

#nwkdev.verbose = 1
nwkdev.handler = aps_handler


boot_time = time.time()
last_beacon = 0
last_join = 0
last_req = 0
announcement_sent = False
query_sent = False

while True:
	zbdev.tick()

	if zbdev.tx_fail != 0:
		print("\n\n\n!!!!! TX FAIL")
		zbdev.tx_fail = 0

	now = time.time()
	if now - last_status > 1.0:
		last_status = now
		print(now)

	if addr[2] is None:
		if now - last_beacon > 0.5 \
		and zbdev.pending_seq is None:
			# no PAN set: send a beacon request
			print("---- BEACON REQUEST ----")
			zbdev.beacon()
			last_beacon = now

	elif addr[1] is None:
		if now - last_join > 1.5 \
		and zbdev.pending_seq is None:
			# PAN is set, but not NWK: send a join request
			print("\n\n\n---- JOIN REQUEST ----")
			zbdev.join()
			last_join = now

	elif not announcement_sent \
	and now - last_join > 3 \
	and zbdev.pending_seq is None:
		print("--- Device announcement ---")
		announcement_sent = True
		nwkdev.tx(dst=ZigbeeNetwork.DEST_BROADCAST, ack_req=False, payload=ZigbeeApplication.ZigbeeApplication(
			profile = ZigbeeApplication.PROFILE_DEVICE,
			cluster = ZigbeeApplication.CLUSTER_DEVICE_ANNOUNCEMENT,
			payload = bytes([
				0x12,
				(addr[1] >> 0) & 0xFF,
				(addr[1] >> 8) & 0xFF,
				addr[0][0],
				addr[0][1],
				addr[0][2],
				addr[0][3],
				addr[0][4],
				addr[0][5],
				addr[0][6],
				addr[0][7],
				0x80,
			]),
		))

	elif not query_sent \
	and False \
	and now - last_join > 4 \
	and zbdev.pending_seq is None:
		print("--- Coordinator query ---")
		query_sent = True
		coord_nwk = 0x0000
		nwkdev.tx(dst=coord_nwk, ack_req=True, payload=ZigbeeApplication.ZigbeeApplication(
			profile = ZigbeeApplication.PROFILE_DEVICE,
			cluster = ZigbeeApplication.CLUSTER_DEVICE_NODE_DESCRIPTOR_REQUEST,
			payload = bytes([
				0x01, # seq
				(coord_nwk >> 0) & 0xFF,
				(coord_nwk >> 8) & 0xFF,
			]),
		))

	elif now - last_req > 0.8 \
	and zbdev.pending_seq is None:
		# NWK is now set; send data requests to keep the party going
		#print("--- Data request ---")
		zbdev.data_request()
		last_req = now