import math
import os
from datetime import datetime
class IOOperations:
"""This class is used to read and write the different file formats used."""
def __init__(self):
pass
def exportDetectedDefects(self, defects_dict, pathToFolder):
"""Exports the found defects
Parameters
------
defects_dict : dict
A dictionary with the defective channel indices
pathToFolder : str
Path to the output folder
"""
if not os.path.exists(pathToFolder + "/results"):
os.mkdir(pathToFolder + "/results")
date = datetime.now()
date_str = date.strftime("%d%m%Y")
time_str = date.strftime("%H%M")
filename = pathToFolder + "/results/quality_result_" + date_str + ".txt"
out_str = ""
numOfDefectsTotal = 0
for asic, data in defects_dict.items():
out_str += "%d : ["%(asic)
for ch in data.get("even"):
out_str += "%d,"%(ch)
numOfDefectsTotal += 1
for ch in data.get("odd"):
out_str += "%d,"%(ch)
numOfDefectsTotal += 1
if out_str[-1] != "[":
out_str = out_str[:-1] + "]\n"
else:
out_str += "]\n"
with open(filename, "w") as file:
file.write("Date : %s\n"%(date_str))
file.write("Time : %s\n"%(time_str))
file.write("Total number of defects : %d\n"%(numOfDefectsTotal))
file.write(out_str)
print("Data successfully exported to file \"%s\""%(filename))
def exportNumOfDefects(self, defects_list, thld, asic_nr, pathToFolder, additionalFilename=""):
"""Exports the number of defects found and the respective threshold used.
Parameters
------
defects_list : list
A list with the different numbers of defects found
thld : float
The threshold step used
asic_nr : int
The number of the ASIC used to generate the data
pathToFolder : str
Path to the output folder
additionalFilename : str
Optional string to be added to the filename
"""
if not os.path.exists(pathToFolder + "/results"):
os.mkdir(pathToFolder + "/results")
filename = pathToFolder + "/results/defects_" + str(asic_nr) + "_" + additionalFilename + ".csv"
with open(filename, "w") as file:
for i in range(len(defects_list)):
line = str(float(i*thld)) + "\t" + str(int(defects_list[i])) + "\n"
file.write(line)
def exportFilterThresholds(self, data, start, step, pathToFolder, asic_nr, additionalFilename=""):
"""Exports the number of filtered channels with the respective threshold.
Parameters
------
data : list
List with the number of filtered channels
start : float
The initial threshold
step : float
The step the threshold was increased for every element
asic_nr : int
The number of the ASIC used to generate the data
pathToFolder : str
Path to the output folder
additionalFilename : str
Optional string to be added to the filename
"""
if not os.path.exists(pathToFolder + "/results"):
os.mkdir(pathToFolder + "/results")
filename = pathToFolder + "/results/" + additionalFilename + "_" + str(asic_nr) + ".csv"
with open(filename, "w") as file:
for i in range(len(data)):
line = str(i*step + start) + "\t" + str(int(data[i])) + "\n"
file.write(line)
def exportCountComparison(self, data, isEven, asic_nr, pathToFolder, additionalFilename=""):
"""This function provides a special output to generate a file with pre and post filtering data for direct comparison.
Parameters
------
data : list
A list with lists of the same length to be printed parallel
isEven : bool
Is true if even channels are processed; Used to display correct channel index
asic_nr : int
The number of the ASIC used to generate the data
pathToFolder : str
Path to the output folder
additionalFilename : str
Optional string to be added to the filename
"""
if not os.path.exists(pathToFolder + "/results"):
os.mkdir(pathToFolder + "/results")
filename = pathToFolder + "/results/" + additionalFilename + "_" + str(asic_nr) + ".csv"
with open(filename, "w") as file:
for ch_index in range(len(data[0])):
if isEven:
line = str(ch_index*2) + "\t"
else:
line = str(ch_index*2+1) + "\t"
for elem in data:
line += str(int(elem[ch_index])) + "\t"
line+= "\n"
file.write(line)
def exportChannelCounts(self, channel_counts, asic_nr, pathToFolder, additionalFilename = ""):
"""This function exports the calculated channel counts.
Parameters
------
channel_counts : list
A list with the average counts for each channel
asic_nr : int
The number of the ASIC this data was generated
pathToFolder : str
A path to the output folder
additionalFilename : str
Optional string to be added to the filename
"""
if not os.path.exists(pathToFolder + "/results"):
os.mkdir(pathToFolder + "/results")
filename = pathToFolder + "/results/counts_" + str(asic_nr) + additionalFilename + ".csv"
with open(filename, "w") as file:
for ch_index in range(len(channel_counts)):
line = str(ch_index) + "\t" + str(int(channel_counts[ch_index])) + "\n"
file.write(line)
def exportSCurves(self, sCurveDict, asic_nr, pathToFolder):
"""This function exports the calculated s curves into a csv file.
Parameters
------
sCurveDict : dict
A dictionary with the averages and sigmas for each discriminator
asic_nr : int
The number of the ASIC this data was generated
pathToFolder : str
A path to the output folder
"""
if not os.path.exists(pathToFolder + "/results"):
os.mkdir(pathToFolder + "/results")
filename = pathToFolder + "/results/scurve_" + str(asic_nr) + ".csv"
with open(filename, "w") as file:
for disc, data in sCurveDict.items():
line = str(disc) + "\t" + str(int(data.get("avg"))) + "\t" + str(int(data.get("sigma"))) + "\n"
file.write(line)
def readConnResults(self, pathToFile):
"""This class reads the results from the connection test stored in the referenced file.
Parameters
------
pathToFolder : str
The path to the file with the raw data
Returns
------
raw_data : dict
A dictionary with a list representing the data and additional information
"""
raw_data = dict()
lines_list = list()
filename = pathToFile.split("/")[-1]
asic_nr = int(filename.split("_")[4])
loop_nr = int(filename.split("_")[-1].split(".")[0])
input_str = ""
with open(pathToFile, "r") as file:
input_str = file.read()
input_str = input_str.split("\n")
if input_str[-1] == "":
input_str = input_str[:-1]
for line in input_str:
line = line.split()
channel_list = line[4:]
lines_list.append(channel_list)
if loop_nr != (len(lines_list) / 128):
print("nloops not matching")
raw_data["data"] = lines_list
raw_data["asic_nr"] = asic_nr
raw_data["nloops"] = loop_nr
return raw_data
class DetectDefects:
"""This class is used for detecting the defective channels"""
def findDefectiveChannels(self, deviation_list:list, asic_nr:int, isEven:bool)-> list:
"""Analyses the deviation data and detects defective channels and their index.
Parameters
------
deviation_list : list
A list with the percentual deviations
asic_nr : int
The number of the ASIC the data was generated with
isEven : bool
Is the data from even or odd channels (used for numbering)
Returns
------
defective_channels : list
A list with the indices of the defective channels
"""
thld = 0.90
defective_channels = []
offset = 0
if not isEven:
offset = 1
if isEven:
print("====================\n||\tASIC %d\t ||\n===================="%(asic_nr))
for i in range(len(deviation_list)):
if deviation_list[i] > thld:
defective_channels.append(i*2+offset)
print("\tCh: %d"%(i*2+offset))
return defective_channels
class ExtremeValues:
"""This class provides functionality to filter extreme values."""
def __init__(self):
self.calc = Calculus()
def calculateNumberOfFilteredValues(self, raw:list, filtered:list)-> int:
"""This function compares two lists and counts the number of different values.
Parameters
------
raw : list
A list with numerical values for the unfiltered counts
filtered : list
A list with the filtered count values
Returns
------
numOfFiltered : int
Number of different (=filtered) values
"""
numOfFiltered = 0
if len(raw) == len(filtered):
for index in range(len(raw)):
if raw[index] != filtered[index]:
numOfFiltered += 1
return numOfFiltered
def filterExtremeValues(self, data:list, thld:float)->list:
"""Filters extreme values based on the percentual deviation between two channels.
To prevent division by 0 errors, any 0 counts will be replaced with 1.
Parameters
------
data : list
The channel counts
thld : float
The threshold the deviation may not exceed
Returns
------
filtered_list : list
A list with the filtered and replaced values.
"""
#Define borders for good values
upperLimit = self.calc.calculateUpperBorder(data)
lowerLimit = self.calc.calculateLowerBorder(data)
#Calculate percentualdeviation between neighbouring channels
deviation_list = []
for index in range(len(data)-1):
if data[index] == 0:
_dev = (data[index+1] - data[index])
else:
_dev = (data[index+1] - data[index])/data[index]
deviation_list.append(_dev)
filtered_list = list(data)
index = 0
leftBorderIndex = 0
rightBorderIndex = 0
replaceMode = False
#Is the first element outside the limits
if (data[index] < lowerLimit) or (data[index] > upperLimit):
replaceMode = True
leftBorderIndex = 0
index = 1
while replaceMode == True:
if (data[index] < upperLimit) and (data[index] > lowerLimit):
rightBorderIndex = index
replaceMode = False
index = index + 1
if index >= len(data):
rightBorderIndex = index
replaceMode = False
for i in range(rightBorderIndex):
filtered_list[i] = data[rightBorderIndex]
while index < (len(data)-1):
if (deviation_list[index] > thld) or (deviation_list[index] < -thld):
leftBorderIndex = index
replaceMode = True
index += 2
if index >= len(data):
index = len(data)-1
while replaceMode == True:
if (data[index] < upperLimit) and (data[index] > lowerLimit):
rightBorderIndex = index
replaceMode = False
index = index + 1
if index >= len(data):
rightBorderIndex = index
replaceMode = False
if rightBorderIndex == len(data): #Last element is extreme - replace complete with left border value
for i in range(leftBorderIndex+1, rightBorderIndex):
filtered_list[i] = data[leftBorderIndex]
else:
_avg = (data[leftBorderIndex] + data[rightBorderIndex])/2
for i in range(leftBorderIndex+1, rightBorderIndex):
filtered_list[i] = _avg
index += 1
return filtered_list
class ReferenceValues:
"""This class is used to calculate the different reference values."""
def __init__(self):
self.calc = Calculus()
def calculateReferenceAverages(self, data):
"""Funtion to calculate different averages on the same data for comparison.
Parameters
------
data : list
List with the counts per channel
"""
average_list = list()
for size in range(3, 14, 2): #every odd number from 2 to 13
average_list.append(self.calc.calculateNAverageOnList(data, size))
return average_list
class ChannelCounts:
"""This class contains different funtions to perform different analysis on
the ASIC level.
"""
def __init__(self):
self.calc = Calculus()
def performAllAnalysis(self, channel_array:list)-> None:
"""Performs all analysis for channel based counts.
Parameters
------
data : list
A two dimensional array with the average discriminator counts for each channel
"""
self.counts = self.calculateAvgCounts(channel_array)
def calculateAvgCounts(self, data:list) -> tuple[list, list, list]:
"""This function takes the per channel discriminator counts and
calculates the channel count as the sum of the discriminator counts.
Parameters
------
data : list
A two dimensional array with the average discriminator counts for each channel
Returns
------
channel_counts : tuple
A list with the 128 channel counts and
"""
channel_counts = list()
for channel in data:
channel_counts.append(sum(channel))
counts_even = list()
counts_odd = list()
for index in range(0,128, 2):
counts_even.append(channel_counts[index])
for index in range(1,128, 2):
counts_odd.append(channel_counts[index])
return (channel_counts, counts_even, counts_odd)
class SCurves:
"""This class performs different calculation on the discriminator level of the data."""
def __init__(self):
self.calc = Calculus()
def performAllAnalysis(self, raw_data_lines:dict)-> None:
self.channel_array = self.calculateAvgDiscriminator(raw_data_lines)
self.sCurve = self.calculateAvgSCurve(self.channel_array)
def calculateAvgDiscriminator(self, data:dict)-> list:
"""This function calculates the average for each channel discriminator over all nloops.
Parameters
------
data : dict
A dictionary with a list of all data lines from the data file
Returns
------
channel_array : list
A two-dimensional list with the channel discriminators
"""
channel_list = list()
for i in range(128):
disc_array = list()
for j in range(31):
disc_array.append(list())
channel_list.append(disc_array)
for line_index in range(len(data.get("data"))):
line = data.get("data")[line_index]
for disc_index in range(len(line)):
channel_list[(line_index%128)][disc_index].append(int(line[disc_index]))
channel_array = list()
for ch_index in range(128):
disc_list = list()
for disc_index in range(31):
_avg = self.calc.calculateAverageOnList(channel_list[ch_index][disc_index])
disc_list.append(_avg)
channel_array.append(disc_list)
return channel_array
def calculateAvgSCurve(self, data:list)-> dict:
"""This function calculates the average and standard deviation for each
discriminator over all channels.
Parameters
------
data : list
A list with the discriminator count for each channels
Returns
------
sCurve : dict
A dictionary with two list: the average counts and the respective sigma
"""
disc_list = list()
sCurve = dict()
for i in range(31):
disc_list.append(list())
for line_index in range(len(data)):
line = data[line_index]
for disc_index in range(len(line)):
disc_list[disc_index].append(int(line[disc_index]))
for disc_index in range(len(disc_list)):
avg = self.calc.calculateAverageOnList(disc_list[disc_index])
sCurve[disc_index] = {"avg" : avg}
return sCurve
class Calculus:
"""A class for calculating different metrics."""
def __init__(self):
pass
def calculateDefectiveChannels(self, deviation:list, thld:float)-> float:
"""Calculates the number of defective channels based on the deviation lists and a threshold.
Parameters
------
deviation : list
A list with the percentual deviations
thld : float
The threshold for the maximum allowed deviation
Returns
------
numOfDefects : int
The number of defects
"""
numOfDefects = 0
for dev in deviation:
if (dev > thld):
numOfDefects += 1
return numOfDefects
def calculateDeviationBetweenLists(self, data1:list, data2:list)-> list:
"""Calculates the percentual deviation between the n-th elements of the two lists.
The data1 list represents the reference value the deviation is relative to.
Parameters
------
data1 : list
List with numerical data. Acts as reference for the calculation
data2 : list
List with numerical data
Returns
------
deviation_list : list
List with the percentual deviation between the elements of both lists
"""
if len(data1) == len(data2):
deviation_list = []
for index in range(len(data1)):
if data1[index] == 0:
_dev = data2[index] - data1[index]
else:
_dev = (data2[index] - data1[index]) / data1[index]
deviation_list.append(_dev)
return deviation_list
def calculatePercentualDeviation(self, data:list)-> list:
"""Caclulates the percentual deviation between the elements of the given list.
Parameters
------
data : list
The list with the data
Returns
------
deviation_list : list
The list with the percentual deviation between the elements
"""
deviation_list = []
for index in range(len(data)-1):
if data[index] == 0:
data[index+1] - data[index]
else:
_dev = (data[index+1] - data[index])/data[index]
deviation_list.append(_dev)
return deviation_list
def calculateNAverageOnList(self, _list:list, size:int)-> list:
"""Calculate a reference average over a defined subset of elements from a given list
for each element in the list.
The used elements are defined by the size parameter and the index of the
referenced element.
Parameters
------
_list : list
A list of numerical elements
size : int
The number of elements to use - needs to be an odd number
Returns
------
avg_list : list
A list with numerical values
"""
numberPerSide = int(size / 2)
avg_list = []
for index in range(len(_list)):
leftBorderIndex = index - numberPerSide
rightBorderIndex = index + numberPerSide
if leftBorderIndex < 0:
leftBorderIndex = 0
if rightBorderIndex >= len(_list):
rightBorderIndex = len(_list)
_avg = self.calculateAverageOnList(_list[leftBorderIndex:rightBorderIndex])
avg_list.append(_avg)
return avg_list
def calculateAverageOnList(self, _list:list)-> float:
"""Calculates the average over a list of numerical values.
Parameters
------
_list : list
A list with numerical values
Returns
------
avg : float
The calculated average
"""
_avg = 0.0
for elem in _list:
_avg += elem
if len(_list) == 0:
avg = 0
else:
avg = float(_avg / len(_list))
return avg
def calculateStdDeviationOnList(self, _list:list)->float:
"""Calculates the standard deviation over all elements of a given list.
Parameters
------
_list : list
A list of numerical values
Returns
------
sigma : float
The sigma of the list
"""
_avg = self.calculateAverageOnList(_list)
_sigma_sum = 0.0
for elem in _list:
_sigma_sum = _sigma_sum + pow((elem - _avg), 2)
_sigma_avg = _sigma_sum / len(_list)
sigma = math.sqrt(_sigma_avg)
return sigma
def calculateUpperBorder(self, _list:list)->float:
"""Caclulates the upper border by calculating the average over all local maxima.
Parameters
------
_list : list
List with numerical values
Returns
------
upperBorder : float
The value of the upper border
"""
localMaxima = []
for index in range(1, len(_list)-1):
if _list[index] > _list[index-1] and _list[index] > _list[index+1]:
localMaxima.append(_list[index])
if _list[0] > _list[1]:
localMaxima.append(_list[0])
if _list[-2] < _list[-1]:
localMaxima.append(_list[-1])
upperBorder = self.calculateAverageOnList(localMaxima)
return upperBorder
def calculateLowerBorder(self, _list:list)-> float:
"""Caclulates the lower border by calculating the average over all local minima.
Parameters
------
_list : list
List with numerical values
Returns
------
lowerBorder : float
The value of the lower border
"""
localMinima = []
for index in range(1, len(_list)-1):
if _list[index] < _list[index-1] and _list[index] < _list[index+1]:
localMinima.append(_list[index])
if _list[0] < _list[1]:
localMinima.append(_list[0])
if _list[-2] > _list[-1]:
localMinima.append(_list[-1])
lowerBorder = self.calculateAverageOnList(localMinima)
return lowerBorder