idiBus
/
IdiBusAgent
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

first commit

Browse Source
main
MichaelK 5 days ago
commit 28a2b34b28
12 changed files with 690 additions and 0 deletions
Show Stats Download Patch File Download Diff File

1
.gitignore vendored
Unescape View File

@ -0,0 +1 @@
./__pycache__

10
CRC16.py
Unescape View File

@ -0,0 +1,10 @@
def modbus_crc16(data: bytearray) -> bytearray:
crc = 0xFFFF
for byte in data:
crc ^= byte
for _ in range(8):
if crc & 0x0001:
crc = (crc >> 1) ^ 0xA001
else:
crc >>= 1
return bytearray([crc >> 8,crc & 0xFF])

50
Executioner.py
Unescape View File

@ -0,0 +1,50 @@
import sys
from time import sleep, time
from source.IdiBus4DC import Module4DC
from source.IdiBusSerialLine import IdiBusSerialLine
class Executioner():
def __init__(self, name):
self.name = name
self.port = ""
self.address = 0
self.speed = 0
def execute(self, command, args):
pass
# Must return string with all data in csv format. without \n or \r
def get_log_string(self):
pass
class Exec4DC(Executioner):
def __init__(self, ):
super().__init__("Module4DC")
def execute(self, command, args):
module = Module4DC(addr=self.address, line=IdiBusSerialLine(port=self.port, baudrate=self.speed))
if command == "set-data":
module.set_channel_data(args[0], args[1], args[2])
elif command == "set-state":
module.set_channel_state(args[0], args[1])
else:
print("Unknown command: " + command)
def get_log_string(self):
module = Module4DC(addr=self.address, line=IdiBusSerialLine(port=self.port, baudrate=self.speed))
str = ""
for i in range(4):
data = module.set_channel_data(i)
if data == None:
return None
else:
str += f"{data["voltage"]},{data["current"]},"
class ExecSystem(Executioner):
def __init__(self):
super().__init__("Module4DC")
def execute(self, command, args):
if command == "sleep":
sleep(args[0])
elif command == "exit":
sys.exit(0)
else:
print("Unknown command: " + command)
def get_log_string(self):
return "System is ok. :P"

103
IdiBusDefs.py
Unescape View File

@ -0,0 +1,103 @@
from enum import Enum
from typing import List, Dict
class comFuncList(Enum):
IDIMMES_COM_C_Init = 220
IDIMMES_COM_C_ShtDown = 221
IDIMMES_COM_C_Freeze = 222
IDIMMES_COM_C_Resume = 223
IDIMMES_COM_C_Dummy = 224
IDIMMES_COM_C_AssignGroup = 225
IDIMMES_COM_C_SetAlarmL12 = 226
IDIMMES_COM_C_SetAlarmL = 227
IDIMMES_COM_C_Virtual = 228
IDIMMES_COM_C_SyncReadChnl = 229
IDIMMES_COM_C_SyncRead = 230
IDIMMES_COM_C_SyncDoChnl = 231
IDIMMES_COM_C_SyncDo = 232
IDIMMES_COM_C_SyncClear = 233
IDIMMES_COM_C_BurstReadCnt = 234
IDIMMES_COM_C_BurstReadTime = 235
IDIMMES_COM_C_SendTimeDate = 236
IDIMMES_COM_C_MkTimedMaster = 237
IDIMMES_COM_C_EnterBootloader = 238
#Skipped for historical reasons
IDIMMES_COM_C_ReadDevFullSN_MS = 241
IDIMMES_COM_C_WriteSnIPv4IPv6 = 242
IDIMMES_COM_C_WriteSnVerifyDates = 243
IDIMMES_COM_C_WriteSnAES256 = 244
IDIMMES_COM_C_SendLongMessage = 245
IDIMMES_COM_C_GetLondMessage = 246
IDIMMES_COM_C_DummyModule = 247
IDIMMES_COM_C_CheckModuleLongOp = 248
IDIMMES_COM_C_CheckChannelLongOp = 249
IDIMMES_COM_C_FmwBootloaderInfo = 250 # Ask for bootloader version
IDIMMES_COM_C_FmwBootloaderStart = 251 # Start upload and recieve fwinfo
IDIMMES_COM_C_FmwBootloaderWrite = 252 # Write chunk
IDIMMES_COM_C_FmwBootloaderEnd = 253 # upload end. Finish operations and reboot
IDIMMES_COM_C_GotoApp = 254 #Goto APP
@classmethod
def values(cls) -> List[int]:
"""Список всех значений"""
return [status.value for status in cls]
@classmethod
def names(cls) -> List[str]:
"""Список всех имен"""
return [status.name for status in cls]
@classmethod
def dict(cls) -> Dict[str, int]:
"""Словарь имя -> значение"""
return {status.name: status.value for status in cls}
@classmethod
def get_by_value(cls, value: int):
"""Получить статус по значению"""
for status in cls:
if status.value == value:
return status
return None
'''
moduleTypeList = {
"TYPE_TST" : "ATmega328PB",
"TYPE_1W1" : "ATmega328PB",
"TYPE_TTT" : "ATmega2560",
"TYPE_2ST" : "ATmega2560",
"TYPE_4DC" : "ATmega2560",
"TYPE_RM8" : "ATmega2560",
}
'''
class IdiBusSpeedCodes(Enum):
CODE_19200B = 19200,
CODE_500K = 576000,
CODE_2400B = 2400,
CODE_9600B = 9600,
CODE_115200B = 115200,
CODE_250K = 256000,
CODE_1M = 921600,
CODE_10M = 10000000
@classmethod
def values(cls) -> List[int]:
"""Список всех значений"""
return [status.value for status in cls]
@classmethod
def names(cls) -> List[int]:
"""Список всех имен"""
return [status.name for status in cls]
@classmethod
def dict(cls) -> Dict[str, int]:
"""Словарь имя -> значение"""
return {status.name: status.value for status in cls}
@classmethod
def get_by_value(cls, value: str):
"""Получить статус по значению"""
for status in cls:
if status.value == value:
return status
return None

118
IdiBusMaster.py
Unescape View File

@ -0,0 +1,118 @@
from .IdiBusDefs import comFuncList
from .IdiBusParse import IdiBusSMESParser
from .IdiBusSerialLine import IdiBusSerialLine
from .IdiBusMessage import *
class IdiBusModule():
def __init__(self, addr, line:IdiBusSerialLine):
self.addr = addr
self.last_mmes_err = 0
if not isinstance(line, IdiBusSerialLine):
raise TypeError("Line is not an IdiBusSerialLine")
self.line = line
def startLongOp(self):
self.line.timeout = 3
def endLongOp(self):
self.line.timeout = self.line.defaultTimeout
def c_Init(self):
return self._sendMMESG(comFuncList.IDIMMES_COM_C_Init.value, await_bytes=5)
def c_Shutdown(self):
return self._sendMMESG(comFuncList.IDIMMES_COM_C_ShtDown.value, await_bytes=5)
def c_Dummy(self):
return self._sendMMESG(comFuncList.IDIMMES_COM_C_Dummy.value, await_bytes=5)
def c_GotoApp(self):
return self._sendMMESG(comFuncList.IDIMMES_COM_C_GotoApp.value, await_bytes=5)
def c_FmwBootloaderInfo(self):
return self._sendMMESG(comFuncList.IDIMMES_COM_C_FmwBootloaderInfo.value, )
def c_EnterBootloader(self, moduleType, hwRev, SN):
moduleType = bytearray(moduleType, 'ascii')
hwRev = bytearray(hwRev, 'ascii')
#hwRev[0]-=48
#hwRev[1]-=48
SN = bytearray(SN, 'ascii')
devdata =bytearray()
devdata.extend(moduleType)
devdata.extend(hwRev)
devdata.extend(SN)
return self._sendMMESG(comFuncList.IDIMMES_COM_C_EnterBootloader.value, data=devdata, await_bytes=5)
def c_ReadDevFullSN_MS(self):
return self._sendMMESG(comFuncList.IDIMMES_COM_C_ReadDevFullSN_MS.value)
def c_BootloaderStart(self, data):
self.startLongOp()
rcv = self._sendMMESG(comFuncList.IDIMMES_COM_C_FmwBootloaderStart.value, data=data, await_bytes=5)
self.endLongOp()
return rcv
def c_BootloaderWrite(self, data):
self.startLongOp()
rcv = self._sendMMESG(comFuncList.IDIMMES_COM_C_FmwBootloaderWrite.value, data=data, await_bytes=5)
self.endLongOp()
return rcv
def c_BootloaderEnd(self):
self.startLongOp()
rcv = self._sendMMESG(comFuncList.IDIMMES_COM_C_FmwBootloaderEnd.value, await_bytes=5)
self.endLongOp()
return rcv
def _sendMMESG(self, func, data=None, await_bytes = 0):
if not func in comFuncList.values():
self.last_mmes_err = "BadFunc"
return 1
init_com = IdiBusMMESG(
self.addr,
IdiBusMMPS(IdiBusMMPS.typeMMESG).getValue(),
func,
)
#Append data if present
if not data is None:
init_com.addData(data)
self.line.writeData(init_com.getMsg())
#Set max packet size if not specified
if await_bytes == 0:
await_bytes = self.line.maxPacketSize
#read all data
SMES = self.line.readData(await_bytes)
#No response from slave
if len(SMES) == 0:
self.last_mmes_err = "NoResponse"
return 1
if len(SMES) < 5 or len(SMES) > 270:
self.last_mmes_err = "InvalidPacketSize"
return 1
else:
self.parser = IdiBusSMESParser(SMES)
if self.parser.validate() == -1:
self.last_mmes_err = "Corrupted"
return 1
else:
#Packet good
return 0
def _sendMMES(self,msg:IdiBusMMES, await_bytes=0 ):
#send it
#print(f"raw message : {msg.getMsg().hex()}")
self.line.writeData(msg.getMsg())
# read all data
SMES = self.line.readData(await_bytes)
#print(f"raw answer : {SMES.hex()}")
# No response from slave
if len(SMES) == 0:
self.last_mmes_err = "NoResponse"
return 1
if len(SMES) < 5 or len(SMES) > 270:
self.last_mmes_err = "InvalidPacketSize"
return 1
else:
self.parser = IdiBusSMESParser(SMES)
if self.parser.validate() == -1:
self.last_mmes_err = "Corrupted"
return 1
else:
# Packet good
self.last_mmes_err = self.parser.getErrcode()
return 0
class IdiBusDevice():
def __init__(self, num):
self.num = num
self.errors = {}
self.commands = {}

86
IdiBusMessage.py
Unescape View File

@ -0,0 +1,86 @@
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

113
IdiBusParse.py
Unescape View File

@ -0,0 +1,113 @@
import struct
from .CRC16 import modbus_crc16
def get_bit(byte, position):
return (byte >> position) & 1
class IdiBusMES:
def __init__(self, data: bytearray):
if len(data) == 0:
raise TypeError("Data is an empty bytearray")
self.data = data
def validate(self):
if self.getCRC() != modbus_crc16(self.data[0:-2]):
return -1
else:
return 0
def getAddress(self):
return int(self.data[0])
def getCRC(self):
return self.data[-2::]
def __repr__(self):
return f"Addr:{self.getAddress()}, Body:{self.data[1::]}"
class IdiBusMMESParser(IdiBusMES):
def __init__(self, data: bytearray):
super().__init__(data)
def getMMPS(self):
return self.data[1]
def getLongMes(self):
return get_bit(int(self.data[1]), 0)
def getMesType(self):
return get_bit(int(self.data[1]), 1)
def getAlarm(self):
return get_bit(int(self.data[1]), 6)
def getEncStatus(self):
return get_bit(int(self.data[1]), 7)
def getDEV(self):
return int(self.data[2])
def getCHAN(self):
return int(self.data[3])
def getCMD(self):
return int(self.data[4])
def getData(self):
return self.data[4:-2]
class IdiBusMMESGParser(IdiBusMES):
def __init__(self, data: bytearray):
super().__init__(data)
def getMMPS(self):
return int(self.data[1])
def getCMD(self):
return int(self.data[2])
def getData(self):
return self.data[3:-2]
class IdiBusSMESParser(IdiBusMES):
def __init__(self, data: bytearray):
super().__init__(data)
def getSPDU(self):
return int(self.data[1])
def getErrorBit(self):
return get_bit(self.data[1], 0)
def getLongMessage(self):
return get_bit(self.data[1], 1)
def getErrcode(self):
return int(self.data[2])
def getData(self):
return self.data[3:-2]
def getDataStruct(self, format_string):
try:
unpacked = struct.unpack(format_string, self.getData())
return unpacked
except struct.error as e:
print(f"Parse error: {e}")
return None
except UnicodeDecodeError as e:
print(f"Parse error: {e}")
return None
class IdiBusDeviceInfoParser():
# Формат: 'B H I f 10s'
# B - unsigned char (1 байт)
# H - unsigned short (2 байта)
# I - unsigned int (4 байта)
# f - float (4 байта)
# 10s - строка из 10 байт
format_string = '<B 3s 6s 3s 2B 7s 6s 2B I'
def __init__(self, data: bytearray):
if len(data) == 0:
raise TypeError("Data is an empty bytearray")
self.data = data
#print(data[-4::])
def getStruct(self):
try:
unpacked = struct.unpack(self.format_string, self.data)
packed = {
'Padding': unpacked[0],
'GS1_country': unpacked[1].decode('utf-8'),
'GS1_company': unpacked[2].decode('utf-8'),
'ModuleType': unpacked[3].decode('utf-8'),
'HW_revison': str(unpacked[4]) +"." + str(unpacked[5]),
'SN': unpacked[6].decode('utf-8'),
'MAC': unpacked[7].decode('utf-8'),
'SW': str(unpacked[8]) + "." + str(unpacked[9]),
'AppCRC': hex(unpacked[10]),
}
return packed
except struct.error as e:
print(f"Parse error: {e}")
return None
except UnicodeDecodeError as e:
print(f"Parse error: {e}")
return None

50
IdiBusSerialLine.py
Unescape View File

@ -0,0 +1,50 @@
import serial
from serial import STOPBITS_TWO
class IdiBusSerialLine(serial.Serial):
def __init__(self, port="", baudrate=0, **kwargs):
if not port or not baudrate:
pass
else:
self.maxPacketSize = 256 + 2 + 5 # From defines in Slave
self.defaultTimeout = 0.025
super().__init__(
port=port,
baudrate=baudrate,
timeout=kwargs.get('timeout', self.defaultTimeout),
parity=kwargs.get('parity', serial.PARITY_NONE),
stopbits=kwargs.get('stopbits', STOPBITS_TWO),
**{k: v for k, v in kwargs.items()
if k not in ['timeout', 'parity', 'stopbits']}
)
#print(f"Serial port opened at {self.port}")
def __del__(self):
if hasattr(self, 'is_open') and self.is_open:
self.close()
#print(f"Serial port closed at {self.port}")
def writeData(self, msg):
"""Отправка данных"""
return self.write(msg) # Возвращаем количество байт
def readData(self, await_bytes=0):
"""Чтение данных"""
if await_bytes == 0:
await_bytes = self.maxPacketSize
return self.read(await_bytes)
# Дополнительные полезные методы
def write_read(self, msg, response_bytes=0):
"""Отправить и прочитать ответ"""
self.writeData(msg)
return self.readData(response_bytes)
def flush_buffers(self):
"""Очистить буферы"""
self.reset_input_buffer()
self.reset_output_buffer()

30
IdiBusSlave.py
Unescape View File

@ -0,0 +1,30 @@
from .IdiBusSerialLine import *
from .IdiBusParse import IdiBusMMESParser, IdiBusMMESGParser
SelfAddress = 254
#Emulates slave module to communicate with master
class IdiBusSlave():
def __init__(self,address,line:IdiBusSerialLine):
self.address = address
self.line = line
def await_message(self, retries):
addr = 0
for i in range(0,retries):
msg = self.line.readData(self.line.maxPacketSize)
# No response
if len(msg) < 3:
continue
parser = IdiBusMMESParser(msg)
# Corrupted
if parser.validate() == -1:
continue
# This one is not for this Slave
if parser.getAddress() != self.address:
continue
# Got a correct message
if parser.getMesType() == 1:
parser = IdiBusMMESGParser(msg)
return parser
return None

75
IdiBusUtils.py
Unescape View File

@ -0,0 +1,75 @@
from .IdiBusMaster import IdiBusSerialLine, IdiBusModule, IdiBusSMES
from .IdiBusSlave import IdiBusSlave
from .IdiBusDefs import comFuncList
class IdiBusUtil():
def __init__(self, line):
if not isinstance(line, IdiBusSerialLine):
raise TypeError("Line is not an IdiBusSerialLine")
self.line = line
def __del__(self):
self.line.close()
#Returns list with adresses of all active devices
def releaseLine(self):
self.line.close()
def scanBus(self,addr):
Dev = IdiBusModule(addr= addr, line=self.line)
if Dev.c_Init() == "OK":
return 1
else:
return 0
def scanBus(self,start, end):
addresses = []
for i in range(start,end):
Dev = IdiBusModule(addr= i, line=self.line)
if Dev.c_Init() == "OK":
addresses.append(i)
return addresses
def shutdownMaster(self):
#print(f"Trying to shutdown Master")
SelfAddress = 254
self.line.timeout = 0.01
SpareMaster = IdiBusSlave(SelfAddress, self.line)
self.line.flush_buffers()
msg = SpareMaster.await_message(10)
#print(msg)
if msg == None:
return -1
if comFuncList.IDIMMES_COM_C_Init.value == int(msg.getCMD()):
#print("Got c_Init")
resp = IdiBusSMES(SelfAddress, 2, 0)
SpareMaster.line.writeData(resp.getMsg())
msg = SpareMaster.await_message(1)
try:
if len(msg.data) == 0:
return -1
except Exception as e:
print(e)
return -1
#print(f'{msg.data}')
if msg.getMMPS() != 4:
#print(f"Got wrong mmps {msg.getMMPS()} != 4")
return -1
#print("Got order request")
resp = IdiBusSMES(SelfAddress, 2, 0)
a = bytearray()
a.append(5)
resp.addData(a)
msg = resp.getMsg()
SpareMaster.line.writeData(msg)
msg = SpareMaster.await_message(1)
try:
if len(msg.data) == 0:
return -1
except Exception as e:
print(e)
return -1
#print(f'{msg.data}')
if msg.getMMPS() != 8:
return -1
#print("Got order ans")
return 0
def recoverMaster(self):
print("Recover Master")

0
__init__.py
Unescape View File

54
atmega_mcu.py
Unescape View File

@ -0,0 +1,54 @@
class AtmegaMcu():
# Словарь с ВСЕМИ МК (сделаем его атрибутом класса)
ATMEGA_MCUS = {}
def __init__(self, name, BlockSize, FlashSize, BootSize, SpecialInfoSizeInPages):
self.name = name
self.BlockSize = BlockSize
self.FlashSize = FlashSize
self.BootSize = BootSize
self.SpecialInfoSizeInPages = SpecialInfoSizeInPages
# Автоматически добавляем в словарь при создании
AtmegaMcu.ATMEGA_MCUS[name] = self
@classmethod
def get_all_names(cls):
"""Получить список всех имен МК"""
return list(cls.ATMEGA_MCUS.keys())
@classmethod
def get_by_name(cls, name):
"""Получить МК по имени"""
return cls.ATMEGA_MCUS.get(name)
@classmethod
def get_all(cls):
"""Получить все МК"""
return list(cls.ATMEGA_MCUS.values())
# Теперь создаем МК - они автоматически добавятся в словарь
ATmega328PB = AtmegaMcu(
"ATmega328PB",
128,
32 * 1024,
4 * 1024,
1,
)
ATmega1280 = AtmegaMcu(
"ATmega1280",
256,
128 * 1024,
8 * 1024,
1,
)
ATmega2560 = AtmegaMcu(
"ATmega2560",
256,
256 * 1024,
8 * 1024,
1,
)
Write Preview
Loading…
Cancel
Save