Skip to content
Snippets Groups Projects
Commit b558fc04 authored by Thomas Juerges's avatar Thomas Juerges
Browse files

Merge branch 'timestamp-logs' into 'master' 

Timestamp logs

See merge request !4
parents cfc2cc16 f88de4cc
Branches
Tags 2020-12-11-Holiday_Season_release
1 merge request!4Timestamp logs
Show whitespace changes
Inline Side-by-side
I2Cbitbang.py
+ 16
− 15
View file @ b558fc04
from I2Cdevices import *; from I2Cdevices import *;
import numpy as np import numpy as np
import logging
class I2Cbitbang(I2Cdevices): class I2Cbitbang(I2Cdevices):
def I2CSet(self,dev,reg,value,I2Ccallback,width=8,bitoffset=0): def I2CSet(self,dev,reg,value,I2Ccallback,width=8,bitoffset=0):
...@@ -10,7 +11,7 @@ class I2Cbitbang(I2Cdevices): ...@@ -10,7 +11,7 @@ class I2Cbitbang(I2Cdevices):
Addr=Dev['dev_address']; Addr=Dev['dev_address'];
# data2=[Addr<<1,regnr]+data; # data2=[Addr<<1,regnr]+data;
data2=[Addr<<1]+int2bytes(regnr)+value data2=[Addr<<1]+int2bytes(regnr)+value
# print(data2) # logging.info("%s",data2)
I2Ccallback('SDO',0) I2Ccallback('SDO',0)
I2Ccallback('SCL',1) I2Ccallback('SCL',1)
if not(I2CbbTX(data2,I2Ccallback)): return False; if not(I2CbbTX(data2,I2Ccallback)): return False;
...@@ -18,9 +19,9 @@ class I2Cbitbang(I2Cdevices): ...@@ -18,9 +19,9 @@ class I2Cbitbang(I2Cdevices):
data3=[0] data3=[0]
if not(I2CbbRX(data2,data3,I2Ccallback)): return False; if not(I2CbbRX(data2,data3,I2Ccallback)): return False;
if data3[0]!=0x80: if data3[0]!=0x80:
print("Expected Ack 0x80, received:",data3); logging.error("Expected Ack 0x80, received: %s",data3);
return False; return False;
print("Si4012 TX success!") logging.info("Si4012 TX success!")
return True; return True;
def I2CGet(self,dev,reg,value,I2Ccallback,width=8,bitoffset=0): def I2CGet(self,dev,reg,value,I2Ccallback,width=8,bitoffset=0):
SPI=Find(self.D['I2C_devices'],'dev_name',dev) SPI=Find(self.D['I2C_devices'],'dev_name',dev)
...@@ -36,7 +37,7 @@ class I2Cbitbang(I2Cdevices): ...@@ -36,7 +37,7 @@ class I2Cbitbang(I2Cdevices):
# data2=[regnr%256]+data2; # data2=[regnr%256]+data2;
# regnr=int(regnr/256); # regnr=int(regnr/256);
# data2=[Addr<<1,regnr]+data2; # data2=[Addr<<1,regnr]+data2;
# print(data2,Addr,Reg['reg_addr']); # logging.info("%s %s %s",data2,Addr,Reg['reg_addr']);
if not(I2CbbTX(data2,I2Ccallback)): return False; if not(I2CbbTX(data2,I2Ccallback)): return False;
# return False; # return False;
data2=[(Addr<<1)+1]; data2=[(Addr<<1)+1];
...@@ -49,14 +50,14 @@ class I2Cbitbang(I2Cdevices): ...@@ -49,14 +50,14 @@ class I2Cbitbang(I2Cdevices):
else: value[:]=data3[:]; else: value[:]=data3[:];
if (width!=l1*8) or (bitoffset>0): if (width!=l1*8) or (bitoffset>0):
value[0]=UnMask(value[0],width,bitoffset) value[0]=UnMask(value[0],width,bitoffset)
#print("Received:",data3); #logging.info("Received: %s",data3);
return True; return True;
def I2CbbTX(data2,SetBit): def I2CbbTX(data2,SetBit):
SetBit('SDIOdir',1) #Input, should be high SetBit('SDIOdir',1) #Input, should be high
if SetBit('SDI',0,True)==0: if SetBit('SDI',0,True)==0:
print("I2C line low!") logging.error("I2C line low!")
return False; return False;
#Start #Start
SetBit('SDIOdir',0) #Output = low SetBit('SDIOdir',0) #Output = low
...@@ -64,18 +65,18 @@ def I2CbbTX(data2,SetBit): ...@@ -64,18 +65,18 @@ def I2CbbTX(data2,SetBit):
for b in data2: for b in data2:
bit_array = "{0:{fill}8b}".format(b, fill='0') bit_array = "{0:{fill}8b}".format(b, fill='0')
#print(bit_array) #logging.info(bit_array)
for bit in bit_array: for bit in bit_array:
SetBit('SDIOdir',int(bit)) #input=high, output=low SetBit('SDIOdir',int(bit)) #input=high, output=low
SetBit('SCL',1) SetBit('SCL',1)
SetBit('SCL',0) SetBit('SCL',0)
SetBit('SDIOdir',1) #Input SetBit('SDIOdir',1) #Input
#print(GetBit(SDI)) #logging.info(GetBit(SDI))
if SetBit('SDI',0,True)==1: if SetBit('SDI',0,True)==1:
print("I2C: Not ACK") logging.error("I2C: Not ACK")
return False; return False;
SetBit('SCL',1) SetBit('SCL',1)
#print(GetBit(SDI)) #logging.info(GetBit(SDI))
SetBit('SCL',0) SetBit('SCL',0)
# SetBit(SDIOdir,0) #Output # SetBit(SDIOdir,0) #Output
...@@ -88,7 +89,7 @@ def I2CbbTX(data2,SetBit): ...@@ -88,7 +89,7 @@ def I2CbbTX(data2,SetBit):
def I2CbbRX(data1,data2,SetBit): def I2CbbRX(data1,data2,SetBit):
SetBit('SDIOdir',1) #Input, should be high SetBit('SDIOdir',1) #Input, should be high
if SetBit('SDI',0,True)==0: if SetBit('SDI',0,True)==0:
print("I2C ack: line low!") logging.error("I2C ack: line low!")
return False; return False;
#Start #Start
SetBit('SDIOdir',0) #Output = low SetBit('SDIOdir',0) #Output = low
...@@ -97,18 +98,18 @@ def I2CbbRX(data1,data2,SetBit): ...@@ -97,18 +98,18 @@ def I2CbbRX(data1,data2,SetBit):
for b in data1: for b in data1:
bit_array = "{0:{fill}8b}".format(b, fill='0') bit_array = "{0:{fill}8b}".format(b, fill='0')
#print(bit_array) #logging.info(bit_array)
for bit in bit_array: for bit in bit_array:
SetBit('SDIOdir',int(bit)) #input=high, output=low SetBit('SDIOdir',int(bit)) #input=high, output=low
SetBit('SCL',1) SetBit('SCL',1)
SetBit('SCL',0) SetBit('SCL',0)
SetBit('SDIOdir',1) #Input SetBit('SDIOdir',1) #Input
#print(GetBit(SDI)) #logging.info(GetBit(SDI))
if SetBit('SDI',0,True)==1: if SetBit('SDI',0,True)==1:
print("I2C: Not ACK") logging.error("I2C: Not ACK")
return False; return False;
SetBit('SCL',1) SetBit('SCL',1)
#print(GetBit(SDI)) #logging.info(GetBit(SDI))
SetBit('SCL',0) SetBit('SCL',0)
for x in range(len(data2)): for x in range(len(data2)):
......
I2Cdevices.py
+ 22
− 21
View file @ b558fc04
from hwdev import *; from hwdev import *;
import numpy as np import numpy as np
import logging
def GetField(D,name,dev_number): def GetField(D,name,dev_number):
...@@ -44,10 +45,10 @@ class I2Cdevices(hwdev): ...@@ -44,10 +45,10 @@ class I2Cdevices(hwdev):
return True return True
def CallMethod(self,name,params,I2Ccallback): def CallMethod(self,name,params,I2Ccallback):
# print("RCU1 call method",name) # logging.info("RCU1 call method %s",name)
for Mth in self.D['Methods']: for Mth in self.D['Methods']:
if Mth['method_name']==name: if Mth['method_name']==name:
print("RCU1 Run method",name) logging.info("RCU1 Run method %s",name)
for regs in Mth['registers']: for regs in Mth['registers']:
for key,value in regs.items(): for key,value in regs.items():
self.SetVarValue(key,value,I2Ccallback) self.SetVarValue(key,value,I2Ccallback)
...@@ -77,11 +78,11 @@ class I2Cdevices(hwdev): ...@@ -77,11 +78,11 @@ class I2Cdevices(hwdev):
a=[0] a=[0]
self.I2CGet(Pin[0],Pin[1],a,I2Ccallback,1,Pin[2]) self.I2CGet(Pin[0],Pin[1],a,I2Ccallback,1,Pin[2])
return a[0]; return a[0];
print("Set Child",child) logging.info("Set Child %s",child)
return child.GetVarValue(name,value,SetGetPin) return child.GetVarValue(name,value,SetGetPin)
def SetVarValue(self,name,value,I2Ccallback): def SetVarValue(self,name,value,I2Ccallback):
print("RCU1 Set ",name,value) logging.info("RCU1 Set %s=%s",name,value)
D=self.D D=self.D
Var1=Find(D['Variables'],'var_name',name) Var1=Find(D['Variables'],'var_name',name)
...@@ -113,18 +114,18 @@ class I2Cdevices(hwdev): ...@@ -113,18 +114,18 @@ class I2Cdevices(hwdev):
if childname==c['child_name']: if childname==c['child_name']:
return self.SetChildValue(c,c['obj'],name[len(childname)+1:],value,I2Ccallback) return self.SetChildValue(c,c['obj'],name[len(childname)+1:],value,I2Ccallback)
# return hwdev.SetVarValue(self,name,value,I2Ccallback) # return hwdev.SetVarValue(self,name,value,I2Ccallback)
print("Unknown variable",name); logging.error("Unknown variable %s",name);
return False; return False;
print(Dev) logging.info("%s",Dev)
if scale: value=int2bytes(int(value[0]/float(scale))) if scale: value=int2bytes(int(value[0]/float(scale)))
if Find(D['I2C_devices'],'dev_name',Dev[0]): if Find(D['I2C_devices'],'dev_name',Dev[0]):
if not(width): width=8 if not(width): width=8
if not(bitoffset): bitoffset=0 if not(bitoffset): bitoffset=0
print('SetVar ',name,'I2C',Dev,value,'(width=',width,'bitoffset=',bitoffset,')') logging.info('SetVar %s I2C %s to %s (width=%s bitoffset=%s)',name,Dev,value,width,bitoffset)
return self.I2CSet(Dev[0],Dev[1],value,I2Ccallback,width,bitoffset) return self.I2CSet(Dev[0],Dev[1],value,I2Ccallback,width,bitoffset)
if Find(D['SPI_devices'],'dev_name',Dev[0]): if Find(D['SPI_devices'],'dev_name',Dev[0]):
print('SetVar ',name,'SPI',Dev,value) logging.info('SetVar %s SPI %S to %s',name,Dev,value)
return SPIset(D,Dev[0],Dev[1],value,I2Ccallback,dev_number) return SPIset(D,Dev[0],Dev[1],value,I2Ccallback,dev_number)
hwdev.SetVarValue(self,name,value,I2Ccallback) hwdev.SetVarValue(self,name,value,I2Ccallback)
# return SetVar(self.D,name,value,I2Ccallback) # return SetVar(self.D,name,value,I2Ccallback)
...@@ -132,7 +133,7 @@ class I2Cdevices(hwdev): ...@@ -132,7 +133,7 @@ class I2Cdevices(hwdev):
def GetVarValue(self,name,value,I2Ccallback): def GetVarValue(self,name,value,I2Ccallback):
#def GetVar(D,name,value,I2Ccallback,dev_number=0): #def GetVar(D,name,value,I2Ccallback,dev_number=0):
dev_number=0;D=self.D dev_number=0;D=self.D
print("RCU1 Get ",name,value) logging.info("RCU1 Get %s=%s",name,value)
Var1=Find(D['Variables'],'var_name',name) Var1=Find(D['Variables'],'var_name',name)
if (Var1): if (Var1):
Dev=GetField(Var1,'var_dev',dev_number) Dev=GetField(Var1,'var_dev',dev_number)
...@@ -151,27 +152,27 @@ class I2Cdevices(hwdev): ...@@ -151,27 +152,27 @@ class I2Cdevices(hwdev):
if childname==c['child_name']: if childname==c['child_name']:
return self.GetChildValue(c,c['obj'],name[len(childname)+1:],value,I2Ccallback) return self.GetChildValue(c,c['obj'],name[len(childname)+1:],value,I2Ccallback)
# return hwdev.SetVarValue(self,name,value,I2Ccallback) # return hwdev.SetVarValue(self,name,value,I2Ccallback)
print("Unknown variable",name); logging.error("Unknown variable %s",name);
return False return False
print(Dev) logging.info("%s",Dev)
if not(width): width=8 if not(width): width=8
if not(bitoffset): bitoffset=0 if not(bitoffset): bitoffset=0
if Find(D['I2C_devices'],'dev_name',Dev[0]): if Find(D['I2C_devices'],'dev_name',Dev[0]):
print('GetVar ',name,'I2C',Dev,value,'(width=',width,'bitoffset=',bitoffset,')') logging.info('GetVar %s I2C %s value %s (width=%s bitoffset=%s)',name,Dev,value,width,bitoffset)
result=self.I2CGet(Dev[0],Dev[1],value,I2Ccallback,width,bitoffset) result=self.I2CGet(Dev[0],Dev[1],value,I2Ccallback,width,bitoffset)
if not(result): return False; if not(result): return False;
if scale: value[0]=float(value[0])*float(scale); if scale: value[0]=float(value[0])*float(scale);
return True return True
if Find(D['SPI_devices'],'dev_name',Dev[0]): if Find(D['SPI_devices'],'dev_name',Dev[0]):
print('GetVar ',name,'SPI',Dev,value) logging.info('GetVar %s SPI %s value %s',name,Dev,value)
return SPIget(D,Dev[0],Dev[1],value,I2Ccallback,dev_number) return SPIget(D,Dev[0],Dev[1],value,I2Ccallback,dev_number)
if Find(D['I2C_bitbang'],'dev_name',Dev[0]): if Find(D['I2C_bitbang'],'dev_name',Dev[0]):
print('GetVar ',name,'I2C bb',Dev,value) logging.info('GetVar %s I2C bb %s value %s',name,Dev,value)
if not(I2Cbbget(D,Dev[0],Dev[1],value,I2Ccallback,dev_number,width,bitoffset)): return False if not(I2Cbbget(D,Dev[0],Dev[1],value,I2Ccallback,dev_number,width,bitoffset)): return False
if scale: value[0]=float(value[0])*float(scale); if scale: value[0]=float(value[0])*float(scale);
return True; return True;
print("Unknown variable",name) logging.error("Unknown variable %s",name)
return False; return False;
# return GetVar(self.D,name,value,I2Ccallback) # return GetVar(self.D,name,value,I2Ccallback)
...@@ -191,11 +192,11 @@ class I2Cdevices(hwdev): ...@@ -191,11 +192,11 @@ class I2Cdevices(hwdev):
def I2CSet(self,dev,reg,value,I2Ccallback,width=8,bitoffset=0): def I2CSet(self,dev,reg,value,I2Ccallback,width=8,bitoffset=0):
#remember of values, add new ones with mask #remember of values, add new ones with mask
# print('I2CSet:',dev,'=',value,'(mask=',mask,')') # logging.info('I2CSet: %s=%s (mask=%s)',dev,value,mask)
Dev=Find(self.D['I2C_devices'],'dev_name',dev) Dev=Find(self.D['I2C_devices'],'dev_name',dev)
Addr=Dev['dev_address'] Addr=Dev['dev_address']
Reg=Find(Dev['dev_registers'],'reg_name',reg) Reg=Find(Dev['dev_registers'],'reg_name',reg)
# print(dev,reg,value) # logging.info("%s %s %s",dev,reg,value)
if (width<8) or (bitoffset>0): if (width<8) or (bitoffset>0):
previous=Reg.get('previous') previous=Reg.get('previous')
if not(previous): if not(previous):
...@@ -203,8 +204,8 @@ class I2Cdevices(hwdev): ...@@ -203,8 +204,8 @@ class I2Cdevices(hwdev):
if not(previous): if not(previous):
previous=0; previous=0;
value[0]=ApplyMask(value[0],width,bitoffset,previous); value[0]=ApplyMask(value[0],width,bitoffset,previous);
# print('masked to',value) # logging.info('masked to %s',value)
# print("Set",dev,reg,"to",value) # logging.info("Set %s %s to %s",dev,reg,value)
result=I2Ccallback(Addr,value,reg=Reg['reg_addr'],) result=I2Ccallback(Addr,value,reg=Reg['reg_addr'],)
#if result: #if result:
Reg['previous']=value[0] Reg['previous']=value[0]
...@@ -215,7 +216,7 @@ class I2Cdevices(hwdev): ...@@ -215,7 +216,7 @@ class I2Cdevices(hwdev):
Addr=Dev['dev_address'] Addr=Dev['dev_address']
Reg=Find(Dev['dev_registers'],'reg_name',reg) Reg=Find(Dev['dev_registers'],'reg_name',reg)
l1=int(np.floor((width+bitoffset+7)/8)) l1=int(np.floor((width+bitoffset+7)/8))
# print(width,bitoffset,l1) # logging.info("%s %s %s",width,bitoffset,l1)
makesinglevalue=((len(value)==1) and (l1>1)); makesinglevalue=((len(value)==1) and (l1>1));
if makesinglevalue: value2=[0 for x in range(l1)] if makesinglevalue: value2=[0 for x in range(l1)]
else: value2=value else: value2=value
...@@ -240,7 +241,7 @@ class I2Cdevices(hwdev): ...@@ -240,7 +241,7 @@ class I2Cdevices(hwdev):
# Addr=Dev['dev_address'] # Addr=Dev['dev_address']
# Reg=Find(Dev['dev_registers'],'reg_name',reg) # Reg=Find(Dev['dev_registers'],'reg_name',reg)
# value[0]=Reg.get('previous') # value[0]=Reg.get('previous')
# print("Get Value=",value[0]) # logging.info("Get Value=%s",value[0])
# if (width<8) or (bitoffset>0): # if (width<8) or (bitoffset>0):
# value[0]=UnMask(value[0],width,bitoffset) # value[0]=UnMask(value[0],width,bitoffset)
# return True; # return True;
......
I2Cswitch.py
+ 8
− 7
View file @ b558fc04
from hwdev import *; from hwdev import *;
import numpy as np import numpy as np
import logging
def I2C_dummy(addr,data,reg=None,read=0): def I2C_dummy(addr,data,reg=None,read=0):
# print("I2C dummy",addr,data) # logging.info("I2C dummy addr=%s data=%s",addr,data)
return True; return True;
class I2Cswitch(hwdev): class I2Cswitch(hwdev):
def SetChannel(self,channel,I2Ccallback): def SetChannel(self,channel,I2Ccallback):
if not(hasattr(self,'CurrentChannel')): self.CurrentChannel=0 if not(hasattr(self,'CurrentChannel')): self.CurrentChannel=0
print("SetChannel",channel,self.CurrentChannel) logging.info("SetChannel %s %s",channel,self.CurrentChannel)
if (channel)==self.CurrentChannel: return True; if (channel)==self.CurrentChannel: return True;
self.CurrentChannel=channel self.CurrentChannel=channel
return I2Ccallback(self.D['dev_address'],[channel]) return I2Ccallback(self.D['dev_address'],[channel])
...@@ -25,7 +26,7 @@ class I2Cswitch(hwdev): ...@@ -25,7 +26,7 @@ class I2Cswitch(hwdev):
def GetVarValue(self,name,value,I2Ccallback): def GetVarValue(self,name,value,I2Ccallback):
def I2Ccallback2(addr,data,reg=None,read=0): def I2Ccallback2(addr,data,reg=None,read=0):
# print("Getvarcallback",x) # logging.info("Getvarcallback %s",x)
if (read==2): if (read==2):
if (cnt>0): return True; if (cnt>0): return True;
if not(self.SetChannel(gchannel,I2Ccallback)): return False; if not(self.SetChannel(gchannel,I2Ccallback)): return False;
...@@ -51,7 +52,7 @@ class I2Cswitch(hwdev): ...@@ -51,7 +52,7 @@ class I2Cswitch(hwdev):
for cnt,childname2 in enumerate(group['group_members']): for cnt,childname2 in enumerate(group['group_members']):
child=Find(self.D['dev_children'],'child_name',childname2) child=Find(self.D['dev_children'],'child_name',childname2)
if not(child): return False; if not(child): return False;
print("Get:",childname2) logging.info("Get: %s",childname2)
cchannel=1<<child['child_addr'] cchannel=1<<child['child_addr']
value2=value[cnt*stride:(cnt+1)*stride] value2=value[cnt*stride:(cnt+1)*stride]
if not(child['obj'].GetVarValue(name[len(childname)+1:],value2,I2Ccallback2)): return False; if not(child['obj'].GetVarValue(name[len(childname)+1:],value2,I2Ccallback2)): return False;
...@@ -74,11 +75,11 @@ class I2Cswitch(hwdev): ...@@ -74,11 +75,11 @@ class I2Cswitch(hwdev):
for y in range(stride): for y in range(stride):
if value[x*stride+y]!=value[(x+1)*stride+y]: AllSame=False; if value[x*stride+y]!=value[(x+1)*stride+y]: AllSame=False;
if AllSame: if AllSame:
print("Allsame"); logging.info("Allsame");
channel=0; channel=0;
for childname2 in group['group_members']: for childname2 in group['group_members']:
channel+=1<<(Find(self.D['dev_children'],'child_name',childname2)['child_addr']) channel+=1<<(Find(self.D['dev_children'],'child_name',childname2)['child_addr'])
print(channel) logging.info("%s",channel)
if not(self.SetChannel(channel,I2Ccallback)): return False; if not(self.SetChannel(channel,I2Ccallback)): return False;
for x,childname2 in enumerate(group['group_members']): for x,childname2 in enumerate(group['group_members']):
child=Find(self.D['dev_children'],'child_name',childname2) child=Find(self.D['dev_children'],'child_name',childname2)
...@@ -90,7 +91,7 @@ class I2Cswitch(hwdev): ...@@ -90,7 +91,7 @@ class I2Cswitch(hwdev):
for x,childname2 in enumerate(group['group_members']): for x,childname2 in enumerate(group['group_members']):
child=Find(self.D['dev_children'],'child_name',childname2) child=Find(self.D['dev_children'],'child_name',childname2)
if not(child): return False; if not(child): return False;
print("Set:",childname2) logging.info("Set: %s",childname2)
if not(self.SetChannel(1<<child['child_addr'],I2Ccallback)): return False; if not(self.SetChannel(1<<child['child_addr'],I2Ccallback)): return False;
if not(child['obj'].SetVarValue(name[len(childname)+1:],value[x*stride:(x+1)*stride],I2Ccallback)): return false; if not(child['obj'].SetVarValue(name[len(childname)+1:],value[x*stride:(x+1)*stride],I2Ccallback)): return false;
return True; return True;
......
SPIbitbang1.py
+ 7
− 6
View file @ b558fc04
from I2Cdevices import *; from I2Cdevices import *;
import numpy as np import numpy as np
import logging
class SPIbitbang1(I2Cdevices): class SPIbitbang1(I2Cdevices):
def I2CSet(self,dev,reg,value,I2Ccallback,width=8,bitoffset=0): def I2CSet(self,dev,reg,value,I2Ccallback,width=8,bitoffset=0):
...@@ -13,7 +14,7 @@ class SPIbitbang1(I2Cdevices): ...@@ -13,7 +14,7 @@ class SPIbitbang1(I2Cdevices):
data2 = ( ADC_rw << 23 ) + ( ADC_bytes << 21 ) + ( ADC_address << 8 ) + value[0] data2 = ( ADC_rw << 23 ) + ( ADC_bytes << 21 ) + ( ADC_address << 8 ) + value[0]
bit_array = "{0:{fill}24b}".format(data2, fill='0') bit_array = "{0:{fill}24b}".format(data2, fill='0')
# print(bit_array) # logging.debug("%s",bit_array)
SetBit('CS',0) #enable SetBit('CS',0) #enable
for bit in bit_array: for bit in bit_array:
SetBit('SDO',int(bit)) SetBit('SDO',int(bit))
...@@ -28,7 +29,7 @@ class SPIbitbang1(I2Cdevices): ...@@ -28,7 +29,7 @@ class SPIbitbang1(I2Cdevices):
SPI=Find(self.D['I2C_devices'],'dev_name',dev) SPI=Find(self.D['I2C_devices'],'dev_name',dev)
Reg=Find(SPI['dev_registers'],'reg_name',reg) Reg=Find(SPI['dev_registers'],'reg_name',reg)
ADC_reg_address=Reg['reg_addr'] ADC_reg_address=Reg['reg_addr']
# print(ADC_reg_address,CS,SDIOdir,SDI,SDO,CLK) # logging.debug("%s %s %s %s %s %s",ADC_reg_address,CS,SDIOdir,SDI,SDO,CLK)
SetBit=I2Ccallback SetBit=I2Ccallback
ADC_bytes = 0x00 ADC_bytes = 0x00
...@@ -46,7 +47,7 @@ class SPIbitbang1(I2Cdevices): ...@@ -46,7 +47,7 @@ class SPIbitbang1(I2Cdevices):
SetBit('CS',1) #disable SetBit('CS',1) #disable
# print("read byte") # logging.debug("read byte")
SetBit('SDIOdir',1) #inpute SetBit('SDIOdir',1) #inpute
SetBit('CS',0) #enable SetBit('CS',0) #enable
a=[0] a=[0]
...@@ -60,8 +61,8 @@ class SPIbitbang1(I2Cdevices): ...@@ -60,8 +61,8 @@ class SPIbitbang1(I2Cdevices):
SetBit('SDIOdir',0) #output SetBit('SDIOdir',0) #output
# SetBit(SDO,1) # SetBit(SDO,1)
# SetBit(CLK,1) # SetBit(CLK,1)
# print(read_bit) # logging.debug("%s",read_bit)
# stri = "Read back data is: {0:2x} ".format(int(read_bit, 2)) # stri = "Read back data is: {0:2x} ".format(int(read_bit, 2))
# print(stri) # logging.debug("%s",stri)
value[0]=int(read_bit, 2) value[0]=int(read_bit, 2)
return True; return True;
SPIbitbang2.py
+ 7
− 6
View file @ b558fc04
from I2Cdevices import *; from I2Cdevices import *;
import numpy as np import numpy as np
import logging
class SPIbitbang2(I2Cdevices): class SPIbitbang2(I2Cdevices):
def I2CSet(self,dev,reg,value,I2Ccallback,width=8,bitoffset=0): def I2CSet(self,dev,reg,value,I2Ccallback,width=8,bitoffset=0):
...@@ -13,7 +14,7 @@ class SPIbitbang2(I2Cdevices): ...@@ -13,7 +14,7 @@ class SPIbitbang2(I2Cdevices):
data2 = ( reg_address << 9 ) + value[0] data2 = ( reg_address << 9 ) + value[0]
bit_array = "{0:{fill}16b}".format(data2, fill='0') bit_array = "{0:{fill}16b}".format(data2, fill='0')
# print(bit_array) # logging.info("%s",bit_array)
SetBit('CS',0) #enable SetBit('CS',0) #enable
for bit in bit_array: for bit in bit_array:
SetBit('SDI',int(bit)) SetBit('SDI',int(bit))
...@@ -28,7 +29,7 @@ class SPIbitbang2(I2Cdevices): ...@@ -28,7 +29,7 @@ class SPIbitbang2(I2Cdevices):
SPI=Find(self.D['I2C_devices'],'dev_name',dev) SPI=Find(self.D['I2C_devices'],'dev_name',dev)
Reg=Find(SPI['dev_registers'],'reg_name',reg) Reg=Find(SPI['dev_registers'],'reg_name',reg)
reg_address=(Reg['reg_addr'] if Reg else int(reg)) reg_address=(Reg['reg_addr'] if Reg else int(reg))
# print(ADC_reg_address,CS,SDIOdir,SDI,SDO,CLK) # logging.info("%s %s %s %s %s %s",ADC_reg_address,CS,SDIOdir,SDI,SDO,CLK)
SetBit=I2Ccallback SetBit=I2Ccallback
ADC_bytes = 0x00 ADC_bytes = 0x00
...@@ -45,7 +46,7 @@ class SPIbitbang2(I2Cdevices): ...@@ -45,7 +46,7 @@ class SPIbitbang2(I2Cdevices):
# SetBit('CS',1) #disable # SetBit('CS',1) #disable
# print("read byte") # logging.info("read byte")
# SetBit('SDIOdir',1) #inpute # SetBit('SDIOdir',1) #inpute
# SetBit('CS',0) #enable # SetBit('CS',0) #enable
a=[0] a=[0]
...@@ -58,8 +59,8 @@ class SPIbitbang2(I2Cdevices): ...@@ -58,8 +59,8 @@ class SPIbitbang2(I2Cdevices):
SetBit('CS',1) #disable SetBit('CS',1) #disable
# SetBit(SDO,1) # SetBit(SDO,1)
# SetBit(CLK,1) # SetBit(CLK,1)
# print(read_bit) # logging.info("%s",read_bit)
stri = "Read back data is: {0:2x} ".format(int(read_bit, 2)) stri = "Read back data is: {0:2x} ".format(int(read_bit, 2))
print(stri) logging.info("%s",stri)
value[0]=int(read_bit, 2) value[0]=int(read_bit, 2)
return True; return True;
hwdev.py
+ 6
− 4
View file @ b558fc04
import yaml; import yaml;
import importlib import importlib
import sys, inspect import sys, inspect
import logging
YAMLDIR='YAML/' YAMLDIR='YAML/'
class hwdev: class hwdev:
def __init__(self,configfile): def __init__(self,configfile):
print("Loading:",configfile) logging.info("Loading: %s",configfile)
self.D=yaml.load(open(YAMLDIR+configfile)) self.D=yaml.load(open(YAMLDIR+configfile))
self.loadchildren(self.D) self.loadchildren(self.D)
def loadchildren(self,D): def loadchildren(self,D):
...@@ -15,12 +17,12 @@ class hwdev: ...@@ -15,12 +17,12 @@ class hwdev:
childname=(c['child_name']+"_" if c.get('child_name') else ''); childname=(c['child_name']+"_" if c.get('child_name') else '');
objnm=c['child_dev'] objnm=c['child_dev']
yamlname=c['child_conf'] yamlname=c['child_conf']
print("Add child:",childname,objnm,yamlname) logging.info("Add child: %s %s %s",childname,objnm,yamlname)
i = importlib.import_module(objnm) i = importlib.import_module(objnm)
for name, obj in inspect.getmembers(i,inspect.isclass): for name, obj in inspect.getmembers(i,inspect.isclass):
if name==objnm: if name==objnm:
# yamlname=(c['conf_name'] if c.get('conf_name') else nm) # yamlname=(c['conf_name'] if c.get('conf_name') else nm)
# print(" Load child:",childname," conf=",yamlname+".yaml") # logging.debug(" Load child: %s conf=%s.yaml",childname,yamlname)
c['obj']=obj(yamlname+".yaml") c['obj']=obj(yamlname+".yaml")
self.children.append([childname,c['obj']]) self.children.append([childname,c['obj']])
...@@ -49,7 +51,7 @@ class hwdev: ...@@ -49,7 +51,7 @@ class hwdev:
def CallMethod(self,name,params,I2Ccallback): def CallMethod(self,name,params,I2Ccallback):
childname=name.split('_')[0] childname=name.split('_')[0]
for cname,child in self.children: for cname,child in self.children:
# print("hwdev callmethod",cname,childname) # logging.debug("hwdev callmethod %s %s",cname,childname)
if cname=="": child.CallMethod(name,params,I2Ccallback) if cname=="": child.CallMethod(name,params,I2Ccallback)
elif cname[:-1]==childname: return child.CallMethod(name[len(childname)+1:],params,I2Ccallback) elif cname[:-1]==childname: return child.CallMethod(name[len(childname)+1:],params,I2Ccallback)
return False return False
......
opcuaserv.py
+ 34
− 26
View file @ b558fc04
...@@ -6,13 +6,21 @@ import time ...@@ -6,13 +6,21 @@ import time
from opcua import ua, Server from opcua import ua, Server
from datetime import datetime; from datetime import datetime;
import argparse import argparse
import logging
parser = argparse.ArgumentParser() parser = argparse.ArgumentParser()
parser.add_argument("-s", "--simulator", help="Do not connect to I2c, but simulate behaviour.", action="store_true") parser.add_argument("-s", "--simulator", help="Do not connect to I2c, but simulate behaviour.", action="store_true")
parser.add_argument("--no-lib-hack", help="Do not require a hacked opcua library. Breaks behaviour.", action="store_true") parser.add_argument("--no-lib-hack", help="Do not require a hacked opcua library. Breaks behaviour.", action="store_true")
parser.add_argument("-p", "--port", help="Port number to listen on [%(default)s].", type=int, default=4842) parser.add_argument("-p", "--port", help="Port number to listen on [%(default)s].", type=int, default=4842)
parser.add_argument("-l", "--loglevel", help="Log level [%(default)s].", type=str, choices=["DEBUG","INFO","WARNING","ERROR"], default="DEBUG")
args = parser.parse_args() args = parser.parse_args()
# set log level
loglevel_nr = getattr(logging, args.loglevel.upper(), None)
if not isinstance(loglevel_nr, int):
raise ValueError('Invalid log level: %s' % args.loglevel)
logging.basicConfig(level=loglevel_nr, format="%(asctime)s [%(levelname)8s] %(message)s")
if args.simulator: if args.simulator:
import pypcc_test as pypcc import pypcc_test as pypcc
else: else:
...@@ -41,13 +49,13 @@ if True: ...@@ -41,13 +49,13 @@ if True:
#Vars={} #Vars={}
#Vars[1]='123' #Vars[1]='123'
#print(Vars) #logging.info("%s", Vars)
#exit() #exit()
def ValCallback(nodeid,force=False): def ValCallback(nodeid,force=False):
vname,myvar,oldvalue=Vars_R[nodeid.Identifier] vname,myvar,oldvalue=Vars_R[nodeid.Identifier]
# print("Value callback",vname) # logging.info("Value callback %s",vname)
# vname=vname[vname.find('_')+1:] # vname=vname[vname.find('_')+1:]
# print("RCU variable:",vname,oldvalue) # logging.info("RCU variable: %s=%s",vname,oldvalue)
# X=RCU1.Getvar2(vname) # X=RCU1.Getvar2(vname)
# if not(running): # if not(running):
if False: if False:
...@@ -56,10 +64,10 @@ def ValCallback(nodeid,force=False): ...@@ -56,10 +64,10 @@ def ValCallback(nodeid,force=False):
return myvar return myvar
timenow=datetime.utcnow() timenow=datetime.utcnow()
timediff=(timenow-myvar.SourceTimestamp).total_seconds() timediff=(timenow-myvar.SourceTimestamp).total_seconds()
# print(timediff) # logging.info("%s",timediff)
if not(force) and timediff<90: return myvar; if not(force) and timediff<90: return myvar;
res=P1.GetVarValue(vname,oldvalue) res=P1.GetVarValue(vname,oldvalue)
print("Read callback",vname,": Result:",res,oldvalue) logging.info("Read callback %s: Result: %s Value: %s",vname,res,oldvalue)
if res: if res:
myvar.Value.Value=(oldvalue[0] if len(oldvalue)==1 else oldvalue) myvar.Value.Value=(oldvalue[0] if len(oldvalue)==1 else oldvalue)
myvar.SourceTimestamp = datetime.utcnow() myvar.SourceTimestamp = datetime.utcnow()
...@@ -74,34 +82,34 @@ class SubHandler(object): ...@@ -74,34 +82,34 @@ class SubHandler(object):
""" """
def datachange_notification(self, node, val, data): def datachange_notification(self, node, val, data):
# print("Python: New data change event", node, val,data) # logging.info("Python: New data change event: %s %s %s", node, val,data)
if not(running): return if not(running): return
vname,myvar=Vars_W[node.nodeid.Identifier] vname,myvar=Vars_W[node.nodeid.Identifier]
val=(val if isinstance(val, list) else [val] ) val=(val if isinstance(val, list) else [val] )
print("Write callback",vname,val) logging.info("Write callback %s=%s",vname,val)
P1.SetVarValue(vname,val) P1.SetVarValue(vname,val)
#readback #readback
# if True: # if True:
# print(Vars_R,Vars_R.values()) # logging.info(Vars_R,Vars_R.values())
for vname2,myvar2,oldvalue in Vars_R.values(): for vname2,myvar2,oldvalue in Vars_R.values():
if vname2==vname: if vname2==vname:
if args.simulator: if args.simulator:
res=True res=True
print("Simulating fallthrough _RW->_R for",vname,": Result:",res,oldvalue) logging.info("Simulating fallthrough _RW->_R for",vname,": Result:",res,oldvalue)
else: else:
res=P1.GetVarValue(vname,val) res=P1.GetVarValue(vname,val)
print("Read callback",vname,": Result:",res,oldvalue) logging.info("Read callback",vname,": Result:",res,oldvalue)
if res: if res:
myvar2.Value.Value=(val[0] if len(val)==1 else val) myvar2.Value.Value=(val[0] if len(val)==1 else val)
myvar2.SourceTimestamp = datetime.utcnow() myvar2.SourceTimestamp = datetime.utcnow()
def event_notification(self, event): def event_notification(self, event):
print("Python: New event", event) logging.debug("Python: New event %s", event)
def CallMethod(ObjectID,name): def CallMethod(ObjectID,name):
print("Callmethod",name) logging.debug("Callmethod %s",name)
P1.CallMethod(name,None) P1.CallMethod(name,None)
...@@ -121,19 +129,19 @@ def AddVar(name,dtype=0,RW=0,cnt=1): ...@@ -121,19 +129,19 @@ def AddVar(name,dtype=0,RW=0,cnt=1):
myvar = (PCCobj.add_variable(idx, vname, 0.0) if cnt<=1 else PCCobj.add_variable(idx, vname, cnt*[0.0])) myvar = (PCCobj.add_variable(idx, vname, 0.0) if cnt<=1 else PCCobj.add_variable(idx, vname, cnt*[0.0]))
else: else:
myvar = (PCCobj.add_variable(idx, vname, 0) if cnt<=1 else PCCobj.add_variable(idx, vname, cnt*[0])) myvar = (PCCobj.add_variable(idx, vname, 0) if cnt<=1 else PCCobj.add_variable(idx, vname, cnt*[0]))
print("Variable added: ",vname) logging.info("Variable added: %s",vname)
Vars_R[myvar.nodeid.Identifier]=[name,myvar.get_data_value(),varvalue2] Vars_R[myvar.nodeid.Identifier]=[name,myvar.get_data_value(),varvalue2]
ValCallback(myvar.nodeid,force=True) ValCallback(myvar.nodeid,force=True)
# print(myvar.get_value()) # logging.info("%s",myvar.get_value())
if not args.no_lib_hack: if not args.no_lib_hack:
server.set_attribute_callback(myvar.nodeid, ValCallback) server.set_attribute_callback(myvar.nodeid, ValCallback)
# varvalue=myvar.get_data_value().Value # varvalue=myvar.get_data_value().Value
# Vars_R[myvar.nodeid.Identifier][2]=varvalue # Vars_R[myvar.nodeid.Identifier][2]=varvalue
# print(varvalue2,varvalue) # logging.info("%s %s",varvalue2,varvalue)
if RW in [2,3]: if RW in [2,3]:
vname=name+"_RW"; vname=name+"_RW";
print("Variable added: ",vname)#,'=',varvalue2) logging.info("Variable added: %s",vname)#,'=',varvalue2)
myvar2 = PCCobj.add_variable(idx, vname, (varvalue2[0] if len(varvalue2)==1 else varvalue2)) myvar2 = PCCobj.add_variable(idx, vname, (varvalue2[0] if len(varvalue2)==1 else varvalue2))
myvar2.set_writable() myvar2.set_writable()
Vars_W[myvar2.nodeid.Identifier]=[name,myvar2.get_data_value()] Vars_W[myvar2.nodeid.Identifier]=[name,myvar2.get_data_value()]
...@@ -143,29 +151,29 @@ def AddVar(name,dtype=0,RW=0,cnt=1): ...@@ -143,29 +151,29 @@ def AddVar(name,dtype=0,RW=0,cnt=1):
def AddMethod(name): def AddMethod(name):
vname=name; vname=name;
myvar = PCCobj.add_method(idx, vname, lambda ObjectId : CallMethod(ObjectId,name), [],[] ) myvar = PCCobj.add_method(idx, vname, lambda ObjectId : CallMethod(ObjectId,name), [],[] )
print("AddMethod:",vname) logging.info("AddMethod: %s",vname)
print("Start server"); logging.info("Start server");
server.start() server.start()
handler = SubHandler() handler = SubHandler()
sub = server.create_subscription(500, handler) sub = server.create_subscription(500, handler)
running=False; running=False;
print("Add variables:") logging.info("Add variables:")
P1.GetVarNames("",AddVar); P1.GetVarNames("",AddVar);
print("Add modes:") logging.info("Add modes:")
P1.GetMethodNames("",AddMethod); P1.GetMethodNames("",AddMethod);
time.sleep(1) time.sleep(1)
running=True; running=True;
print("Server started") logging.info("Server started")
try: try:
while True: while True:
time.sleep(1) time.sleep(1)
finally: finally:
print("Stop server"); logging.info("Stop server");
server.stop() server.stop()
...@@ -190,16 +198,16 @@ finally: ...@@ -190,16 +198,16 @@ finally:
#RCU.Setvar2('Power_Ant1',[0]) #RCU.Setvar2('Power_Ant1',[0])
#print(RCU.Getvar2('Amp_Gain0')) #logging.info(RCU.Getvar2('Amp_Gain0'))
#print(RCU.Getvar2('ID')) #logging.info(RCU.Getvar2('ID'))
#print(RCU.Getvar2('ADC_lock1')) #logging.info(RCU.Getvar2('ADC_lock1'))
#print(RCU.Getvar2('HBA_DelayX1',element=1)) #logging.info(RCU.Getvar2('HBA_DelayX1',element=1))
......
pypcc.py
+ 2
− 1
View file @ b558fc04
from hwdev import hwdev; from hwdev import hwdev;
import pylibi2c; import pylibi2c;
import time import time
import logging
#bus = pylibi2c.I2CDevice('/dev/i2c-1' #bus = pylibi2c.I2CDevice('/dev/i2c-1'
#read=0: write to register #read=0: write to register
#read=1: read from register #read=1: read from register
...@@ -11,7 +12,7 @@ def I2C1server(addr,data,reg=None,read=0): ...@@ -11,7 +12,7 @@ def I2C1server(addr,data,reg=None,read=0):
if read==3: if read==3:
time.sleep(data[0]/1000.) time.sleep(data[0]/1000.)
return True return True
# print("I2C",addr,reg,data,read) # logging.debug("I2C addr=%s reg=%s data=%s read=%s",addr,reg,data,read)
bus=pylibi2c.I2CDevice('/dev/i2c-1',addr) bus=pylibi2c.I2CDevice('/dev/i2c-1',addr)
if read==1: if read==1:
length=len(data) length=len(data)
......
pypcc_test.py
+ 2
− 1
View file @ b558fc04
from hwdev import hwdev; from hwdev import hwdev;
import time import time
import logging
def I2C1server(addr,data,reg=None,read=0): def I2C1server(addr,data,reg=None,read=0):
print("I2C:",addr,data,reg,read) logging.info("I2C: addr=%s data=%s reg=%s read=%s",addr,data,reg,read)
return True; return True;
class pypcc(hwdev): class pypcc(hwdev):
......
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment