##// END OF EJS Templates
schain
RSS
Add sleep for reload in online mode jroIO_digitalRF.py
avaldez -
r1266:6ae8a5868751
parent child
Show More
Side by Side Unified
Context file:
r1266:6ae8a5868751 Loading diff...:
Show file before | Show file after | Hide comments
@@ -1,792 +1,794
1 1
2 2 '''
3 3 Created on Jul 3, 2014
4 4
5 5 @author: roj-idl71
6 6 '''
7 7 # SUBCHANNELS EN VEZ DE CHANNELS
8 8 # BENCHMARKS -> PROBLEMAS CON ARCHIVOS GRANDES -> INCONSTANTE EN EL TIEMPO
9 9 # ACTUALIZACION DE VERSION
10 10 # HEADERS
11 11 # MODULO DE ESCRITURA
12 12 # METADATA
13 13
14 14 import os
15 15 import time
16 16 import datetime
17 17 import numpy
18 18 import timeit
19 19 from fractions import Fraction
20 20 from time import time
21 from time import sleep
21 22
22 23 import schainpy.admin
23 24 from schainpy.model.data.jroheaderIO import RadarControllerHeader, SystemHeader
24 25 from schainpy.model.data.jrodata import Voltage
25 26 from schainpy.model.proc.jroproc_base import ProcessingUnit, Operation, MPDecorator
26 27
27 28 import pickle
28 29 try:
29 30 import digital_rf
30 31 except:
31 32 print('You should install "digital_rf" module if you want to read Digital RF data')
32 33
33 34 @MPDecorator
34 35 class DigitalRFReader(ProcessingUnit):
35 36 '''
36 37 classdocs
37 38 '''
38 39
39 40 def __init__(self):
40 41 '''
41 42 Constructor
42 43 '''
43 44
44 45 ProcessingUnit.__init__(self)
45 46
46 47 self.dataOut = Voltage()
47 48 self.__printInfo = True
48 49 self.__flagDiscontinuousBlock = False
49 50 self.__bufferIndex = 9999999
50 51 self.__codeType = 0
51 52 self.__ippKm = None
52 53 self.__nCode = None
53 54 self.__nBaud = None
54 55 self.__code = None
55 56 self.dtype = None
56 57 self.oldAverage = None
57 58 self.path = None
58 59
59 60 def close(self):
60 61 print('Average of writing to digital rf format is ', self.oldAverage * 1000)
61 62 return
62 63
63 64 def __getCurrentSecond(self):
64 65
65 66 return self.__thisUnixSample / self.__sample_rate
66 67
67 68 thisSecond = property(__getCurrentSecond, "I'm the 'thisSecond' property.")
68 69
69 70 def __setFileHeader(self):
70 71 '''
71 72 In this method will be initialized every parameter of dataOut object (header, no data)
72 73 '''
73 74 ippSeconds = 1.0 * self.__nSamples / self.__sample_rate
74 75
75 76 nProfiles = 1.0 / ippSeconds # Number of profiles in one second
76 77
77 78 try:
78 79 self.dataOut.radarControllerHeaderObj = RadarControllerHeader(
79 80 self.__radarControllerHeader)
80 81 except:
81 82 self.dataOut.radarControllerHeaderObj = RadarControllerHeader(
82 83 txA=0,
83 84 txB=0,
84 85 nWindows=1,
85 86 nHeights=self.__nSamples,
86 87 firstHeight=self.__firstHeigth,
87 88 deltaHeight=self.__deltaHeigth,
88 89 codeType=self.__codeType,
89 90 nCode=self.__nCode, nBaud=self.__nBaud,
90 91 code=self.__code)
91 92
92 93 try:
93 94 self.dataOut.systemHeaderObj = SystemHeader(self.__systemHeader)
94 95 except:
95 96 self.dataOut.systemHeaderObj = SystemHeader(nSamples=self.__nSamples,
96 97 nProfiles=nProfiles,
97 98 nChannels=len(
98 99 self.__channelList),
99 100 adcResolution=14)
100 101 self.dataOut.type = "Voltage"
101 102
102 103 self.dataOut.data = None
103 104
104 105 self.dataOut.dtype = self.dtype
105 106
106 107 # self.dataOut.nChannels = 0
107 108
108 109 # self.dataOut.nHeights = 0
109 110
110 111 self.dataOut.nProfiles = int(nProfiles)
111 112
112 113 self.dataOut.heightList = self.__firstHeigth + \
113 114 numpy.arange(self.__nSamples, dtype=numpy.float) * \
114 115 self.__deltaHeigth
115 116
116 117 self.dataOut.channelList = list(range(self.__num_subchannels))
117 118
118 119 self.dataOut.blocksize = self.dataOut.getNChannels() * self.dataOut.getNHeights()
119 120
120 121 # self.dataOut.channelIndexList = None
121 122
122 123 self.dataOut.flagNoData = True
123 124
124 125 self.dataOut.flagDataAsBlock = False
125 126 # Set to TRUE if the data is discontinuous
126 127 self.dataOut.flagDiscontinuousBlock = False
127 128
128 129 self.dataOut.utctime = None
129 130
130 131 # timezone like jroheader, difference in minutes between UTC and localtime
131 132 self.dataOut.timeZone = self.__timezone / 60
132 133
133 134 self.dataOut.dstFlag = 0
134 135
135 136 self.dataOut.errorCount = 0
136 137
137 138 try:
138 139 self.dataOut.nCohInt = self.fixed_metadata_dict.get(
139 140 'nCohInt', self.nCohInt)
140 141
141 142 # asumo que la data esta decodificada
142 143 self.dataOut.flagDecodeData = self.fixed_metadata_dict.get(
143 144 'flagDecodeData', self.flagDecodeData)
144 145
145 146 # asumo que la data esta sin flip
146 147 self.dataOut.flagDeflipData = self.fixed_metadata_dict['flagDeflipData']
147 148
148 149 self.dataOut.flagShiftFFT = self.fixed_metadata_dict['flagShiftFFT']
149 150
150 151 self.dataOut.useLocalTime = self.fixed_metadata_dict['useLocalTime']
151 152 except:
152 153 pass
153 154
154 155 self.dataOut.ippSeconds = ippSeconds
155 156
156 157 # Time interval between profiles
157 158 # self.dataOut.timeInterval = self.dataOut.ippSeconds * self.dataOut.nCohInt
158 159
159 160 self.dataOut.frequency = self.__frequency
160 161
161 162 self.dataOut.realtime = self.__online
162 163
163 164 def findDatafiles(self, path, startDate=None, endDate=None):
164 165
165 166 if not os.path.isdir(path):
166 167 return []
167 168
168 169 try:
169 170 digitalReadObj = digital_rf.DigitalRFReader(
170 171 path, load_all_metadata=True)
171 172 except:
172 173 digitalReadObj = digital_rf.DigitalRFReader(path)
173 174
174 175 channelNameList = digitalReadObj.get_channels()
175 176
176 177 if not channelNameList:
177 178 return []
178 179
179 180 metadata_dict = digitalReadObj.get_rf_file_metadata(channelNameList[0])
180 181
181 182 sample_rate = metadata_dict['sample_rate'][0]
182 183
183 184 this_metadata_file = digitalReadObj.get_metadata(channelNameList[0])
184 185
185 186 try:
186 187 timezone = this_metadata_file['timezone'].value
187 188 except:
188 189 timezone = 0
189 190
190 191 startUTCSecond, endUTCSecond = digitalReadObj.get_bounds(
191 192 channelNameList[0]) / sample_rate - timezone
192 193
193 194 startDatetime = datetime.datetime.utcfromtimestamp(startUTCSecond)
194 195 endDatatime = datetime.datetime.utcfromtimestamp(endUTCSecond)
195 196
196 197 if not startDate:
197 198 startDate = startDatetime.date()
198 199
199 200 if not endDate:
200 201 endDate = endDatatime.date()
201 202
202 203 dateList = []
203 204
204 205 thisDatetime = startDatetime
205 206
206 207 while(thisDatetime <= endDatatime):
207 208
208 209 thisDate = thisDatetime.date()
209 210
210 211 if thisDate < startDate:
211 212 continue
212 213
213 214 if thisDate > endDate:
214 215 break
215 216
216 217 dateList.append(thisDate)
217 218 thisDatetime += datetime.timedelta(1)
218 219
219 220 return dateList
220 221
221 222 def setup(self, path=None,
222 223 startDate=None,
223 224 endDate=None,
224 225 startTime=datetime.time(0, 0, 0),
225 226 endTime=datetime.time(23, 59, 59),
226 227 channelList=None,
227 228 nSamples=None,
228 229 online=False,
229 230 delay=60,
230 231 buffer_size=1024,
231 232 ippKm=None,
232 233 nCohInt=1,
233 234 nCode=1,
234 235 nBaud=1,
235 236 flagDecodeData=False,
236 237 code=numpy.ones((1, 1), dtype=numpy.int),
237 238 **kwargs):
238 239 '''
239 240 In this method we should set all initial parameters.
240 241
241 242 Inputs:
242 243 path
243 244 startDate
244 245 endDate
245 246 startTime
246 247 endTime
247 248 set
248 249 expLabel
249 250 ext
250 251 online
251 252 delay
252 253 '''
253 254 self.path = path
254 255 self.nCohInt = nCohInt
255 256 self.flagDecodeData = flagDecodeData
256 257 self.i = 0
257 258 if not os.path.isdir(path):
258 259 raise ValueError("[Reading] Directory %s does not exist" % path)
259 260
260 261 try:
261 262 self.digitalReadObj = digital_rf.DigitalRFReader(
262 263 path, load_all_metadata=True)
263 264 except:
264 265 self.digitalReadObj = digital_rf.DigitalRFReader(path)
265 266
266 267 channelNameList = self.digitalReadObj.get_channels()
267 268
268 269 if not channelNameList:
269 270 raise ValueError("[Reading] Directory %s does not have any files" % path)
270 271
271 272 if not channelList:
272 273 channelList = list(range(len(channelNameList)))
273 274
274 275 ########## Reading metadata ######################
275 276
276 277 top_properties = self.digitalReadObj.get_properties(
277 278 channelNameList[channelList[0]])
278 279
279 280 self.__num_subchannels = top_properties['num_subchannels']
280 281 self.__sample_rate = 1.0 * \
281 282 top_properties['sample_rate_numerator'] / \
282 283 top_properties['sample_rate_denominator']
283 284 # self.__samples_per_file = top_properties['samples_per_file'][0]
284 285 self.__deltaHeigth = 1e6 * 0.15 / self.__sample_rate # why 0.15?
285 286
286 287 this_metadata_file = self.digitalReadObj.get_digital_metadata(
287 288 channelNameList[channelList[0]])
288 289 metadata_bounds = this_metadata_file.get_bounds()
289 290 self.fixed_metadata_dict = this_metadata_file.read(
290 291 metadata_bounds[0])[metadata_bounds[0]] # GET FIRST HEADER
291 292
292 293 try:
293 294 self.__processingHeader = self.fixed_metadata_dict['processingHeader']
294 295 self.__radarControllerHeader = self.fixed_metadata_dict['radarControllerHeader']
295 296 self.__systemHeader = self.fixed_metadata_dict['systemHeader']
296 297 self.dtype = pickle.loads(self.fixed_metadata_dict['dtype'])
297 298 except:
298 299 pass
299 300
300 301 self.__frequency = None
301 302
302 303 self.__frequency = self.fixed_metadata_dict.get('frequency', 1)
303 304
304 305 self.__timezone = self.fixed_metadata_dict.get('timezone', 18000)
305 306
306 307 try:
307 308 nSamples = self.fixed_metadata_dict['nSamples']
308 309 except:
309 310 nSamples = None
310 311
311 312 self.__firstHeigth = 0
312 313
313 314 try:
314 315 codeType = self.__radarControllerHeader['codeType']
315 316 except:
316 317 codeType = 0
317 318
318 319 try:
319 320 if codeType:
320 321 nCode = self.__radarControllerHeader['nCode']
321 322 nBaud = self.__radarControllerHeader['nBaud']
322 323 code = self.__radarControllerHeader['code']
323 324 except:
324 325 pass
325 326
326 327 if not ippKm:
327 328 try:
328 329 # seconds to km
329 330 ippKm = self.__radarControllerHeader['ipp']
330 331 except:
331 332 ippKm = None
332 333 ####################################################
333 334 self.__ippKm = ippKm
334 335 startUTCSecond = None
335 336 endUTCSecond = None
336 337
337 338 if startDate:
338 339 startDatetime = datetime.datetime.combine(startDate, startTime)
339 340 startUTCSecond = (
340 341 startDatetime - datetime.datetime(1970, 1, 1)).total_seconds() + self.__timezone
341 342
342 343 if endDate:
343 344 endDatetime = datetime.datetime.combine(endDate, endTime)
344 345 endUTCSecond = (endDatetime - datetime.datetime(1970,
345 346 1, 1)).total_seconds() + self.__timezone
346 347
347 348 start_index, end_index = self.digitalReadObj.get_bounds(
348 349 channelNameList[channelList[0]])
349 350
350 351 if not startUTCSecond:
351 352 startUTCSecond = start_index / self.__sample_rate
352 353
353 354 if start_index > startUTCSecond * self.__sample_rate:
354 355 startUTCSecond = start_index / self.__sample_rate
355 356
356 357 if not endUTCSecond:
357 358 endUTCSecond = end_index / self.__sample_rate
358 359
359 360 if end_index < endUTCSecond * self.__sample_rate:
360 361 endUTCSecond = end_index / self.__sample_rate
361 362 if not nSamples:
362 363 if not ippKm:
363 364 raise ValueError("[Reading] nSamples or ippKm should be defined")
364 365 nSamples = int(ippKm / (1e6 * 0.15 / self.__sample_rate))
365 366 channelBoundList = []
366 367 channelNameListFiltered = []
367 368
368 369 for thisIndexChannel in channelList:
369 370 thisChannelName = channelNameList[thisIndexChannel]
370 371 start_index, end_index = self.digitalReadObj.get_bounds(
371 372 thisChannelName)
372 373 channelBoundList.append((start_index, end_index))
373 374 channelNameListFiltered.append(thisChannelName)
374 375
375 376 self.profileIndex = 0
376 377 self.i = 0
377 378 self.__delay = delay
378 379
379 380 self.__codeType = codeType
380 381 self.__nCode = nCode
381 382 self.__nBaud = nBaud
382 383 self.__code = code
383 384
384 385 self.__datapath = path
385 386 self.__online = online
386 387 self.__channelList = channelList
387 388 self.__channelNameList = channelNameListFiltered
388 389 self.__channelBoundList = channelBoundList
389 390 self.__nSamples = nSamples
390 391 self.__samples_to_read = int(nSamples) # FIJO: AHORA 40
391 392 self.__nChannels = len(self.__channelList)
392 393
393 394 self.__startUTCSecond = startUTCSecond
394 395 self.__endUTCSecond = endUTCSecond
395 396
396 397 self.__timeInterval = 1.0 * self.__samples_to_read / \
397 398 self.__sample_rate # Time interval
398 399
399 400 if online:
400 401 # self.__thisUnixSample = int(endUTCSecond*self.__sample_rate - 4*self.__samples_to_read)
401 402 startUTCSecond = numpy.floor(endUTCSecond)
402 403
403 404 # por que en el otro metodo lo primero q se hace es sumar samplestoread
404 405 self.__thisUnixSample = int(startUTCSecond * self.__sample_rate) - self.__samples_to_read
405 406
406 407 self.__data_buffer = numpy.zeros(
407 408 (self.__num_subchannels, self.__samples_to_read), dtype=numpy.complex)
408 409
409 410 self.__setFileHeader()
410 411 self.isConfig = True
411 412
412 413 print("[Reading] Digital RF Data was found from %s to %s " % (
413 414 datetime.datetime.utcfromtimestamp(
414 415 self.__startUTCSecond - self.__timezone),
415 416 datetime.datetime.utcfromtimestamp(
416 417 self.__endUTCSecond - self.__timezone)
417 418 ))
418 419
419 420 print("[Reading] Starting process from %s to %s" % (datetime.datetime.utcfromtimestamp(startUTCSecond - self.__timezone),
420 421 datetime.datetime.utcfromtimestamp(
421 422 endUTCSecond - self.__timezone)
422 423 ))
423 424 self.oldAverage = None
424 425 self.count = 0
425 426 self.executionTime = 0
426 427
427 428 def __reload(self):
428 429 # print
429 430 # print "%s not in range [%s, %s]" %(
430 431 # datetime.datetime.utcfromtimestamp(self.thisSecond - self.__timezone),
431 432 # datetime.datetime.utcfromtimestamp(self.__startUTCSecond - self.__timezone),
432 433 # datetime.datetime.utcfromtimestamp(self.__endUTCSecond - self.__timezone)
433 434 # )
434 435 print("[Reading] reloading metadata ...")
435 436
436 437 try:
437 438 self.digitalReadObj.reload(complete_update=True)
438 439 except:
439 440 self.digitalReadObj = digital_rf.DigitalRFReader(self.path)
440 441
441 442 start_index, end_index = self.digitalReadObj.get_bounds(
442 443 self.__channelNameList[self.__channelList[0]])
443 444
444 445 if start_index > self.__startUTCSecond * self.__sample_rate:
445 446 self.__startUTCSecond = 1.0 * start_index / self.__sample_rate
446 447
447 448 if end_index > self.__endUTCSecond * self.__sample_rate:
448 449 self.__endUTCSecond = 1.0 * end_index / self.__sample_rate
449 450 print()
450 451 print("[Reading] New timerange found [%s, %s] " % (
451 452 datetime.datetime.utcfromtimestamp(
452 453 self.__startUTCSecond - self.__timezone),
453 454 datetime.datetime.utcfromtimestamp(
454 455 self.__endUTCSecond - self.__timezone)
455 456 ))
456 457
457 458 return True
458 459
459 460 return False
460 461
461 462 def timeit(self, toExecute):
462 463 t0 = time.time()
463 464 toExecute()
464 465 self.executionTime = time.time() - t0
465 466 if self.oldAverage is None:
466 467 self.oldAverage = self.executionTime
467 468 self.oldAverage = (self.executionTime + self.count *
468 469 self.oldAverage) / (self.count + 1.0)
469 470 self.count = self.count + 1.0
470 471 return
471 472
472 473 def __readNextBlock(self, seconds=30, volt_scale=1):
473 474 '''
474 475 '''
475 476
476 477 # Set the next data
477 478 self.__flagDiscontinuousBlock = False
478 479 self.__thisUnixSample += self.__samples_to_read
479 480
480 481 if self.__thisUnixSample + 2 * self.__samples_to_read > self.__endUTCSecond * self.__sample_rate:
481 482 print ("[Reading] There are no more data into selected time-range")
482 483 if self.__online:
484 sleep(3)
483 485 self.__reload()
484 486 else:
485 487 return False
486 488
487 489 if self.__thisUnixSample + 2 * self.__samples_to_read > self.__endUTCSecond * self.__sample_rate:
488 490 return False
489 491 self.__thisUnixSample -= self.__samples_to_read
490 492
491 493 indexChannel = 0
492 494
493 495 dataOk = False
494 496
495 497 for thisChannelName in self.__channelNameList: # TODO VARIOS CHANNELS?
496 498 for indexSubchannel in range(self.__num_subchannels):
497 499 try:
498 500 t0 = time()
499 501 result = self.digitalReadObj.read_vector_c81d(self.__thisUnixSample,
500 502 self.__samples_to_read,
501 503 thisChannelName, sub_channel=indexSubchannel)
502 504 self.executionTime = time() - t0
503 505 if self.oldAverage is None:
504 506 self.oldAverage = self.executionTime
505 507 self.oldAverage = (
506 508 self.executionTime + self.count * self.oldAverage) / (self.count + 1.0)
507 509 self.count = self.count + 1.0
508 510
509 511 except IOError as e:
510 512 # read next profile
511 513 self.__flagDiscontinuousBlock = True
512 514 print("[Reading] %s" % datetime.datetime.utcfromtimestamp(self.thisSecond - self.__timezone), e)
513 515 break
514 516
515 517 if result.shape[0] != self.__samples_to_read:
516 518 self.__flagDiscontinuousBlock = True
517 519 print("[Reading] %s: Too few samples were found, just %d/%d samples" % (datetime.datetime.utcfromtimestamp(self.thisSecond - self.__timezone),
518 520 result.shape[0],
519 521 self.__samples_to_read))
520 522 break
521 523
522 524 self.__data_buffer[indexSubchannel, :] = result * volt_scale
523 525 indexChannel+=1
524 526
525 527 dataOk = True
526 528
527 529 self.__utctime = self.__thisUnixSample / self.__sample_rate
528 530
529 531 if not dataOk:
530 532 return False
531 533
532 534 print("[Reading] %s: %d samples <> %f sec" % (datetime.datetime.utcfromtimestamp(self.thisSecond - self.__timezone),
533 535 self.__samples_to_read,
534 536 self.__timeInterval))
535 537
536 538 self.__bufferIndex = 0
537 539
538 540 return True
539 541
540 542 def __isBufferEmpty(self):
541 543 return self.__bufferIndex > self.__samples_to_read - self.__nSamples # 40960 - 40
542 544
543 545 def getData(self, seconds=30, nTries=5):
544 546 '''
545 547 This method gets the data from files and put the data into the dataOut object
546 548
547 549 In addition, increase el the buffer counter in one.
548 550
549 551 Return:
550 552 data : retorna un perfil de voltages (alturas * canales) copiados desde el
551 553 buffer. Si no hay mas archivos a leer retorna None.
552 554
553 555 Affected:
554 556 self.dataOut
555 557 self.profileIndex
556 558 self.flagDiscontinuousBlock
557 559 self.flagIsNewBlock
558 560 '''
559 561 #print("getdata")
560 562 err_counter = 0
561 563 self.dataOut.flagNoData = True
562 564
563 565 if self.__isBufferEmpty():
564 566 #print("hi")
565 567 self.__flagDiscontinuousBlock = False
566 568
567 569 while True:
568 570 #print ("q ha pasado")
569 571 if self.__readNextBlock():
570 572 break
571 573 if self.__thisUnixSample > self.__endUTCSecond * self.__sample_rate:
572 574 raise schainpy.admin.SchainError('Error')
573 575 return
574 576
575 577 if self.__flagDiscontinuousBlock:
576 578 raise schainpy.admin.SchainError('discontinuous block found')
577 579 return
578 580
579 581 if not self.__online:
580 582 raise schainpy.admin.SchainError('Online?')
581 583 return
582 584
583 585 err_counter += 1
584 586 if err_counter > nTries:
585 587 raise schainpy.admin.SchainError('Max retrys reach')
586 588 return
587 589
588 590 print('[Reading] waiting %d seconds to read a new block' % seconds)
589 591 time.sleep(seconds)
590 592
591 593 self.dataOut.data = self.__data_buffer[:, self.__bufferIndex:self.__bufferIndex + self.__nSamples]
592 594 self.dataOut.utctime = ( self.__thisUnixSample + self.__bufferIndex) / self.__sample_rate
593 595 self.dataOut.flagNoData = False
594 596 self.dataOut.flagDiscontinuousBlock = self.__flagDiscontinuousBlock
595 597 self.dataOut.profileIndex = self.profileIndex
596 598
597 599 self.__bufferIndex += self.__nSamples
598 600 self.profileIndex += 1
599 601
600 602 if self.profileIndex == self.dataOut.nProfiles:
601 603 self.profileIndex = 0
602 604
603 605 return True
604 606
605 607 def printInfo(self):
606 608 '''
607 609 '''
608 610 if self.__printInfo == False:
609 611 return
610 612
611 613 # self.systemHeaderObj.printInfo()
612 614 # self.radarControllerHeaderObj.printInfo()
613 615
614 616 self.__printInfo = False
615 617
616 618 def printNumberOfBlock(self):
617 619 '''
618 620 '''
619 621 return
620 622 # print self.profileIndex
621 623
622 624 def run(self, **kwargs):
623 625 '''
624 626 This method will be called many times so here you should put all your code
625 627 '''
626 628
627 629 if not self.isConfig:
628 630 self.setup(**kwargs)
629 631 #self.i = self.i+1
630 632 self.getData(seconds=self.__delay)
631 633
632 634 return
633 635
634 636
635 637 class DigitalRFWriter(Operation):
636 638 '''
637 639 classdocs
638 640 '''
639 641
640 642 def __init__(self, **kwargs):
641 643 '''
642 644 Constructor
643 645 '''
644 646 Operation.__init__(self, **kwargs)
645 647 self.metadata_dict = {}
646 648 self.dataOut = None
647 649 self.dtype = None
648 650 self.oldAverage = 0
649 651
650 652 def setHeader(self):
651 653
652 654 self.metadata_dict['frequency'] = self.dataOut.frequency
653 655 self.metadata_dict['timezone'] = self.dataOut.timeZone
654 656 self.metadata_dict['dtype'] = pickle.dumps(self.dataOut.dtype)
655 657 self.metadata_dict['nProfiles'] = self.dataOut.nProfiles
656 658 self.metadata_dict['heightList'] = self.dataOut.heightList
657 659 self.metadata_dict['channelList'] = self.dataOut.channelList
658 660 self.metadata_dict['flagDecodeData'] = self.dataOut.flagDecodeData
659 661 self.metadata_dict['flagDeflipData'] = self.dataOut.flagDeflipData
660 662 self.metadata_dict['flagShiftFFT'] = self.dataOut.flagShiftFFT
661 663 self.metadata_dict['useLocalTime'] = self.dataOut.useLocalTime
662 664 self.metadata_dict['nCohInt'] = self.dataOut.nCohInt
663 665 self.metadata_dict['type'] = self.dataOut.type
664 666 self.metadata_dict['flagDataAsBlock']= getattr(
665 667 self.dataOut, 'flagDataAsBlock', None) # chequear
666 668
667 669 def setup(self, dataOut, path, frequency, fileCadence, dirCadence, metadataCadence, set=0, metadataFile='metadata', ext='.h5'):
668 670 '''
669 671 In this method we should set all initial parameters.
670 672 Input:
671 673 dataOut: Input data will also be outputa data
672 674 '''
673 675 self.setHeader()
674 676 self.__ippSeconds = dataOut.ippSeconds
675 677 self.__deltaH = dataOut.getDeltaH()
676 678 self.__sample_rate = 1e6 * 0.15 / self.__deltaH
677 679 self.__dtype = dataOut.dtype
678 680 if len(dataOut.dtype) == 2:
679 681 self.__dtype = dataOut.dtype[0]
680 682 self.__nSamples = dataOut.systemHeaderObj.nSamples
681 683 self.__nProfiles = dataOut.nProfiles
682 684
683 685 if self.dataOut.type != 'Voltage':
684 686 raise 'Digital RF cannot be used with this data type'
685 687 self.arr_data = numpy.ones((1, dataOut.nFFTPoints * len(
686 688 self.dataOut.channelList)), dtype=[('r', self.__dtype), ('i', self.__dtype)])
687 689 else:
688 690 self.arr_data = numpy.ones((self.__nSamples, len(
689 691 self.dataOut.channelList)), dtype=[('r', self.__dtype), ('i', self.__dtype)])
690 692
691 693 file_cadence_millisecs = 1000
692 694
693 695 sample_rate_fraction = Fraction(self.__sample_rate).limit_denominator()
694 696 sample_rate_numerator = int(sample_rate_fraction.numerator)
695 697 sample_rate_denominator = int(sample_rate_fraction.denominator)
696 698 start_global_index = dataOut.utctime * self.__sample_rate
697 699
698 700 uuid = 'prueba'
699 701 compression_level = 0
700 702 checksum = False
701 703 is_complex = True
702 704 num_subchannels = len(dataOut.channelList)
703 705 is_continuous = True
704 706 marching_periods = False
705 707
706 708 self.digitalWriteObj = digital_rf.DigitalRFWriter(path, self.__dtype, dirCadence,
707 709 fileCadence, start_global_index,
708 710 sample_rate_numerator, sample_rate_denominator, uuid, compression_level, checksum,
709 711 is_complex, num_subchannels, is_continuous, marching_periods)
710 712 metadata_dir = os.path.join(path, 'metadata')
711 713 os.system('mkdir %s' % (metadata_dir))
712 714 self.digitalMetadataWriteObj = digital_rf.DigitalMetadataWriter(metadata_dir, dirCadence, 1, # 236, file_cadence_millisecs / 1000
713 715 sample_rate_numerator, sample_rate_denominator,
714 716 metadataFile)
715 717 self.isConfig = True
716 718 self.currentSample = 0
717 719 self.oldAverage = 0
718 720 self.count = 0
719 721 return
720 722
721 723 def writeMetadata(self):
722 724 start_idx = self.__sample_rate * self.dataOut.utctime
723 725
724 726 self.metadata_dict['processingHeader'] = self.dataOut.processingHeaderObj.getAsDict(
725 727 )
726 728 self.metadata_dict['radarControllerHeader'] = self.dataOut.radarControllerHeaderObj.getAsDict(
727 729 )
728 730 self.metadata_dict['systemHeader'] = self.dataOut.systemHeaderObj.getAsDict(
729 731 )
730 732 self.digitalMetadataWriteObj.write(start_idx, self.metadata_dict)
731 733 return
732 734
733 735 def timeit(self, toExecute):
734 736 t0 = time()
735 737 toExecute()
736 738 self.executionTime = time() - t0
737 739 if self.oldAverage is None:
738 740 self.oldAverage = self.executionTime
739 741 self.oldAverage = (self.executionTime + self.count *
740 742 self.oldAverage) / (self.count + 1.0)
741 743 self.count = self.count + 1.0
742 744 return
743 745
744 746 def writeData(self):
745 747 if self.dataOut.type != 'Voltage':
746 748 raise 'Digital RF cannot be used with this data type'
747 749 for channel in self.dataOut.channelList:
748 750 for i in range(self.dataOut.nFFTPoints):
749 751 self.arr_data[1][channel * self.dataOut.nFFTPoints +
750 752 i]['r'] = self.dataOut.data[channel][i].real
751 753 self.arr_data[1][channel * self.dataOut.nFFTPoints +
752 754 i]['i'] = self.dataOut.data[channel][i].imag
753 755 else:
754 756 for i in range(self.dataOut.systemHeaderObj.nSamples):
755 757 for channel in self.dataOut.channelList:
756 758 self.arr_data[i][channel]['r'] = self.dataOut.data[channel][i].real
757 759 self.arr_data[i][channel]['i'] = self.dataOut.data[channel][i].imag
758 760
759 761 def f(): return self.digitalWriteObj.rf_write(self.arr_data)
760 762 self.timeit(f)
761 763
762 764 return
763 765
764 766 def run(self, dataOut, frequency=49.92e6, path=None, fileCadence=1000, dirCadence=36000, metadataCadence=1, **kwargs):
765 767 '''
766 768 This method will be called many times so here you should put all your code
767 769 Inputs:
768 770 dataOut: object with the data
769 771 '''
770 772 # print dataOut.__dict__
771 773 self.dataOut = dataOut
772 774 if not self.isConfig:
773 775 self.setup(dataOut, path, frequency, fileCadence,
774 776 dirCadence, metadataCadence, **kwargs)
775 777 self.writeMetadata()
776 778
777 779 self.writeData()
778 780
779 781 ## self.currentSample += 1
780 782 # if self.dataOut.flagDataAsBlock or self.currentSample == 1:
781 783 # self.writeMetadata()
782 784 ## if self.currentSample == self.__nProfiles: self.currentSample = 0
783 785
784 786 return dataOut# en la version 2.7 no aparece este return
785 787
786 788 def close(self):
787 789 print('[Writing] - Closing files ')
788 790 print('Average of writing to digital rf format is ', self.oldAverage * 1000)
789 791 try:
790 792 self.digitalWriteObj.close()
791 793 except:
792 794 pass
General Comments 0
You need to be logged in to leave comments. Login now