Source code for ual.edge

from ualdef import *
import ual_low_level as ull
import numpy
import edge_types as types

class edge:
[docs] ''' class edge CPO for edge/SOL plasma description. Time-dependent CPO Attributes: - datainfo : class datainfo Generic information on a data item - grid : class complexgrid Grid description - species : class species_desc: array of species_descObj objects Description of ion species. Array of structures(nspecies) - fluid : class edge_fluid Fluid description of edge plasma. Time-dependent. - kinetic : class edge_kinetic Kinetic description of edge plasma. Time-dependent. - codeparam : class codeparam Code parameters - time : float Time [s]; Time-dependent; Scalar ''' def __init__(self):
[docs] self.base_path = 'edge' self.idx = EMPTY_INT self.datainfo = types.datainfo('datainfo') self.grid = types.complexgrid('grid') self.species = types.species_desc('species') self.fluid = types.edge_fluid('fluid') self.kinetic = types.edge_kinetic('kinetic') self.codeparam = types.codeparam('codeparam') self.time = EMPTY_DOUBLE def __str__(self, depth=0):
[docs] space = depth*'\t' ret = space + 'class edge\n' ret = ret + space + 'Attribute datainfo\n ' + self.datainfo.__str__(depth+1) ret = ret + space + 'Attribute grid\n ' + self.grid.__str__(depth+1) ret = ret + space + 'Attribute species\n ' + self.species.__str__(depth+1) ret = ret + space + 'Attribute fluid\n ' + self.fluid.__str__(depth+1) ret = ret + space + 'Attribute kinetic\n ' + self.kinetic.__str__(depth+1) ret = ret + space + 'Attribute codeparam\n ' + self.codeparam.__str__(depth+1) ret = ret + space + 'Attribute time: ' + str(self.time) + '\n' return ret def setExpIdx(self, idx):
[docs] self.idx = idx self.datainfo.setExpIdx(idx) self.grid.setExpIdx(idx) self.species.setExpIdx(idx) self.fluid.setExpIdx(idx) self.kinetic.setExpIdx(idx) self.codeparam.setExpIdx(idx) def putSlice(self, occurence=0):
[docs] if occurence==0: path = self.base_path else: path = self.base_path + '/' + str(occurence) cpopath = '' status = ull.beginCPOPutSlice(self.idx, path) self.datainfo.time = self.time self.datainfo.putSlice(path, cpopath) self.grid.time = self.time self.grid.putSlice(path, cpopath) self.species.time = self.time self.species.putSlice(path, cpopath) self.fluid.time = self.time self.fluid.putSlice(path, cpopath) self.kinetic.time = self.time self.kinetic.putSlice(path, cpopath) self.codeparam.time = self.time self.codeparam.putSlice(path, cpopath) status = ull.putDoubleSlice(self.idx, path, cpopath + 'time', self.time, self.time) check_status(status) status = ull.endCPOPutSlice(self.idx, path) def replaceLastSlice(self, occurence=0):
[docs] if occurence==0: path = self.base_path else: path = self.base_path + '/' + str(occurence) cpopath = '' status = ull.beginCPOReplaceLastSlice(self.idx, path) self.datainfo.replaceLastSlice(path, cpopath) self.grid.replaceLastSlice(path, cpopath) self.species.replaceLastSlice(path, cpopath) self.fluid.replaceLastSlice(path, cpopath) self.kinetic.replaceLastSlice(path, cpopath) self.codeparam.replaceLastSlice(path, cpopath) status = ull.replaceLastDoubleSlice(self.idx, path, cpopath + 'time', self.time) check_status(status) status = ull.endCPOReplaceLastSlice(self.idx, path) def putNonTimed(self, occurence=0):
[docs] if occurence==0: path = self.base_path else: path = self.base_path + '/' + str(occurence) cpopath = '' self.deleteData(occurence) self.discardMemCache(occurence) status = ull.beginCPOPutNonTimed(self.idx, path) self.datainfo.putNonTimed(path, cpopath) self.grid.putNonTimed(path, cpopath) self.species.putNonTimed(path, cpopath) self.fluid.putNonTimed(path, cpopath) self.kinetic.putNonTimed(path, cpopath) self.codeparam.putNonTimed(path, cpopath) status = ull.endCPOPutNonTimed(self.idx, path) def getSlice(self, inTime, interpolMode, occurence=0):
[docs] if occurence==0: path = self.base_path else: path = self.base_path + '/' + str(occurence) cpopath = '' status = ull.beginCPOGetSlice(self.idx, path, self.time) self.datainfo.getSlice(path, cpopath, inTime, interpolMode) self.grid.getSlice(path, cpopath, inTime, interpolMode) self.species.getSlice(path, cpopath, inTime, interpolMode) self.fluid.getSlice(path, cpopath, inTime, interpolMode) self.kinetic.getSlice(path, cpopath, inTime, interpolMode) self.codeparam.getSlice(path, cpopath, inTime, interpolMode) status, ret_time, retTime = ull.getDoubleSlice(self.idx, path, cpopath + 'time', inTime, interpolMode) check_status(status) if not status: self.time = ret_time self.time = retTime status = ull.endCPOGetSlice(self.idx, path) def build_non_resampled_data(self, occurence=0):
[docs] if occurence==0: path = self.base_path else: path = self.base_path + '/' + str(occurence) cpopath = '' status, nbslice = ull.beginCPOGet(self.idx, path, True) check_status(status) array=None if nbslice > 0: datainfoList = self.datainfo.build_non_resampled_data(path, cpopath, nbslice) gridList = self.grid.build_non_resampled_data(path, cpopath, nbslice) speciesList = self.species.build_non_resampled_data(path, cpopath, nbslice) fluidList = self.fluid.build_non_resampled_data(path, cpopath, nbslice) kineticList = self.kinetic.build_non_resampled_data(path, cpopath, nbslice) codeparamList = self.codeparam.build_non_resampled_data(path, cpopath, nbslice) status, timeList = ull.getVect1DDouble(self.idx, path, cpopath + 'time') if len(timeList) == 0: timeList = numpy.resize(timeList, (nbslice)) check_status(status) array = [] for i in range(nbslice): slice = edge() slice.setExpIdx(self.idx) slice.datainfo = datainfoList[i] slice.grid = gridList[i] slice.species = speciesList[i] slice.fluid = fluidList[i] slice.kinetic = kineticList[i] slice.codeparam = codeparamList[i] slice.time = timeList[i].copy().astype(float) array.append(slice) else: print 'error, nbslice must be > 0 and got:' + str(nbslice) status = ull.endCPOGet(self.idx, path) return array def flushCache(self, occurence=0):
[docs] if occurence==0: path = self.base_path else: path = self.base_path + '/' + str(occurence) cpopath = '' self.datainfo.flushCache(path, cpopath) self.grid.flushCache(path, cpopath) ull.euitm_flush(self.idx, path, cpopath + 'species') self.fluid.flushCache(path, cpopath) self.kinetic.flushCache(path, cpopath) self.codeparam.flushCache(path, cpopath) ull.euitm_flush(self.idx, path, cpopath + 'time') def discardMemCache(self, occurence=0):
[docs] if occurence==0: path = self.base_path else: path = self.base_path + '/' + str(occurence) cpopath = '' self.datainfo.discardMemCache(path, cpopath) self.grid.discardMemCache(path, cpopath) ull.euitm_discard_mem(self.idx, path, cpopath + 'species') self.fluid.discardMemCache(path, cpopath) self.kinetic.discardMemCache(path, cpopath) self.codeparam.discardMemCache(path, cpopath) ull.euitm_discard_mem(self.idx, path, cpopath + 'time') def deleteData(self, occurence=0):
[docs] if occurence==0: path = self.base_path else: path = self.base_path + '/' + str(occurence) cpopath = '' self.datainfo.deleteData(path, cpopath) self.grid.deleteData(path, cpopath) ull.deleteData(self.idx, path, cpopath + 'species') self.fluid.deleteData(path, cpopath) self.kinetic.deleteData(path, cpopath) self.codeparam.deleteData(path, cpopath) ull.deleteData(self.idx, path, cpopath + 'time') class edgeArray:
[docs] ''' class edgeArray CPO for edge/SOL plasma description. Time-dependent CPO Attributes: - array : list of edge Each list element correspond to one time slice. ''' def __init__(self):
[docs] self.array = [] self.idx = EMPTY_INT def __str__(self, depth=0):
[docs] space = depth*'\t' ret = space + 'class edgeArray nb_cpos=%d\n' %(len(self.array)) for i in range(len(self.array)): ret = ret + space + 'edge cpos=%d\n' %(i) + self.array[i].__str__(depth+1) return ret def setExpIdx(self, idx):
[docs] self.idx = idx def put(self, occurence=0):
[docs] if (len(self.array)>0): self.array[0].putNonTimed(occurence) for i in self.array: i.putSlice(occurence) else: print('no time slice to be put') def get(self, occurence=0):
[docs] cpo = edge() cpo.setExpIdx(self.idx) self.array = cpo.build_non_resampled_data(occurence) def resize(self, nb_cpos):
[docs] self.array = [] for i in range(nb_cpos): self.array.append(edge()) self.array[i].setExpIdx(self.idx) def deleteAll(self, occurence=0):
[docs] cpo = edge() cpo.setExpIdx(self.idx) cpo.deleteData(occurence)