diff --git a/app_window.py b/app_window.py new file mode 100644 index 0000000..f2e33bb --- /dev/null +++ b/app_window.py @@ -0,0 +1,212 @@ +import tkinter as tk +from PIL import Image, ImageTk + +import gauge as SDR_gauge +import sensor as SDR_sensor +import engine_display as SDR_engine_display +from button_factory import ButtonFactory + +# Valve state constants +VALVE_OPEN = True +VALVE_CLOSED = False + +# Layout constant +_GRID_LEN = 85 + + +class DashboardWindow: + """Encapsulates all Tkinter details for the engine dashboard. + + The rest of the application never imports tkinter directly; it interacts + with the GUI exclusively through this class's public interface: + + update_valve_states(valve_states) — sync valve button visuals + update() — tick the GUI event loop + close() — destroy both windows + + Widget collections are exposed as plain attributes: + gauges : list of SDR_gauge.gauge, in _GAUGE_CONFIGS order + valve_buttons : dict[str, SDR_valve.Buttons], keyed by valve-state key + """ + + def __init__( self, callbacks, valve_buttons ): + """ + Parameters + ---------- + callbacks : CallbackHandler — provides all button callbacks + valve_buttons : dict — shared mutable dict; populated here + so external code (TelemetryProcessor, + CallbackHandler closures) can reference it + """ + self._callbacks = callbacks + self.valve_buttons = valve_buttons # populated during _build_widgets + self.gauges = [] + + self._root = self._build_root() + self._frames = self._build_frames() + self.plumbing = SDR_engine_display.Engine_Display(_GRID_LEN) + self._build_widgets() + self._layout() + + # ------------------------------------------------------------------------- + # Public interface + # ------------------------------------------------------------------------- + + def update_valve_states( self, valve_states ): + """Sync every valve button's visual state with the latest telemetry.""" + for key, btn in self.valve_buttons.items(): + state_str = valve_states.get(key) + if state_str == "OPEN": + btn.symbol.setState(VALVE_OPEN); btn.state = VALVE_OPEN + elif state_str == "CLOSED": + btn.symbol.setState(VALVE_CLOSED); btn.state = VALVE_CLOSED + btn.updateText() + btn.updateColor() + btn.configButton() + + def update( self ): + """Advance the Tkinter and P&ID event loops by one tick.""" + self.plumbing.updatePipeStatus() + self._root.update() + self.plumbing.getWindow().update() + + def close( self ): + """Destroy both the dashboard window and the P&ID window.""" + self._root.destroy() + self.plumbing.win.destroy() + + # ------------------------------------------------------------------------- + # Private construction helpers + # ------------------------------------------------------------------------- + + def _build_root( self ): + root = tk.Tk() + root.title("Engine Dashboard") + root.configure( background="black", borderwidth=10 ) + root.geometry("900x1000") + root.protocol( "WM_DELETE_WINDOW", self._callbacks.close_window ) + + # Load logo images (kept on self to prevent garbage collection) + self._sdr_logo = tk.PhotoImage(file='images/SDRLogo5.png') + sdr_img = Image.open("images/SDRlogont2.png") + sdr_img = sdr_img.resize( + (int(0.8*140), int(0.8*125)), Image.Resampling.LANCZOS + ) + self._sdr_photo = ImageTk.PhotoImage(sdr_img) + return root + + def _build_frames( self ): + r = self._root + valve_frame = tk.Label( r, bg='black' ) + frames = { + "main_title" : tk.Label( r, bg='black' ), + "valve_button" : valve_frame, + "valve_col1" : tk.Label( valve_frame, bg='black' ), + "valve_col2" : tk.Label( valve_frame, bg='black' ), + "valve_col3" : tk.Label( valve_frame, bg='black' ), + "valve_col4" : tk.Label( valve_frame, bg='black' ), + "sequence_row1" : tk.Label( r, bg='black' ), + "sequence_row2" : tk.Label( r, bg='black' ), + "gauge_row1" : tk.Label( r, bg='black' ), + "gauge_row2" : tk.Label( r, bg='black' ), + } + return frames + + def _build_widgets( self ): + f = self._frames + cb = self._callbacks + + # --- Valve buttons --------------------------------------------------- + # (valve_state_key, label, col_frame_key, plumbing_sym, seq_fn_name) + _VALVE_CONFIGS = [ + ("oxPress", "LOX Pressure (1)", "valve_col1", self.plumbing.one, "manual_lox_press" ), + ("oxVent", "LOX Vent (2)", "valve_col1", self.plumbing.two, "manual_lox_vent" ), + ("oxPurge", "LOX Purge (5)", "valve_col1", self.plumbing.five, "manual_lox_purge" ), + ("oxMain", "LOX Main", "valve_col1", self.plumbing.s2, "manual_lox_main" ), + ("fuelPress", "Fuel Pressure (3)", "valve_col2", self.plumbing.three, "manual_fuel_press"), + ("fuelVent", "Fuel Vent (4)", "valve_col2", self.plumbing.four, "manual_fuel_vent" ), + ("fuelPurge", "Fuel Purge (6)", "valve_col2", self.plumbing.six, "manual_fuel_purge"), + ("fuelMain", "Fuel Main Valve", "valve_col2", self.plumbing.s1, "manual_fuel_main" ), + ] + for key, label, col_key, sym, seq_fn in _VALVE_CONFIGS: + self.valve_buttons[key] = ButtonFactory.valve_button( + f[col_key], label, sym, cb.make_valve_callback(key, seq_fn) + ) + + # --- Sequence / control buttons -------------------------------------- + # (dict_key, label, frame_key, callback, fg, outline, text_color) + _BUTTON_CONFIGS = [ + ("pre_fire_purge", "Pre-Fire Purge", "sequence_row1", cb.pre_fire_purge, 'white', 'white', 'white'), + ("fill_chill", "Fill/Chill", "sequence_row1", cb.fill_and_chill, 'white', 'white', 'white'), + ("standby", "Standby", "sequence_row1", cb.standby, 'white', 'white', 'white'), + ("ignite", "Ignite", "sequence_row1", cb.fire_engine, 'white', 'white', 'white'), + ("stop_hotfire", "Stop Hotfire", "sequence_row2", cb.stop_hotfire, 'white', 'white', 'white'), + ("stop_purge", "Disarm", "sequence_row2", cb.stop_purge, 'white', 'white', 'white'), + ("lox_purge", "LOX Purge", "sequence_row2", cb.lox_purge, 'white', 'white', 'white'), + ("kbottle_close", "KBottle Close", "sequence_row2", cb.kbottle_close, 'white', 'white', 'white'), + ("getstate", "Get State", "valve_col3", cb.get_state, 'white', 'white', 'white'), + ("manual", "Manual", "valve_col3", cb.manual_mode, 'white', 'white', 'white'), + ("abort", "ABORT", "valve_col4", cb.hotfire_abort, 'red', 'red', 'red' ), + ("reset", "Reset", "valve_col4", cb.reset, 'white', 'white', 'white'), + ] + self._sequence_buttons = {} + for name, label, frame_key, callback, fg, outline, text in _BUTTON_CONFIGS: + self._sequence_buttons[name] = ButtonFactory.sequence_button( + f[frame_key], label, callback, + fg_color=fg, outline_color=outline, text_color=text + ) + + # --- Sensor gauges --------------------------------------------------- + # (sensor_key, display_label, row_frame_key) + _GAUGE_CONFIGS = [ + ("pt7", "Fuel Tank Pressure", "gauge_row1"), + ("ffr", "Fuel Flow Rate", "gauge_row1"), + ("pt2", "None", "gauge_row1"), + ("lc", "Thrust", "gauge_row1"), + ("pt0", "LOX Pressure", "gauge_row2"), + ("oxfr", "LOX Flow Rate", "gauge_row2"), + ("pt4", "Engine Pressure", "gauge_row2"), + ("tc", "LOX Temperature", "gauge_row2"), + ] + for sensor_key, label, row_key in _GAUGE_CONFIGS: + g = SDR_gauge.gauge( + f[row_key], + background = 'black', + max_sensor_val = SDR_sensor.max_sensor_vals[sensor_key] + ) + g.setText("Nan", label) + self.gauges.append(g) + + def _layout( self ): + f = self._frames + sb = self._sequence_buttons + + f["main_title"].pack() + + f["valve_col1"].pack( side='left' ) + f["valve_col2"].pack( side='left' ) + f["valve_col3"].pack( side='left' ) + f["valve_col4"].pack( side='left' ) + f["valve_button"].pack() + + f["sequence_row1"].pack() + f["sequence_row2"].pack() + + sb["pre_fire_purge"].pack( side="left", padx=30 ) + sb["fill_chill" ].pack( side="left", padx=30 ) + sb["standby" ].pack( side="left", padx=30 ) + sb["ignite" ].pack( side="left", padx=30 ) + sb["getstate" ].pack( side="top", padx=30 ) + sb["manual" ].pack( side="top", padx=30 ) + sb["abort" ].pack( side="top", padx=30 ) + sb["stop_hotfire" ].pack( side="left", padx=30 ) + sb["stop_purge" ].pack( side="left", padx=30 ) + sb["lox_purge" ].pack( side="left", padx=30 ) + sb["kbottle_close" ].pack( side="left", padx=30 ) + sb["reset" ].pack( side="top", padx=30 ) + + f["gauge_row1"].pack() + f["gauge_row2"].pack() + for g in self.gauges: + g.getWidget().pack(side='left') +## DashboardWindow ## diff --git a/button_factory.py b/button_factory.py new file mode 100644 index 0000000..f6638d0 --- /dev/null +++ b/button_factory.py @@ -0,0 +1,32 @@ +import valve as SDR_valve +import buttons as SDR_buttons + + +class ButtonFactory: + """Abstract factory that centralises construction of valve and sequence buttons. + + Having widget constructor details in one place means a single edit propagates + to every button rather than hunting through the initialisation block.""" + + @staticmethod + def valve_button(parent, label, plumbing_sym, callback): + """Create an SDR_valve.Buttons widget.""" + return SDR_valve.Buttons(parent, label, 'top', plumbing_sym, + f_callback=callback) + + @staticmethod + def sequence_button(parent, label, callback, + fg_color='white', outline_color='white', + text_color='white', size=(135, 45)): + """Create an SDR_buttons.Button widget with the dashboard's default styling.""" + return SDR_buttons.Button( + parent, + text = label, + bg_color = 'black', + fg_color = fg_color, + outline_color = outline_color, + text_color = text_color, + size = size, + f_callback = callback, + ) +## ButtonFactory ## diff --git a/callbacks.py b/callbacks.py new file mode 100644 index 0000000..0b83ad2 --- /dev/null +++ b/callbacks.py @@ -0,0 +1,101 @@ +import sequence as SDR_sequence + + +class CallbackHandler: + """Owns every button callback in the dashboard. + + Dependencies are injected at construction so callbacks are not coupled to + module-level globals. The window reference is set separately via + set_window() to break the initialization cycle: + callbacks → window (needs callbacks to build buttons) + window → callbacks (needs window reference to close it) + """ + + def __init__( self, engine_state, terminal, valve_buttons, exit_flag ): + """ + Parameters + ---------- + engine_state : Liquid_Engine_State + terminal : sdec.terminalData + valve_buttons : dict — shared mutable dict populated by DashboardWindow + exit_flag : list — single-element mutable bool, e.g. [False] + """ + self._engine_state = engine_state + self._terminal = terminal + self._valve_buttons = valve_buttons + self._exit_flag = exit_flag + self._window = None + + def set_window( self, window ): + """Late-bind the window reference after DashboardWindow is created.""" + self._window = window + + # ------------------------------------------------------------------------- + # Window lifecycle + # ------------------------------------------------------------------------- + + def close_window( self ): + self._window.close() + self._exit_flag[0] = True + + # ------------------------------------------------------------------------- + # Sequence / control callbacks + # ------------------------------------------------------------------------- + + def pre_fire_purge( self ): + SDR_sequence.pre_fire_purge( self._engine_state, self._terminal ) + + def fill_and_chill( self ): + SDR_sequence.fill_and_chill( self._engine_state, self._terminal ) + + def standby( self ): + SDR_sequence.standby ( self._engine_state, self._terminal ) + + def fire_engine( self ): + SDR_sequence.fire_engine ( self._engine_state, self._terminal ) + + def hotfire_abort( self ): + SDR_sequence.hotfire_abort ( self._engine_state, self._terminal ) + + def get_state( self ): + SDR_sequence.get_state ( self._engine_state, self._terminal ) + + def stop_hotfire( self ): + SDR_sequence.stop_hotfire ( self._engine_state, self._terminal ) + + def stop_purge( self ): + SDR_sequence.stop_purge ( self._engine_state, self._terminal ) + + def lox_purge( self ): + SDR_sequence.lox_purge ( self._engine_state, self._terminal ) + + def kbottle_close( self ): + SDR_sequence.kbottle_close ( self._engine_state, self._terminal ) + + def manual_mode( self ): + SDR_sequence.manual ( self._engine_state, self._terminal ) + + def reset( self ): + pass + + # ------------------------------------------------------------------------- + # Valve callback factory + # ------------------------------------------------------------------------- + + def make_valve_callback( self, valve_key, seq_fn_name ): + """Return a closure that toggles a valve button and fires the matching + sequence function. Captured variables are resolved at click-time so + valve_buttons can be populated after the closure is created.""" + def callback(): + btn = self._valve_buttons[valve_key] + getattr(SDR_sequence, seq_fn_name)( + self._engine_state, self._terminal, btn.state + ) + if btn.symbol is not None and self._engine_state.state == "Manual State": + btn.state = not btn.state + btn.symbol.setState(btn.state) + btn.updateText() + btn.updateColor() + btn.configButton() + return callback +## CallbackHandler ## diff --git a/engine_display.py b/engine_display.py index 670da66..db65cc5 100644 --- a/engine_display.py +++ b/engine_display.py @@ -19,16 +19,15 @@ ############################################################### # Project Modules # ############################################################### -import component_template -import solenoid as SDR_solenoid -import ball_valve as SDR_ball_valve -import orifice as SDR_orifice -import pressure_sensor as SDR_pressure_sensor -import temp_sensor as SDR_temp_sensor -import tank as SDR_tank -import pipe as SDR_pipe -import nozzle as SDR_nozzle -import main +from plumbing import component_template +from plumbing import solenoid as SDR_solenoid +from plumbing import ball_valve as SDR_ball_valve +from plumbing import orifice as SDR_orifice +from plumbing import pressure_sensor as SDR_pressure_sensor +from plumbing import temp_sensor as SDR_temp_sensor +from plumbing import tank as SDR_tank +from plumbing import pipe as SDR_pipe +from plumbing import nozzle as SDR_nozzle ############################################################### diff --git a/liquid_engine_state.py b/liquid_engine_state.py new file mode 100644 index 0000000..15b93f8 --- /dev/null +++ b/liquid_engine_state.py @@ -0,0 +1,12 @@ +class Liquid_Engine_State: + """Holds and transitions the current engine state string.""" + + def __init__( self ): + self.state = "Initialization State" + + def get_engine_state( self ): + return self.state + + def set_engine_state( self, new_engine_state ): + self.state = new_engine_state +## Liquid_Engine_State ## diff --git a/main.py b/main.py index 9422e96..b7ac8f0 100644 --- a/main.py +++ b/main.py @@ -1,6 +1,6 @@ #################################################################################### # # -# main.py -- main interface program for SDR's liquid engine # +# main.py -- application entry point for SDR's liquid engine dashboard # # # # Author: Nitish Chennoju, Colton Acosta # # Date: 6/12/2022 # @@ -13,408 +13,95 @@ # Developers # #################################################################################### __author__ = "Nitish Chennoju" -__credits__ = ["Colton Acosta" , +__hall_of_fame__ = ["Colton Acosta" , "Katie Herrington", "Ian Chandra" ] - -__credits_2023__ = ["Nick Nguyen" , - "Anton Sazonov" , - "Brian Lew" , - "Alexander Linderman"] - - #################################################################################### # Standard Imports # #################################################################################### -# General import time import os -import sys import datetime -import math - -# Serial (USB) +import sys import serial import serial.tools.list_ports -from serial import SerialException - -# Interface/GUI -import tkinter as tk -from tkinter import ttk -from PIL import Image, ImageTk #################################################################################### # Project Modules # #################################################################################### -# Setup path -sys.path.insert(0, './plumbing') -sys.path.insert(0, './sdec') +_BASE_DIR = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, os.path.join(_BASE_DIR, 'plumbing')) +sys.path.insert(0, os.path.join(_BASE_DIR, 'sdec')) -# Imports -import solenoid -import gauge as SDR_gauge -import valve as SDR_valve -import engine_display as SDR_engine_display -import sequence as SDR_sequence -import buttons as SDR_buttons -import sensor as SDR_sensor - -# SDEC import sdec import commands -import hw_commands -import sensor_conv import engineController - -#################################################################################### -# Defines # -#################################################################################### - -# Valve open/close states -VALVE_OPEN = True -VALVE_CLOSED = False - -# Filter enable for gauge readouts -GAUGE_FILTER_ENABLED = True +from liquid_engine_state import Liquid_Engine_State +from sensor_data_buffer import Sensor_Data_Buffer +from callbacks import CallbackHandler +from app_window import DashboardWindow +from telemetry import TelemetryProcessor #################################################################################### -# Objects # +# Global State # #################################################################################### +liquid_engine_state = Liquid_Engine_State() +sensor_data_buffer = Sensor_Data_Buffer( tau=1 ) -#################################################################################### -# # -# OBJECT: # -# liquid_engine_state # -# # -# DESCRIPTION: # -# Contains state information on the liquid engine # -# # -#################################################################################### -class Liquid_Engine_State: - - # Initialzation Function - def __init__( self ): - self.state = "Initialization State" - ## __init__ ## - - # Get the engine state - def get_engine_state( self ): - return self.state - ## get_engine_state ## - - # Set the engine state - def set_engine_state( self, new_engine_state ): - self.state = new_engine_state - ## set_engine_state ## -## liquid_engine_state ## - - -#################################################################################### -# # -# OBJECT: # -# Sensor_Data_Buffer # -# # -# DESCRIPTION: # -# Contains a queue of previous sensor data for filtering of gauge readouts # -# # -#################################################################################### -class Sensor_Data_Buffer: - - # Initialization - # max_buffer_size: size of the FIFO buffer - # tau : time constant of low-pass filter - def __init__( self, max_buffer_size = 10, tau = 1 ): - self.data_buffer = [] - self.filter_buffer = [] - self.buffer_size = 0 - self.max_buffer_size = max_buffer_size - self.tau = tau - self.sensors = [ "pt0", "pt1", "pt2", "pt3", "pt4", "pt5", "pt6", - "pt7", "lc", "tc"] - ## __init__ ## - - # Add data to the buffer - def add_data( self, sensor_data ): - if ( self.buffer_size < self.max_buffer_size ): - self.data_buffer.append( sensor_data ) - self.buffer_size += 1 - if ( self.buffer_size == 1 ): - self.filter_buffer.append( sensor_data ) - else: - self.filter_buffer.append( self.filter_data() ) - else: - self.data_buffer.pop( 0 ) - self.data_buffer.append( sensor_data ) - self.filter_buffer.append( self.filter_data() ) - ## add_data ## - - # Filter incoming data - def filter_data( self ): - # Calculate filter coefficients - T = self.data_buffer[-1]["t"] - self.data_buffer[-2]["t"] # Time period - alpha = 2*self.tau + T - beta = T - 2*self.tau - - # Difference equation - filtered_readouts = {} - for sensor in self.sensors: - filtered_readouts[sensor] = ( (T/alpha)*( self.data_buffer[-1][sensor] + - self.data_buffer[-2][sensor] ) - (beta/alpha)*(self.filter_buffer[-1][sensor]) ) - return filtered_readouts - ## filter_data ## - - ## Get the most recent filter data - def get_data( self ): - return self.filter_buffer[-1] -## Sensor_Data_Buffer ## - - -#################################################################################### -# Callbacks # -#################################################################################### - -# Close all GUI windows -def close_window_callback(): - global exitFlag - root.destroy() - plumbing.win.destroy() - exitFlag = True - -# Sequencing/Control callbacks -def pre_fire_purge_callback(): - SDR_sequence.pre_fire_purge( liquid_engine_state, terminalSerObj ) - -def fill_and_chill_callback(): - SDR_sequence.fill_and_chill( liquid_engine_state, terminalSerObj ) - -def standby_callback(): - SDR_sequence.standby ( liquid_engine_state, terminalSerObj ) - -def fire_engine_callback(): - SDR_sequence.fire_engine ( liquid_engine_state, terminalSerObj ) - -def hotfire_abort_callback(): - SDR_sequence.hotfire_abort ( liquid_engine_state, terminalSerObj ) - -def get_state_callback(): - SDR_sequence.get_state ( liquid_engine_state, terminalSerObj ) - -def stop_hotfire_callback(): - SDR_sequence.stop_hotfire ( liquid_engine_state, terminalSerObj ) - -def stop_purge_callback(): - SDR_sequence.stop_purge ( liquid_engine_state, terminalSerObj ) - -def lox_purge_callback(): - SDR_sequence.lox_purge ( liquid_engine_state, terminalSerObj ) - -def kbottle_close_callback(): - SDR_sequence.kbottle_close ( liquid_engine_state, terminalSerObj ) - -def manual_mode_callback(): - SDR_sequence.manual ( liquid_engine_state, terminalSerObj ) - -def reset_callback(): - pass - -def lox_press_callback(): - SDR_sequence.manual_lox_press ( liquid_engine_state, terminalSerObj, solenoid1_buttons.state ) - if ( ( solenoid1_buttons.symbol != None ) and ( liquid_engine_state.state == "Manual State" ) ): - solenoid1_buttons.state = not solenoid1_buttons.state - solenoid1_buttons.symbol.setState( solenoid1_buttons.state ) - solenoid1_buttons.updateText() - solenoid1_buttons.updateColor() - solenoid1_buttons.configButton() - -def lox_vent_callback(): - SDR_sequence.manual_lox_vent ( liquid_engine_state, terminalSerObj, solenoid2_buttons.state ) - if ( ( solenoid2_buttons.symbol != None ) and ( liquid_engine_state.state == "Manual State" ) ): - solenoid2_buttons.state = not solenoid2_buttons.state - solenoid2_buttons.symbol.setState( solenoid2_buttons.state ) - solenoid2_buttons.updateText() - solenoid2_buttons.updateColor() - solenoid2_buttons.configButton() - -def manual_lox_purge_callback(): - SDR_sequence.manual_lox_purge( liquid_engine_state, terminalSerObj, solenoid3_buttons.state ) - if ( ( solenoid3_buttons.symbol != None ) and ( liquid_engine_state.state == "Manual State" ) ): - solenoid3_buttons.state = not solenoid3_buttons.state - solenoid3_buttons.symbol.setState( solenoid3_buttons.state ) - solenoid3_buttons.updateText() - solenoid3_buttons.updateColor() - solenoid3_buttons.configButton() - -def fuel_purge_callback(): - SDR_sequence.manual_fuel_purge( liquid_engine_state, terminalSerObj, solenoid4_buttons.state ) - if ( ( solenoid4_buttons.symbol != None ) and ( liquid_engine_state.state == "Manual State" ) ): - solenoid4_buttons.state = not solenoid4_buttons.state - solenoid4_buttons.symbol.setState( solenoid4_buttons.state ) - solenoid4_buttons.updateText() - solenoid4_buttons.updateColor() - solenoid4_buttons.configButton() - -def fuel_press_callback(): - SDR_sequence.manual_fuel_press( liquid_engine_state, terminalSerObj, solenoid5_buttons.state ) - if ( ( solenoid5_buttons.symbol != None ) and ( liquid_engine_state.state == "Manual State" ) ): - solenoid5_buttons.state = not solenoid5_buttons.state - solenoid5_buttons.symbol.setState( solenoid5_buttons.state ) - solenoid5_buttons.updateText() - solenoid5_buttons.updateColor() - solenoid5_buttons.configButton() - -def fuel_vent_callback(): - SDR_sequence.manual_fuel_vent( liquid_engine_state, terminalSerObj, solenoid6_buttons.state ) - if ( ( solenoid6_buttons.symbol != None ) and ( liquid_engine_state.state == "Manual State" ) ): - solenoid6_buttons.state = not solenoid6_buttons.state - solenoid6_buttons.symbol.setState( solenoid6_buttons.state ) - solenoid6_buttons.updateText() - solenoid6_buttons.updateColor() - solenoid6_buttons.configButton() - -def lox_main_callback(): - SDR_sequence.manual_lox_main( liquid_engine_state, terminalSerObj, ball_valve1_buttons.state ) - if ( ( ball_valve1_buttons.symbol != None ) and ( liquid_engine_state.state == "Manual State" ) ): - ball_valve1_buttons.state = not ball_valve1_buttons.state - ball_valve1_buttons.symbol.setState( ball_valve1_buttons.state ) - ball_valve1_buttons.updateText() - ball_valve1_buttons.updateColor() - ball_valve1_buttons.configButton() - -def fuel_main_callback(): - SDR_sequence.manual_fuel_main( liquid_engine_state, terminalSerObj, ball_valve2_buttons.state ) - if ( ( ball_valve2_buttons.symbol != None ) and ( liquid_engine_state.state == "Manual State" ) ): - ball_valve2_buttons.state = not ball_valve2_buttons.state - ball_valve2_buttons.symbol.setState( ball_valve2_buttons.state ) - ball_valve2_buttons.updateText() - ball_valve2_buttons.updateColor() - ball_valve2_buttons.configButton() +engine_state_filenames = { + "Initialization State": "init" , + "Ready State" : "ready" , + "Pre-Fire Purge State": "pfpurge" , + "Fill and Chill State": "fillchill", + "Standby State" : "standby" , + "Fire State" : "fire" , + "Disarm State" : "disarm" , + "Post-Fire State" : "postfire" , + "Manual State" : "manual" , + "Abort State" : "abort" , +} #################################################################################### -# Procedures # +# Setup functions # #################################################################################### -# Update the valves state after telreq -def update_valve_states( valve_states ): - # LOX Pressure - if ( valve_states["oxPress"] == "OPEN" ): - solenoid1_buttons.symbol.setState( VALVE_OPEN ) - solenoid1_buttons.state = VALVE_OPEN - elif ( valve_states["oxPress"] == "CLOSED" ): - solenoid1_buttons.symbol.setState( VALVE_CLOSED ) - solenoid1_buttons.state = VALVE_CLOSED - solenoid1_buttons.updateText() - solenoid1_buttons.updateColor() - solenoid1_buttons.configButton() +def setup_serial_port( engine_state ): + """Connect to the first available CP2102 USB device and transition to Ready State. - # LOX Vent - if ( valve_states["oxVent"] == "OPEN" ): - solenoid2_buttons.symbol.setState( VALVE_OPEN ) - solenoid2_buttons.state = VALVE_OPEN - elif ( valve_states["oxVent"] == "CLOSED" ): - solenoid2_buttons.symbol.setState( VALVE_CLOSED ) - solenoid2_buttons.state = VALVE_CLOSED - solenoid2_buttons.updateText() - solenoid2_buttons.updateColor() - solenoid2_buttons.configButton() + Returns the initialised terminalSerObj.""" + terminal = sdec.terminalData() + for _, port in enumerate( serial.tools.list_ports.comports() ): + if 'CP2102' in port.description: + commands.connect( ['-p', port.device], terminal ) + engine_state.set_engine_state( "Ready State" ) + return terminal - # LOX Purge - if ( valve_states["oxPurge"] == "OPEN" ): - solenoid3_buttons.symbol.setState( VALVE_OPEN ) - solenoid3_buttons.state = VALVE_OPEN - elif ( valve_states["oxPurge"] == "CLOSED" ): - solenoid3_buttons.symbol.setState( VALVE_CLOSED ) - solenoid3_buttons.state = VALVE_CLOSED - solenoid3_buttons.updateText() - solenoid3_buttons.updateColor() - solenoid3_buttons.configButton() - # Fuel Purge - if ( valve_states["fuelPurge"] == "OPEN" ): - solenoid4_buttons.symbol.setState( VALVE_OPEN ) - solenoid4_buttons.state = VALVE_OPEN - elif ( valve_states["fuelPurge"] == "CLOSED" ): - solenoid4_buttons.symbol.setState( VALVE_CLOSED ) - solenoid4_buttons.state = VALVE_CLOSED - solenoid4_buttons.updateText() - solenoid4_buttons.updateColor() - solenoid4_buttons.configButton() +def setup_data_logging( engine_state ): + """Create a dated output directory and find the next unused test-run number. + Writes an initial marker file for the Initialization State. - # Fuel Pressure - if ( valve_states["fuelPress"] == "OPEN" ): - solenoid5_buttons.symbol.setState( VALVE_OPEN ) - solenoid5_buttons.state = VALVE_OPEN - elif ( valve_states["fuelPress"] == "CLOSED" ): - solenoid5_buttons.symbol.setState( VALVE_CLOSED ) - solenoid5_buttons.state = VALVE_CLOSED - solenoid5_buttons.updateText() - solenoid5_buttons.updateColor() - solenoid5_buttons.configButton() + Returns (output_dir, test_num).""" + run_date = datetime.date.today().strftime("%m-%d-%Y") + output_dir = os.path.join("output", "hotfire", run_date) + os.makedirs(output_dir, exist_ok=True) - # Fuel Vent - if ( valve_states["fuelVent"] == "OPEN" ): - solenoid6_buttons.symbol.setState( VALVE_OPEN ) - solenoid6_buttons.state = VALVE_OPEN - elif ( valve_states["fuelVent"] == "CLOSED" ): - solenoid6_buttons.symbol.setState( VALVE_CLOSED ) - solenoid6_buttons.state = VALVE_CLOSED - solenoid6_buttons.updateText() - solenoid6_buttons.updateColor() - solenoid6_buttons.configButton() + base = os.path.join(output_dir, engine_state_filenames[engine_state.state]) + test_num = 0 + while os.path.exists( base + str(test_num) + ".txt" ): + test_num += 1 - # LOX Main - if ( valve_states["oxMain"] == "OPEN" ): - ball_valve1_buttons.symbol.setState( VALVE_OPEN ) - ball_valve1_buttons.state = VALVE_OPEN - elif ( valve_states["oxMain"] == "CLOSED" ): - ball_valve1_buttons.symbol.setState( VALVE_CLOSED ) - ball_valve1_buttons.state = VALVE_CLOSED - ball_valve1_buttons.updateText() - ball_valve1_buttons.updateColor() - ball_valve1_buttons.configButton() + with open( base + str(test_num) + ".txt", "a" ) as f: + f.write( "Initialization State" ) - # Fuel Main - if ( valve_states["fuelMain"] == "OPEN" ): - ball_valve2_buttons.symbol.setState( VALVE_OPEN ) - ball_valve2_buttons.state = VALVE_OPEN - elif ( valve_states["fuelMain"] == "CLOSED" ): - ball_valve2_buttons.symbol.setState( VALVE_CLOSED ) - ball_valve2_buttons.state = VALVE_CLOSED - ball_valve2_buttons.updateText() - ball_valve2_buttons.updateColor() - ball_valve2_buttons.configButton() -## update_valve_states ## - - -#################################################################################### -# Global Variables # -#################################################################################### - -# State of the engine -liquid_engine_state = Liquid_Engine_State() - -# Buffer for filtering data prior to displaying on gauges -sensor_data_buffer = Sensor_Data_Buffer( tau = 1 ) - -# File Name outputs -engine_state_filenames = { - "Initialization State": "init" , - "Ready State" : "ready" , - "Pre-Fire Purge State": "pfpurge" , - "Fill and Chill State": "fillchill", - "Standby State" : "standby" , - "Fire State" : "fire" , - "Disarm State" : "disarm" , - "Post-Fire State" : "postfire", - "Manual State" : "manual" , - "Abort State" : "abort" - } + return output_dir, test_num #################################################################################### @@ -422,1582 +109,67 @@ def update_valve_states( valve_states ): #################################################################################### if __name__ == '__main__': - ################################################################################ - # Serial Port Setup # - ################################################################################ - - # Initialize Serial Port Object - terminalSerObj = sdec.terminalData() - - # Look for possible connections - avail_ports = serial.tools.list_ports.comports() - for port_num, port in enumerate( avail_ports ): - if ( 'CP2102' in port.description ): - # Connect - port_num = port.device - connect_args = [ '-p', port_num] - commands.connect( connect_args, terminalSerObj ) - - # Enter the ready state - liquid_engine_state.set_engine_state( "Ready State" ) - - ################################################################################ - # Data logging setup # - ################################################################################ - - # Get Date - run_date = datetime.date.today() - run_date = run_date.strftime("%m-%d-%Y") - - # Create Output directory - if ( not ( os.path.exists( "output" ) ) ): - os.mkdir( "output" ) - output_dir = "output/" + "hotfire/" - if ( not ( os.path.exists( output_dir ) ) ): - os.mkdir( output_dir ) - output_dir += run_date - if ( not ( os.path.exists( output_dir ) ) ): - os.mkdir( output_dir ) - - # Determine output filename based on existing files - base_output_filename = output_dir + "/" + engine_state_filenames[liquid_engine_state.state] - test_num = 0 - output_filename = base_output_filename + str( test_num ) + ".txt" - while ( os.path.exists( output_filename ) ): - test_num += 1 - output_filename = base_output_filename + str( test_num ) + ".txt" - - # Create an initial File - with open( output_filename, "a" ) as file: - file.write( "Initialization State" ) - - - ################################################################################ - # Global variables # - ################################################################################ - - # Declarations - global exitFlag # Flag set when window closes - - # Initializations - exitFlag = False - - # Timing - safe_tanks_timing_cnt = 1 - - ################################################################################ - # Local variables # - ################################################################################ - - # Spacing constants within GUI - pad = 10 - gridLen = 85 - - - ################################################################################ - # Window frames # - ################################################################################ - - # root window - root = tk.Tk() - root.title("Engine Dashboard") - root.configure( - background="black", - borderwidth=10 - ) - # root.geometry("900x1000") - root.geometry("900x1000") - - root.protocol( - "WM_DELETE_WINDOW", - close_window_callback - ) - - # Program icon - SDRlogo = tk.PhotoImage(file='images/SDRLogo5.png') - SDRImage = Image.open("images/SDRlogont2.png") - SDRImage = SDRImage.resize( - (int(0.8*140),int(0.8*125)), - Image.Resampling.LANCZOS - ) - SDR = ImageTk.PhotoImage(SDRImage) - #root.iconphoto(True, SDRlogo) - - # Logo and dashboard label frame - main_title_frame = tk.Label( - root, - bg = 'black' - ) - - # Valve button frame - valve_button_frame = tk.Label( root, bg = 'black' ) - - # Valve button row frames - valve_button_col1 = tk.Label( valve_button_frame, bg = 'black' ) - valve_button_col2 = tk.Label( valve_button_frame, bg = 'black' ) - valve_button_col3 = tk.Label( valve_button_frame, bg = 'black' ) - valve_button_col4 = tk.Label( valve_button_frame, bg = 'black' ) - - - # Sequence button frames - sequence_frame_row1 = tk.Label( - root, - bg='black' - ) - sequence_frame_row2 = tk.Label( - root, - bg='black' - ) - - # Gauge row frames - gauge_frame_row1 = tk.Label( - root, - bg='black' - ) - - gauge_frame_row2 = tk.Label( - root, - bg='black' - ) - + # --- Setup ------------------------------------------------------------------ + terminalSerObj = setup_serial_port( liquid_engine_state ) + output_dir, test_num = setup_data_logging( liquid_engine_state ) - ################################################################################ - # Widget initializations # - ################################################################################ + # --- Wire components -------------------------------------------------------- + exit_flag = [False] # mutable container so CallbackHandler can set it + valve_buttons = {} # shared dict; DashboardWindow populates it - # P&ID diagram window - plumbing = SDR_engine_display.Engine_Display(gridLen) - - # SDR logo - #SDRlabel = tk.Label( - # main_title_frame, - # image = SDR, - # bg = 'black' - # ) - - # Main window label - #main_window_title = tk.Label( - # main_title_frame, - # text="Engine Dashboard", - # bg="black", - # fg="white", - # font="Arial 30" - # ) - - # Valve buttons - solenoid1_buttons = SDR_valve.Buttons( - valve_button_col1, - "LOX Pressure (1)", - 'top' , - plumbing.one, - f_callback = lox_press_callback - ) - - solenoid2_buttons = SDR_valve.Buttons( - valve_button_col1, - "LOX Vent (2)" , - 'top' , - plumbing.two, - f_callback = lox_vent_callback - ) - - solenoid3_buttons = SDR_valve.Buttons( - valve_button_col1, - "LOX Purge (5)" , - 'top' , - plumbing.five, - f_callback = manual_lox_purge_callback - ) - - ball_valve1_buttons = SDR_valve.Buttons( - valve_button_col1, - "LOX Main" , - 'top' , - plumbing.s2, - f_callback = lox_main_callback - ) - - solenoid5_buttons = SDR_valve.Buttons( - valve_button_col2, - "Fuel Pressure (3)", - 'top' , - plumbing.three, - f_callback = fuel_press_callback - ) - - solenoid6_buttons = SDR_valve.Buttons( - valve_button_col2, - "Fuel Vent (4)", - 'top' , - plumbing.four, - f_callback = fuel_vent_callback - ) - - solenoid4_buttons = SDR_valve.Buttons( - valve_button_col2 , - "Fuel Purge (6)", - 'top' , - plumbing.six, - f_callback = fuel_purge_callback - ) - - - ball_valve2_buttons = SDR_valve.Buttons( - valve_button_col2 , - "Fuel Main Valve", - 'top' , - plumbing.s1, - f_callback = fuel_main_callback - ) - - # Pre-Fire purge button - pre_fire_purge_button = SDR_buttons.Button( - sequence_frame_row1 , - text = "Pre-Fire Purge", - bg_color = 'black' , - fg_color = 'white' , - outline_color = 'white' , - text_color = 'white' , - size = ( 135, 45 ), - f_callback = pre_fire_purge_callback - ) - - # Fill and chill button - fill_chill_button = SDR_buttons.Button( - sequence_frame_row1 , - text = "Fill/Chill", - bg_color = 'black' , - fg_color = 'white' , - outline_color = 'white' , - text_color = 'white' , - size = ( 135, 45 ) , - f_callback = fill_and_chill_callback - ) - - # Standby button - standby_button = SDR_buttons.Button( - sequence_frame_row1 , - text = "Standby", - bg_color = 'black' , - fg_color = 'white' , - outline_color = 'white' , - text_color = 'white' , - size = ( 135, 45 ) , - f_callback = standby_callback - ) - - # Standby button - ignite_button = SDR_buttons.Button( - sequence_frame_row1 , - text = "Ignite" , - bg_color = 'black' , - fg_color = 'white' , - outline_color = 'white' , - text_color = 'white' , - size = ( 135, 45 ) , - f_callback = fire_engine_callback - ) - - # Stop hotfire button - stop_hotfire_button = SDR_buttons.Button( - sequence_frame_row2 , - text = "Stop Hotfire" , - bg_color = 'black' , - fg_color = 'white' , - outline_color = 'white' , - text_color = 'white' , - size = ( 135, 45 ), - f_callback = stop_hotfire_callback - ) - # Stop purge button - stop_purge_button = SDR_buttons.Button( - sequence_frame_row2 , - text = "Disarm" , - bg_color = 'black' , - fg_color = 'white' , - outline_color = 'white' , - text_color = 'white' , - size = ( 135, 45 ), - f_callback = stop_purge_callback - ) - # LOX purge button - lox_purge_button = SDR_buttons.Button( - sequence_frame_row2 , - text = "LOX Purge" , - bg_color = 'black' , - fg_color = 'white' , - outline_color = 'white' , - text_color = 'white' , - size = ( 135, 45 ), - f_callback = lox_purge_callback - ) - - # Kbottle close button - kbottle_close_button = SDR_buttons.Button( - sequence_frame_row2 , - text = "KBottle Close" , - bg_color = 'black' , - fg_color = 'white' , - outline_color = 'white' , - text_color = 'white' , - size = ( 135, 45 ), - f_callback = kbottle_close_callback - ) - - # Get state button - getstate_button = SDR_buttons.Button( - valve_button_col3 , - text = "Get State", - bg_color = 'black' , - fg_color = 'white' , - outline_color = 'white' , - text_color = 'white' , - size = ( 135, 45 ), - f_callback = get_state_callback - ) - - # Abort button - abort_button = SDR_buttons.Button( - valve_button_col4 , - text = "ABORT" , - bg_color = 'black' , - fg_color = 'red' , - outline_color = 'red' , - text_color = 'red' , - size = ( 135, 45 ), - f_callback = hotfire_abort_callback - ) - - # Abort button - manual_mode_button = SDR_buttons.Button( - valve_button_col3 , - text = "Manual" , - bg_color = 'black' , - fg_color = 'white' , - outline_color = 'white' , - text_color = 'white' , - size = ( 135, 45 ), - f_callback = manual_mode_callback - ) - - # New Button - reset_button = SDR_buttons.Button( - valve_button_col4, - text = "Reset", - bg_color = "black", - fg_color = "white", - outline_color = "white", - text_color = "white", - size = (135, 45), - f_callback = reset_callback + callbacks = CallbackHandler( + liquid_engine_state, terminalSerObj, valve_buttons, exit_flag + ) + window = DashboardWindow( callbacks, valve_buttons ) + callbacks.set_window( window ) # break init cycle: window now set on handler + + processor = TelemetryProcessor( + terminal = terminalSerObj, + sensor_buffer = sensor_data_buffer, + gauges = window.gauges, + update_valve_fn = window.update_valve_states, + engine_state = liquid_engine_state, + engine_state_filenames = engine_state_filenames, + output_dir = output_dir, + test_num = test_num, + start_time = time.perf_counter(), + gauge_filter_enabled = True, ) - gauges_size = 140 - # Sensor gauges - gauge1 = SDR_gauge.gauge( # Fuel Tank Pressure - gauge_frame_row1 , - background = 'black', - max_sensor_val = SDR_sensor.max_sensor_vals["pt0"] - ) - - gauge2 = SDR_gauge.gauge( # Fuel Flow Rate - gauge_frame_row1, - background = 'black', - max_sensor_val = SDR_sensor.max_sensor_vals["ffr"] - ) - - gauge3 = SDR_gauge.gauge( # Fuel Injection Pressure - gauge_frame_row1, - background = 'black', - max_sensor_val = SDR_sensor.max_sensor_vals["pt2"] - ) - - gauge4 = SDR_gauge.gauge( # Thrust - gauge_frame_row1, - background = 'black', - max_sensor_val = SDR_sensor.max_sensor_vals["lc"] - ) - - gauge5 = SDR_gauge.gauge( # LOX Pressure - gauge_frame_row2, - background = 'black', - max_sensor_val = SDR_sensor.max_sensor_vals["pt4"] - ) - - gauge6 = SDR_gauge.gauge( # LOX Flow Rate - gauge_frame_row2, - background = 'black', - max_sensor_val = SDR_sensor.max_sensor_vals["oxfr"] - ) - - gauge7 = SDR_gauge.gauge( # Engine Pressure - gauge_frame_row2, - background = 'black', - max_sensor_val = SDR_sensor.max_sensor_vals["pt6"] - ) - - gauge8 = SDR_gauge.gauge( # LOX Temperature - gauge_frame_row2, - background = 'black', - max_sensor_val = SDR_sensor.max_sensor_vals["tc"] - ) - - gauge1.setText("Nan", "Fuel Tank Pressure" ) - gauge2.setText("Nan", "Fuel Flow Rate" ) - gauge3.setText("Nan", "None" ) - gauge4.setText("Nan", "Thrust" ) - gauge5.setText("Nan", "LOX Pressure" ) - gauge6.setText("Nan", "LOX Flow Rate" ) - gauge7.setText("Nan", "Engine Pressure" ) - gauge8.setText("Nan", "LOX Temperature" ) - - - ################################################################################ - # Draw initial window # - ################################################################################ - - # Main window title - main_title_frame.pack() - #SDRlabel.pack(side='left', padx=30) - #main_window_title.pack(pady=30, side='right') - - # Valve buttons - valve_button_col1.pack( side = 'left' ) - valve_button_col2.pack( side = 'left' ) - valve_button_col3.pack( side = 'left' ) - valve_button_col4.pack(side='left') - valve_button_frame.pack() - - # Sequence button frames - sequence_frame_row1.pack() - sequence_frame_row2.pack() - - # Sequence frames - pre_fire_purge_button.pack( side = "left", padx = 30 ) - fill_chill_button.pack ( side = "left", padx = 30 ) - standby_button.pack ( side = "left", padx = 30 ) - ignite_button.pack ( side = "left", padx = 30 ) - getstate_button.pack ( side = "top" , padx = 30 ) - manual_mode_button.pack ( side = "top" , padx = 30 ) - abort_button.pack ( side = "top" , padx = 30 ) - stop_hotfire_button.pack ( side = "left", padx = 30 ) - stop_purge_button.pack ( side = "left", padx = 30 ) - lox_purge_button.pack ( side = "left", padx = 30 ) - kbottle_close_button.pack ( side = "left", padx = 30 ) - reset_button.pack (side = "top", padx = 30) - - # Gauge frame rows - gauge_frame_row1.pack() - gauge_frame_row2.pack() - - # Gauges - gauge1.getWidget().pack(side='left') - gauge2.getWidget().pack(side='left') - gauge3.getWidget().pack(side='left') - gauge4.getWidget().pack(side='left') - gauge5.getWidget().pack(side='left') - gauge6.getWidget().pack(side='left') - gauge7.getWidget().pack(side='left') - gauge8.getWidget().pack(side='left') - - - ################################################################################ - # Main Program Loop # - ################################################################################ - prevCon = True - - # Start timer - start_time = time.perf_counter() - - # Update GUI - while (not exitFlag): - # Look for new serial connections - if ( terminalSerObj.comport == None ): - avail_ports = serial.tools.list_ports.comports() - for port_num, port in enumerate( avail_ports ): - if ( ( 'CP2102' in port.description ) or - ( 'CP210x' in port.description ) ): - # Connect - port_num = port.device - connect_args = [ '-p', port_num] - commands.connect( connect_args, terminalSerObj ) - - # Transition into the ready state + # --- Main loop -------------------------------------------------------------- + while not exit_flag[0]: + if terminalSerObj.comport is None: + for _, port in enumerate( serial.tools.list_ports.comports() ): + if ( 'CP2102' in port.description ) or ( 'CP210x' in port.description ): + commands.connect( ['-p', port.device], terminalSerObj ) liquid_engine_state.set_engine_state( "Ready State" ) - - # Connection is made else: - # Record time of data reception - time_sec = time.perf_counter() - start_time - - #################################################################### - # Ready State # - #################################################################### - if ( liquid_engine_state.get_engine_state() == "Ready State" ): - # Get telemetry - engineController.telreq( [], terminalSerObj, show_output = False ) - terminalSerObj.sensor_readouts["t"] = time_sec - sensor_readouts_formatted = {} - for sensor in terminalSerObj.sensor_readouts: - sensor_readouts_formatted[sensor] = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - sensor , - terminalSerObj.sensor_readouts[sensor] ) - - # Filter the data - if ( GAUGE_FILTER_ENABLED ): - sensor_data_buffer.add_data( terminalSerObj.sensor_readouts ) - sensor_data_filtered = sensor_data_buffer.get_data() - filtered_sensor_readouts_formatted = {} - for sensor in sensor_data_filtered: - filtered_sensor_readouts_formatted[sensor] = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - sensor , - sensor_data_filtered[sensor] ) - - # Calculate Flow Rates - if ( GAUGE_FILTER_ENABLED ): - ox_flow_rate = sensor_conv.ox_pressure_to_flow( - sensor_data_filtered["pt1"] - - sensor_data_filtered["pt2"] ) - fuel_flow_rate = sensor_conv.fuel_pressure_to_flow( - sensor_data_filtered["pt6"] - - sensor_data_filtered["pt5"] ) - ox_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "oxfr" , - ox_flow_rate ) - fuel_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "ffr" , - fuel_flow_rate ) - else: - ox_flow_rate = sensor_conv.ox_pressure_to_flow( - terminalSerObj.sensor_readouts["pt1"] - - terminalSerObj.sensor_readouts["pt2"] ) - fuel_flow_rate = sensor_conv.fuel_pressure_to_flow( - terminalSerObj.sensor_readouts["pt6"] - - terminalSerObj.sensor_readouts["pt5"] ) - ox_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "oxfr" , - ox_flow_rate ) - fuel_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "ffr" , - fuel_flow_rate ) - - # Update GUI - if ( GAUGE_FILTER_ENABLED ): - gauge1.setText( filtered_sensor_readouts_formatted["pt7"], "Fuel Tank Pressure" ) - gauge2.setText( fuel_flow_rate_formatted , "Fuel Flow Rate" ) - gauge3.setText( "NaN" , "None" ) - gauge4.setText( filtered_sensor_readouts_formatted["lc"] , "Thrust" ) - gauge5.setText( filtered_sensor_readouts_formatted["pt0"], "LOX Pressure" ) - gauge6.setText( ox_flow_rate_formatted , "LOX Flow Rate" ) - gauge7.setText( filtered_sensor_readouts_formatted["pt4"], "Engine Pressure" ) - gauge8.setText( filtered_sensor_readouts_formatted["tc" ], "LOX Temperature" ) - gauge1.setAngle( sensor_data_filtered["pt7"] ) - gauge2.setAngle( fuel_flow_rate ) - gauge3.setAngle( 0 ) - gauge4.setAngle( sensor_data_filtered["lc" ] ) - gauge5.setAngle( sensor_data_filtered["pt0"] ) - gauge6.setAngle( ox_flow_rate ) - gauge7.setAngle( sensor_data_filtered["pt4"] ) - gauge8.setAngle( sensor_data_filtered["tc" ] ) - else: - gauge1.setText( sensor_readouts_formatted["pt7"], "Fuel Tank Pressure" ) - gauge2.setText( fuel_flow_rate_formatted , "Fuel Flow Rate" ) - gauge3.setText( "NaN" , "None" ) - gauge4.setText( sensor_readouts_formatted["lc"] , "Thrust" ) - gauge5.setText( sensor_readouts_formatted["pt0"], "LOX Pressure" ) - gauge6.setText( ox_flow_rate_formatted , "LOX Flow Rate" ) - gauge7.setText( sensor_readouts_formatted["pt4"], "Engine Pressure" ) - gauge8.setText( sensor_readouts_formatted["tc" ], "LOX Temperature" ) - gauge1.setAngle( terminalSerObj.sensor_readouts["pt7"] ) - gauge2.setAngle( fuel_flow_rate ) - gauge3.setAngle( 0 ) - gauge4.setAngle( terminalSerObj.sensor_readouts["lc" ] ) - gauge5.setAngle( terminalSerObj.sensor_readouts["pt0"] ) - gauge6.setAngle( ox_flow_rate ) - gauge7.setAngle( terminalSerObj.sensor_readouts["pt4"] ) - gauge8.setAngle( terminalSerObj.sensor_readouts["tc" ] ) - update_valve_states( terminalSerObj.valve_states ) - - - # Log Data - base_output_filename = output_dir + "/" + engine_state_filenames[liquid_engine_state.state] - output_filename = base_output_filename + str( test_num ) + ".txt" - with open( output_filename, "a" ) as file: - file.write(str(time_sec) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt0"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt1"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt2"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt3"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt4"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt5"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt6"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt7"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["lc" ]) + " ") - file.write(str(terminalSerObj.sensor_readouts["tc" ]) + " ") - file.write(terminalSerObj.valve_states["oxPress" ] + " ") - file.write(terminalSerObj.valve_states["oxVent" ] + " ") - file.write(terminalSerObj.valve_states["oxPurge" ] + " ") - file.write(terminalSerObj.valve_states["fuelPress"] + " ") - file.write(terminalSerObj.valve_states["fuelVent" ] + " ") - file.write(terminalSerObj.valve_states["fuelPurge"] + " ") - file.write(terminalSerObj.valve_states["oxMain" ] + " ") - file.write(terminalSerObj.valve_states["fuelMain" ] + " ") - file.write("\n") - - #################################################################### - # Pre-Fire Engine Purge # - #################################################################### - elif ( liquid_engine_state.get_engine_state() == "Pre-Fire Purge State" ): - # Get telemetry - engineController.telreq( [], terminalSerObj, show_output = False ) - terminalSerObj.sensor_readouts["t"] = time_sec - sensor_readouts_formatted = {} - for sensor in terminalSerObj.sensor_readouts: - sensor_readouts_formatted[sensor] = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - sensor , - terminalSerObj.sensor_readouts[sensor] ) - - # Filter the data - if ( GAUGE_FILTER_ENABLED ): - sensor_data_buffer.add_data( terminalSerObj.sensor_readouts ) - sensor_data_filtered = sensor_data_buffer.get_data() - filtered_sensor_readouts_formatted = {} - for sensor in sensor_data_filtered: - filtered_sensor_readouts_formatted[sensor] = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - sensor , - sensor_data_filtered[sensor] ) - - # Calculate Flow Rates - if ( GAUGE_FILTER_ENABLED ): - ox_flow_rate = sensor_conv.ox_pressure_to_flow( - sensor_data_filtered["pt1"] - - sensor_data_filtered["pt2"] ) - fuel_flow_rate = sensor_conv.fuel_pressure_to_flow( - sensor_data_filtered["pt6"] - - sensor_data_filtered["pt5"] ) - ox_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "oxfr" , - ox_flow_rate ) - fuel_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "ffr" , - fuel_flow_rate ) - else: - ox_flow_rate = sensor_conv.ox_pressure_to_flow( - terminalSerObj.sensor_readouts["pt1"] - - terminalSerObj.sensor_readouts["pt2"] ) - fuel_flow_rate = sensor_conv.fuel_pressure_to_flow( - terminalSerObj.sensor_readouts["pt6"] - - terminalSerObj.sensor_readouts["pt5"] ) - ox_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "oxfr" , - ox_flow_rate ) - fuel_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "ffr" , - fuel_flow_rate ) - - # Update GUI - if ( GAUGE_FILTER_ENABLED ): - gauge1.setText( filtered_sensor_readouts_formatted["pt7"], "Fuel Tank Pressure" ) - gauge2.setText( fuel_flow_rate_formatted , "Fuel Flow Rate" ) - gauge3.setText( "NaN" , "None" ) - gauge4.setText( filtered_sensor_readouts_formatted["lc"] , "Thrust" ) - gauge5.setText( filtered_sensor_readouts_formatted["pt0"], "LOX Pressure" ) - gauge6.setText( ox_flow_rate_formatted , "LOX Flow Rate" ) - gauge7.setText( filtered_sensor_readouts_formatted["pt4"], "Engine Pressure" ) - gauge8.setText( filtered_sensor_readouts_formatted["tc" ], "LOX Temperature" ) - gauge1.setAngle( sensor_data_filtered["pt7"] ) - gauge2.setAngle( fuel_flow_rate ) - gauge3.setAngle( 0 ) - gauge4.setAngle( sensor_data_filtered["lc" ] ) - gauge5.setAngle( sensor_data_filtered["pt0"] ) - gauge6.setAngle( ox_flow_rate ) - gauge7.setAngle( sensor_data_filtered["pt4"] ) - gauge8.setAngle( sensor_data_filtered["tc" ] ) - else: - gauge1.setText( sensor_readouts_formatted["pt7"], "Fuel Tank Pressure" ) - gauge2.setText( fuel_flow_rate_formatted , "Fuel Flow Rate" ) - gauge3.setText( "NaN" , "None" ) - gauge4.setText( sensor_readouts_formatted["lc"] , "Thrust" ) - gauge5.setText( sensor_readouts_formatted["pt0"], "LOX Pressure" ) - gauge6.setText( ox_flow_rate_formatted , "LOX Flow Rate" ) - gauge7.setText( sensor_readouts_formatted["pt4"], "Engine Pressure" ) - gauge8.setText( sensor_readouts_formatted["tc" ], "LOX Temperature" ) - gauge1.setAngle( terminalSerObj.sensor_readouts["pt7"] ) - gauge2.setAngle( fuel_flow_rate ) - gauge3.setAngle( 0 ) - gauge4.setAngle( terminalSerObj.sensor_readouts["lc" ] ) - gauge5.setAngle( terminalSerObj.sensor_readouts["pt0"] ) - gauge6.setAngle( ox_flow_rate ) - gauge7.setAngle( terminalSerObj.sensor_readouts["pt4"] ) - gauge8.setAngle( terminalSerObj.sensor_readouts["tc" ] ) - update_valve_states( terminalSerObj.valve_states ) - - # Log Data - base_output_filename = output_dir + "/" + engine_state_filenames[liquid_engine_state.state] - output_filename = base_output_filename + str( test_num ) + ".txt" - with open( output_filename, "a" ) as file: - file.write(str(time_sec) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt0"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt1"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt2"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt3"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt4"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt5"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt6"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt7"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["lc" ]) + " ") - file.write(str(terminalSerObj.sensor_readouts["tc" ]) + " ") - file.write(terminalSerObj.valve_states["oxPress" ] + " ") - file.write(terminalSerObj.valve_states["oxVent" ] + " ") - file.write(terminalSerObj.valve_states["oxPurge" ] + " ") - file.write(terminalSerObj.valve_states["fuelPress"] + " ") - file.write(terminalSerObj.valve_states["fuelVent" ] + " ") - file.write(terminalSerObj.valve_states["fuelPurge"] + " ") - file.write(terminalSerObj.valve_states["oxMain" ] + " ") - file.write(terminalSerObj.valve_states["fuelMain" ] + " ") - file.write("\n") - - #################################################################### - # Fill and Chill # - #################################################################### - elif ( liquid_engine_state.get_engine_state() == "Fill and Chill State"): - # Check tank pressures - engineController.tankstat( [], terminalSerObj ) - - # Get telemetry - engineController.telreq( [], terminalSerObj, show_output = False ) - terminalSerObj.sensor_readouts["t"] = time_sec - sensor_readouts_formatted = {} - for sensor in terminalSerObj.sensor_readouts: - sensor_readouts_formatted[sensor] = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - sensor , - terminalSerObj.sensor_readouts[sensor] ) - - # Filter the data - if ( GAUGE_FILTER_ENABLED ): - sensor_data_buffer.add_data( terminalSerObj.sensor_readouts ) - sensor_data_filtered = sensor_data_buffer.get_data() - filtered_sensor_readouts_formatted = {} - for sensor in sensor_data_filtered: - filtered_sensor_readouts_formatted[sensor] = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - sensor , - sensor_data_filtered[sensor] ) - - # Calculate Flow Rates - if ( GAUGE_FILTER_ENABLED ): - ox_flow_rate = sensor_conv.ox_pressure_to_flow( - sensor_data_filtered["pt1"] - - sensor_data_filtered["pt2"] ) - fuel_flow_rate = sensor_conv.fuel_pressure_to_flow( - sensor_data_filtered["pt6"] - - sensor_data_filtered["pt5"] ) - ox_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "oxfr" , - ox_flow_rate ) - fuel_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "ffr" , - fuel_flow_rate ) - else: - ox_flow_rate = sensor_conv.ox_pressure_to_flow( - terminalSerObj.sensor_readouts["pt1"] - - terminalSerObj.sensor_readouts["pt2"] ) - fuel_flow_rate = sensor_conv.fuel_pressure_to_flow( - terminalSerObj.sensor_readouts["pt6"] - - terminalSerObj.sensor_readouts["pt5"] ) - ox_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "oxfr" , - ox_flow_rate ) - fuel_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "ffr" , - fuel_flow_rate ) - - # Update GUI - if ( GAUGE_FILTER_ENABLED ): - gauge1.setText( filtered_sensor_readouts_formatted["pt7"], "Fuel Tank Pressure" ) - gauge2.setText( fuel_flow_rate_formatted , "Fuel Flow Rate" ) - gauge3.setText( "NaN" , "None" ) - gauge4.setText( filtered_sensor_readouts_formatted["lc"] , "Thrust" ) - gauge5.setText( filtered_sensor_readouts_formatted["pt0"], "LOX Pressure" ) - gauge6.setText( ox_flow_rate_formatted , "LOX Flow Rate" ) - gauge7.setText( filtered_sensor_readouts_formatted["pt4"], "Engine Pressure" ) - gauge8.setText( filtered_sensor_readouts_formatted["tc" ], "LOX Temperature" ) - gauge1.setAngle( sensor_data_filtered["pt7"] ) - gauge2.setAngle( fuel_flow_rate ) - gauge3.setAngle( 0 ) - gauge4.setAngle( sensor_data_filtered["lc" ] ) - gauge5.setAngle( sensor_data_filtered["pt0"] ) - gauge6.setAngle( ox_flow_rate ) - gauge7.setAngle( sensor_data_filtered["pt4"] ) - gauge8.setAngle( sensor_data_filtered["tc" ] ) - else: - gauge1.setText( sensor_readouts_formatted["pt7"], "Fuel Tank Pressure" ) - gauge2.setText( fuel_flow_rate_formatted , "Fuel Flow Rate" ) - gauge3.setText( "NaN" , "None" ) - gauge4.setText( sensor_readouts_formatted["lc"] , "Thrust" ) - gauge5.setText( sensor_readouts_formatted["pt0"], "LOX Pressure" ) - gauge6.setText( ox_flow_rate_formatted , "LOX Flow Rate" ) - gauge7.setText( sensor_readouts_formatted["pt4"], "Engine Pressure" ) - gauge8.setText( sensor_readouts_formatted["tc" ], "LOX Temperature" ) - gauge1.setAngle( terminalSerObj.sensor_readouts["pt7"] ) - gauge2.setAngle( fuel_flow_rate ) - gauge3.setAngle( 0 ) - gauge4.setAngle( terminalSerObj.sensor_readouts["lc" ] ) - gauge5.setAngle( terminalSerObj.sensor_readouts["pt0"] ) - gauge6.setAngle( ox_flow_rate ) - gauge7.setAngle( terminalSerObj.sensor_readouts["pt4"] ) - gauge8.setAngle( terminalSerObj.sensor_readouts["tc" ] ) - update_valve_states( terminalSerObj.valve_states ) - - # Log Data - base_output_filename = output_dir + "/" + engine_state_filenames[liquid_engine_state.state] - output_filename = base_output_filename + str( test_num ) + ".txt" - with open( output_filename, "a" ) as file: - file.write(str(time_sec) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt0"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt1"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt2"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt3"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt4"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt5"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt6"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt7"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["lc" ]) + " ") - file.write(str(terminalSerObj.sensor_readouts["tc" ]) + " ") - file.write(terminalSerObj.valve_states["oxPress" ] + " ") - file.write(terminalSerObj.valve_states["oxVent" ] + " ") - file.write(terminalSerObj.valve_states["oxPurge" ] + " ") - file.write(terminalSerObj.valve_states["fuelPress"] + " ") - file.write(terminalSerObj.valve_states["fuelVent" ] + " ") - file.write(terminalSerObj.valve_states["fuelPurge"] + " ") - file.write(terminalSerObj.valve_states["oxMain" ] + " ") - file.write(terminalSerObj.valve_states["fuelMain" ] + " ") - file.write("\n") - - #################################################################### - # Standby # - #################################################################### - elif ( liquid_engine_state.get_engine_state() == "Standby State"): - # Get telemetry - engineController.telreq( [], terminalSerObj, show_output = False ) - terminalSerObj.sensor_readouts["t"] = time_sec - sensor_readouts_formatted = {} - for sensor in terminalSerObj.sensor_readouts: - sensor_readouts_formatted[sensor] = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - sensor , - terminalSerObj.sensor_readouts[sensor] ) - - # Filter the data - if ( GAUGE_FILTER_ENABLED ): - sensor_data_buffer.add_data( terminalSerObj.sensor_readouts ) - sensor_data_filtered = sensor_data_buffer.get_data() - filtered_sensor_readouts_formatted = {} - for sensor in sensor_data_filtered: - filtered_sensor_readouts_formatted[sensor] = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - sensor , - sensor_data_filtered[sensor] ) - - # Calculate Flow Rates - if ( GAUGE_FILTER_ENABLED ): - ox_flow_rate = sensor_conv.ox_pressure_to_flow( - sensor_data_filtered["pt1"] - - sensor_data_filtered["pt2"] ) - fuel_flow_rate = sensor_conv.fuel_pressure_to_flow( - sensor_data_filtered["pt6"] - - sensor_data_filtered["pt5"] ) - ox_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "oxfr" , - ox_flow_rate ) - fuel_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "ffr" , - fuel_flow_rate ) - else: - ox_flow_rate = sensor_conv.ox_pressure_to_flow( - terminalSerObj.sensor_readouts["pt1"] - - terminalSerObj.sensor_readouts["pt2"] ) - fuel_flow_rate = sensor_conv.fuel_pressure_to_flow( - terminalSerObj.sensor_readouts["pt6"] - - terminalSerObj.sensor_readouts["pt5"] ) - ox_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "oxfr" , - ox_flow_rate ) - fuel_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "ffr" , - fuel_flow_rate ) - - # Update GUI - if ( GAUGE_FILTER_ENABLED ): - gauge1.setText( filtered_sensor_readouts_formatted["pt7"], "Fuel Tank Pressure" ) - gauge2.setText( fuel_flow_rate_formatted , "Fuel Flow Rate" ) - gauge3.setText( "NaN" , "None" ) - gauge4.setText( filtered_sensor_readouts_formatted["lc"] , "Thrust" ) - gauge5.setText( filtered_sensor_readouts_formatted["pt0"], "LOX Pressure" ) - gauge6.setText( ox_flow_rate_formatted , "LOX Flow Rate" ) - gauge7.setText( filtered_sensor_readouts_formatted["pt4"], "Engine Pressure" ) - gauge8.setText( filtered_sensor_readouts_formatted["tc" ], "LOX Temperature" ) - gauge1.setAngle( sensor_data_filtered["pt7"] ) - gauge2.setAngle( fuel_flow_rate ) - gauge3.setAngle( 0 ) - gauge4.setAngle( sensor_data_filtered["lc" ] ) - gauge5.setAngle( sensor_data_filtered["pt0"] ) - gauge6.setAngle( ox_flow_rate ) - gauge7.setAngle( sensor_data_filtered["pt4"] ) - gauge8.setAngle( sensor_data_filtered["tc" ] ) - else: - gauge1.setText( sensor_readouts_formatted["pt7"], "Fuel Tank Pressure" ) - gauge2.setText( fuel_flow_rate_formatted , "Fuel Flow Rate" ) - gauge3.setText( "NaN" , "None" ) - gauge4.setText( sensor_readouts_formatted["lc"] , "Thrust" ) - gauge5.setText( sensor_readouts_formatted["pt0"], "LOX Pressure" ) - gauge6.setText( ox_flow_rate_formatted , "LOX Flow Rate" ) - gauge7.setText( sensor_readouts_formatted["pt4"], "Engine Pressure" ) - gauge8.setText( sensor_readouts_formatted["tc" ], "LOX Temperature" ) - gauge1.setAngle( terminalSerObj.sensor_readouts["pt7"] ) - gauge2.setAngle( fuel_flow_rate ) - gauge3.setAngle( 0 ) - gauge4.setAngle( terminalSerObj.sensor_readouts["lc" ] ) - gauge5.setAngle( terminalSerObj.sensor_readouts["pt0"] ) - gauge6.setAngle( ox_flow_rate ) - gauge7.setAngle( terminalSerObj.sensor_readouts["pt4"] ) - gauge8.setAngle( terminalSerObj.sensor_readouts["tc" ] ) - update_valve_states( terminalSerObj.valve_states ) - - # Log Data - base_output_filename = output_dir + "/" + engine_state_filenames[liquid_engine_state.state] - output_filename = base_output_filename + str( test_num ) + ".txt" - with open( output_filename, "a" ) as file: - file.write(str(time_sec) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt0"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt1"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt2"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt3"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt4"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt5"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt6"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt7"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["lc" ]) + " ") - file.write(str(terminalSerObj.sensor_readouts["tc" ]) + " ") - file.write(terminalSerObj.valve_states["oxPress" ] + " ") - file.write(terminalSerObj.valve_states["oxVent" ] + " ") - file.write(terminalSerObj.valve_states["oxPurge" ] + " ") - file.write(terminalSerObj.valve_states["fuelPress"] + " ") - file.write(terminalSerObj.valve_states["fuelVent" ] + " ") - file.write(terminalSerObj.valve_states["fuelPurge"] + " ") - file.write(terminalSerObj.valve_states["oxMain" ] + " ") - file.write(terminalSerObj.valve_states["fuelMain" ] + " ") - file.write("\n") - - #################################################################### - # Engine Hotfire # - #################################################################### - elif ( liquid_engine_state.get_engine_state() == "Fire State"): - # Get telemetry - engineController.telreq( [], terminalSerObj, show_output = False ) - terminalSerObj.sensor_readouts["t"] = time_sec - sensor_readouts_formatted = {} - for sensor in terminalSerObj.sensor_readouts: - sensor_readouts_formatted[sensor] = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - sensor , - terminalSerObj.sensor_readouts[sensor] ) - - # Filter the data - if ( GAUGE_FILTER_ENABLED ): - sensor_data_buffer.add_data( terminalSerObj.sensor_readouts ) - sensor_data_filtered = sensor_data_buffer.get_data() - filtered_sensor_readouts_formatted = {} - for sensor in sensor_data_filtered: - filtered_sensor_readouts_formatted[sensor] = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - sensor , - sensor_data_filtered[sensor] ) - - # Calculate Flow Rates - if ( GAUGE_FILTER_ENABLED ): - ox_flow_rate = sensor_conv.ox_pressure_to_flow( - sensor_data_filtered["pt1"] - - sensor_data_filtered["pt2"] ) - fuel_flow_rate = sensor_conv.fuel_pressure_to_flow( - sensor_data_filtered["pt6"] - - sensor_data_filtered["pt5"] ) - ox_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "oxfr" , - ox_flow_rate ) - fuel_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "ffr" , - fuel_flow_rate ) - else: - ox_flow_rate = sensor_conv.ox_pressure_to_flow( - terminalSerObj.sensor_readouts["pt1"] - - terminalSerObj.sensor_readouts["pt2"] ) - fuel_flow_rate = sensor_conv.fuel_pressure_to_flow( - terminalSerObj.sensor_readouts["pt6"] - - terminalSerObj.sensor_readouts["pt5"] ) - ox_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "oxfr" , - ox_flow_rate ) - fuel_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "ffr" , - fuel_flow_rate ) - - # Update GUI - if ( GAUGE_FILTER_ENABLED ): - gauge1.setText( filtered_sensor_readouts_formatted["pt7"], "Fuel Tank Pressure" ) - gauge2.setText( fuel_flow_rate_formatted , "Fuel Flow Rate" ) - gauge3.setText( "NaN" , "None" ) - gauge4.setText( filtered_sensor_readouts_formatted["lc"] , "Thrust" ) - gauge5.setText( filtered_sensor_readouts_formatted["pt0"], "LOX Pressure" ) - gauge6.setText( ox_flow_rate_formatted , "LOX Flow Rate" ) - gauge7.setText( filtered_sensor_readouts_formatted["pt4"], "Engine Pressure" ) - gauge8.setText( filtered_sensor_readouts_formatted["tc" ], "LOX Temperature" ) - gauge1.setAngle( sensor_data_filtered["pt7"] ) - gauge2.setAngle( fuel_flow_rate ) - gauge3.setAngle( 0 ) - gauge4.setAngle( sensor_data_filtered["lc" ] ) - gauge5.setAngle( sensor_data_filtered["pt0"] ) - gauge6.setAngle( ox_flow_rate ) - gauge7.setAngle( sensor_data_filtered["pt4"] ) - gauge8.setAngle( sensor_data_filtered["tc" ] ) - else: - gauge1.setText( sensor_readouts_formatted["pt7"], "Fuel Tank Pressure" ) - gauge2.setText( fuel_flow_rate_formatted , "Fuel Flow Rate" ) - gauge3.setText( "NaN" , "None" ) - gauge4.setText( sensor_readouts_formatted["lc"] , "Thrust" ) - gauge5.setText( sensor_readouts_formatted["pt0"], "LOX Pressure" ) - gauge6.setText( ox_flow_rate_formatted , "LOX Flow Rate" ) - gauge7.setText( sensor_readouts_formatted["pt4"], "Engine Pressure" ) - gauge8.setText( sensor_readouts_formatted["tc" ], "LOX Temperature" ) - gauge1.setAngle( terminalSerObj.sensor_readouts["pt7"] ) - gauge2.setAngle( fuel_flow_rate ) - gauge3.setAngle( 0 ) - gauge4.setAngle( terminalSerObj.sensor_readouts["lc" ] ) - gauge5.setAngle( terminalSerObj.sensor_readouts["pt0"] ) - gauge6.setAngle( ox_flow_rate ) - gauge7.setAngle( terminalSerObj.sensor_readouts["pt4"] ) - gauge8.setAngle( terminalSerObj.sensor_readouts["tc" ] ) - update_valve_states( terminalSerObj.valve_states ) - - # Log Data - base_output_filename = output_dir + "/" + engine_state_filenames[liquid_engine_state.state] - output_filename = base_output_filename + str( test_num ) + ".txt" - with open( output_filename, "a" ) as file: - file.write(str(time_sec) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt0"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt1"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt2"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt3"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt4"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt5"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt6"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt7"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["lc" ]) + " ") - file.write(str(terminalSerObj.sensor_readouts["tc" ]) + " ") - file.write(terminalSerObj.valve_states["oxPress" ] + " ") - file.write(terminalSerObj.valve_states["oxVent" ] + " ") - file.write(terminalSerObj.valve_states["oxPurge" ] + " ") - file.write(terminalSerObj.valve_states["fuelPress"] + " ") - file.write(terminalSerObj.valve_states["fuelVent" ] + " ") - file.write(terminalSerObj.valve_states["fuelPurge"] + " ") - file.write(terminalSerObj.valve_states["oxMain" ] + " ") - file.write(terminalSerObj.valve_states["fuelMain" ] + " ") - file.write("\n") - - #################################################################### - # Disarm Sequence # - #################################################################### - elif ( liquid_engine_state.get_engine_state() == "Disarm State" ): - # Check tank pressures - engineController.tankstat( [], terminalSerObj ) - - # Get telemetry - engineController.telreq( [], terminalSerObj, show_output = False ) - terminalSerObj.sensor_readouts["t"] = time_sec - sensor_readouts_formatted = {} - for sensor in terminalSerObj.sensor_readouts: - sensor_readouts_formatted[sensor] = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - sensor , - terminalSerObj.sensor_readouts[sensor] ) - - # Filter the data - if ( GAUGE_FILTER_ENABLED ): - sensor_data_buffer.add_data( terminalSerObj.sensor_readouts ) - sensor_data_filtered = sensor_data_buffer.get_data() - filtered_sensor_readouts_formatted = {} - for sensor in sensor_data_filtered: - filtered_sensor_readouts_formatted[sensor] = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - sensor , - sensor_data_filtered[sensor] ) - - # Calculate Flow Rates - if ( GAUGE_FILTER_ENABLED ): - ox_flow_rate = sensor_conv.ox_pressure_to_flow( - sensor_data_filtered["pt1"] - - sensor_data_filtered["pt2"] ) - fuel_flow_rate = sensor_conv.fuel_pressure_to_flow( - sensor_data_filtered["pt6"] - - sensor_data_filtered["pt5"] ) - ox_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "oxfr" , - ox_flow_rate ) - fuel_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "ffr" , - fuel_flow_rate ) - else: - ox_flow_rate = sensor_conv.ox_pressure_to_flow( - terminalSerObj.sensor_readouts["pt1"] - - terminalSerObj.sensor_readouts["pt2"] ) - fuel_flow_rate = sensor_conv.fuel_pressure_to_flow( - terminalSerObj.sensor_readouts["pt6"] - - terminalSerObj.sensor_readouts["pt5"] ) - ox_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "oxfr" , - ox_flow_rate ) - fuel_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "ffr" , - fuel_flow_rate ) - - # Update GUI - if ( GAUGE_FILTER_ENABLED ): - gauge1.setText( filtered_sensor_readouts_formatted["pt7"], "Fuel Tank Pressure" ) - gauge2.setText( fuel_flow_rate_formatted , "Fuel Flow Rate" ) - gauge3.setText( "NaN" , "None" ) - gauge4.setText( filtered_sensor_readouts_formatted["lc"] , "Thrust" ) - gauge5.setText( filtered_sensor_readouts_formatted["pt0"], "LOX Pressure" ) - gauge6.setText( ox_flow_rate_formatted , "LOX Flow Rate" ) - gauge7.setText( filtered_sensor_readouts_formatted["pt4"], "Engine Pressure" ) - gauge8.setText( filtered_sensor_readouts_formatted["tc" ], "LOX Temperature" ) - gauge1.setAngle( sensor_data_filtered["pt7"] ) - gauge2.setAngle( fuel_flow_rate ) - gauge3.setAngle( 0 ) - gauge4.setAngle( sensor_data_filtered["lc" ] ) - gauge5.setAngle( sensor_data_filtered["pt0"] ) - gauge6.setAngle( ox_flow_rate ) - gauge7.setAngle( sensor_data_filtered["pt4"] ) - gauge8.setAngle( sensor_data_filtered["tc" ] ) - else: - gauge1.setText( sensor_readouts_formatted["pt7"], "Fuel Tank Pressure" ) - gauge2.setText( fuel_flow_rate_formatted , "Fuel Flow Rate" ) - gauge3.setText( "NaN" , "None" ) - gauge4.setText( sensor_readouts_formatted["lc"] , "Thrust" ) - gauge5.setText( sensor_readouts_formatted["pt0"], "LOX Pressure" ) - gauge6.setText( ox_flow_rate_formatted , "LOX Flow Rate" ) - gauge7.setText( sensor_readouts_formatted["pt4"], "Engine Pressure" ) - gauge8.setText( sensor_readouts_formatted["tc" ], "LOX Temperature" ) - gauge1.setAngle( terminalSerObj.sensor_readouts["pt7"] ) - gauge2.setAngle( fuel_flow_rate ) - gauge3.setAngle( 0 ) - gauge4.setAngle( terminalSerObj.sensor_readouts["lc" ] ) - gauge5.setAngle( terminalSerObj.sensor_readouts["pt0"] ) - gauge6.setAngle( ox_flow_rate ) - gauge7.setAngle( terminalSerObj.sensor_readouts["pt4"] ) - gauge8.setAngle( terminalSerObj.sensor_readouts["tc" ] ) - update_valve_states( terminalSerObj.valve_states ) - - # Log Data - base_output_filename = output_dir + "/" + engine_state_filenames[liquid_engine_state.state] - output_filename = base_output_filename + str( test_num ) + ".txt" - with open( output_filename, "a" ) as file: - file.write(str(time_sec) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt0"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt1"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt2"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt3"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt4"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt5"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt6"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt7"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["lc" ]) + " ") - file.write(str(terminalSerObj.sensor_readouts["tc" ]) + " ") - file.write(terminalSerObj.valve_states["oxPress" ] + " ") - file.write(terminalSerObj.valve_states["oxVent" ] + " ") - file.write(terminalSerObj.valve_states["oxPurge" ] + " ") - file.write(terminalSerObj.valve_states["fuelPress"] + " ") - file.write(terminalSerObj.valve_states["fuelVent" ] + " ") - file.write(terminalSerObj.valve_states["fuelPurge"] + " ") - file.write(terminalSerObj.valve_states["oxMain" ] + " ") - file.write(terminalSerObj.valve_states["fuelMain" ] + " ") - file.write("\n") - -#<<<<<<< HEAD - #################################################################### - # Post-Fire Engine State # - #################################################################### - elif ( liquid_engine_state.get_engine_state() == "Post-Fire State"): - # Get telemetry - engineController.telreq( [], terminalSerObj, show_output = False ) - terminalSerObj.sensor_readouts["t"] = time_sec - sensor_readouts_formatted = {} - for sensor in terminalSerObj.sensor_readouts: - sensor_readouts_formatted[sensor] = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - sensor , - terminalSerObj.sensor_readouts[sensor] ) - - # Filter the data - if ( GAUGE_FILTER_ENABLED ): - sensor_data_buffer.add_data( terminalSerObj.sensor_readouts ) - sensor_data_filtered = sensor_data_buffer.get_data() - filtered_sensor_readouts_formatted = {} - for sensor in sensor_data_filtered: - filtered_sensor_readouts_formatted[sensor] = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - sensor , - sensor_data_filtered[sensor] ) - - # Calculate Flow Rates - if ( GAUGE_FILTER_ENABLED ): - ox_flow_rate = sensor_conv.ox_pressure_to_flow( - sensor_data_filtered["pt1"] - - sensor_data_filtered["pt2"] ) - fuel_flow_rate = sensor_conv.fuel_pressure_to_flow( - sensor_data_filtered["pt6"] - - sensor_data_filtered["pt5"] ) - ox_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "oxfr" , - ox_flow_rate ) - fuel_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "ffr" , - fuel_flow_rate ) - else: - ox_flow_rate = sensor_conv.ox_pressure_to_flow( - terminalSerObj.sensor_readouts["pt1"] - - terminalSerObj.sensor_readouts["pt2"] ) - fuel_flow_rate = sensor_conv.fuel_pressure_to_flow( - terminalSerObj.sensor_readouts["pt6"] - - terminalSerObj.sensor_readouts["pt5"] ) - ox_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "oxfr" , - ox_flow_rate ) - fuel_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "ffr" , - fuel_flow_rate ) - - # Update GUI - if ( GAUGE_FILTER_ENABLED ): - gauge1.setText( filtered_sensor_readouts_formatted["pt7"], "Fuel Tank Pressure" ) - gauge2.setText( fuel_flow_rate_formatted , "Fuel Flow Rate" ) - gauge3.setText( "NaN" , "None" ) - gauge4.setText( filtered_sensor_readouts_formatted["lc"] , "Thrust" ) - gauge5.setText( filtered_sensor_readouts_formatted["pt0"], "LOX Pressure" ) - gauge6.setText( ox_flow_rate_formatted , "LOX Flow Rate" ) - gauge7.setText( filtered_sensor_readouts_formatted["pt4"], "Engine Pressure" ) - gauge8.setText( filtered_sensor_readouts_formatted["tc" ], "LOX Temperature" ) - gauge1.setAngle( sensor_data_filtered["pt7"] ) - gauge2.setAngle( fuel_flow_rate ) - gauge3.setAngle( 0 ) - gauge4.setAngle( sensor_data_filtered["lc" ] ) - gauge5.setAngle( sensor_data_filtered["pt0"] ) - gauge6.setAngle( ox_flow_rate ) - gauge7.setAngle( sensor_data_filtered["pt4"] ) - gauge8.setAngle( sensor_data_filtered["tc" ] ) - else: - gauge1.setText( sensor_readouts_formatted["pt7"], "Fuel Tank Pressure" ) - gauge2.setText( fuel_flow_rate_formatted , "Fuel Flow Rate" ) - gauge3.setText( "NaN" , "None" ) - gauge4.setText( sensor_readouts_formatted["lc"] , "Thrust" ) - gauge5.setText( sensor_readouts_formatted["pt0"], "LOX Pressure" ) - gauge6.setText( ox_flow_rate_formatted , "LOX Flow Rate" ) - gauge7.setText( sensor_readouts_formatted["pt4"], "Engine Pressure" ) - gauge8.setText( sensor_readouts_formatted["tc" ], "LOX Temperature" ) - gauge1.setAngle( terminalSerObj.sensor_readouts["pt7"] ) - gauge2.setAngle( fuel_flow_rate ) - gauge3.setAngle( 0 ) - gauge4.setAngle( terminalSerObj.sensor_readouts["lc" ] ) - gauge5.setAngle( terminalSerObj.sensor_readouts["pt0"] ) - gauge6.setAngle( ox_flow_rate ) - gauge7.setAngle( terminalSerObj.sensor_readouts["pt4"] ) - gauge8.setAngle( terminalSerObj.sensor_readouts["tc" ] ) - update_valve_states( terminalSerObj.valve_states ) - - # Log Data - base_output_filename = output_dir + "/" + engine_state_filenames[liquid_engine_state.state] - output_filename = base_output_filename + str( test_num ) + ".txt" - with open( output_filename, "a" ) as file: - file.write(str(time_sec) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt0"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt1"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt2"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt3"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt4"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt5"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt6"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt7"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["lc" ]) + " ") - file.write(str(terminalSerObj.sensor_readouts["tc" ]) + " ") - file.write(terminalSerObj.valve_states["oxPress" ] + " ") - file.write(terminalSerObj.valve_states["oxVent" ] + " ") - file.write(terminalSerObj.valve_states["oxPurge" ] + " ") - file.write(terminalSerObj.valve_states["fuelPress"] + " ") - file.write(terminalSerObj.valve_states["fuelVent" ] + " ") - file.write(terminalSerObj.valve_states["fuelPurge"] + " ") - file.write(terminalSerObj.valve_states["oxMain" ] + " ") - file.write(terminalSerObj.valve_states["fuelMain" ] + " ") - file.write("\n") - - #################################################################### - # Manual State # - #################################################################### - elif ( liquid_engine_state.get_engine_state() == "Manual State"): - # Get telemetry - engineController.telreq( [], terminalSerObj, show_output = True ) - terminalSerObj.sensor_readouts["t"] = time_sec - sensor_readouts_formatted = {} - for sensor in terminalSerObj.sensor_readouts: - sensor_readouts_formatted[sensor] = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - sensor , - terminalSerObj.sensor_readouts[sensor] ) - - # Filter the data - if ( GAUGE_FILTER_ENABLED ): - sensor_data_buffer.add_data( terminalSerObj.sensor_readouts ) - sensor_data_filtered = sensor_data_buffer.get_data() - filtered_sensor_readouts_formatted = {} - for sensor in sensor_data_filtered: - filtered_sensor_readouts_formatted[sensor] = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - sensor , - sensor_data_filtered[sensor] ) - - # Calculate Flow Rates - if ( GAUGE_FILTER_ENABLED ): - ox_flow_rate = sensor_conv.ox_pressure_to_flow( - sensor_data_filtered["pt1"] - - sensor_data_filtered["pt2"] ) - fuel_flow_rate = sensor_conv.fuel_pressure_to_flow( - sensor_data_filtered["pt6"] - - sensor_data_filtered["pt5"] ) - ox_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "oxfr" , - ox_flow_rate ) - fuel_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "ffr" , - fuel_flow_rate ) - else: - ox_flow_rate = sensor_conv.ox_pressure_to_flow( - terminalSerObj.sensor_readouts["pt1"] - - terminalSerObj.sensor_readouts["pt2"] ) - fuel_flow_rate = sensor_conv.fuel_pressure_to_flow( - terminalSerObj.sensor_readouts["pt6"] - - terminalSerObj.sensor_readouts["pt5"] ) - ox_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "oxfr" , - ox_flow_rate ) - fuel_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "ffr" , - fuel_flow_rate ) - - # Update GUI - if ( GAUGE_FILTER_ENABLED ): - gauge1.setText( filtered_sensor_readouts_formatted["pt7"], "Fuel Tank Pressure" ) - gauge2.setText( fuel_flow_rate_formatted , "Fuel Flow Rate" ) - gauge3.setText( "NaN" , "None" ) - gauge4.setText( filtered_sensor_readouts_formatted["lc"] , "Thrust" ) - gauge5.setText( filtered_sensor_readouts_formatted["pt0"], "LOX Pressure" ) - gauge6.setText( ox_flow_rate_formatted , "LOX Flow Rate" ) - gauge7.setText( filtered_sensor_readouts_formatted["pt4"], "Engine Pressure" ) - gauge8.setText( filtered_sensor_readouts_formatted["tc" ], "LOX Temperature" ) - gauge1.setAngle( sensor_data_filtered["pt7"] ) - gauge2.setAngle( fuel_flow_rate ) - gauge3.setAngle( 0 ) - gauge4.setAngle( sensor_data_filtered["lc" ] ) - gauge5.setAngle( sensor_data_filtered["pt0"] ) - gauge6.setAngle( ox_flow_rate ) - gauge7.setAngle( sensor_data_filtered["pt4"] ) - gauge8.setAngle( sensor_data_filtered["tc" ] ) - else: - gauge1.setText( sensor_readouts_formatted["pt7"], "Fuel Tank Pressure" ) - gauge2.setText( fuel_flow_rate_formatted , "Fuel Flow Rate" ) - gauge3.setText( "NaN" , "None" ) - gauge4.setText( sensor_readouts_formatted["lc"] , "Thrust" ) - gauge5.setText( sensor_readouts_formatted["pt0"], "LOX Pressure" ) - gauge6.setText( ox_flow_rate_formatted , "LOX Flow Rate" ) - gauge7.setText( sensor_readouts_formatted["pt4"], "Engine Pressure" ) - gauge8.setText( sensor_readouts_formatted["tc" ], "LOX Temperature" ) - gauge1.setAngle( terminalSerObj.sensor_readouts["pt7"] ) - gauge2.setAngle( fuel_flow_rate ) - gauge3.setAngle( 0 ) - gauge4.setAngle( terminalSerObj.sensor_readouts["lc" ] ) - gauge5.setAngle( terminalSerObj.sensor_readouts["pt0"] ) - gauge6.setAngle( ox_flow_rate ) - gauge7.setAngle( terminalSerObj.sensor_readouts["pt4"] ) - gauge8.setAngle( terminalSerObj.sensor_readouts["tc" ] ) - update_valve_states( terminalSerObj.valve_states ) - - - #################################################################### - # Abort State # - #################################################################### - elif ( liquid_engine_state.get_engine_state() == "Abort State" ): - # Get telemetry - engineController.telreq( [], terminalSerObj, show_output = False ) - terminalSerObj.sensor_readouts["t"] = time_sec - sensor_readouts_formatted = {} - for sensor in terminalSerObj.sensor_readouts: - sensor_readouts_formatted[sensor] = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - sensor , - terminalSerObj.sensor_readouts[sensor] ) - - # Filter the data - if ( GAUGE_FILTER_ENABLED ): - sensor_data_buffer.add_data( terminalSerObj.sensor_readouts ) - sensor_data_filtered = sensor_data_buffer.get_data() - filtered_sensor_readouts_formatted = {} - for sensor in sensor_data_filtered: - filtered_sensor_readouts_formatted[sensor] = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - sensor , - sensor_data_filtered[sensor] ) - - # Calculate Flow Rates - if ( GAUGE_FILTER_ENABLED ): - ox_flow_rate = sensor_conv.ox_pressure_to_flow( - sensor_data_filtered["pt1"] - - sensor_data_filtered["pt2"] ) - fuel_flow_rate = sensor_conv.fuel_pressure_to_flow( - sensor_data_filtered["pt6"] - - sensor_data_filtered["pt5"] ) - ox_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "oxfr" , - ox_flow_rate ) - fuel_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "ffr" , - fuel_flow_rate ) - else: - ox_flow_rate = sensor_conv.ox_pressure_to_flow( - terminalSerObj.sensor_readouts["pt1"] - - terminalSerObj.sensor_readouts["pt2"] ) - fuel_flow_rate = sensor_conv.fuel_pressure_to_flow( - terminalSerObj.sensor_readouts["pt6"] - - terminalSerObj.sensor_readouts["pt5"] ) - ox_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "oxfr" , - ox_flow_rate ) - fuel_flow_rate_formatted = SDR_sensor.format_sensor_readout( - terminalSerObj.controller, - "ffr" , - fuel_flow_rate ) - - # Update GUI - if ( GAUGE_FILTER_ENABLED ): - gauge1.setText( filtered_sensor_readouts_formatted["pt7"], "Fuel Tank Pressure" ) - gauge2.setText( fuel_flow_rate_formatted , "Fuel Flow Rate" ) - gauge3.setText( "NaN" , "None" ) - gauge4.setText( filtered_sensor_readouts_formatted["lc"] , "Thrust" ) - gauge5.setText( filtered_sensor_readouts_formatted["pt0"], "LOX Pressure" ) - gauge6.setText( ox_flow_rate_formatted , "LOX Flow Rate" ) - gauge7.setText( filtered_sensor_readouts_formatted["pt4"], "Engine Pressure" ) - gauge8.setText( filtered_sensor_readouts_formatted["tc" ], "LOX Temperature" ) - gauge1.setAngle( sensor_data_filtered["pt7"] ) - gauge2.setAngle( fuel_flow_rate ) - gauge3.setAngle( 0 ) - gauge4.setAngle( sensor_data_filtered["lc" ] ) - gauge5.setAngle( sensor_data_filtered["pt0"] ) - gauge6.setAngle( ox_flow_rate ) - gauge7.setAngle( sensor_data_filtered["pt4"] ) - gauge8.setAngle( sensor_data_filtered["tc" ] ) - else: - gauge1.setText( sensor_readouts_formatted["pt7"], "Fuel Tank Pressure" ) - gauge2.setText( fuel_flow_rate_formatted , "Fuel Flow Rate" ) - gauge3.setText( "NaN" , "None" ) - gauge4.setText( sensor_readouts_formatted["lc"] , "Thrust" ) - gauge5.setText( sensor_readouts_formatted["pt0"], "LOX Pressure" ) - gauge6.setText( ox_flow_rate_formatted , "LOX Flow Rate" ) - gauge7.setText( sensor_readouts_formatted["pt4"], "Engine Pressure" ) - gauge8.setText( sensor_readouts_formatted["tc" ], "LOX Temperature" ) - gauge1.setAngle( terminalSerObj.sensor_readouts["pt7"] ) - gauge2.setAngle( fuel_flow_rate ) - gauge3.setAngle( 0 ) - gauge4.setAngle( terminalSerObj.sensor_readouts["lc" ] ) - gauge5.setAngle( terminalSerObj.sensor_readouts["pt0"] ) - gauge6.setAngle( ox_flow_rate ) - gauge7.setAngle( terminalSerObj.sensor_readouts["pt4"] ) - gauge8.setAngle( terminalSerObj.sensor_readouts["tc" ] ) - update_valve_states( terminalSerObj.valve_states ) - # Update engin e schematic - plumbing.updatePipeStatus() - - # Log Data - base_output_filename = output_dir + "/" + engine_state_filenames[liquid_engine_state.state] - output_filename = base_output_filename + str( test_num ) + ".txt" - with open( output_filename, "a" ) as file: - file.write(str(time_sec) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt0"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt1"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt2"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt3"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt4"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt5"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt6"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["pt7"]) + " ") - file.write(str(terminalSerObj.sensor_readouts["lc" ]) + " ") - file.write(str(terminalSerObj.sensor_readouts["tc" ]) + " ") - file.write(terminalSerObj.valve_states["oxPress" ] + " ") - file.write(terminalSerObj.valve_states["oxVent" ] + " ") - file.write(terminalSerObj.valve_states["oxPurge" ] + " ") - file.write(terminalSerObj.valve_states["fuelPress"] + " ") - file.write(terminalSerObj.valve_states["fuelVent" ] + " ") - file.write(terminalSerObj.valve_states["fuelPurge"] + " ") - file.write(terminalSerObj.valve_states["oxMain" ] + " ") - file.write(terminalSerObj.valve_states["fuelMain" ] + " ") - file.write("\n") - - - # Update engine schematic - plumbing.updatePipeStatus() - - # Draw to main window - root.update() - - # Draw to plumbing window - plumbing.getWindow().update() + state = liquid_engine_state.get_engine_state() + + if state == "Ready State": + processor.process(show_output=True) + elif state == "Pre-Fire Purge State": + processor.process() + elif state == "Fill and Chill State": + engineController.tankstat([], terminalSerObj) + processor.process() + elif state == "Standby State": + processor.process() + elif state == "Fire State": + processor.process() + elif state == "Disarm State": + engineController.tankstat([], terminalSerObj) + processor.process() + elif state == "Post-Fire State": + processor.process() + elif state == "Manual State": + processor.process(show_output=True) + elif state == "Abort State": + processor.process() + + window.update() #################################################################################### # END OF FILE # -#################################################################################### \ No newline at end of file +#################################################################################### diff --git a/sensor_data_buffer.py b/sensor_data_buffer.py new file mode 100644 index 0000000..956e3b3 --- /dev/null +++ b/sensor_data_buffer.py @@ -0,0 +1,55 @@ +#################################################################################### +# # +# sensor_data_buffer.py -- FIFO low-pass filter buffer for gauge readouts # +# # +# Author: Nitish Chennoju, Colton Acosta # +# Sun Devil Rocketry Avionics # +# # +#################################################################################### + + +class Sensor_Data_Buffer: + """FIFO buffer that applies a low-pass filter to incoming sensor data. + + Parameters + ---------- + max_buffer_size : int — maximum number of samples to keep + tau : float — time constant of the low-pass filter + """ + + def __init__( self, max_buffer_size=10, tau=1 ): + self.data_buffer = [] + self.filter_buffer = [] + self.buffer_size = 0 + self.max_buffer_size = max_buffer_size + self.tau = tau + self.sensors = ["pt0","pt1","pt2","pt3","pt4","pt5","pt6","pt7","lc","tc"] + + def add_data( self, sensor_data ): + if self.buffer_size < self.max_buffer_size: + self.data_buffer.append( sensor_data ) + self.buffer_size += 1 + if self.buffer_size == 1: + self.filter_buffer.append( sensor_data ) + else: + self.filter_buffer.append( self.filter_data() ) + else: + self.data_buffer.pop( 0 ) + self.data_buffer.append( sensor_data ) + self.filter_buffer.append( self.filter_data() ) + + def filter_data( self ): + T = self.data_buffer[-1]["t"] - self.data_buffer[-2]["t"] + alpha = 2*self.tau + T + beta = T - 2*self.tau + filtered_readouts = {} + for sensor in self.sensors: + filtered_readouts[sensor] = ( + (T/alpha) * (self.data_buffer[-1][sensor] + self.data_buffer[-2][sensor]) + - (beta/alpha) * self.filter_buffer[-1][sensor] + ) + return filtered_readouts + + def get_data( self ): + return self.filter_buffer[-1] +## Sensor_Data_Buffer ## diff --git a/telemetry.py b/telemetry.py new file mode 100644 index 0000000..86d5e69 --- /dev/null +++ b/telemetry.py @@ -0,0 +1,133 @@ +import time +import os +import sys + +import sensor_conv +import engineController +import sensor as SDR_sensor + +class TelemetryProcessor: + """Encapsulates the per-loop telemetry pipeline: + + request → format → filter → flow rates → gauge update → valve sync → log + + By accepting all dependencies through the constructor this class has no + knowledge of Tkinter, serial setup, or application state beyond what it + needs to do its job. + """ + + def __init__( self, + terminal, + sensor_buffer, + gauges, + update_valve_fn, + engine_state, + engine_state_filenames, + output_dir, + test_num, + start_time, + gauge_filter_enabled=True ): + """ + Parameters + ---------- + terminal : sdec.terminalData + sensor_buffer : Sensor_Data_Buffer + gauges : list[SDR_gauge.gauge] — ordered to match gauge layout + update_valve_fn : callable(valve_states) — e.g. window.update_valve_states + engine_state : Liquid_Engine_State + engine_state_filenames : dict[str, str] + output_dir : str + test_num : int + start_time : float — value of time.perf_counter() at app start + gauge_filter_enabled : bool + """ + self._terminal = terminal + self._sensor_buffer = sensor_buffer + self._gauges = gauges + self._update_valve_fn = update_valve_fn + self._engine_state = engine_state + self._engine_state_filenames = engine_state_filenames + self._output_dir = output_dir + self._test_num = test_num + self._start_time = start_time + self._gauge_filter_enabled = gauge_filter_enabled + + def process( self, show_output=False ): + """Run one full telemetry cycle for the current engine state.""" + time_sec = time.perf_counter() - self._start_time + + engineController.telreq([], self._terminal, show_output=show_output) + self._terminal.sensor_readouts["t"] = time_sec + + src, src_fmt = self._get_sensor_source() + + ox_fr, fuel_fr, ox_fr_fmt, fuel_fr_fmt = self._compute_flow_rates(src) + + self._update_gauges(src, src_fmt, ox_fr, fuel_fr, ox_fr_fmt, fuel_fr_fmt) + self._update_valve_fn(self._terminal.valve_states) + self._log_data(time_sec) + + # ------------------------------------------------------------------------- + # Private helpers + # ------------------------------------------------------------------------- + + def _format_readouts( self, readout_dict ): + return { + s: SDR_sensor.format_sensor_readout(self._terminal.controller, s, v) + for s, v in readout_dict.items() + } + + def _get_sensor_source( self ): + """Return (source_values, formatted_source_values), applying the + low-pass filter when enabled.""" + fmt = self._format_readouts(self._terminal.sensor_readouts) + + if self._gauge_filter_enabled: + self._sensor_buffer.add_data(self._terminal.sensor_readouts) + filtered = self._sensor_buffer.get_data() + return filtered, self._format_readouts(filtered) + + return self._terminal.sensor_readouts, fmt + + def _compute_flow_rates( self, src ): + ox_fr = sensor_conv.ox_pressure_to_flow(src["pt1"] - src["pt2"]) + fuel_fr = sensor_conv.fuel_pressure_to_flow(src["pt6"] - src["pt5"]) + ox_fr_fmt = SDR_sensor.format_sensor_readout( + self._terminal.controller, "oxfr", ox_fr) + fuel_fr_fmt = SDR_sensor.format_sensor_readout( + self._terminal.controller, "ffr", fuel_fr) + return ox_fr, fuel_fr, ox_fr_fmt, fuel_fr_fmt + + def _update_gauges( self, src, src_fmt, ox_fr, fuel_fr, ox_fr_fmt, fuel_fr_fmt ): + # Order must match the _GAUGE_CONFIGS table in DashboardWindow + gauge_updates = [ + (src_fmt["pt7"], "Fuel Tank Pressure", src["pt7"]), + (fuel_fr_fmt, "Fuel Flow Rate", fuel_fr ), + ("NaN", "None", 0 ), + (src_fmt["lc"], "Thrust", src["lc"] ), + (src_fmt["pt0"], "LOX Pressure", src["pt0"]), + (ox_fr_fmt, "LOX Flow Rate", ox_fr ), + (src_fmt["pt4"], "Engine Pressure", src["pt4"]), + (src_fmt["tc"], "LOX Temperature", src["tc"] ), + ] + for gauge, (text_val, label, angle_val) in zip(self._gauges, gauge_updates): + gauge.setText(text_val, label) + gauge.setAngle(angle_val) + + def _log_data( self, time_sec ): + state_name = self._engine_state.state + log_path = os.path.join( + self._output_dir, + self._engine_state_filenames[state_name] + str(self._test_num) + ".txt" + ) + _SENSOR_KEYS = ["pt0","pt1","pt2","pt3","pt4","pt5","pt6","pt7","lc","tc"] + _VALVE_KEYS = ["oxPress","oxVent","oxPurge","fuelPress","fuelVent", + "fuelPurge","oxMain","fuelMain"] + with open(log_path, "a") as f: + f.write(str(time_sec) + " ") + for s in _SENSOR_KEYS: + f.write(str(self._terminal.sensor_readouts[s]) + " ") + for v in _VALVE_KEYS: + f.write(self._terminal.valve_states[v] + " ") + f.write("\n") +## TelemetryProcessor ##