aleajactaest/clientbot
1
0
Fork 0
mirror of https://port.numenaute.org/aleajactaest/clientbot.git synced 2024-09-19 11:10:34 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions
clientbot/tools/DecodeDatabase.py

532 lines
19 KiB
Python
Raw Blame History

#!/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):
print("." * level, ele , ":")
if isinstance(dico[ele], dict):
show_dico(dico[ele], level+1)
else:
print("." * level, ele, ':', dico[ele])
except:
print("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)
print(k, 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("%" * 80)
#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)]
print(list_ele)
last_key = max(list_ele)
print(last_key )
print("%" * 80)
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):
print("id:", id)
for ele in head['child']:
print("ele:", ele)
print("head:", head)
print("max:", head['child'][ele]['max'])
print("min:", 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:
return self.count
else:
return 1
def show(self, level=1):
print(" " * level, level, " Leaf ", self.name, ":", self.count, ":", self.type)
def execute(self, msgin, name=""):
if name:
tmp = "/" + name
else:
tmp = "/" + self.name
if self.type[0] == 'I':
logging.getLogger(LOGGER).debug("Read:" + str (self.type))
value = int(self.type[1:])
_ = msgin.readSerial64(value, name='DatabaseXML' + tmp, typeName=self.type)
else:
logging.getLogger(LOGGER).debug("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:])
_ = msgin.readSerial(value, name='DatabaseXML' + idname, typeName=self.type)
return level+1
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:
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):
print(" " * level, level, "pos:", pos, " Branch ", self.name, ":", self.count, ":", self.atom, 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):
nbchild = self.countLeaves()
#nbBit = getPowerOf2.getPowerOf2_Bis(nbchild)
logging.getLogger(LOGGER).debug("needRead:" + str(msgin.needRead()) + " nbchild:" + str(nbchild))
logging.getLogger(LOGGER).debug(msgin.showAllData())
cBitSet = msgin.readCBitSet(nbchild, self.name + ":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)
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"
def execute_normal(self, msgin):
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='DatabaseXML', 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:
idname = '/' + self.name + '/' + ele.name
else:
idname = '/' + ele.name
_= msgin.readSerial(nbBit, name='DatabaseXML' + 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)
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:
idname = '/' + self.name + '/' + ele.name + comment
else:
idname = '/' + ele.name + comment
logging.getLogger(LOGGER).debug(str(i) + " id:" + str(id) + " name:" + str(ele.name))
_ = msgin.readSerial(nbBit, name='DatabaseXML' + 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=idname)
return
ii = i
def execute(self, msgin):
if self.atom:
self.execute_atom(msgin)
else:
self.execute_normal(msgin)
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")
print("-"*80)
base = BranchDatabase()
print("-"*80)
#base.loadXml(databaseXml, 'PLR')
base.loadRootXml(databaseXml, 'PLR')
print("-"*80)
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)
print(tmp.showAllData())
print("-"*80)
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)
print(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
print("-" * 80)
show_dico(self.databasePlr)
print("-" * 80)
print("max", self.databasePlr['max'] )
print("-" * 80)
#raise "Decode"
def execute2(self, msgin, world):
logging.getLogger(LOGGER).debug("execute")
head = self.databasePlr
listpath = []
print("="*80)
print(head)
print("="*80)
while True:
print("---")
logging.getLogger(LOGGER).debug("count_elements:" + str(count_elements(head)))
nbchild = count_elements(head)
print("count_elements(head):", nbchild)
if nbchild == 0:
print("Ahhhh", "+"*80)
return True
print("nbchild:", nbchild)
nbBit = getPowerOf2.getPowerOf2_Bis(nbchild)
print("nbBit:", nbBit)
# res=1;
# ret=0;
# print("ret:", ret, "res:", res)
# while res<nbchild:
# ret += 1
# res *= 2
# print("ret:", ret, "res:", res)
# print("ret:", ret)
logging.getLogger(LOGGER).debug("nbBit:" + str(nbBit))
id = msgin.readSerial(nbBit, name='DatabaseXML', typeName='Number:'+str(nbBit), emulate=True)
print("id:", id)
logging.getLogger(LOGGER).debug("XML DECODE : %3d -> %s" % (nbBit, ':'.join(listpath)) )
print("Ahhhh a", "-"*80)
show_dico(head)
ele = get_element(head, int(id))
print(ele)
show_dico(ele)
print(msgin.showAllData())
name = ele['name']
id = msgin.readSerial(nbBit, name='DatabaseXML', typeName='Number:'+str(nbBit), commentValue=name)
listpath.append(name)
fullname = ':'.join(listpath)
print("fullname:", fullname)
logging.getLogger(LOGGER).debug(fullname)
if 'count' in ele:
nbBit = getPowerOf2.getPowerOf2_Bis(int(ele['count']))
count = msgin.readSerial(nbBit, name='DatabaseXML', typeName='count:'+str(ele['count']+':'+str(nbBit)))
listpath.append(str(count))
if 'type' in ele:
print("Ahhhh 2", "+"*80)
print(ele['type'])
if ele['type'] == 'I32':
_ = msgin.readSerial(32, name='DatabaseXML', typeName=ele['type'])
else:
pass
print(msgin.showAllData())
return True
head = ele
print("Ahhhh 3", "-"*80)
return False
Reference in a new issue View git blame Copy permalink