openrvdas.logger.writers.record_screen_writer
No module-level documentation available.
1#!/usr/bin/env python3 2 3import shutil 4 5from logger.utils.das_record import DASRecord # noqa: E402 6from logger.writers.writer import Writer # noqa: E402 7 8 9class RecordScreenWriter(Writer): 10 """Write DASRecords to terminal screen in some survivable 11 format. Mostly intended for debugging.""" 12 13 def __init__(self, **kwargs): 14 super().__init__(**kwargs) # processes 'quiet' and type hints 15 16 self.values = {} 17 self.timestamps = {} 18 self.latest = 0 19 20 ############################ 21 def move_cursor(self, x, y): 22 print('\033[{};{}f'.format(str(x), str(y))) 23 24 ############################ 25 # receives a DASRecord 26 def write(self, record: DASRecord): 27 28 # See if it's something we can process, and if not, try digesting 29 if not self.can_process_record(record): # inherited from BaseModule() 30 self.digest_record(record) # inherited from BaseModule() 31 return 32 33 # Incorporate the values from the record 34 self.latest = record.timestamp 35 for field in record.fields: 36 self.values[field] = record.fields[field] 37 self.timestamps[field] = self.latest 38 39 # Get term size, in case it's been resized 40 (cols, rows) = shutil.get_terminal_size() 41 self.move_cursor(0, 0) 42 # Redraw stuff 43 keys = sorted(self.values.keys()) 44 for i in range(rows): 45 # Go through keys in alpha order 46 if i < len(keys): 47 key = keys[i] 48 line = '{} : {}'.format(key, self.values[key]) 49 50 # Fill the rest of the screen with blank lines 51 else: 52 line = '' 53 54 # Pad the lines out to screen width to overwrite old stuff 55 pad_size = cols - len(line) 56 line += ' ' * pad_size 57 58 print(line)
class
RecordScreenWriter(logger.writers.writer.Writer):
10class RecordScreenWriter(Writer): 11 """Write DASRecords to terminal screen in some survivable 12 format. Mostly intended for debugging.""" 13 14 def __init__(self, **kwargs): 15 super().__init__(**kwargs) # processes 'quiet' and type hints 16 17 self.values = {} 18 self.timestamps = {} 19 self.latest = 0 20 21 ############################ 22 def move_cursor(self, x, y): 23 print('\033[{};{}f'.format(str(x), str(y))) 24 25 ############################ 26 # receives a DASRecord 27 def write(self, record: DASRecord): 28 29 # See if it's something we can process, and if not, try digesting 30 if not self.can_process_record(record): # inherited from BaseModule() 31 self.digest_record(record) # inherited from BaseModule() 32 return 33 34 # Incorporate the values from the record 35 self.latest = record.timestamp 36 for field in record.fields: 37 self.values[field] = record.fields[field] 38 self.timestamps[field] = self.latest 39 40 # Get term size, in case it's been resized 41 (cols, rows) = shutil.get_terminal_size() 42 self.move_cursor(0, 0) 43 # Redraw stuff 44 keys = sorted(self.values.keys()) 45 for i in range(rows): 46 # Go through keys in alpha order 47 if i < len(keys): 48 key = keys[i] 49 line = '{} : {}'.format(key, self.values[key]) 50 51 # Fill the rest of the screen with blank lines 52 else: 53 line = '' 54 55 # Pad the lines out to screen width to overwrite old stuff 56 pad_size = cols - len(line) 57 line += ' ' * pad_size 58 59 print(line)
Write DASRecords to terminal screen in some survivable format. Mostly intended for debugging.
RecordScreenWriter(**kwargs)
14 def __init__(self, **kwargs): 15 super().__init__(**kwargs) # processes 'quiet' and type hints 16 17 self.values = {} 18 self.timestamps = {} 19 self.latest = 0
Abstract base class for data Writers.
def
write(self, record: logger.utils.das_record.DASRecord):
27 def write(self, record: DASRecord): 28 29 # See if it's something we can process, and if not, try digesting 30 if not self.can_process_record(record): # inherited from BaseModule() 31 self.digest_record(record) # inherited from BaseModule() 32 return 33 34 # Incorporate the values from the record 35 self.latest = record.timestamp 36 for field in record.fields: 37 self.values[field] = record.fields[field] 38 self.timestamps[field] = self.latest 39 40 # Get term size, in case it's been resized 41 (cols, rows) = shutil.get_terminal_size() 42 self.move_cursor(0, 0) 43 # Redraw stuff 44 keys = sorted(self.values.keys()) 45 for i in range(rows): 46 # Go through keys in alpha order 47 if i < len(keys): 48 key = keys[i] 49 line = '{} : {}'.format(key, self.values[key]) 50 51 # Fill the rest of the screen with blank lines 52 else: 53 line = '' 54 55 # Pad the lines out to screen width to overwrite old stuff 56 pad_size = cols - len(line) 57 line += ' ' * pad_size 58 59 print(line)
Core method - write a record that we've been passed.