You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
87 lines
2.6 KiB
87 lines
2.6 KiB
from .CRC16 import modbus_crc16
|
|
|
|
#Fast Master message type -I hate this-
|
|
|
|
#Usage
|
|
#1. Construct message
|
|
#2. Append data if needed
|
|
#3. get message body (CRC will be added automatically)
|
|
class IdiBusMMESG:
|
|
def __init__(self, address, mmps, cmd):
|
|
self.MMESG = bytearray()
|
|
self.MMESG.extend(int(address).to_bytes(1, byteorder="little"))
|
|
self.MMESG.extend(int(mmps).to_bytes(1, byteorder="little"))
|
|
self.MMESG.extend(int(cmd).to_bytes(1, byteorder="little"))
|
|
|
|
def addData(self, data):
|
|
self.MMESG.extend(data)
|
|
#Appends CRC16 and returns whole packet
|
|
def getMsg(self):
|
|
crc = modbus_crc16(self.MMESG)
|
|
self.MMESG.extend(crc)
|
|
return self.MMESG
|
|
|
|
def __repr__(self):
|
|
return f"MMESG({self.MMESG})"
|
|
|
|
#Usage
|
|
#1. Construct message
|
|
#2. Append data if needed
|
|
#3. get message body (CRC will be added automatically)
|
|
#Main Master message type
|
|
class IdiBusMMES:
|
|
def __init__(self, address, mmps, dev, ch,cmd):
|
|
self.MMES = bytearray()
|
|
if (cmd <=15):
|
|
mmps |= (cmd << 2)
|
|
self.MMES.extend(address.to_bytes(1, byteorder="little"))
|
|
self.MMES.extend(mmps.to_bytes(1, byteorder="little"))
|
|
self.MMES.extend(dev.to_bytes(1, byteorder="little"))
|
|
self.MMES.extend(ch.to_bytes(1, byteorder="little"))
|
|
if (cmd > 15):
|
|
self.MMES.extend(cmd.to_bytes(1, byteorder="little"))
|
|
def addData(self, data):
|
|
self.MMES.extend(bytearray(data))
|
|
|
|
def getMsg(self):
|
|
crc = modbus_crc16(self.MMES)
|
|
self.MMES.extend(crc)
|
|
return self.MMES
|
|
|
|
def __repr__(self):
|
|
return f"Addr:{self.getAddr()}"
|
|
|
|
|
|
#Slave message constructor
|
|
class IdiBusSMES:
|
|
def __init__(self, address, spdu, errcode):
|
|
self.SMES = bytearray()
|
|
self.SMES.extend(address.to_bytes(1, byteorder="little"))
|
|
self.SMES.extend(spdu.to_bytes(1, byteorder="little"))
|
|
self.SMES.extend(errcode.to_bytes(1, byteorder="little"))
|
|
def addData(self, data):
|
|
self.SMES.extend(bytearray(data))
|
|
|
|
def getMsg(self):
|
|
crc = modbus_crc16(self.SMES)
|
|
self.SMES.extend(crc)
|
|
return self.SMES
|
|
|
|
def __repr__(self):
|
|
return f"SMES: {self.SMES}"
|
|
|
|
|
|
class IdiBusMMPS:
|
|
typeMMESG = 1
|
|
typeMMES = 0
|
|
def __init__(self, MesType = typeMMESG,):
|
|
self.LongMessage = 0
|
|
self.MesType = MesType
|
|
self.CommandType = 0
|
|
self.AlarmFrame = 0
|
|
self.EncriptedAES = 0
|
|
def getValue(self):
|
|
val = 0
|
|
val |= (self.LongMessage << 0) | (self.MesType << 1) | (self.CommandType << 2) | (self.AlarmFrame << 6) | (self.EncriptedAES << 7)
|
|
return val
|