# SNIFF
##########################
- def sniff(self,freq,duration,description, verbose=True, comment=None, filename=None, standardid=None, debug=False, faster=False, parsed=True, data = None):
+ def sniff(self,freq,duration,description, verbose=True, comment=None, filename=None, standardid=None, debug=False, faster=False, parsed=True, data = None,writeToFile=True):
#reset eveything on the chip
self.client.serInit()
self.client.MCPsetrate(freq);
# This will handle the files so that we do not loose them. each day we will create a new csv file
- if( filename==None):
+ if( filename==None and writeToFile == True):
#get folder information (based on today's date)
now = datetime.datetime.now()
datestr = now.strftime("%Y%m%d")
path = self.DATALOCATION+datestr+".csv"
filename = path
-
- outfile = open(filename,'a');
- dataWriter = csv.writer(outfile,delimiter=',');
- dataWriter.writerow(['# Time Error Bytes 1-13']);
- dataWriter.writerow(['#' + description])
-
+ if( writeToFile == True):
+ outfile = open(filename,'a');
+ dataWriter = csv.writer(outfile,delimiter=',');
+ dataWriter.writerow(['# Time Error Bytes 1-13']);
+ dataWriter.writerow(['#' + description])
+
self.client.MCPreqstatNormal();
print "Listening...";
packetcount = 0;
elif(messagestat == 0x00):
print "No messages in buffers."
- if packet!=None:
+ if (packet!=None and writeToFile == True):
packetcount+=1;
row = [];
return packetcount
- def filterStdSweep(self, freq, low, high, time = 5):
- msgIDs = []
- self.client.serInit()
- self.client.MCPsetup()
- for i in range(low, high+1, 6):
- print "sniffing id: %d, %d, %d, %d, %d, %d" % (i,i+1,i+2,i+3,i+4,i+5)
- comment= "sweepFilter: "
- #comment = "sweepFilter_%d_%d_%d_%d_%d_%d" % (i,i+1,i+2,i+3,i+4,i+5)
- description = "Running a sweep filer for all the possible standard IDs. This run filters for: %d, %d, %d, %d, %d, %d" % (i,i+1,i+2,i+3,i+4,i+5)
- count = self.sniff(freq=freq, duration = time, description = description,comment = comment, standardid = [i, i+1, i+2, i+3, i+4, i+5])
- if( count != 0):
- for j in range(i,i+5):
- comment = "sweepFilter: "
- #comment = "sweepFilter: %d" % (j)
- description = "Running a sweep filer for all the possible standard IDs. This run filters for: %d " % j
- count = self.sniff(freq=freq, duration = time, description = description,comment = comment, standardid = [j, j, j, j])
- if( count != 0):
- msgIDs.append(j)
- return msgIDs
-
- def sweepRandom(self, freq, number = 5, time = 200):
- msgIDs = []
- ids = []
- self.client.serInit()
- self.client.MCPsetup()
- for i in range(0,number+1,6):
- idsTemp = []
- comment = "sweepFilter: "
- for j in range(0,6,1):
- id = randrange(2047)
- #comment += "_%d" % id
- idsTemp.append(id)
- ids.append(id)
- print comment
- description = "Running a sweep filer for all the possible standard IDs. This runs the following : " + comment
- count = self.sniff(freq=freq, duration=time, description=description, comment = comment, standardid = idsTemp)
- if( count != 0):
- for element in idsTemp:
- #comment = "sweepFilter: %d" % (element)
- comment="sweepFilter: "
- description = "Running a sweep filer for all the possible standard IDs. This run filters for: %d " % element
- count = self.sniff(freq=freq, duration = time, description = description,comment = comment, standardid = [element, element, element])
- if( count != 0):
- msgIDs.append(j)
- return msgIDs, ids
+# def filterStdSweep(self, freq, low, high, time = 5):
+# msgIDs = []
+# self.client.serInit()
+# self.client.MCPsetup()
+# for i in range(low, high+1, 6):
+# print "sniffing id: %d, %d, %d, %d, %d, %d" % (i,i+1,i+2,i+3,i+4,i+5)
+# comment= "sweepFilter: "
+# #comment = "sweepFilter_%d_%d_%d_%d_%d_%d" % (i,i+1,i+2,i+3,i+4,i+5)
+# description = "Running a sweep filer for all the possible standard IDs. This run filters for: %d, %d, %d, %d, %d, %d" % (i,i+1,i+2,i+3,i+4,i+5)
+# count = self.sniff(freq=freq, duration = time, description = description,comment = comment, standardid = [i, i+1, i+2, i+3, i+4, i+5])
+# if( count != 0):
+# for j in range(i,i+5):
+# comment = "sweepFilter: "
+# #comment = "sweepFilter: %d" % (j)
+# description = "Running a sweep filer for all the possible standard IDs. This run filters for: %d " % j
+# count = self.sniff(freq=freq, duration = time, description = description,comment = comment, standardid = [j, j, j, j])
+# if( count != 0):
+# msgIDs.append(j)
+# return msgIDs
+
+# def sweepRandom(self, freq, number = 5, time = 200):
+# msgIDs = []
+# ids = []
+# self.client.serInit()
+# self.client.MCPsetup()
+# for i in range(0,number+1,6):
+# idsTemp = []
+# comment = "sweepFilter: "
+# for j in range(0,6,1):
+# id = randrange(2047)
+# #comment += "_%d" % id
+# idsTemp.append(id)
+# ids.append(id)
+# print comment
+# description = "Running a sweep filer for all the possible standard IDs. This runs the following : " + comment
+# count = self.sniff(freq=freq, duration=time, description=description, comment = comment, standardid = idsTemp)
+# if( count != 0):
+# for element in idsTemp:
+# #comment = "sweepFilter: %d" % (element)
+# comment="sweepFilter: "
+# description = "Running a sweep filer for all the possible standard IDs. This run filters for: %d " % element
+# count = self.sniff(freq=freq, duration = time, description = description,comment = comment, standardid = [element, element, element])
+# if( count != 0):
+# msgIDs.append(j)
+# return msgIDs, ids
def sniffTest(self, freq):
self.client.MCPreqstatNormal();
- # this will sweep through the given ids to request a packet and then sniff on that
- # id for a given amount duration. This will be repeated the number of attempts time
-
- #at the moment this is set to switch to the next id once a message is identified
- def rtrSweep(self,freq,lowID,highID, attempts = 1,duration = 1, verbose = True):
- #set up file
- now = datetime.datetime.now()
- datestr = now.strftime("%Y%m%d")
- path = self.DATALOCATION+datestr+"_rtr.csv"
- filename = path
- outfile = open(filename,'a');
- dataWriter = csv.writer(outfile,delimiter=',');
- dataWriter.writerow(['# Time Error Bytes 1-13']);
- dataWriter.writerow(['#' + "rtr sweep from %d to %d"%(lowID,highID)])
- print "started"
- #self.client.serInit()
- #self.spitSetup(freq)
- for i in range(lowID,highID+1, 1):
- self.client.serInit()
- self.spitSetup(freq)
- standardid = [i, i, i, i]
- #set filters
- self.addFilter(standardid, verbose = True)
-
- #### split SID into different areas
- SIDlow = (standardid[0] & 0x07) << 5; # get SID bits 2:0, rotate them to bits 7:5
- SIDhigh = (standardid[0] >> 3) & 0xFF; # get SID bits 10:3, rotate them to bits 7:0
- #create RTR packet
- packet = [SIDhigh, SIDlow, 0x00,0x00,0x40]
- dataWriter.writerow(["#requested id %d"%i])
- #self.client.poke8(0x2C,0x00); #clear the CANINTF register; we care about bits 0 and 1 (RXnIF flags) which indicate a message is being held
- #clear buffer
- packet1 = self.client.rxpacket();
- packet2 = self.client.rxpacket();
- #send in rtr request
- self.client.txpacket(packet)
- ## listen for 2 packets. one should be the rtr we requested the other should be
- ## a new packet response
- starttime = time.time()
- while ((time.time() - starttime) < duration):
- packet = self.client.rxpacket()
- if( packet == None):
- continue
- row = []
- row.append("%f"%time.time()) #timestamp
- row.append(0) #error flag (not checkign)
- row.append("rtrRequest_%d"%i) #comment
- row.append(duration) #sniff time
- row.append(1) # filtering boolean
- for byte in packet:
- row.append("%02x"%ord(byte));
- dataWriter.writerow(row)
- print self.client.packet2parsedstr(packet)
-# packet1=self.client.rxpacket();
-# packet2=self.client.rxpacket();
-# if( packet1 != None and packet2 != None):
-# print "packets recieved :\n "
-# print self.client.packet2parsedstr(packet1);
-# print self.client.packet2parsedstr(packet2);
-# continue
-# elif( packet1 != None):
-# print self.client.packet2parsedstr(packet1)
-# elif( packet2 != None):
-# print self.client.packet2parsedstr(packet2)
- trial= 2;
- # for each trial
- while( trial <= attempts):
- print "trial: ", trial
- self.client.MCPrts(TXB0=True);
- starttime = time.time()
- # this time we will sniff for the given amount of time to see if there is a
- # time till the packets come in
- while( (time.time()-starttime) < duration):
- packet=self.client.rxpacket();
- row = []
- row.append("%f"%time.time()) #timestamp
- row.append(0) #error flag (not checking)
- row.append("rtrRequest_%d"%i) #comment
- row.append(duration) #sniff time
- row.append(1) # filtering boolean
- for byte in packet:
- row.append("%02x"%ord(byte));
- dataWriter.writerow(row)
- print self.client.packet2parsedstr(packet)
-# packet2=self.client.rxpacket();
-#
-# if( packet1 != None and packet2 != None):
-# print "packets recieved :\n "
-# print self.client.packet2parsedstr(packet1);
-# print self.client.packet2parsedstr(packet2);
-# #break
-# elif( packet1 != None):
-# print "just packet1"
-# print self.client.packet2parsedstr(packet1)
-# elif( packet2 != None):
-# print "just packet2"
-# print self.client.packet2parsedstr(packet2)
- trial += 1
- print "sweep complete"
- outfile.close()
+
def spitSetup(self,freq):
self.reset();
self.client.MCPsetrate(freq);
-
+ # This will write the data provided in the packets which is expected to be a list of lists
+ # of the following form:
+ # for a given row = packets[i]
+ # row[0] time delay relative to the last packet. if 0 or empty there will be no delay
+ # row[1] = Standard ID (integer)
+ # row[2] = Data Length (0-8) (if it is zero we assume an Remote Transmit Request)
+ # row[3] = Data Byte 0
+ # row[4] = Data Byte 1
+ # .... up to Data Byte 8 ( THIS ASSUMES A PACKET OF LENGTH 8!!!
+ def writeData(self,packets,freq):
+ self.client.serInit()
+ self.spitSetup(freq)
+ for row in packets:
+ if( row[0] != 0 and row[0] != ""):
+ time.sleep(row[0])
+ sID = row[1]
+ #### split SID into different regs
+ SIDlow = (sID & 0x07) << 5; # get SID bits 2:0, rotate them to bits 7:5
+ SIDhigh = (sID >> 3) & 0xFF; # get SID bits 10:3, rotate them to bits 7:0
+ packet = [SIDhigh,SIDlow,0x00,0x00,0x08]
+ #dlc = row[2]
+ for i in range(4,dlc+4):
+ packet.append(row[i])
+ self.client.txpacket(packet)
+
+
+
+
if __name__ == "__main__":
projects / goodfet / blobdiff