mirror of
https://port.numenaute.org/aleajactaest/clientbot.git
synced 2024-11-22 07:06:14 +00:00
493 lines
18 KiB
Python
493 lines
18 KiB
Python
#!/usr/bin/python3
|
|
# -*- coding: utf-8 -*-
|
|
#
|
|
# module DecodeDatabase
|
|
#
|
|
# Copyright (C) 2019 AleaJactaEst
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU General Public License as published by
|
|
# the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
import logging
|
|
from tools import getPowerOf2
|
|
from tools import BitStream
|
|
|
|
LOGGER='DecodeDatabase'
|
|
|
|
def show_dico(dico, level=1):
|
|
try:
|
|
for ele in dico:
|
|
if isinstance(dico[ele], dict):
|
|
logging.getLogger(LOGGER).debug("%s %s %s" % ("." * level, ele , ":"))
|
|
if isinstance(dico[ele], dict):
|
|
show_dico(dico[ele], level+1)
|
|
else:
|
|
logging.getLogger(LOGGER).debug("%s %s %s" % ("." * level, ele, ':', dico[ele]))
|
|
except:
|
|
logging.getLogger(LOGGER).debug("empty")
|
|
|
|
def child(ele):
|
|
ret = {}
|
|
ret_branch = {}
|
|
ref_other = {}
|
|
id_branch = 0
|
|
id_other = 0
|
|
min_i = -1
|
|
max_i = -1
|
|
for k in ele.keys():
|
|
ret[k] = ele.get(k)
|
|
logging.getLogger(LOGGER).debug("%s %s" % (str(k), str(ele.get(k))))
|
|
for _child in list(ele):
|
|
x = child(_child)
|
|
if x['name'] == 'branch':
|
|
ret_branch.setdefault(id_branch, x)
|
|
id_branch += 1
|
|
else:
|
|
ref_other.setdefault(id_other, x)
|
|
id_other += 1
|
|
if ret_branch or ref_other:
|
|
ret['child'] = {}
|
|
id = 0
|
|
for x in ret_branch:
|
|
min_i = max_i + 1
|
|
max_i = min_i
|
|
#show_dico( ret_branch[x])
|
|
if 'count' in ret_branch[x]:
|
|
max_i = min_i + int(ret_branch[x]['count']) - 1
|
|
ret_branch[x]['min'] = min_i
|
|
ret_branch[x]['max'] = max_i
|
|
ret['child'].setdefault(id, ret_branch[x])
|
|
id += 1
|
|
for x in ref_other:
|
|
min_i = max_i + 1
|
|
max_i = min_i
|
|
#show_dico( ref_other[x])
|
|
if 'count' in ref_other[x]:
|
|
max_i = min_i + int(ref_other[x]['count']) - 1
|
|
ref_other[x]['min'] = min_i
|
|
ref_other[x]['max'] = max_i
|
|
ret['child'].setdefault(id, ref_other[x])
|
|
id += 1
|
|
return ret
|
|
|
|
def count_elements(head):
|
|
#print(head)
|
|
if 'child' in head.keys():
|
|
list_ele = [x for x in list(head['child'].keys()) if isinstance(x, int)]
|
|
else:
|
|
list_ele = [x for x in list(head.keys()) if isinstance(x, int)]
|
|
logging.getLogger(LOGGER).debug(str(list_ele))
|
|
last_key = max(list_ele)
|
|
logging.getLogger(LOGGER).debug(str(last_key ))
|
|
try:
|
|
last_key = max(list_ele)
|
|
if 'child' in head.keys():
|
|
return head['child'][last_key]['max'] + 1
|
|
else:
|
|
return head[last_key]['max'] + 1
|
|
except TypeError:
|
|
return 0
|
|
|
|
def get_element(head, id):
|
|
logging.getLogger(LOGGER).debug("id: %s" % str(id))
|
|
for ele in head['child']:
|
|
logging.getLogger(LOGGER).debug("ele:%s" % str(ele))
|
|
logging.getLogger(LOGGER).debug("head:%s" % str(head))
|
|
logging.getLogger(LOGGER).debug("max:%s", str(head['child'][ele]['max']))
|
|
logging.getLogger(LOGGER).debug("min:%s", str(head['child'][ele]['min']))
|
|
if id <= head['child'][ele]['max'] and id >= head['child'][ele]['min']:
|
|
return head['child'][ele]
|
|
return None
|
|
|
|
|
|
class LeafDatabase():
|
|
def __init__(self):
|
|
self.name = ""
|
|
self.type = ""
|
|
self.count = None
|
|
|
|
def loadXml(self, xmldata):
|
|
self.name = xmldata.get('name')
|
|
self.type = xmldata.get('type')
|
|
if xmldata.get('count'):
|
|
self.count = int(xmldata.get('count'))
|
|
|
|
def countLeaves(self):
|
|
if self.count:
|
|
logging.getLogger(LOGGER).debug("countLeaves leaf %s (nb:%s)" % (self.name, str(self.count)))
|
|
return self.count
|
|
else:
|
|
logging.getLogger(LOGGER).debug("countLeaves leaf %s (nb:1)" % (self.name))
|
|
return 1
|
|
|
|
def show(self, level=1):
|
|
logging.getLogger(LOGGER).debug("%s %s Leaf %s : %s : %s" % (" " * level, str(level), self.name, str(self.count), str(self.type)))
|
|
|
|
def execute(self, msgin, name=""):
|
|
if name:
|
|
tmp = name
|
|
else:
|
|
tmp = self.name
|
|
if self.type[0] == 'I': # Unsigned
|
|
logging.getLogger(LOGGER).debug("Read:" + str (self.type))
|
|
value = int(self.type[1:])
|
|
_ = msgin.readSerialUint64(value, name=tmp, typeName=self.type)
|
|
elif self.type[0] == 'S': # Signed
|
|
logging.getLogger(LOGGER).debug("Read:" + str (self.type))
|
|
value = int(self.type[1:])
|
|
# _ = msgin.readNbChar(value, name='DatabaseXML' + tmp)
|
|
_ = msgin.readSerialUint64(value, name=tmp)
|
|
elif self.type == 'TEXT':
|
|
logging.getLogger(LOGGER).debug("type:" + str (self.type))
|
|
value = 32
|
|
_ = msgin.readSerialSint64(value, name=idname)
|
|
else:
|
|
logging.getLogger(LOGGER).error("Type inconnu:" + str (self.type))
|
|
raise "type not managed"
|
|
|
|
def execute_atom(self, level, pos, msgin, name=""):
|
|
if name:
|
|
extraName = name + "/" + self.name
|
|
else:
|
|
extraName = self.name
|
|
if self.count:
|
|
step = self.count
|
|
else:
|
|
step = 1
|
|
for step in range(0, step):
|
|
if self.count:
|
|
idname = extraName + str(step)
|
|
else:
|
|
idname = extraName
|
|
logging.getLogger(LOGGER).debug('step:' + str(step) + ' level:'+ str(level) + ' pos:' + str(pos) + ' idname:' + idname)
|
|
if level < pos:
|
|
level += 1
|
|
continue
|
|
if self.type[0] == 'I':
|
|
value = int(self.type[1:])
|
|
if value > 64:
|
|
raise "type not managed"
|
|
logging.getLogger(LOGGER).debug('value:' + str(value))
|
|
_ = msgin.readSerialUint64(value, name=idname, typeName=self.type)
|
|
return level+1
|
|
elif self.type[0] == 'S':
|
|
logging.getLogger(LOGGER).debug("type:" + str (self.type))
|
|
value = int(self.type[1:])
|
|
_ = msgin.readSerialSint64(value, name=idname)
|
|
elif self.type == 'TEXT':
|
|
logging.getLogger(LOGGER).debug("type:" + str (self.type))
|
|
value = 32
|
|
_ = msgin.readSerialUint64(value, name=idname)
|
|
else:
|
|
logging.getLogger(LOGGER).debug("Type inconnu:", self.type)
|
|
raise "type not managed"
|
|
return level
|
|
|
|
|
|
class BranchDatabase():
|
|
def __init__(self):
|
|
self.name = None
|
|
self.branch = []
|
|
self.leaf = []
|
|
self.min = None
|
|
self.max = None
|
|
self.atom = False
|
|
self.count = None
|
|
|
|
def loadXml(self, xmldata, filter=None):
|
|
# print("xmldata:", xmldata)
|
|
# print("keys:", xmldata.keys())
|
|
# print("filter:", filter)
|
|
#print("bank:", xmldata['bank'])
|
|
if filter:
|
|
if 'bank' in xmldata:
|
|
if filter != xmldata['bank']:
|
|
return
|
|
else:
|
|
return
|
|
if xmldata.get('atom'):
|
|
self.atom = True
|
|
if xmldata.get('count'):
|
|
self.count = int(xmldata.get('count'))
|
|
self.name = xmldata.get('name')
|
|
# print(self.name)
|
|
for ele in xmldata:
|
|
if ele.tag == 'branch':
|
|
newbranch = BranchDatabase()
|
|
newbranch.loadXml(ele)
|
|
self.branch.append(newbranch)
|
|
elif ele.tag == 'leaf':
|
|
newleaf = LeafDatabase()
|
|
newleaf.loadXml(ele)
|
|
self.leaf.append(newleaf)
|
|
|
|
def loadRootXml(self, xmldata, filter=None):
|
|
for ele in xmldata:
|
|
if ele.tag == 'branch':
|
|
if filter:
|
|
if ele.get('bank') == filter:
|
|
# print(ele.get('bank'), filter)
|
|
newbranch = BranchDatabase()
|
|
newbranch.loadXml(ele)
|
|
self.branch.append(newbranch)
|
|
|
|
def getIdBits(self):
|
|
count = 0
|
|
for ele in self.branch:
|
|
if ele.count:
|
|
count += ele.count
|
|
else:
|
|
count += 1
|
|
for ele in self.leaf:
|
|
if ele.count:
|
|
count += ele.count
|
|
else:
|
|
count += 1
|
|
return count
|
|
|
|
def countLeaves(self):
|
|
count = 0
|
|
for ele in self.branch:
|
|
count += ele.countLeaves()
|
|
for ele in self.leaf:
|
|
count += ele.countLeaves()
|
|
if self.count and self.atom:
|
|
# logging.getLogger(LOGGER).debug("countLeaves branch <- %s (nb:%s)" % (self.name, str(self.count)))
|
|
count *= self.count
|
|
return count
|
|
|
|
def execute_atom_found(self, level, pos, msgin, name=""):
|
|
if self.count:
|
|
step = self.count
|
|
else:
|
|
step=1
|
|
if name:
|
|
extraName = name + self.name
|
|
else:
|
|
extraName = self.name
|
|
for step in range(0, step):
|
|
logging.getLogger(LOGGER).debug('step:' + str(step)+ ' level:' + str(level) + ' pos:' + str(pos) + ' name:' + name)
|
|
if self.count:
|
|
idname = extraName + str(step)
|
|
else:
|
|
idname = extraName
|
|
for ele in self.branch:
|
|
level = ele.execute_atom_found(level, pos, msgin, idname)
|
|
logging.getLogger(LOGGER).debug('step:' + str(step)+ ' level:' + str(level) + ' pos:' + str(pos) + ' name:' + name + ' idname:' + str(idname))
|
|
if level > pos:
|
|
return level
|
|
for ele in self.leaf:
|
|
level = ele.execute_atom(level, pos, msgin, idname)
|
|
logging.getLogger(LOGGER).debug('step:' + str(step)+ ' level:' + str(level) + ' pos:' + str(pos) + ' name:' + name + ' idname:' + str(idname))
|
|
if level > pos:
|
|
return level
|
|
return level
|
|
|
|
def show(self, level=0, pos=0, filterlevel=None):
|
|
logging.getLogger(LOGGER).debug( "%s %s pos:%s Branch:%s : %s : %s - %s" % (" " * level, str(level), str(pos), str(self.name), str(self.count), str(self.atom), str(self.getIdBits())))
|
|
if filterlevel is not None:
|
|
if filterlevel <= level:
|
|
return
|
|
i = 0
|
|
for ele in self.branch:
|
|
ele.show(level + 1, i, filterlevel)
|
|
i += 1
|
|
for ele in self.leaf:
|
|
ele.show(level + 1)
|
|
|
|
def execute_atom(self, msgin, parent):
|
|
nbchild = self.countLeaves()
|
|
# nbchild = len(self.leaf) + len(self.branch)
|
|
#nbBit = getPowerOf2.getPowerOf2_Bis(nbchild)
|
|
|
|
logging.getLogger(LOGGER).debug("needRead:" + str(msgin.needRead()) + " nbchild:" + str(nbchild))
|
|
logging.getLogger(LOGGER).debug(msgin.showAllData())
|
|
if self.name:
|
|
idname = parent + self.name +'/'
|
|
else:
|
|
idname = parent
|
|
cBitSet = msgin.readCBitSet(nbchild, idname + ":Param", typeName = 'I' + str(nbchild))
|
|
#cBitSet = CBitSet.CBitSet()
|
|
#cBitSet.readSerialExtra(msgin, nbchild, self.name + "/Param")
|
|
logging.getLogger(LOGGER).debug(msgin.showAllData())
|
|
todelete_count_true = 0
|
|
for i in range(0, nbchild):
|
|
#ii = nbchild - i
|
|
logging.getLogger(LOGGER).debug(str(i) + " - " + str(cBitSet.get(i)))
|
|
if cBitSet.get(i):
|
|
todelete_count_true += 1
|
|
_ = self.execute_atom_found(0, i, msgin, parent)
|
|
if todelete_count_true > 1:
|
|
logging.getLogger(LOGGER).debug(msgin.showAllData())
|
|
#raise "A controler"
|
|
logging.getLogger(LOGGER).debug(msgin.showAllDataRaw())
|
|
logging.getLogger(LOGGER).debug(msgin.showAllData())
|
|
#raise "A faire"
|
|
#if self.name == "GROUP":
|
|
# raise "A controler"
|
|
|
|
def execute_normal(self, msgin, parent):
|
|
nbchild = self.getIdBits()
|
|
#nbBit = getPowerOf2.getPowerOf2_Bis(nbchild)
|
|
nbBit = getPowerOf2.getPowerOf2_ter(nbchild)
|
|
if nbBit > msgin.needRead() :
|
|
return
|
|
logging.getLogger(LOGGER).debug("needRead:" + str(msgin.needRead()) + " nbBit:" + str(nbBit))
|
|
id = msgin.readSerial(nbBit, name=parent, typeName='Number:'+str(nbBit), emulate=True)
|
|
i = 0
|
|
ii = 0
|
|
for ele in self.branch:
|
|
#logging.getLogger(LOGGER).debug(str(i) + " id:" + str(id) + " name:" + str(ele.name))
|
|
if ele.count:
|
|
i += ele.count
|
|
else:
|
|
i += 1
|
|
if i > id:
|
|
comment = ""
|
|
if ele.count:
|
|
pos = id - ii
|
|
comment = str(pos)
|
|
logging.getLogger(LOGGER).debug(str(i) + " id:" + str(id) + " name:" + str(ele.name))
|
|
if self.name:
|
|
idnameshort = parent + self.name + '/'
|
|
else:
|
|
idnameshort = parent
|
|
idname = idnameshort + ele.name
|
|
_= msgin.readSerial(nbBit, name=idname, typeName='Number:'+str(nbBit), emulate=False, commentValue=ele.name+comment)
|
|
logging.getLogger(LOGGER).debug("name:" + ele.name + ", count:" + str(ele.count) + ", atom:" + str(ele.atom))
|
|
ele.execute(msgin, idnameshort)
|
|
return
|
|
ii = i
|
|
for ele in self.leaf:
|
|
logging.getLogger(LOGGER).debug(str(i) + " id:" + str(id) + " name:" + str(ele.name))
|
|
if ele.count:
|
|
i += ele.count
|
|
else:
|
|
i += 1
|
|
if i > id:
|
|
comment = ""
|
|
if ele.count:
|
|
pos = id - ii
|
|
comment = str(pos)
|
|
if self.name:
|
|
idnameshort = parent + self.name + '/'
|
|
else:
|
|
idnameshort = parent
|
|
idname = idnameshort + ele.name + comment
|
|
logging.getLogger(LOGGER).debug(str(i) + " id:" + str(id) + " name:" + str(ele.name))
|
|
_ = msgin.readSerial(nbBit, name= idname, typeName='Number:'+str(nbBit), emulate=False, commentValue=ele.name+comment)
|
|
logging.getLogger(LOGGER).debug("name:" + ele.name + ", count:" + str(ele.count))
|
|
ele.execute(msgin, name=idnameshort)
|
|
return
|
|
ii = i
|
|
|
|
def execute(self, msgin, parent='DatabaseXML/'):
|
|
if self.atom:
|
|
self.execute_atom(msgin, parent)
|
|
else:
|
|
self.execute_normal(msgin, parent)
|
|
|
|
def execute_root(self, msgin):
|
|
if self.atom:
|
|
nbchild = self.countLeaves()
|
|
#nbBit = getPowerOf2.getPowerOf2_Bis(nbchild)
|
|
nbBit = getPowerOf2.getPowerOf2(nbchild)
|
|
raise "A faire"
|
|
else:
|
|
nbchild = self.getIdBits()
|
|
#nbBit = getPowerOf2.getPowerOf2_Bis(nbchild)
|
|
nbBit = getPowerOf2.getPowerOf2(nbchild)
|
|
while msgin.needRead() > nbBit:
|
|
logging.getLogger(LOGGER).debug("needRead:" + str(msgin.needRead()) + ", nbBit:" + str(nbBit))
|
|
self.execute(msgin)
|
|
|
|
|
|
class DecodeDatabase():
|
|
def __init__(self):
|
|
self.databaseXml = None
|
|
self.databasePlr = None
|
|
|
|
def loadDatabase(self, databaseXml):
|
|
self.databasePlr = BranchDatabase()
|
|
self.databasePlr.loadRootXml(databaseXml, 'PLR')
|
|
|
|
def execute(self, msgin, world):
|
|
self.databasePlr.execute(msgin)
|
|
|
|
def loadDatabase2(self, databaseXml):
|
|
logging.getLogger(LOGGER).debug("loadDatabase")
|
|
base = BranchDatabase()
|
|
#base.loadXml(databaseXml, 'PLR')
|
|
base.loadRootXml(databaseXml, 'PLR')
|
|
base.show(filterlevel=2)
|
|
base.show()
|
|
tmp = BitStream.BitStream()
|
|
tmp.internalSerial(23, 5)
|
|
tmp.internalSerial(0,1)
|
|
tmp.internalSerial(13237038,32)
|
|
base.execute_root(tmp)
|
|
logging.getLogger(LOGGER).debug(tmp.showAllData())
|
|
tmp = BitStream.BitStream()
|
|
tmp.internalSerial(21, 5)
|
|
tmp.internalSerial(8,5)
|
|
tmp.internalSerial(2,3)
|
|
tmp.internalSerial(0,3)
|
|
tmp.internalSerial(0,5)
|
|
tmp.internalSerial(3,9)
|
|
tmp.internalSerial(69,9)
|
|
tmp.internalSerial(10,5)
|
|
tmp.internalSerial(10,5)
|
|
tmp.internalSerial(0,5)
|
|
tmp.internalSerial(0,9)
|
|
tmp.internalSerial(200,9)
|
|
tmp.internalSerial(21,5)
|
|
tmp.internalSerial(10,5)
|
|
tmp.internalSerial(8,5)
|
|
tmp.internalSerial(0,1)
|
|
tmp.internalSerial(821933,32)
|
|
tmp.internalSerial(0,5)
|
|
tmp.internalSerial(0,9)
|
|
tmp.internalSerial(100,9)
|
|
tmp.internalSerial(14,5)
|
|
tmp.internalSerial(16,5)
|
|
tmp.internalSerial(4,5)
|
|
tmp.internalSerial(0,2)
|
|
tmp.internalSerial(15,8)
|
|
tmp.internalSerial(64,8)
|
|
tmp.internalSerial(0,8)
|
|
tmp.internalSerial(1,2)
|
|
base.execute_root(tmp)
|
|
logging.getLogger(LOGGER).debug(tmp.showAllData())
|
|
|
|
raise "Stop"
|
|
self.databaseXml = databaseXml
|
|
id = 0
|
|
self.databasePlr = {'name': 'root', 'bank': 'PLR', 'child': {}, 'min':0, 'max':0 , 'atom': False}
|
|
for ele in self.databaseXml:
|
|
if ele.tag == 'branch':
|
|
if ele.get('bank') == "PLR":
|
|
if 'atom' in ele:
|
|
if int(ele.get('atom')) > 0:
|
|
#self.databasePlr
|
|
pass
|
|
self.databasePlr['child'][id] = child(ele)
|
|
self.databasePlr['child'][id]['min'] = id
|
|
self.databasePlr['child'][id]['max'] = id
|
|
id += 1
|
|
#print(dir(ele))
|
|
nb = len(self.databasePlr) - 1
|
|
self.databasePlr['min'] = 0
|
|
self.databasePlr['max'] = nb
|
|
show_dico(self.databasePlr)
|
|
logging.getLogger(LOGGER).debug("max %d" % str(self.databasePlr['max']))
|
|
#raise "Decode"
|