forked from minjunJchoi/fluctana
-
Notifications
You must be signed in to change notification settings - Fork 1
/
diiiddata.py
114 lines (91 loc) · 3.34 KB
/
diiiddata.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
# Author : Minjun J. Choi ([email protected])
#
# Description : This code reads the DIII-D data using gadat.pro code
#
# Last updated
# 2018.10.08 :
import numpy as np
import matplotlib.pyplot as plt
import pidly
#### VAR to NODE
# TECE01--TECE40 : calibrated ECE
# Bt : toroidal field
# DENSITY : plasma density from EFIT01
# TAUE : energy confinement time
# BETAN : normalized beta
# CERQROTCT1--T48 : toroidal velocity from CER
VAR_NODE = {'ne':'DENR0F', 'Da04':'FS04', 'NBI_15L':'PINJ_15L', 'NBI_15R':'PINJ_15R', 'n1_amp':'n1rms', 'n2_amp':'n2rms'}
class DiiidData():
def __init__(self, shot ,clist):
self.shot = shot
self.clist = clist
def get_data(self, trange, norm=0, atrange=[1.0, 1.1], res=0, verbose=1):
if norm == 0:
if verbose == 1: print('Data is not normalized')
elif norm == 1:
if verbose == 1: print('Data is normalized by trange average')
elif norm == 2:
if verbose == 1: print('Data is normalized by atrange average')
self.trange = trange
# open idl
idl = pidly.IDL('/fusion/usc/opt/idl/idl84/bin/idl')
# --- loop starts --- #
clist_temp = self.clist.copy()
for i, cname in enumerate(clist_temp):
# set node
if cname in VAR_NODE:
node ='{:s}'.format(VAR_NODE[cname])
else:
node = cname
# load data
try:
#idl.pro('gadat,time,data,/alldata',node,self.shot,XMIN=self.trange[0]*1000.0,XMAX=self.trange[1]*1000.0)
idl.pro('gadat2,time,data,/alldata',node,self.shot,XMIN=self.trange[0]*1000.0,XMAX=self.trange[1]*1000.0)
time, v = idl.time, idl.data
if verbose == 1: print("Read {:d} - {:s} (number of data points = {:d})".format(self.shot, node, len(v)))
except:
self.clist.remove(cname)
if verbose == 1: print("Failed {:s}".format(node))
continue
# [ms] -> [s]
time = time/1000.0
# set data size
idx = np.where((time >= trange[0])*(time <= trange[1]))
idx1 = int(idx[0][0])
idx2 = int(idx[0][-1]+2)
time = time[idx1:idx2]
v = v[idx1:idx2]
if norm == 1:
v = v/np.mean(v) - 1
# expand dimension - concatenate
v = np.expand_dims(v, axis=0)
if i == 0:
data = v
else:
data = np.concatenate((data, v), axis=0)
# --- loop ends --- #
self.time = time
self.fs = round(1/(time[1] - time[0])/1000)*1000
self.data = data
# get channel position
self.channel_position()
# close idl
idl.close()
return time, data
def channel_position(self): # Needs updates ####################
cnum = len(self.clist)
self.rpos = np.arange(cnum) # R [m]
self.zpos = np.zeros(cnum) # z [m]
self.apos = np.arange(cnum) # angle [rad]
for c in range(cnum):
# Mirnov coils
# ECE
pass
if __name__ == "__main__":
pass
# g = DiiidData(shot=174964,clist=['ne'])
# g.get_data(trange=[0,10])
# plt.plot(g.time, g.data[0,:], color='k')
# plt.show()
# g.close
# DisconnectFromMds(g.socket)