h3n3/nspanel-lovelace-ui
1
0
Fork 0
mirror of https://github.com/joBr99/nspanel-lovelace-ui.git synced 2025-12-19 14:14:12 +01:00
Code Issues Packages Projects Releases Wiki Activity

swap default tasmota driver

Browse Source
This commit is contained in:
joBr99
2022-03-20 17:18:15 +01:00
parent 6dd1205f8d
commit 9aaed49a2f
6 changed files with 737 additions and 704 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
README.md
View File

@@ -192,8 +192,6 @@ See Tasmota [MQTT Documentation](https://tasmota.github.io/docs/MQTT/) for more
Upload the nspanel.tft from the lastest release to a Webserver (for example www folder of Home Assistant) and execute the following command in Tasmota Console. (Development Version: [tft file from HMI folder](HMI/nspanel.tft))
**Webserver needs to support HTTP Range Header Requests, python2/3 http server doesn't work**
**Webserver must be HTTP, HTTPS is not supported, due to limitations of berry lang on tasmota**
`FlashNextion http://ip-address-of-your-homeassistant:8123/local/nspanel.tft`
@@ -346,8 +344,7 @@ Since release 1.1 you can update the berry driver directly from the Tasmota Cons
### Flashing of the Display Firmware with FlashNextion doesn't work
1. Make sure to use the [tasmota32-nspanel.bin](https://github.com/tasmota/install/raw/main/firmware/unofficial/tasmota32-nspanel.bin) Tasmota build.
2. Make sure to use an WebServer which supports http range requests like HomeAssistant, apache2 or nginx for exmaple.
3. Make sure to use HTTP and **not HTTPS**
2. Make sure to use HTTP and **not HTTPS**
### My flashing doesn't start at all

49
apps/nspanel-lovelace-ui/nspanel-lovelace-ui.py
View File

@@ -10,7 +10,6 @@ babel_spec = importlib.util.find_spec("babel")
if babel_spec is not None:
import babel.dates
class NsPanelLovelaceUIManager(hass.Hass):
def initialize(self):
@@ -77,7 +76,7 @@ class NsPanelLovelaceUI:
# Parse Json Message from Tasmota and strip out message from nextion display
data = json.loads(data["payload"])
if("CustomRecv" not in data):
self.api.log("Received Message from Tasmota: %s", data, level="DEBUG")
self.api.log("Received Message from Tasmota, but not from nextion screen: %s", data, level="DEBUG")
return
msg = data["CustomRecv"]
self.api.log("Received Message from Tasmota: %s", msg) #, level="DEBUG"
@@ -86,17 +85,11 @@ class NsPanelLovelaceUI:
msg = msg.split(",")
# run action based on received command
# TODO: replace with match case after appdeamon container swiched to python 3.10 - https://pakstech.com/blog/python-switch-case/ - https://www.python.org/dev/peps/pep-0636/
if msg[0] == "event":
if msg[1] == "startup":
self.api.log("Handling startup event", level="DEBUG")
# grab version from screen
current_panel_version = int(msg[2])
self.api.log("Nextion Display reports version: %s; Checking for updates ...", current_panel_version)
# send date and time
self.update_time("")
self.update_date("")
@@ -205,60 +198,64 @@ class NsPanelLovelaceUI:
def handle_button_press(self, entity_id, btype, optVal=None):
if(btype == "OnOff"):
if(optVal == "1"):
if entity_id == "updateNoYes" and optVal == "yes":
self.api.log("Sending update command")
self.mqtt.mqtt_publish(self.config["panelSendTopic"].replace("CustomSend", "FlashNextion"), release_config_desired_display_firmware_url)
if btype == "OnOff":
if optVal == "1":
self.api.turn_on(entity_id)
else:
self.api.turn_off(entity_id)
if(btype == "up"):
if btype == "up":
self.api.get_entity(entity_id).call_service("open_cover")
if(btype == "stop"):
if btype == "stop":
self.api.get_entity(entity_id).call_service("stop_cover")
if(btype == "down"):
if btype == "down":
self.api.get_entity(entity_id).call_service("close_cover")
if(btype == "button"):
if(entity_id.startswith('scene')):
if btype == "button":
if entity_id.startswith('scene'):
self.api.get_entity(entity_id).call_service("turn_on")
if(entity_id.startswith('light') or entity_id.startswith('switch')):
if entity_id.startswith('light') or entity_id.startswith('switch'):
self.api.get_entity(entity_id).call_service("toggle")
else:
self.api.get_entity(entity_id).call_service("press")
if(btype == "media-next"):
if btype == "media-next":
self.api.get_entity(entity_id).call_service("media_next_track")
if(btype == "media-back"):
if btype == "media-back":
self.api.get_entity(entity_id).call_service("media_previous_track")
if(btype == "media-pause"):
if btype == "media-pause":
self.api.get_entity(entity_id).call_service("media_play_pause")
if(btype == "hvac_action"):
if btype == "hvac_action":
self.api.get_entity(entity_id).call_service("set_hvac_mode", hvac_mode=optVal)
if(btype == "brightnessSlider"):
if btype == "brightnessSlider":
# scale 0-100 to ha brightness range
brightness = int(scale(int(optVal),(0,100),(0,255)))
self.api.get_entity(entity_id).call_service("turn_on", brightness=brightness)
if(btype == "colorTempSlider"):
if btype == "colorTempSlider":
entity = self.api.get_entity(entity_id)
#scale 0-100 from slider to color range of lamp
color_val = scale(int(optVal), (0, 100), (entity.attributes.min_mireds, entity.attributes.max_mireds))
self.api.get_entity(entity_id).call_service("turn_on", color_temp=color_val)
if(btype == "colorWheel"):
if btype == "colorWheel":
self.api.log(optVal)
optVal = optVal.split('|')
color = pos_to_color(int(optVal[0]), int(optVal[1]))
self.api.log(color)
self.api.get_entity(entity_id).call_service("turn_on", rgb_color=color)
if(btype == "positionSlider"):
if btype == "positionSlider":
pos = int(optVal)
self.api.get_entity(entity_id).call_service("set_cover_position", position=pos)
if(btype == "volumeSlider"):
if btype == "volumeSlider":
pos = int(optVal)
# HA wants this value between 0 and 1 as float
pos = pos/100
@@ -575,5 +572,5 @@ class NsPanelLovelaceUI:
def send_message_page(self, id, heading, msg, b1, b2):
self.send_mqtt_msg(f"pageType,popupNotify")
self.send_mqtt_msg(f"entityUpdateDetail,{id},{heading},65535,{b2},65535,{b2},65535,{msg},65535")
self.send_mqtt_msg(f"entityUpdateDetail,|{id}|{heading}|65535|{b1}|65535|{b2}|65535|{msg}|65535|0")

39
tasmota/README.md
View File

@@ -1,5 +1,44 @@
# Nextion Berry Driver
This berry driver is intended for the usage with a custom HMI/TFT firmware on nspanel and is a customisted version form [peepshow-21's ns-flash](https://github.com/peepshow-21/ns-flash)
It adds the following commands to Tasmota:
- `Nextion Payload`
Send's normal Nextion Commands to the Screen (suffixed by 0xFFFFFF)
- `CustomSend Payload`
Send's normal Custom Commands to the Screen in the following format:
`55 BB [payload length] [payload] [crc] [crc]`
- `FlashNextion URL`
Start's flashing a tft file to the nextion screen via Nextion Upload Protocol 1.2
Webserver must be reachable via HTTP
Example: `FlashNextion http://192.168.75.30:8123/local/nspanel.tft`
- `GetDriverVersion`
Returns the version currently defined in the berry script
- `UpdateDriverVersion URL`
Downloads the autoexec.be script from the specified URL and loads it.
Besides the commands, serial input will be published on 'RESULT' Topic, depending on the input in one of the following formats:
- `{"CustomRecv":%s}`
- `{"nextion":%s}`
# Nextion Berry Driver Legacy Range (Old version with HTTP Range Method)
This berry driver is intended for the usage with a custom HMI/TFT firmware on nspanel.
It adds the following commands to Tasmota:

388
tasmota/autoexec-beta.be
View File

@@ -1,388 +0,0 @@
# Sonoff NSPanel Tasmota Lovelace UI Berry Driver | code by joBr99
# based on;
# Sonoff NSPanel Tasmota (Nextion with Flashing) driver | code by peepshow-21
# based on;
# Sonoff NSPanel Tasmota driver v0.47 | code by blakadder and s-hadinger
# Example Flash
# FlashNextion http://ip-address-of-your-homeassistant:8123/local/nspanel.tft
# FlashNextion http://nspanel.pky.eu/lui.tft
class Nextion : Driver
static VERSION = "1.1.3"
static header = bytes('55BB')
static flash_block_size = 4096
var flash_mode
var flash_size
var flash_written
var flash_buff
var flash_offset
var awaiting_offset
var tcp
var ser
var last_per
def split_55(b)
var ret = []
var s = size(b)
var i = s-2 # start from last-1
while i > 0
if b[i] == 0x55 && b[i+1] == 0xBB
ret.push(b[i..s-1]) # push last msg to list
b = b[(0..i-1)] # write the rest back to b
end
i -= 1
end
ret.push(b)
return ret
end
def crc16(data, poly)
if !poly poly = 0xA001 end
# CRC-16 MODBUS HASHING ALGORITHM
var crc = 0xFFFF
for i:0..size(data)-1
crc = crc ^ data[i]
for j:0..7
if crc & 1
crc = (crc >> 1) ^ poly
else
crc = crc >> 1
end
end
end
return crc
end
# encode using custom protocol 55 BB [payload length] [payload length] [payload] [crc] [crc]
def encode(payload)
var b = bytes()
b += self.header
b.add(size(payload), 2) # add size as 2 bytes, little endian
b += bytes().fromstring(payload)
var msg_crc = self.crc16(b)
b.add(msg_crc, 2) # crc 2 bytes, little endian
return b
end
def encodenx(payload)
var b = bytes().fromstring(payload)
b += bytes('FFFFFF')
return b
end
def sendnx(payload)
import string
var payload_bin = self.encodenx(payload)
self.ser.write(payload_bin)
log(string.format("NXP: Nextion command sent = %s",str(payload_bin)), 3)
end
def send(payload)
var payload_bin = self.encode(payload)
if self.flash_mode==1
log("NXP: skipped command becuase still flashing", 3)
else
self.ser.write(payload_bin)
log("NXP: payload sent = " + str(payload_bin), 3)
end
end
def write_to_nextion(b)
self.ser.write(b)
end
def screeninit()
log("NXP: Screen Initialized")
self.sendnx("recmod=1")
end
def write_block()
import string
log("FLH: Read block",3)
while size(self.flash_buff)<self.flash_block_size && self.tcp.connected()
if self.tcp.available()>0
self.flash_buff += self.tcp.readbytes()
else
tasmota.delay(50)
log("FLH: Wait for available...",3)
end
end
log("FLH: Buff size "+str(size(self.flash_buff)),3)
var to_write
if size(self.flash_buff)>self.flash_block_size
to_write = self.flash_buff[0..self.flash_block_size-1]
self.flash_buff = self.flash_buff[self.flash_block_size..]
else
to_write = self.flash_buff
self.flash_buff = bytes()
end
log("FLH: Writing "+str(size(to_write)),3)
var per = (self.flash_written*100)/self.flash_size
if (self.last_per!=per)
self.last_per = per
tasmota.publish_result(string.format("{\"Flashing\":{\"complete\": %d}}",per), "RESULT")
end
if size(to_write)>0
self.flash_written += size(to_write)
if self.flash_offset==0 || self.flash_written>self.flash_offset
self.ser.write(to_write)
self.flash_offset = 0
else
tasmota.set_timer(10,/->self.write_block())
end
end
log("FLH: Total "+str(self.flash_written),3)
if (self.flash_written==self.flash_size)
log("FLH: Flashing complete")
self.flash_mode = 0
end
end
def every_100ms()
import string
if self.ser.available() > 0
var msg = self.ser.read()
if size(msg) > 0
log(string.format("NXP: Received Raw = %s",str(msg)), 3)
if (self.flash_mode==1)
var strv = msg[0..-4].asstring()
if string.find(strv,"comok 2")>=0
log("FLH: Send (High Speed) flash start")
self.sendnx(string.format("whmi-wris %d,115200,res0",self.flash_size))
elif size(msg)==1 && msg[0]==0x08
log("FLH: Waiting offset...",3)
self.awaiting_offset = 1
elif size(msg)==4 && self.awaiting_offset==1
self.awaiting_offset = 0
self.flash_offset = msg.get(0,4)
log("FLH: Flash offset marker "+str(self.flash_offset),3)
self.write_block()
elif size(msg)==1 && msg[0]==0x05
self.write_block()
else
log("FLH: Something has gone wrong flashing display firmware ["+str(msg)+"]",2)
end
else
var msg_list = self.split_55(msg)
for i:0..size(msg_list)-1
msg = msg_list[i]
if size(msg) > 0
if msg == bytes('000000FFFFFF88FFFFFF')
self.screeninit()
elif size(msg)>=2 && msg[0]==0x55 && msg[1]==0xBB
var jm = string.format("{\"CustomRecv\":\"%s\"}",msg[4..-3].asstring())
tasmota.publish_result(jm, "RESULT")
elif msg[0]==0x07 && size(msg)==1 # BELL/Buzzer
tasmota.cmd("buzzer 1,1")
else
var jm = string.format("{\"nextion\":\"%s\"}",str(msg[0..-4]))
tasmota.publish_result(jm, "RESULT")
end
end
end
end
end
end
end
def begin_nextion_flash()
self.flash_written = 0
self.awaiting_offset = 0
self.flash_offset = 0
self.sendnx('DRAKJHSUYDGBNCJHGJKSHBDN')
self.sendnx('recmod=0')
self.sendnx('recmod=0')
self.flash_mode = 1
self.sendnx("connect")
end
def open_url(url)
import string
var host
var port
var s1 = string.split(url,7)[1]
var i = string.find(s1,":")
var sa
if i<0
port = 80
i = string.find(s1,"/")
sa = string.split(s1,i)
host = sa[0]
else
sa = string.split(s1,i)
host = sa[0]
s1 = string.split(sa[1],1)[1]
i = string.find(s1,"/")
sa = string.split(s1,i)
port = int(sa[0])
end
var get = sa[1]
log(string.format("FLH: host: %s, port: %s, get: %s",host,port,get))
self.tcp = tcpclient()
self.tcp.connect(host,port)
log("FLH: Connected:"+str(self.tcp.connected()),3)
var get_req = "GET "+get+" HTTP/1.0\r\n"
get_req += string.format("HOST: %s:%s\r\n\r\n",host,port)
self.tcp.write(get_req)
var a = self.tcp.available()
i = 1
while a==0 && i<5
tasmota.delay(100*i)
tasmota.yield()
i += 1
log("FLH: Retry "+str(i),3)
a = self.tcp.available()
end
if a==0
log("FLH: Nothing available to read!",3)
return
end
var b = self.tcp.readbytes()
i = 0
var end_headers = false;
var headers
while i<size(b) && headers==nil
if b[i..(i+3)]==bytes().fromstring("\r\n\r\n")
headers = b[0..(i+3)].asstring()
self.flash_buff = b[(i+4)..]
else
i += 1
end
end
#print(headers)
# check http respose for code 200
var tag = "200 OK"
i = string.find(headers,tag)
if (i>0)
log("FLH: HTTP Respose is 200 OK",3)
else
log("FLH: HTTP Respose is not 200 OK",3)
print(headers)
return
end
# check http respose for content-length
tag = "Content-Length: "
i = string.find(headers,tag)
if (i>0)
var i2 = string.find(headers,"\r\n",i)
var s = headers[i+size(tag)..i2-1]
self.flash_size=int(s)
end
if self.flash_size==0
log("FLH: No size header, counting ...",3)
self.flash_size = size(self.flash_buff)
#print("counting start ...")
while self.tcp.connected()
while self.tcp.available()>0
self.flash_size += size(self.tcp.readbytes())
end
tasmota.delay(50)
end
#print("counting end ...",self.flash_size)
self.tcp.close()
self.open_url(url)
else
log("FLH: Size found in header, skip count",3)
end
log("FLH: Flash file size: "+str(self.flash_size),3)
end
def flash_nextion(url)
self.flash_size = 0
self.open_url(url)
self.begin_nextion_flash()
end
def version_number(str)
import string
var i1 = string.find(str,".",0)
var i2 = string.find(str,".",i1+1)
var num = int(str[0..i1-1])*10000+int(str[i1+1..i2-1])*100+int(str[i2+1..])
return num
end
def init()
log("NXP: Initializing Driver")
self.ser = serial(17, 16, 115200, serial.SERIAL_8N1)
self.flash_mode = 0
end
end
var nextion = Nextion()
tasmota.add_driver(nextion)
def get_current_version(cmd, idx, payload, payload_json)
import string
var version_of_this_script = 2
var jm = string.format("{\"nlui_driver_version\":\"%s\"}", version_of_this_script)
tasmota.publish_result(jm, "RESULT")
end
tasmota.add_cmd('GetDriverVersion', get_current_version)
def update_berry_driver(cmd, idx, payload, payload_json)
def task()
import string
var cl = webclient()
cl.begin(payload)
var r = cl.GET()
if r == 200
print("Sucessfully downloaded nspanel-lovelace-ui berry driver")
else
print("Error while downloading nspanel-lovelace-ui berry driver")
end
r = cl.write_file("autoexec.be")
if r < 0
print("Error while writeing nspanel-lovelace-ui berry driver")
else
print("Scucessfully written nspanel-lovelace-ui berry driver")
var s = load('autoexec.be')
if s == true
var jm = string.format("{\"nlui_driver_update\":\"%s\"}", "succeeded")
tasmota.publish_result(jm, "RESULT")
else
var jm = string.format("{\"nlui_driver_update\":\"%s\"}", "failed")
tasmota.publish_result(jm, "RESULT")
end
end
end
tasmota.set_timer(0,task)
tasmota.resp_cmnd_done()
end
tasmota.add_cmd('UpdateDriverVersion', update_berry_driver)
def flash_nextion(cmd, idx, payload, payload_json)
def task()
nextion.flash_nextion(payload)
end
tasmota.set_timer(0,task)
tasmota.resp_cmnd_done()
end
def send_cmd(cmd, idx, payload, payload_json)
nextion.sendnx(payload)
tasmota.resp_cmnd_done()
end
def send_cmd2(cmd, idx, payload, payload_json)
nextion.send(payload)
tasmota.resp_cmnd_done()
end
tasmota.add_cmd('Nextion', send_cmd)
tasmota.add_cmd('CustomSend', send_cmd2)
tasmota.add_cmd('FlashNextion', flash_nextion)

409
tasmota/autoexec-legacy-range.be Normal file
View File

@@ -0,0 +1,409 @@
# Nextion Serial Protocol driver by joBr99 + nextion upload protocol 1.2 (the fast one yay) implementation using http range and tcpclient
# based on;
# Sonoff NSPanel Tasmota driver v0.47 | code by blakadder and s-hadinger
class TftDownloader
var tcp
var host
var port
var file
var s
var b
var tft_file_size
var current_chunk
var current_chunk_start
var download_range
def init(host, port, file, download_range)
self.tft_file_size = 0
self.host = host
self.port = port
self.file = file
self.download_range = download_range #32768
end
def download_chunk(b_start, b_length)
import string
self.tcp = tcpclient()
self.tcp.connect(self.host, self.port)
print("connected:", self.tcp.connected())
self.s = "GET " + self.file + " HTTP/1.0\r\n"
self.s += "HOST: " + self.host + "\r\n"
self.s += string.format("Range: bytes=%d-%d\r\n", b_start, (b_start+b_length-1))
print(string.format("Downloading Byte %d - %d", b_start, (b_start+b_length-1)))
self.s += "\r\n"
self.tcp.write(self.s)
#read one char after another until we reached end of http header
var end_of_header = false
var header = ""
while !end_of_header
if self.tcp.available() > 0
header += self.tcp.read(1)
if(string.find(header, '\r\n\r\n') != -1)
end_of_header = true
end
end
end
var content_length = 0
# check for 206 status code
if(string.find(header, '206 Partial Content') != -1)
# download was sucessful
else
print("Error while downloading")
print(header)
return nil
end
# convert header to list
header = string.split(header, '\r\n')
for i : header.iter()
#print(i)
if(string.find(i, 'Content-Range:') != -1)
if self.tft_file_size == 0
print(i)
self.tft_file_size = number(string.split(i, '/')[1])
end
end
if(string.find(i, 'Content-Length:') != -1)
content_length = number(string.split(i, 16)[1])
end
end
#print(content_length)
# read bytes until content_length is reached
var content = bytes()
while content.size() != content_length
if self.tcp.available() > 0
content += self.tcp.readbytes()
end
end
#print(content.size())
return content
end
def get_file_size()
self.download_chunk(0, 1)
return self.tft_file_size
end
# returns the next 4096 bytes after pos of the tft file
def next_chunk(pos)
if(self.current_chunk == nil)
print("current chunk empty")
self.current_chunk = self.download_chunk(pos, self.download_range)
self.current_chunk_start = pos
end
if(pos < self.current_chunk_start)
print("Requested pos is below start point of chunk in memory, not implemented")
end
if(pos >= (self.current_chunk_start+self.download_range))
print("Requested pos is after the end of chunk in memory, downloading new range")
self.current_chunk = self.download_chunk(pos, self.download_range)
self.current_chunk_start = pos
end
var start_within_current_chunk = pos - self.current_chunk_start
return self.current_chunk[start_within_current_chunk..(start_within_current_chunk+4095)]
end
end
class Nextion : Driver
var ser
var flash_size
var flash_mode
var flash_version
var flash_skip
var flash_current_byte
var tftd
var progress_percentage_last
static header = bytes('55BB')
def init()
log("NSP: Initializing Driver")
self.ser = serial(17, 16, 115200, serial.SERIAL_8N1)
self.flash_mode = 0
self.flash_version = 1
self.flash_skip = false
tasmota.add_driver(self)
end
def crc16(data, poly)
if !poly poly = 0xA001 end
# CRC-16 MODBUS HASHING ALGORITHM
var crc = 0xFFFF
for i:0..size(data)-1
crc = crc ^ data[i]
for j:0..7
if crc & 1
crc = (crc >> 1) ^ poly
else
crc = crc >> 1
end
end
end
return crc
end
def split_55(b)
var ret = []
var s = size(b)
var i = s-2 # start from last-1
while i > 0
if b[i] == 0x55 && b[i+1] == 0xBB
ret.push(b[i..s-1]) # push last msg to list
b = b[(0..i-1)] # write the rest back to b
end
i -= 1
end
ret.push(b)
return ret
end
# encode using custom protocol 55 BB [payload length] [payload length] [payload] [crc] [crc]
def encode(payload)
var b = bytes()
b += self.header
b.add(size(payload), 2) # add size as 2 bytes, little endian
b += bytes().fromstring(payload)
var msg_crc = self.crc16(b)
b.add(msg_crc, 2) # crc 2 bytes, little endian
return b
end
# send a nextion payload
def encodenx(payload)
var b = bytes().fromstring(payload)
b += bytes('FFFFFF')
return b
end
def sendnx(payload)
var payload_bin = self.encodenx(payload)
self.ser.write(payload_bin)
print("NSP: Sent =", payload_bin)
log("NSP: Nextion command sent = " + str(payload_bin), 3)
end
def send(payload)
var payload_bin = self.encode(payload)
if self.flash_mode==1
log("NSP: skipped command becuase still flashing", 3)
else
self.ser.write(payload_bin)
log("NSP: payload sent = " + str(payload_bin), 3)
end
end
def start_flash(url)
import string
var host
var port
var s1 = string.split(url,7)[1]
var i = string.find(s1,":")
var sa
if i<0
port = 80
i = string.find(s1,"/")
sa = string.split(s1,i)
host = sa[0]
else
sa = string.split(s1,i)
host = sa[0]
s1 = string.split(sa[1],1)[1]
i = string.find(s1,"/")
sa = string.split(s1,i)
port = int(sa[0])
end
var file = sa[1]
#print(host,port,file)
self.tftd = TftDownloader(host, port, file, 32768)
# get size of tft file
self.flash_size = self.tftd.get_file_size()
self.flash_mode = 1
self.sendnx('DRAKJHSUYDGBNCJHGJKSHBDN')
self.sendnx('recmod=0')
self.sendnx('recmod=0')
self.sendnx("connect")
self.sendnx("connect")
self.flash_current_byte = 0
end
def write_chunk(b_start)
var chunk = self.tftd.next_chunk(b_start)
#import string
#print(string.format("Sending Byte %d - %d with size of %d", b_start, b_start+4095, chunk.size()))
self.ser.write(chunk)
return chunk.size()
end
def flash_nextion()
import string
var x = self.write_chunk(self.flash_current_byte)
self.flash_current_byte = self.flash_current_byte + x
var progress_percentage = (self.flash_current_byte*100/self.flash_size)
if (self.progress_percentage_last!=progress_percentage)
print(string.format("Flashing Progress ( %d / %d ) [ %d ]", self.flash_current_byte, self.flash_size, progress_percentage))
self.progress_percentage_last = progress_percentage
tasmota.publish_result(string.format("{\"Flashing\":{\"complete\": %d}}",progress_percentage), "RESULT")
end
if (self.flash_current_byte==self.flash_size)
log("NSP: Flashing complete")
self.flash_mode = 0
end
tasmota.yield()
end
def every_100ms()
import string
if self.ser.available() > 0
var msg = self.ser.read()
if size(msg) > 0
print("NSP: Received Raw =", msg)
if self.flash_mode==1
var str = msg[0..-4].asstring()
log(str, 3)
# TODO: add check for firmware versions < 126 and send proto 1.1 command for thoose
if (string.find(str,"comok 2")==0)
if self.flash_version==1
log("NSP: Flashing 1.1")
self.sendnx(string.format("whmi-wri %d,115200,1",self.flash_size)) # Nextion Upload Protocol 1.1
else
log("NSP: Flashing 1.2")
self.sendnx(string.format("whmi-wris %d,115200,1",self.flash_size)) # Nextion Upload Protocol 1.2
end
# skip to byte (upload protocol 1.2)
elif (size(msg)==1 && msg[0]==0x08)
self.flash_skip = true
print("rec 0x08")
elif (size(msg)==4 && self.flash_skip)
var skip_to_byte = msg[0..4].get(0,4)
if(skip_to_byte == 0)
print("don't skip, offset is 0")
else
print("skip to ", skip_to_byte)
self.flash_current_byte = skip_to_byte
end
self.flash_nextion()
# send next 4096 bytes (proto 1.1/1.2)
elif (size(msg)==1 && msg[0]==0x05)
print("rec 0x05")
self.flash_nextion()
end
else
# Recive messages using custom protocol 55 BB [payload length] [payload length] [payload] [crc] [crc]
if msg[0..1] == self.header
var lst = self.split_55(msg)
for i:0..size(lst)-1
msg = lst[i]
#var j = msg[2]+2
var j = size(msg) - 3
msg = msg[4..j]
if size(msg) > 2
var jm = string.format("{\"CustomRecv\":\"%s\"}",msg.asstring())
tasmota.publish_result(jm, "RESULT")
end
end
elif msg == bytes('000000FFFFFF88FFFFFF')
log("NSP: Screen Initialized")
else
var jm = string.format("{\"nextion\":\"%s\"}",str(msg[0..-4]))
tasmota.publish_result(jm, "RESULT")
end
end
end
end
end
end
def get_current_version(cmd, idx, payload, payload_json)
import string
var version_of_this_script = 2
var jm = string.format("{\"nlui_driver_version\":\"%s\"}", version_of_this_script)
tasmota.publish_result(jm, "RESULT")
end
tasmota.add_cmd('GetDriverVersion', get_current_version)
def update_berry_driver(cmd, idx, payload, payload_json)
def task()
import string
var cl = webclient()
cl.begin(payload)
var r = cl.GET()
if r == 200
print("Sucessfully downloaded nspanel-lovelace-ui berry driver")
else
print("Error while downloading nspanel-lovelace-ui berry driver")
end
r = cl.write_file("autoexec.be")
if r < 0
print("Error while writeing nspanel-lovelace-ui berry driver")
else
print("Scucessfully written nspanel-lovelace-ui berry driver")
var s = load('autoexec.be')
if s == true
var jm = string.format("{\"nlui_driver_update\":\"%s\"}", "succeeded")
tasmota.publish_result(jm, "RESULT")
else
var jm = string.format("{\"nlui_driver_update\":\"%s\"}", "failed")
tasmota.publish_result(jm, "RESULT")
end
end
end
tasmota.set_timer(0,task)
tasmota.resp_cmnd_done()
end
tasmota.add_cmd('UpdateDriverVersion', update_berry_driver)
var nextion = Nextion()
def flash_nextion(cmd, idx, payload, payload_json)
def task()
nextion.flash_version = 1
nextion.start_flash(payload)
end
tasmota.set_timer(0,task)
tasmota.resp_cmnd_done()
end
tasmota.add_cmd('FlashNextion', flash_nextion)
def flash_nextion_1_2(cmd, idx, payload, payload_json)
def task()
nextion.flash_version = 2
nextion.start_flash(payload)
end
tasmota.set_timer(0,task)
tasmota.resp_cmnd_done()
end
tasmota.add_cmd('FlashNextionFast', flash_nextion_1_2)
def send_cmd(cmd, idx, payload, payload_json)
nextion.sendnx(payload)
tasmota.resp_cmnd_done()
end
tasmota.add_cmd('Nextion', send_cmd)
def send_cmd2(cmd, idx, payload, payload_json)
nextion.send(payload)
tasmota.resp_cmnd_done()
end
tasmota.add_cmd('CustomSend', send_cmd2)

505
tasmota/autoexec.be
View File

@@ -1,138 +1,43 @@
# Nextion Serial Protocol driver by joBr99 + nextion upload protocol 1.2 (the fast one yay) implementation using http range and tcpclient
# Sonoff NSPanel Tasmota Lovelace UI Berry Driver | code by joBr99
# based on;
# Sonoff NSPanel Tasmota (Nextion with Flashing) driver | code by peepshow-21
# based on;
# Sonoff NSPanel Tasmota driver v0.47 | code by blakadder and s-hadinger
class TftDownloader
var tcp
var host
var port
var file
var s
var b
var tft_file_size
var current_chunk
var current_chunk_start
var download_range
def init(host, port, file, download_range)
self.tft_file_size = 0
self.host = host
self.port = port
self.file = file
self.download_range = download_range #32768
end
def download_chunk(b_start, b_length)
import string
self.tcp = tcpclient()
self.tcp.connect(self.host, self.port)
print("connected:", self.tcp.connected())
self.s = "GET " + self.file + " HTTP/1.0\r\n"
self.s += "HOST: " + self.host + "\r\n"
self.s += string.format("Range: bytes=%d-%d\r\n", b_start, (b_start+b_length-1))
print(string.format("Downloading Byte %d - %d", b_start, (b_start+b_length-1)))
self.s += "\r\n"
self.tcp.write(self.s)
#read one char after another until we reached end of http header
var end_of_header = false
var header = ""
while !end_of_header
if self.tcp.available() > 0
header += self.tcp.read(1)
if(string.find(header, '\r\n\r\n') != -1)
end_of_header = true
end
end
end
var content_length = 0
# check for 206 status code
if(string.find(header, '206 Partial Content') != -1)
# download was sucessful
else
print("Error while downloading")
print(header)
return nil
end
# convert header to list
header = string.split(header, '\r\n')
for i : header.iter()
#print(i)
if(string.find(i, 'Content-Range:') != -1)
if self.tft_file_size == 0
print(i)
self.tft_file_size = number(string.split(i, '/')[1])
end
end
if(string.find(i, 'Content-Length:') != -1)
content_length = number(string.split(i, 16)[1])
end
end
#print(content_length)
# read bytes until content_length is reached
var content = bytes()
while content.size() != content_length
if self.tcp.available() > 0
content += self.tcp.readbytes()
end
end
#print(content.size())
return content
end
def get_file_size()
self.download_chunk(0, 1)
return self.tft_file_size
end
# returns the next 4096 bytes after pos of the tft file
def next_chunk(pos)
if(self.current_chunk == nil)
print("current chunk empty")
self.current_chunk = self.download_chunk(pos, self.download_range)
self.current_chunk_start = pos
end
if(pos < self.current_chunk_start)
print("Requested pos is below start point of chunk in memory, not implemented")
end
if(pos >= (self.current_chunk_start+self.download_range))
print("Requested pos is after the end of chunk in memory, downloading new range")
self.current_chunk = self.download_chunk(pos, self.download_range)
self.current_chunk_start = pos
end
var start_within_current_chunk = pos - self.current_chunk_start
return self.current_chunk[start_within_current_chunk..(start_within_current_chunk+4095)]
end
end
# Example Flash
# FlashNextion http://ip-address-of-your-homeassistant:8123/local/nspanel.tft
# FlashNextion http://nspanel.pky.eu/lui.tft
class Nextion : Driver
var ser
var flash_size
var flash_mode
var flash_version
var flash_skip
var flash_current_byte
var tftd
var progress_percentage_last
static VERSION = "1.1.3"
static header = bytes('55BB')
def init()
log("NSP: Initializing Driver")
self.ser = serial(17, 16, 115200, serial.SERIAL_8N1)
self.flash_mode = 0
self.flash_version = 1
self.flash_skip = false
tasmota.add_driver(self)
static flash_block_size = 4096
var flash_mode
var flash_size
var flash_written
var flash_buff
var flash_offset
var awaiting_offset
var tcp
var ser
var last_per
def split_55(b)
var ret = []
var s = size(b)
var i = s-2 # start from last-1
while i > 0
if b[i] == 0x55 && b[i+1] == 0xBB
ret.push(b[i..s-1]) # push last msg to list
b = b[(0..i-1)] # write the rest back to b
end
i -= 1
end
ret.push(b)
return ret
end
def crc16(data, poly)
@@ -152,21 +57,6 @@ class Nextion : Driver
return crc
end
def split_55(b)
var ret = []
var s = size(b)
var i = s-2 # start from last-1
while i > 0
if b[i] == 0x55 && b[i+1] == 0xBB
ret.push(b[i..s-1]) # push last msg to list
b = b[(0..i-1)] # write the rest back to b
end
i -= 1
end
ret.push(b)
return ret
end
# encode using custom protocol 55 BB [payload length] [payload length] [payload] [crc] [crc]
def encode(payload)
var b = bytes()
@@ -178,7 +68,6 @@ class Nextion : Driver
return b
end
# send a nextion payload
def encodenx(payload)
var b = bytes().fromstring(payload)
b += bytes('FFFFFF')
@@ -186,23 +75,135 @@ class Nextion : Driver
end
def sendnx(payload)
import string
var payload_bin = self.encodenx(payload)
self.ser.write(payload_bin)
print("NSP: Sent =", payload_bin)
log("NSP: Nextion command sent = " + str(payload_bin), 3)
log(string.format("NXP: Nextion command sent = %s",str(payload_bin)), 3)
end
def send(payload)
var payload_bin = self.encode(payload)
if self.flash_mode==1
log("NSP: skipped command becuase still flashing", 3)
log("NXP: skipped command becuase still flashing", 3)
else
self.ser.write(payload_bin)
log("NSP: payload sent = " + str(payload_bin), 3)
log("NXP: payload sent = " + str(payload_bin), 3)
end
end
def start_flash(url)
def write_to_nextion(b)
self.ser.write(b)
end
def screeninit()
log("NXP: Screen Initialized")
self.sendnx("recmod=1")
end
def write_block()
import string
log("FLH: Read block",3)
while size(self.flash_buff)<self.flash_block_size && self.tcp.connected()
if self.tcp.available()>0
self.flash_buff += self.tcp.readbytes()
else
tasmota.delay(50)
log("FLH: Wait for available...",3)
end
end
log("FLH: Buff size "+str(size(self.flash_buff)),3)
var to_write
if size(self.flash_buff)>self.flash_block_size
to_write = self.flash_buff[0..self.flash_block_size-1]
self.flash_buff = self.flash_buff[self.flash_block_size..]
else
to_write = self.flash_buff
self.flash_buff = bytes()
end
log("FLH: Writing "+str(size(to_write)),3)
var per = (self.flash_written*100)/self.flash_size
if (self.last_per!=per)
self.last_per = per
tasmota.publish_result(string.format("{\"Flashing\":{\"complete\": %d}}",per), "RESULT")
end
if size(to_write)>0
self.flash_written += size(to_write)
if self.flash_offset==0 || self.flash_written>self.flash_offset
self.ser.write(to_write)
self.flash_offset = 0
else
tasmota.set_timer(10,/->self.write_block())
end
end
log("FLH: Total "+str(self.flash_written),3)
if (self.flash_written==self.flash_size)
log("FLH: Flashing complete")
self.flash_mode = 0
end
end
def every_100ms()
import string
if self.ser.available() > 0
var msg = self.ser.read()
if size(msg) > 0
log(string.format("NXP: Received Raw = %s",str(msg)), 3)
if (self.flash_mode==1)
var strv = msg[0..-4].asstring()
if string.find(strv,"comok 2")>=0
log("FLH: Send (High Speed) flash start")
self.sendnx(string.format("whmi-wris %d,115200,res0",self.flash_size))
elif size(msg)==1 && msg[0]==0x08
log("FLH: Waiting offset...",3)
self.awaiting_offset = 1
elif size(msg)==4 && self.awaiting_offset==1
self.awaiting_offset = 0
self.flash_offset = msg.get(0,4)
log("FLH: Flash offset marker "+str(self.flash_offset),3)
self.write_block()
elif size(msg)==1 && msg[0]==0x05
self.write_block()
else
log("FLH: Something has gone wrong flashing display firmware ["+str(msg)+"]",2)
end
else
var msg_list = self.split_55(msg)
for i:0..size(msg_list)-1
msg = msg_list[i]
if size(msg) > 0
if msg == bytes('000000FFFFFF88FFFFFF')
self.screeninit()
elif size(msg)>=2 && msg[0]==0x55 && msg[1]==0xBB
var jm = string.format("{\"CustomRecv\":\"%s\"}",msg[4..-3].asstring())
tasmota.publish_result(jm, "RESULT")
elif msg[0]==0x07 && size(msg)==1 # BELL/Buzzer
tasmota.cmd("buzzer 1,1")
else
var jm = string.format("{\"nextion\":\"%s\"}",str(msg[0..-4]))
tasmota.publish_result(jm, "RESULT")
end
end
end
end
end
end
end
def begin_nextion_flash()
self.flash_written = 0
self.awaiting_offset = 0
self.flash_offset = 0
self.sendnx('DRAKJHSUYDGBNCJHGJKSHBDN')
self.sendnx('recmod=0')
self.sendnx('recmod=0')
self.flash_mode = 1
self.sendnx("connect")
end
def open_url(url)
import string
var host
var port
@@ -222,115 +223,109 @@ class Nextion : Driver
sa = string.split(s1,i)
port = int(sa[0])
end
var file = sa[1]
#print(host,port,file)
var get = sa[1]
log(string.format("FLH: host: %s, port: %s, get: %s",host,port,get))
self.tcp = tcpclient()
self.tcp.connect(host,port)
log("FLH: Connected:"+str(self.tcp.connected()),3)
var get_req = "GET "+get+" HTTP/1.0\r\n"
get_req += string.format("HOST: %s:%s\r\n\r\n",host,port)
self.tcp.write(get_req)
var a = self.tcp.available()
i = 1
while a==0 && i<5
tasmota.delay(100*i)
tasmota.yield()
i += 1
log("FLH: Retry "+str(i),3)
a = self.tcp.available()
end
if a==0
log("FLH: Nothing available to read!",3)
return
end
var b = self.tcp.readbytes()
i = 0
var end_headers = false;
var headers
while i<size(b) && headers==nil
if b[i..(i+3)]==bytes().fromstring("\r\n\r\n")
headers = b[0..(i+3)].asstring()
self.flash_buff = b[(i+4)..]
else
i += 1
end
end
#print(headers)
# check http respose for code 200
var tag = "200 OK"
i = string.find(headers,tag)
if (i>0)
log("FLH: HTTP Respose is 200 OK",3)
else
log("FLH: HTTP Respose is not 200 OK",3)
print(headers)
return
end
# check http respose for content-length
tag = "Content-Length: "
i = string.find(headers,tag)
if (i>0)
var i2 = string.find(headers,"\r\n",i)
var s = headers[i+size(tag)..i2-1]
self.flash_size=int(s)
end
if self.flash_size==0
log("FLH: No size header, counting ...",3)
self.flash_size = size(self.flash_buff)
#print("counting start ...")
while self.tcp.connected()
while self.tcp.available()>0
self.flash_size += size(self.tcp.readbytes())
end
tasmota.delay(50)
end
#print("counting end ...",self.flash_size)
self.tcp.close()
self.open_url(url)
else
log("FLH: Size found in header, skip count",3)
end
log("FLH: Flash file size: "+str(self.flash_size),3)
self.tftd = TftDownloader(host, port, file, 32768)
# get size of tft file
self.flash_size = self.tftd.get_file_size()
self.flash_mode = 1
self.sendnx('DRAKJHSUYDGBNCJHGJKSHBDN')
self.sendnx('recmod=0')
self.sendnx('recmod=0')
self.sendnx("connect")
self.sendnx("connect")
self.flash_current_byte = 0
end
def write_chunk(b_start)
var chunk = self.tftd.next_chunk(b_start)
#import string
#print(string.format("Sending Byte %d - %d with size of %d", b_start, b_start+4095, chunk.size()))
self.ser.write(chunk)
return chunk.size()
def flash_nextion(url)
self.flash_size = 0
self.open_url(url)
self.begin_nextion_flash()
end
def flash_nextion()
def version_number(str)
import string
var x = self.write_chunk(self.flash_current_byte)
self.flash_current_byte = self.flash_current_byte + x
var progress_percentage = (self.flash_current_byte*100/self.flash_size)
if (self.progress_percentage_last!=progress_percentage)
print(string.format("Flashing Progress ( %d / %d ) [ %d ]", self.flash_current_byte, self.flash_size, progress_percentage))
self.progress_percentage_last = progress_percentage
tasmota.publish_result(string.format("{\"Flashing\":{\"complete\": %d}}",progress_percentage), "RESULT")
var i1 = string.find(str,".",0)
var i2 = string.find(str,".",i1+1)
var num = int(str[0..i1-1])*10000+int(str[i1+1..i2-1])*100+int(str[i2+1..])
return num
end
if (self.flash_current_byte==self.flash_size)
log("NSP: Flashing complete")
def init()
log("NXP: Initializing Driver")
self.ser = serial(17, 16, 115200, serial.SERIAL_8N1)
self.flash_mode = 0
end
tasmota.yield()
end
def every_100ms()
import string
if self.ser.available() > 0
var msg = self.ser.read()
if size(msg) > 0
print("NSP: Received Raw =", msg)
if self.flash_mode==1
var str = msg[0..-4].asstring()
log(str, 3)
# TODO: add check for firmware versions < 126 and send proto 1.1 command for thoose
if (string.find(str,"comok 2")==0)
if self.flash_version==1
log("NSP: Flashing 1.1")
self.sendnx(string.format("whmi-wri %d,115200,1",self.flash_size)) # Nextion Upload Protocol 1.1
else
log("NSP: Flashing 1.2")
self.sendnx(string.format("whmi-wris %d,115200,1",self.flash_size)) # Nextion Upload Protocol 1.2
end
# skip to byte (upload protocol 1.2)
elif (size(msg)==1 && msg[0]==0x08)
self.flash_skip = true
print("rec 0x08")
elif (size(msg)==4 && self.flash_skip)
var skip_to_byte = msg[0..4].get(0,4)
if(skip_to_byte == 0)
print("don't skip, offset is 0")
else
print("skip to ", skip_to_byte)
self.flash_current_byte = skip_to_byte
end
self.flash_nextion()
# send next 4096 bytes (proto 1.1/1.2)
elif (size(msg)==1 && msg[0]==0x05)
print("rec 0x05")
self.flash_nextion()
end
else
# Recive messages using custom protocol 55 BB [payload length] [payload length] [payload] [crc] [crc]
if msg[0..1] == self.header
var lst = self.split_55(msg)
for i:0..size(lst)-1
msg = lst[i]
#var j = msg[2]+2
var j = size(msg) - 3
msg = msg[4..j]
if size(msg) > 2
var jm = string.format("{\"CustomRecv\":\"%s\"}",msg.asstring())
tasmota.publish_result(jm, "RESULT")
end
end
elif msg == bytes('000000FFFFFF88FFFFFF')
log("NSP: Screen Initialized")
else
var jm = string.format("{\"nextion\":\"%s\"}",str(msg[0..-4]))
tasmota.publish_result(jm, "RESULT")
end
end
end
end
end
end
var nextion = Nextion()
tasmota.add_driver(nextion)
def get_current_version(cmd, idx, payload, payload_json)
import string
var version_of_this_script = 2
var version_of_this_script = 3
var jm = string.format("{\"nlui_driver_version\":\"%s\"}", version_of_this_script)
tasmota.publish_result(jm, "RESULT")
end
@@ -352,7 +347,7 @@ def update_berry_driver(cmd, idx, payload, payload_json)
if r < 0
print("Error while writeing nspanel-lovelace-ui berry driver")
else
print("Scucessfully written nspanel-lovelace-ui berry driver")
print("Sucessfully written nspanel-lovelace-ui berry driver")
var s = load('autoexec.be')
if s == true
var jm = string.format("{\"nlui_driver_update\":\"%s\"}", "succeeded")
@@ -370,40 +365,24 @@ end
tasmota.add_cmd('UpdateDriverVersion', update_berry_driver)
var nextion = Nextion()
def flash_nextion(cmd, idx, payload, payload_json)
def task()
nextion.flash_version = 1
nextion.start_flash(payload)
nextion.flash_nextion(payload)
end
tasmota.set_timer(0,task)
tasmota.resp_cmnd_done()
end
tasmota.add_cmd('FlashNextion', flash_nextion)
def flash_nextion_1_2(cmd, idx, payload, payload_json)
def task()
nextion.flash_version = 2
nextion.start_flash(payload)
end
tasmota.set_timer(0,task)
tasmota.resp_cmnd_done()
end
tasmota.add_cmd('FlashNextionFast', flash_nextion_1_2)
def send_cmd(cmd, idx, payload, payload_json)
nextion.sendnx(payload)
tasmota.resp_cmnd_done()
end
tasmota.add_cmd('Nextion', send_cmd)
def send_cmd2(cmd, idx, payload, payload_json)
nextion.send(payload)
tasmota.resp_cmnd_done()
end
tasmota.add_cmd('Nextion', send_cmd)
tasmota.add_cmd('CustomSend', send_cmd2)
tasmota.add_cmd('FlashNextion', flash_nextion)