Package empro :: Package toolkit :: Module citifile
[frames] | no frames]

Source Code for Module empro.toolkit.citifile

  1  # Copyright 1983-2019 Keysight Technologies, Inc , Keysight Confidential 
  2   
  3  from UserDict import IterableUserDict as _IterableUserDict 
  4  from empro.toolkit import dataset as _dataset 
5 6 -class CitiFile(_IterableUserDict):
7 ''' 8 Class to hold the content of a CITI file in memory as EMPro DataSets. 9 '''
10 - def __init__(self, data_or_path=None, name=None):
11 ''' 12 CitiFile(path) -> reads a CITI file from disk. 13 CitiFile(name='DATA') -> an empty instance with a name 14 CitiFile({'S[1,1]': DataSet, 'S[1,2]': DataSet, ...}, name='DATA'): -> 15 a CitiFile made from a dictionary of EMPro DataSets. The keys 16 of the dictionary shall be strings and shall be valid names for 17 data series in a CITI file. The values of the dictionary shall 18 be one-dimensional EMPro datasets, and their dimension shall be 19 identical. This dimension will be used as the independent 20 variable of the CITI file. 21 CitiFile(matrix) -> a CitiFile from a matrix where matrix is a 22 L{empro.toolkit.dataset.DataSetMatrix} instance. 23 ''' 24 _IterableUserDict.__init__(self) 25 self.__dimension = None 26 if data_or_path is None: 27 self.name = name or 'DATA' 28 return 29 if isinstance(data_or_path, basestring): 30 self.__read(data_or_path) 31 return 32 if _dataset._isDataSet(data_or_path): 33 self[data_or_path.name] = data_or_path 34 else: 35 if isinstance(data_or_path, _dataset.DataSetMatrix): 36 keyFmt = "%s[%%s,%%s]" % data_or_path.name 37 else: 38 keyFmt = "%s" 39 for key, dataset in data_or_path.items(): 40 self[keyFmt % key] = dataset 41 self.name = name or _name(data_or_path) or 'DATA'
42 43 @property
44 - def var(self):
45 ''' 46 The independent variable of the CITI file as a EMPro DataSet. 47 This is also the dimension of all data DataSets. 48 ''' 49 return self.__dimension
50 51 @property
52 - def size(self):
53 ''' 54 length of the independent variable or dimension 55 ''' 56 return len(self.__dimension)
57
58 - def __setitem__(self, key, dataset):
59 assert dataset.numberOfDimensions() == 1, "datasets shall be one-dimensional" 60 if self.__dimension is None: 61 self.__dimension = dataset.dimension(0) 62 else: 63 assert self.__dimension == dataset.dimension(0), "all datasets shall have the same dimension" 64 _IterableUserDict.__setitem__(self, key, dataset)
65
66 - def getVarName(self):
67 varName = self.__dimension.name.upper() 68 return varName
69
70 - def write(self, path):
71 ''' 72 Write the CITI file to disk 73 ''' 74 from empro import units 75 self.__checkSize() 76 f = open(path, 'w') 77 f.write("CITIFILE A.01.00\n") 78 f.write("NAME %s\n" % self.name) 79 varName = self.__dimension.name.upper() 80 if 'FREQUENCY'.startswith(varName) or varName == 'FFT DIMENSION': 81 varName = 'FREQ' 82 f.write("VAR %s MAG %d\n" % (varName, len(self.__dimension))) 83 for key in sorted(self): 84 f.write("DATA %s %s\n" % (key, _dataType(self[key]))) 85 self.__writeBlock(f, self.__dimension, "MAG", "VAR_LIST_") 86 for key in sorted(self): 87 self.__writeBlock(f, self[key], _dataType(self[key])) 88 f.close()
89
90 - def writeAdsDataSet(self, path):
91 ''' 92 Write the data to disk as a Ads DataSet (".ds") file 93 ''' 94 self.__checkSize() 95 import empro 96 varName = self.getVarName() 97 if 'FREQUENCY'.startswith(varName): 98 varName = 'freq' 99 depVars = [] 100 numSorted = sorted(self, key=_numeric_key) 101 for key in numSorted: 102 depVars.append(key) 103 dsd = empro.DataSetDescription(self.name,varName,depVars) 104 indepVarValues = self.var 105 depVarValueLists = [] 106 for key in numSorted: 107 depVarValueLists.append(self[key]) 108 ds = empro.DataSet() 109 ds.create(path) 110 dsds = ds.newDataSet(dsd) 111 dsds.fill(indepVarValues,depVarValueLists) 112 ds.close()
113
114 - def asMatrix(self, dataName=None):
115 ''' 116 self.asMatrix('S') -> DataSetMatrix 117 CITI files often have data series of the form 'S[i,j]' or 'PORTZ[i]' 118 that form a matrix of data series. Using this function you can get them 119 as a DataSetMatrix in one call. The dataName argument is the name 120 without the brackets like 'S' or 'PORTZ'. If dataName is None, it 121 tries to guess it, if only one dataName is possible. 122 ''' 123 import re 124 if not dataName: 125 dataName = self.__guessDataName() 126 datasets = {} 127 for key in self.keys(): 128 match = re.match(r"^%s\[(\d)+,(\d+)\]$" % re.escape(dataName), key) 129 if match: 130 i, j = int(match.group(1)), int(match.group(2)) 131 datasets[i, j] = self[key] 132 continue 133 match = re.match(r"^%s\[(\d)+\]$" % re.escape(dataName), key) 134 if match: 135 i = int(match.group(1)) 136 datasets[i, i] = self[key] 137 assert datasets, "No subscripted %r data arrays found" % dataName 138 return _dataset.DataSetMatrix(datasets, name=dataName)
139
140 - def getPermutatedCiti(self, pdict):
141 ''' 142 Returns a CitiFile datastructure with the mapping pdict performed on the indices 143 self : input CitiFile datastructure 144 pdict : mapping to be performed on the indices of the form { fromIndex : toIndex, ... } 145 E.g. : 146 self.getPermutatedCiti({1:2, 2:3, 3:1}) -> CitiFile 147 citiOut.data['S[2,2]'] = self.data['S[1,1]'] 148 citiOut.data['S[2,1]'] = self.data['S[1,3]'] 149 citiOut.data['S[3,3]'] = self.data['S[2,2]'] 150 citiOut.data['S[1,1]'] = self.data['S[3,3]'] 151 citiOut.data['S[4,4]'] = self.data['S[4,4]'] # Indices not listed in the mapping are mapped to themselves 152 citiOut.data['S[4,1]'] = self.data['S[4,3]'] # Indices not listed in the mapping are mapped to themselves 153 ... 154 ''' 155 import re 156 citiOut = CitiFile(self) 157 for k in self.keys(): 158 match = re.match("(.*)\[(\d+),(\d+)\]$" , k) 159 if match: 160 i, j = int(match.group(2)), int(match.group(3)) 161 if not pdict.has_key(i): 162 pdict[i]=i 163 if not pdict.has_key(j): 164 pdict[j]=j 165 newk = "%s[%s,%s]" % (match.group(1), pdict[i], pdict[j]) 166 citiOut.data[newk]=self[k] 167 else: 168 match = re.match("(.*)\[(\d+)\]$" , k) 169 if match: 170 i = int(match.group(2)) 171 if not pdict.has_key(i): 172 pdict[i]=i 173 newk = "%s[%s]" % (match.group(1), pdict[i]) 174 citiOut.data[newk]=self[k] 175 else: 176 pass 177 return citiOut
178 179
180 - def __read(self, path):
181 import math, cmath 182 from empro import units 183 magangle = lambda mag, angle: mag * cmath.exp(1j * math.radians(angle)) 184 dbangle = lambda mag, angle: magangle(10.0 ** (mag / 20.0), angle) 185 paired = lambda caster: lambda s, c=caster: caster(*map(float, s.split(","))) 186 casterDict = { 187 'MAG': float, 188 'RI': paired(complex), 189 'MAGANGLE': paired(magangle), 190 'DBANGLE': paired(dbangle), 191 } 192 dataSpecs = [] 193 f = open(path) 194 for line in f: 195 fields = line.rstrip('\n').split() 196 if not fields: 197 continue 198 keyword, fields = fields[0], fields[1:] 199 if keyword == 'NAME': 200 self.name, = fields 201 if keyword == 'VAR': 202 dimensionName, dimensionType, dimensionSize = fields 203 elif keyword == 'DATA': 204 dataName, dataType = fields 205 dataSpecs.append((dataName, casterDict[dataType])) 206 elif keyword == 'VAR_LIST_BEGIN': 207 assert self.__dimension is None, "can only have one independent variable table/segment" 208 self.__dimension = self.__readBlock(f, 'VAR_LIST_END', dimensionName, casterDict[dimensionType]) 209 for unit in units.FREQUENCY, units.TIME: 210 if unit.startswith(dimensionName.upper()): 211 self.__dimension.unitClass = unit 212 break 213 elif keyword == 'BEGIN': 214 (name, caster), dataSpecs = dataSpecs[0], dataSpecs[1:] 215 self.data[name] = self.__readBlock(f, 'END', name, caster) 216 f.close() 217 assert not self.__dimension is None, "CITI file must have an independent variable" 218 for dataset in self.values(): 219 try: 220 re, im = dataset.real, dataset.imag 221 except AttributeError: 222 dataset.addDimension(self.__dimension) 223 else: 224 re.addDimension(self.__dimension) 225 im.addDimension(self.__dimension) 226 self.__checkSize(n=int(dimensionSize))
227
228 - def __readBlock(self, f, endTag, name, caster):
229 Xs = [] 230 for line in f: 231 if line.startswith(endTag): 232 break 233 Xs.append(caster(line.strip())) 234 return _dataset.makeDataSet(Xs, id=name)
235
236 - def __writeBlock(self, f, Xs, dataType, tagPrefix=''):
237 f.write("%sBEGIN\n" % tagPrefix) 238 if dataType == 'RI': 239 for x in Xs: 240 f.write("%r,%r\n" % (x.real, x.imag)) 241 else: 242 assert dataType == 'MAG' 243 for x in Xs: 244 f.write("%r\n" % x) 245 f.write("%sEND\n" % tagPrefix)
246
247 - def __checkSize(self, n=None):
248 assert all(dataset.numberOfDimensions() == 1 for dataset in self.values()), 'all value datasets must be one dimensional' 249 if not self.__dimension: 250 if not self.values(): 251 return 252 self.__dimension = self.values().dimension(0) 253 for dataset in self.values(): 254 assert dataset.dimension(0) == self.__dimension, "'%s' has incompatible dimension" 255 assert len(dataset) == self.size, "'%s' has incompatible length"
256
257 - def __guessDataName(self):
258 import re 259 possibleDataNames = tuple(set(re.match("^(.*?)(\[.+\])?$", key).group(1) for key in self.keys())) 260 assert len(possibleDataNames) > 0, "Unable to guess any data name" 261 assert len(possibleDataNames) == 1, "Unable to guess data name, try one of %r" % (possibleDataNames,) 262 return possibleDataNames[0]
263
264 265 -def read(path):
266 ''' 267 read(path) -> CitiFile 268 269 Reads a CITI file from disk. 270 ''' 271 return CitiFile(path)
272
273 274 -def write(path, data, name=None):
275 ''' 276 write(path, data [, name]) -> None 277 278 Writes a L{empro.toolkit.DataSetMatrix} or a dictionary of 279 L{empro.datasource.DataSet} to a CITI file. 280 ''' 281 CitiFile(data, name).write(path)
282
283 284 -def getPortNames(ctiFile):
285 """ 286 Returns the port names as specified in <ctiFile>. 287 Port name format: 288 #eesof Port[1] name1 289 #eesof Port[2] name2 290 ... 291 If not specified, throws an exception. 292 """ 293 portNames = [] 294 reader = open(ctiFile, 'r') 295 tag = "#eesof Port[" 296 line = reader.readline() 297 while not tag in line and line!="": 298 line = reader.readline() 299 portId = 0 300 while tag in line: 301 portId = portId + 1 302 def _extractPortName(line, portId): 303 toBeRemoved = "#eesof Port[%d] " % portId 304 portName = line.replace(toBeRemoved, "") 305 portName = portName.replace("\n", "") 306 portName = portName.replace("\r", "") 307 return portName
308 portName = _extractPortName(line, portId) 309 portNames.append(portName) 310 line = reader.readline() 311 reader.close() 312 return portNames 313
314 -def stos(originalCtiFile, newCtiFile, vecPortZref, verbose=True, useZprm=False, showProgress=False):
315 """ 316 Changes the reference impedance (PORTZ) from originalCtiFile 317 to vecPortZref, and writes the resulting S-parameters and PORTZ 318 to newCtiFile. 319 320 @param vecPortZref is a list of (port name, Zref) pairs 321 Example: vecPortZref = [("port1", 50), ("port2", 10)] 322 323 @param useZprm if True: stos = (stoz, renorm, ztos) 324 else stos = (stoy, renorm, ytos) 325 326 If the port names in vecPortZref don't match getPortNames(originalCtiFile) 327 an exception is thrown 328 """ 329 330 import empro.enparams 331 evaluator = empro.enparams.CitiEvaluator(originalCtiFile, False) 332 nbPorts = evaluator.nbPorts() 333 nbFreq = evaluator.nbStoredFrequencies() 334 if verbose: 335 print "LOADING <%s>" % originalCtiFile 336 print "\t# ports :", nbPorts 337 print "\t# frequencies :", nbFreq 338 339 # extracting the port names 340 # if not availble, the user-specified port names in vecPortZref will be used instead 341 originalPortNames = getPortNames(originalCtiFile) 342 if len(originalPortNames) == 0: 343 originalPortNames = [] 344 for name, Zref in vecPortZref: 345 originalPortNames.append(name) 346 347 # cti header 348 writer = open(newCtiFile, 'w') 349 writer.write("CITIFILE A.01.01") 350 writer.write("\n\n# FEM : 310.RC1 2015-11-27") 351 writer.write("\n# Date : Wed Dec 2 11:39:23 2015") 352 writer.write("\n\nNAME DATA\n") 353 for portId in range(nbPorts): 354 writer.write("\n#eesof Port[%d] %s" % (portId+1,originalPortNames[portId])) 355 356 # checking the ports and building Zref 357 newZref = [] 358 if len(originalPortNames) != len(vecPortZref): 359 raise Exception("%d (portName,Zref) pairs expected, %d found" % (len(originalPortNames),len(vecPortZref)) ) 360 cnt = 0 361 if verbose: print "\tChanging Zref:" 362 for name, Zref in vecPortZref: 363 if verbose: print "\t%3d. %25s : %f + i%f Ohm" % (cnt+1, originalPortNames[cnt], Zref.real, Zref.imag) 364 cnt = cnt + 1 365 newZref.append(Zref) 366 367 if showProgress: 368 import empro.gui 369 progressDialog = empro.gui.ProgressDialog("Converting", "Converting", "Converting", 0, nbFreq) 370 progressDialog.show() 371 372 # frequencies 373 writer.write("\nVAR freq MAG %d\n" % nbFreq) 374 for r in range(nbPorts): 375 for c in range(nbPorts): 376 writer.write("\nDATA S[%d,%d] RI" % (r+1,c+1)) 377 for r in range(nbPorts): 378 writer.write("\nDATA PORTZ[%d] RI" % (r+1)) 379 writer.write("\n") 380 381 # frequencies 382 writer.write("\nVAR_LIST_BEGIN") 383 for freq in evaluator.storedFrequencies(): 384 writer.write("\n%23.17G"%freq) 385 writer.write("\nVAR_LIST_END\n") 386 387 # S-parameters 388 Sorig = empro.enparams.VectorOfScatteringMatrices(nbPorts, nbFreq) 389 Snew = empro.enparams.VectorOfScatteringMatrices(nbPorts, nbFreq) 390 evaluator.loadSampledSparameters(Sorig) 391 if showProgress: 392 progressDialog.labelText = "Converting" 393 progressDialog.value = 0 394 for f in range(nbFreq): 395 Snew.setS(f, Sorig.S(f).stos(newZref)) 396 if showProgress and f % 25 == 0: 397 progressDialog.value = int(f * (100.0 / nbFreq)) 398 if showProgress: 399 progressDialog.labelText = "Writing" 400 progressDialog.value = 0 401 for r in range(nbPorts): 402 if showProgress and r % 10 == 0: 403 progressDialog.value = int(r * (100.0 / nbPorts)) 404 for c in range(nbPorts): 405 writer.write("\nBEGIN") 406 for f in range(nbFreq): 407 Sfrc = Snew.S(f).param(r,c) 408 writer.write("\n%23.17G, %23.17G" % (Sfrc.real, Sfrc.imag)) 409 writer.write("\nEND\n") 410 411 # Zref 412 for r in range(nbPorts): 413 Zrefr = newZref[r] 414 writer.write("\nBEGIN") 415 for f in range(nbFreq): 416 writer.write("\n%23.17G, %23.17G" % (Zrefr, 0.0)) 417 writer.write("\nEND\n") 418 419 if showProgress: 420 progressDialog.value = 1 421 422 writer.close() 423 if verbose: print "WRITTEN <%s>" % newCtiFile
424
425 # --- implementation ----------------------------------------------------------------------------- 426 427 -def _dataType(dataset):
428 assert _dataset._isDataSet(dataset), dataset 429 try: 430 re, im = dataset.real, dataset.imag 431 except AttributeError: 432 return 'MAG' 433 else: 434 return 'RI'
435
436 -def _name(x):
437 try: 438 return x.name 439 except AttributeError: 440 return None
441
442 -def _numeric_key(key):
443 ''' 444 The default sort method uses the alphanumeric lexicographical order. 445 For numbers larger than 9 this does not correspond to the desired numerical 446 order. E.g., S(10,1) comes lexicographically before S(1,1), while it should 447 come after it when the elements of a matrix are sorted. 448 449 This function, splits the key in a tuple of a alphanumeric prefix, followed 450 by 0, 1 or 2 numeric indices. Use as sorted(arg,key=_numeric_key) to sort 451 1D and 2D matrices (and of course scalars) in the correct way. 452 ''' 453 import re 454 match = re.match(r"^(.*)[\(\[](\d+),(\d+)[\)\]]$", key) 455 if match: 456 return (match.group(1), int(match.group(2)), int(match.group(3))) 457 match = re.match(r"^(.*)[\(\[](\d+)[\)\]]$", key) 458 if match: 459 return (match.group(1), int(match.group(2))) 460 return (key, )
461