sv/tsun-gen3-proxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add send_modbus_cmd()

Browse Source
This commit is contained in:
Stefan Allius
2024-05-03 18:24:06 +02:00
parent a2f67e7d3e
commit 763af8b4cf
1 changed files with 12 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
app/src/gen3plus/solarman_v5.py
View File

@@ -6,12 +6,14 @@ from datetime import datetime
if __name__ == "app.src.gen3plus.solarman_v5": if __name__ == "app.src.gen3plus.solarman_v5":
from app.src.messages import hex_dump_memory, Message from app.src.messages import hex_dump_memory, Message
from app.src.modbus import Modbus
from app.src.config import Config from app.src.config import Config
from app.src.gen3plus.infos_g3p import InfosG3P from app.src.gen3plus.infos_g3p import InfosG3P
from app.src.infos import Register from app.src.infos import Register
else: # pragma: no cover else: # pragma: no cover
from messages import hex_dump_memory, Message from messages import hex_dump_memory, Message
from config import Config from config import Config
from modbus import Modbus
from gen3plus.infos_g3p import InfosG3P from gen3plus.infos_g3p import InfosG3P
from infos import Register from infos import Register
# import traceback # import traceback
@@ -56,6 +58,8 @@ class SolarmanV5(Message):
self.snr = 0 self.snr = 0
self.db = InfosG3P() self.db = InfosG3P()
self.time_ofs = 0 self.time_ofs = 0
self.mb = Modbus()
self.forward_modbus_rep = False
self.switch = { self.switch = {
0x4210: self.msg_data_ind, # real time data 0x4210: self.msg_data_ind, # real time data
@@ -293,6 +297,14 @@ class SolarmanV5(Message):
self._heartbeat()) self._heartbeat())
self.__finish_send_msg() self.__finish_send_msg()
async def send_modbus_cmd(self, func, addr, val) -> None:
self.forward_modbus_rep = False
self.__build_header(0x4510)
self._send_buffer += struct.pack('<BHLLL', 2, 0x2b0, 0, 0, 0)
self._send_buffer += self.mb.build_msg(1, func, addr, val)
self.__finish_send_msg()
await self.async_write('Send Modbus Command:')
def send_at_cmd(self, AT_cmd: str) -> None: def send_at_cmd(self, AT_cmd: str) -> None:
self.__build_header(0x4510) self.__build_header(0x4510)
self._send_buffer += struct.pack(f'<BHLLL{len(AT_cmd)}sc', 1, 2, self._send_buffer += struct.pack(f'<BHLLL{len(AT_cmd)}sc', 1, 2,