Source code for autopilot.core.terminal

"""Methods for running the Terminal GUI"""

import typing
import argparse
import json
import sys
import os
from pathlib import Path
from pprint import pformat
import time

import datetime
import logging
import threading
from collections import OrderedDict as odict
import numpy as np

from PySide2 import QtCore, QtGui, QtSvg, QtWidgets

from autopilot import prefs
from autopilot.core import styles

if __name__ == '__main__':
    # Parse arguments - this should have been called with a .json prefs file passed
    # We'll try to look in the default location first
    parser = argparse.ArgumentParser(description="Run an autopilot Terminal")
    parser.add_argument('-f', '--prefs', help="Location of .json prefs file (created during setup_autopilot.py)")
    args = parser.parse_args()

    if not args.prefs:
        prefs_file = '/usr/autopilot/prefs.json'

        if not os.path.exists(prefs_file):
            raise Exception("No Prefs file passed, and file not in default location")

        raise Warning('No prefs file passed, loaded from default location. Should pass explicitly with -p')

    else:
        prefs_file = args.prefs

    # init prefs for module access
    prefs.init(prefs_file)

from autopilot.core.subject import Subject
from autopilot.core.plots import Plot_Widget
from autopilot.networking import Net_Node, Terminal_Station
from autopilot.utils.invoker import get_invoker
from autopilot.core.gui import Control_Panel, Protocol_Wizard, Weights, Reassign, Calibrate_Water, Bandwidth_Test, pop_dialog, Stream_Video, Plugins
from autopilot.core.loggers import init_logger

# Try to import viz, but continue if that doesn't work
IMPORTED_VIZ = False
VIZ_ERROR = None
try:
    from autopilot import viz
    IMPORTED_VIZ = True
except ImportError as e:
    VIZ_ERROR = str(e)
import pdb


# TODO: Be more complete about generating logs
# TODO: Make exit graceful
# TODO: Make 'edit subject' button
# TODO: Make experiment tags, save and populate?

# http://zetcode.com/gui/pysidetutorial/layoutmanagement/
# https://wiki.qt.io/PySide_Tutorials

_TERMINAL = None

[docs]class Terminal(QtWidgets.QMainWindow): """ Central host to a swarm of :class:`.Pilot` s and user-facing :mod:`~.core.gui` objects. Called as a module with the -f flag to give the location of a prefs file, eg:: python terminal.py -f prefs_file.json if the -f flag is not passed, looks in the default location for prefs (ie. `/usr/autopilot/prefs.json`) **Listens used by the internal :class:`.Net_Node` ** +---------------+--------------------------------+--------------------------------------------------------+ | Key | Method | Description | +===============+================================+========================================================+ | `'STATE'` | :meth:`~.Terminal.l_state` | A Pi has changed state | +---------------+--------------------------------+--------------------------------------------------------+ | `'PING'` | :meth:`~.Terminal.l_ping` | Someone wants to know if we're alive | +---------------+--------------------------------+--------------------------------------------------------+ | `'DATA'` | :meth:`~.Terminal.l_data` | Receiving data to store | +---------------+--------------------------------+--------------------------------------------------------+ | `'HANDSHAKE'` | :meth:`~.Terminal.l_handshake` | Pilot first contact, telling us it's alive and its IP | +---------------+--------------------------------+--------------------------------------------------------+ .. note:: See :mod:`autopilot.prefs` for full list of prefs needed by terminal! Attributes: node (:class:`~.networking.Net_Node`): Our Net_Node we use to communicate with our main networking object networking (:class:`~.networking.Terminal_Station`): Our networking object to communicate with the outside world subjects (dict): A dictionary mapping subject ID to :class:`~.subject.Subject` object. layout (:class:`QtWidgets.QGridLayout`): Layout used to organize widgets control_panel (:class:`~.gui.Control_Panel`): Control Panel to manage pilots and subjects data_panel (:class:`~.plots.Plot_Widget`): Plots for each pilot and subject. logo (:class:`QtWidgets.QLabel`): Label holding our beautiful logo ;X logger (:class:`logging.Logger`): Used to log messages and network events. settings (:class:`PySide2.QtCore.QSettings`): QSettings used to store pyside configuration like window size, stored in ``prefs.get("TERMINAL_SETTINGS_FN")`` """ def __init__(self): # type: () -> None super(Terminal, self).__init__() # store instance globals()['_TERMINAL'] = self # Load settings # Currently, the only setting is "geometry", but loading here # in case we start to use other ones in the future self.settings = QtCore.QSettings(prefs.get("TERMINAL_SETTINGS_FN"), QtCore.QSettings.NativeFormat) # networking self.node = None self.networking = None self.heartbeat_dur = 10 # check every n seconds whether our pis are around still # data self.subjects = {} # Dict of our open subject objects # gui self.layout = None self.widget = None self.file_menu = None self.tool_menu = None self.control_panel = None self.data_panel = None self.logo = None # property private attributes self._pilots = None # logging self.logger = init_logger(self) # Listen dictionary - which methods to call for different messages # Methods are spawned in new threads using handle_message self.listens = { 'STATE': self.l_state, # A Pi has changed state 'PING' : self.l_ping, # Someone wants to know if we're alive 'DATA' : self.l_data, 'CONTINUOUS': self.l_data, # handle continuous data same way as other data 'STREAM': self.l_data, 'HANDSHAKE': self.l_handshake # a pi is making first contact, telling us its IP } # Make invoker object to send GUI events back to the main thread # self.invoker = Invoker() self.invoker = get_invoker() # prefs.add('INVOKER', self.invoker) self.initUI() # Start Networking # Networking is in two parts, # "internal" networking for messages sent to and from the Terminal object itself # "external" networking for messages to and from all the other components, # The split is so the external networking can run in another process, do potentially time-consuming tasks # like resending & confirming message delivery without blocking or missing messages # Start external communications in own process # Has to be after init_network so it makes a new context self.networking = Terminal_Station(self.pilots) self.networking.start() self.logger.info("Station object Initialized") self.node = Net_Node(id="_T", upstream='T', port=prefs.get('MSGPORT'), listens=self.listens, instance=False) self.logger.info("Net Node Initialized") # send an initial ping looking for our pilots self.node.send('T', 'INIT') # start beating ur heart # self.heartbeat_timer = threading.Timer(self.heartbeat_dur, self.heartbeat) # self.heartbeat_timer.daemon = True # self.heartbeat_timer.start() #self.heartbeat(once=True) self.logger.info('Terminal Initialized') # if we don't have any pilots, pop a dialogue to declare one if len(self.pilots) == 0: box = pop_dialog( 'No Pilots', 'No pilots were found in the pilot_db, add one now?', buttons=('Yes', 'No') ) ret = box.exec_() if ret == box.Yes: self.new_pilot()
[docs] def initUI(self): """ Initializes graphical elements of Terminal. Including... * Toolbar * :class:`.gui.Control_Panel` * :class:`.plots.Plot_Widget` """ # Set central widget self.widget = QtWidgets.QWidget() self.setCentralWidget(self.widget) # Set the layout self.layout = QtWidgets.QGridLayout() self.layout.setSpacing(0) self.layout.setContentsMargins(0,0,0,0) self.widget.setLayout(self.layout) # Set title self.setWindowTitle('Terminal') #self.menuBar().setFixedHeight(40) # This is the pixel resolution of the entire screen if 'pytest' in sys.modules: primary_display = None terminal_winsize_behavior = 'custom' custom_size=[0,0,1000,480] else: terminal_winsize_behavior = prefs.get('TERMINAL_WINSIZE_BEHAVIOR') custom_size = prefs.get('TERMINAL_CUSTOM_SIZE') app = QtWidgets.QApplication.instance() screensize = app.primaryScreen().size() # This is the available geometry of the primary screen, excluding # window manager reserved areas such as task bars and system menus. primary_display = app.primaryScreen().availableGeometry() ## Initalize the menuBar # Linux: Set the menuBar to a fixed height # Darwin: Don't worry about menuBar if sys.platform == 'darwin': bar_height = 0 else: if primary_display is None: self.menuBar().setFixedHeight(30) else: bar_height = (primary_display.height()/30)+5 self.menuBar().setFixedHeight(bar_height) # Create a File menu self.file_menu = self.menuBar().addMenu("&File") self.file_menu.setObjectName("file") # Add "New Pilot" and "New Protocol" actions to File menu new_pilot_act = QtWidgets.QAction("New &Pilot", self, triggered=self.new_pilot) new_prot_act = QtWidgets.QAction("New Pro&tocol", self, triggered=self.new_protocol) #batch_create_subjects = QtWidgets.QAction("Batch &Create subjects", self, triggered=self.batch_subjects) # TODO: Update pis self.file_menu.addAction(new_pilot_act) self.file_menu.addAction(new_prot_act) #self.file_menu.addAction(batch_create_subjects) # Create a Tools menu self.tool_menu = self.menuBar().addMenu("&Tools") # Add actions to Tools menu subject_weights_act = QtWidgets.QAction("View Subject &Weights", self, triggered=self.subject_weights) update_protocol_act = QtWidgets.QAction("Update Protocols", self, triggered=self.update_protocols) reassign_act = QtWidgets.QAction("Batch Reassign Protocols", self, triggered=self.reassign_protocols) calibrate_act = QtWidgets.QAction("Calibrate &Water Ports", self, triggered=self.calibrate_ports) self.tool_menu.addAction(subject_weights_act) self.tool_menu.addAction(update_protocol_act) self.tool_menu.addAction(reassign_act) self.tool_menu.addAction(calibrate_act) # Swarm menu # (tools 4 administering and interacting with agents in swarm) self.swarm_menu = self.menuBar().addMenu("S&warm") stream_video = QtWidgets.QAction("Stream Video", self, triggered=self.stream_video) self.swarm_menu.addAction(stream_video) # Plots menu self.plots_menu = self.menuBar().addMenu("&Plots") psychometric = QtWidgets.QAction("Psychometric Curve", self, triggered=self.plot_psychometric) self.plots_menu.addAction(psychometric) # Create a Tests menu and add a Test Bandwidth action self.tests_menu = self.menuBar().addMenu("Test&s") bandwidth_test_act = QtWidgets.QAction("Test Bandwidth", self, triggered=self.test_bandwidth) self.tests_menu.addAction(bandwidth_test_act) # Create a Plugins menu to manage plugins and provide a hook to give them additional terminal actions self.plugins_menu = self.menuBar().addMenu("Plugins") plugin = QtWidgets.QAction("Manage Plugins", self, triggered=self.manage_plugins) self.plugins_menu.addAction(plugin) ## Init main panels and add to layout # Control panel sits on the left, controls pilots & subjects self.control_panel = Control_Panel(pilots=self.pilots, subjects=self.subjects, start_fn=self.toggle_start, ping_fn=self.ping_pilot) # Data panel sits on the right, plots stuff. self.data_panel = Plot_Widget() self.data_panel.init_plots(self.pilots.keys()) # Set logo to corner widget # if sys.platform != 'darwin': # self.menuBar().setCornerWidget(self.logo, QtCore.Qt.TopRightCorner) # self.menuBar().adjustSize() # Add Control Panel and Data Panel to main layout #self.layout.addWidget(self.logo, 0,0,1,2) self.layout.addWidget(self.control_panel, 0,0,1,1) self.layout.addWidget(self.data_panel, 0,1,1,1) self.layout.setColumnStretch(0, 1) self.layout.setColumnStretch(1, 3) ## Set window size # The window size behavior depends on TERMINAL_WINSIZE_BEHAVIOR pref # If 'remember': restore to the geometry from the last close # If 'maximum': restore to fill the entire screen # If 'moderate': restore to a reasonable size of (1000, 400) pixels # Set geometry according to pref if terminal_winsize_behavior == 'maximum': # Set geometry to available geometry self.setGeometry(primary_display) # Set SizePolicy to maximum self.setSizePolicy( QtWidgets.QSizePolicy.Maximum, QtWidgets.QSizePolicy.Maximum) # Move to top left corner of primary display self.move(primary_display.left(), primary_display.top()) # Also set the maximum height of each panel self.control_panel.setMaximumHeight(primary_display.height()) self.data_panel.setMaximumHeight(primary_display.height()) elif terminal_winsize_behavior == 'remember': # Attempt to restore previous geometry if self.settings.value("geometry") is None: # It was never saved, for instance, this is the first time # this app has been run # So default to the moderate size self.move(primary_display.left(), primary_display.top()) self.resize(1000, 400) else: # It was saved, so restore the last geometry self.restoreGeometry(self.settings.value("geometry")) elif terminal_winsize_behavior == "custom": self.move(custom_size[0], custom_size[1]) self.resize(custom_size[2], custom_size[3]) else: if terminal_winsize_behavior != 'moderate': self.logger.warning(f'TERMINAL_WINSIZE_BEHAVIOR {terminal_winsize_behavior} is not implemented, defaulting to "moderate"') # The moderate size self.move(primary_display.left(), primary_display.top()) self.resize(1000, 400) ## Finalize some aesthetics # set stylesheet for main window self.setStyleSheet(styles.TERMINAL) # set fonts to antialias self.setFont(self.font().setStyleStrategy(QtGui.QFont.PreferAntialias)) ## Show, and log that initialization is complete self.show() logging.info('UI Initialized')
[docs] def reset_ui(self): """ Clear Layout and call :meth:`~.Terminal.initUI` again """ # type: () -> None self.layout = QtWidgets.QGridLayout() self.layout.setSpacing(0) self.layout.setContentsMargins(0,0,0,0) self.widget.setLayout(self.layout) self.setCentralWidget(self.widget) self.initUI()
################ # Properties @property def pilots(self) -> odict: """ A dictionary mapping pilot ID to its attributes, including a list of its subjects assigned to it, its IP, etc. Returns: dict: like ``self.pilots['pilot_id'] = {'subjects': ['subject_0', 'subject_1'], 'ip': '192.168.0.101'}`` """ # try to load, if none exists make one if self._pilots is None: pilot_db_fn = Path(prefs.get('PILOT_DB')) # if pilot file doesn't exist, make blank one if not pilot_db_fn.exists(): self.logger.warning(f'No pilot_db.json file was found at {pilot_db_fn}, creating a new one') self._pilots = odict() with open(pilot_db_fn, 'w') as pilot_file: json.dump(self._pilots, pilot_file) # otherwise, try to load it else: try: # Load pilots db as ordered dictionary with open(pilot_db_fn, 'r') as pilot_file: self._pilots = json.load(pilot_file, object_pairs_hook=odict) self.logger.info(f'successfully loaded pilot_db.json file from {pilot_db_fn}') self.logger.debug(pformat(self._pilots)) except Exception as e: self.logger.exception((f"Exception opening pilot_db.json file at {pilot_db_fn}, got exception: {e}.\n", "Not proceeding to prevent possibly overwriting corrupt pilot_db.file")) raise e return self._pilots @property def protocols(self) -> list: """ List of protocol names available in ``PROTOCOLDIR`` Returns: list: list of protocol names in ``prefs.get('PROTOCOLDIR')`` """ # get list of protocol files protocols = os.listdir(prefs.get('PROTOCOLDIR')) protocols = [os.path.splitext(p)[0] for p in protocols if p.endswith('.json')] return protocols @property def subject_protocols(self) -> dict: """ Returns: subject_protocols (dict): a dictionary of subjects: [protocol, step] """ # get subjects and current protocols subjects = self.subject_list subjects_protocols = {} for subject in subjects: if subject not in self.subjects.keys(): self.subjects[subject] = Subject(subject) subjects_protocols[subject] = [self.subjects[subject].protocol_name, self.subjects[subject].step] return subjects_protocols @property def subject_list(self) -> list: """ Get a list of all subject IDs Returns: list: list of all subject IDs present in :attr:`.Terminal.pilots` """ subjects = [] for pilot, vals in self.pilots.items(): subjects.extend(vals['subjects']) # use sets to get a unique list subjects = list(set(subjects)) return subjects ##########################3 # Listens & inter-object methods
[docs] def ping_pilot(self, pilot): self.node.send(pilot, 'PING')
[docs] def heartbeat(self, once=False): """ Perioducally send an ``INIT`` message that checks the status of connected pilots sent with frequency according to :attr:`.Terminal.heartbeat_dur` Args: once (bool): if True, do a single heartbeat but don't start a thread to do more. """ self.node.send('T', 'INIT', repeat=False, flags={'NOREPEAT': True}) if not once: self.heartbeat_timer = threading.Timer(self.heartbeat_dur, self.heartbeat) self.heartbeat_timer.daemon = True self.heartbeat_timer.start()
[docs] def toggle_start(self, starting, pilot, subject=None): """Start or Stop running the currently selected subject's task. Sends a message containing the task information to the concerned pilot. Each :class:`Pilot_Panel` is given a lambda function that calls this one with the arguments specified See :class:`Pilot_Button`, as it is what calls this function. Args: starting (bool): Does this button press mean we are starting (True) or stopping (False) the task? pilot: Which Pilot is starting or stopping? subject: Which Subject is currently selected? """ # stopping is the enemy of starting so we put them in the same function to learn about each other if starting is True: # Get Weights start_weight, ok = QtWidgets.QInputDialog.getDouble(self, "Set Starting Weight", "Starting Weight:") if ok: # Ope'nr up if she aint if subject not in self.subjects.keys(): self.subjects[subject] = Subject(subject) task = self.subjects[subject].prepare_run() task['pilot'] = pilot self.subjects[subject].update_weights(start=float(start_weight)) self.node.send(to=pilot, key="START", value=task) # also let the plot know to start self.node.send(to="P_{}".format(pilot), key="START", value=task) else: # pressed cancel, don't start return else: # Get Weights stop_weight, ok = QtWidgets.QInputDialog.getDouble(self, "Set Stopping Weight", "Stopping Weight:") if ok: # Send message to pilot to stop running, # it should initiate a coherence checking routine to make sure # its data matches what the Terminal got, # so the terminal will handle closing the subject object self.node.send(to=pilot, key="STOP") # also let the plot know to start self.node.send(to="P_{}".format(pilot), key="STOP") # TODO: Start coherence checking ritual # TODO: Auto-select the next subject in the list. self.subjects[subject].stop_run() self.subjects[subject].update_weights(stop=float(stop_weight)) else: # pressed cancel return
############################ # MESSAGE HANDLING METHODS
[docs] def l_data(self, value): """ A Pilot has sent us data. `value` field of message should have `subject` and `pilot` added to dictionary for identification. Any key in `value` that matches a column in the subject's trial data table will be saved. If the subject graduates after receiving this piece of data, stop the current task running on the Pilot and send the new one. Args: value (dict): A dict of field-value pairs to save """ # A Pi has sent us data, let's save it huh? subject_name = value['subject'] self.subjects[subject_name].save_data(value) if self.subjects[subject_name].did_graduate.is_set() is True: self.node.send(to=value['pilot'], key="STOP", value={'graduation':True}) self.subjects[subject_name].stop_run() self.subjects[subject_name].graduate() task = self.subjects[subject_name].prepare_run() task['pilot'] = value['pilot'] # FIXME: Don't hardcode wait time, wait until we get confirmation that the running task has fully unloaded time.sleep(5) self.node.send(to=value['pilot'], key="START", value=task)
[docs] def l_ping(self, value): """ TODO: Reminder to implement heartbeating. Note: Currently unused, as Terminal Net_Node stability hasn't been a problem and no universal system of heartbeating has been established (global stability has not been an issue). Args: value: (unused) """ # Only our Station object should ever ping us, because # we otherwise want it handling any pings on our behalf. # self.send_message('ALIVE', value=b'T') pass
[docs] def l_state(self, value): """A Pilot has changed state, keep track of it. Args: value (dict): dict containing `state` . """ # TODO: If we are stopping, we enter into a cohere state # TODO: If we are stopped, close the subject object. # TODO: Also tell the relevant dataview to clear # update the pilot button self.logger.debug(f'updating pilot state: {value}') if value['pilot'] not in self.pilots.keys(): self.logger.info('Got state info from an unknown pilot, adding...') self.new_pilot(name=value['pilot']) self.pilots[value['pilot']]['state'] = value['state'] self.control_panel.panels[value['pilot']].button.set_state(value['state'])
[docs] def l_handshake(self, value): """ Pilot is sending its IP and state on startup. If we haven't heard of this pilot before, make a new entry in :attr:`~.Terminal.pilots` and :meth:`.gui.Control_Panel.update_db` . Args: value (dict): dict containing `ip` and `state` """ if value['pilot'] in self.pilots.keys(): self.pilots[value['pilot']]['ip'] = value.get('ip', '') self.pilots[value['pilot']]['state'] = value.get('state', '') self.pilots[value['pilot']]['prefs'] = value.get('prefs', {}) else: self.new_pilot(name=value['pilot'], ip=value.get('ip', ''), pilot_prefs=value.get('prefs', {})) # update the pilot button if value['pilot'] in self.control_panel.panels.keys(): self.control_panel.panels[value['pilot']].button.set_state(value['state']) self.control_panel.update_db()
############################# # GUI & etc. methods
[docs] def new_pilot(self, name:typing.Optional[str]=None, ip:str='', pilot_prefs:typing.Optional[dict]=None): """ Make a new entry in :attr:`.Terminal.pilots` and make appropriate GUI elements. Args: ip (str): Optional. if given, stored in db. name (str): If None, prompted for a name, otherwise used for entry in pilot DB. """ if name is None: name, ok = QtWidgets.QInputDialog.getText(self, "Pilot ID", "Pilot ID:") if not ok or not name: self.logger.info('Cancel button clicked, not adding new pilot') return # Warn if we're going to overwrite if name in self.pilots.keys(): self.logger.warning(f'pilot with id {name} already in pilot db, overwriting...') if pilot_prefs is None: pilot_prefs = {} self.control_panel.add_pilot(name) new_pilot = {name:{'subjects':[], 'ip':ip, 'prefs':pilot_prefs}} self.control_panel.update_db(new=new_pilot) self.logger.info(f'added new pilot {name}')
[docs] def new_protocol(self): """ Open a :class:`.gui.Protocol_Wizard` to create a new protocol. Prompts for name of protocol, then saves in `prefs.get('PROTOCOLDIR')` """ self.new_protocol_window = Protocol_Wizard() self.new_protocol_window.exec_() if self.new_protocol_window.result() == 1: steps = self.new_protocol_window.steps # The values useful to the step functions are stored with a 'value' key in the param_dict save_steps = [] for s in steps: param_values = {} for k, v in s.items(): if 'value' in v.keys(): param_values[k] = v['value'] elif k == 'stim': # TODO: Super hacky - don't do this. Refactor params already. param_values[k] = {} for stimtype, stim in v.items(): param_values[k][stimtype] = stim save_steps.append(param_values) # Name the protocol name, ok = QtWidgets.QInputDialog.getText(self, "Name Protocol", "Protocol Name:") if ok and name != '': protocol_file = os.path.join(prefs.get('PROTOCOLDIR'), name + '.json') with open(protocol_file, 'w') as pfile_open: json.dump(save_steps, pfile_open, indent=4, separators=(',', ': '), sort_keys=True) elif name == '' or not ok: placeholder_name = 'protocol_created_{}'.format(datetime.date.today().isoformat()) protocol_file = os.path.join(prefs.get('PROTOCOLDIR'), placeholder_name + '.json') with open(protocol_file, 'w') as pfile_open: json.dump(save_steps, pfile_open, indent=4, separators=(',', ': '), sort_keys=True)
[docs] def subject_weights(self): """ Gets recent weights from all :attr:`~.Terminal.subjects` and open a :class:`.gui.Weights` window to view or set weights. """ subjects = self.subject_list # open objects if not already for subject in subjects: if subject not in self.subjects.keys(): self.subjects[subject] = Subject(subject) # for each subject, get weight weights = [] for subject in subjects: weight = self.subjects[subject].get_weight(include_baseline=True) weight['subject'] = subject weights.append(weight) self.weight_widget = Weights(weights, self.subjects) self.weight_widget.show()
[docs] def update_protocols(self): """ If we change the protocol file, update the stored version in subject files """ # # get list of protocol files protocols = os.listdir(prefs.get('PROTOCOLDIR')) protocols = [p for p in protocols if p.endswith('.json')] updated_subjects = [] subjects = self.subject_list for subject in subjects: if subject not in self.subjects.keys(): self.subjects[subject] = Subject(subject) protocol_bool = [self.subjects[subject].protocol_name == os.path.splitext(p)[0] for p in protocols] if any(protocol_bool): which_prot = np.where(protocol_bool)[0][0] protocol = protocols[which_prot] self.subjects[subject].assign_protocol(os.path.join(prefs.get('PROTOCOLDIR'), protocol), step_n=self.subjects[subject].step) updated_subjects.append(subject) msgbox = QtWidgets.QMessageBox() msgbox.setText("Subject Protocols Updated for:") msgbox.setDetailedText("\n".join(sorted(updated_subjects))) msgbox.exec_()
[docs] def reassign_protocols(self): """ Batch reassign protocols and steps. Opens a :class:`.gui.Reassign` window after getting protocol data, and applies any changes made in the window. """ reassign_window = Reassign(self.subject_protocols, self.protocols) reassign_window.exec_() if reassign_window.result() == 1: subject_protocols = reassign_window.subjects for subject, protocol in subject_protocols.items(): step = protocol[1] protocol = protocol[0] # since assign_protocol also changes the step, stash the step number here to tell if it's changed subject_orig_step = self.subjects[subject].step # if the protocol is the blank protocol, do nothing if not protocol: self.logger.info(f'Protocol for {subject} set to blank, not setting') continue if self.subjects[subject].protocol_name != protocol or subject_orig_step != step: self.logger.info('Setting {} protocol from {} to {}'.format(subject, self.subjects[subject].protocol_name, protocol)) protocol_file = os.path.join(prefs.get('PROTOCOLDIR'), protocol + '.json') self.subjects[subject].assign_protocol(protocol_file, step) if subject_orig_step != step: self.logger.info('Setting {} step from {} to {}'.format(subject, subject_orig_step, step)) # this now happens in .assign_protocol #step_name = self.subjects[subject].current[step]['step_name'] #update history also flushes current - aka it also actually changes the step number #self.subjects[subject].update_history('step', step_name, step) else: self.logger.debug('reassign cancelled')
[docs] def calibrate_ports(self): """ Calibrate :class:`.hardware.gpio.Solenoid` objects. See :class:`.gui.Calibrate_Water`. After calibration routine, send results to pilot for storage. """ calibrate_window = Calibrate_Water(self.pilots) calibrate_window.exec_() if calibrate_window.result() == 1: for pilot, p_widget in calibrate_window.pilot_widgets.items(): p_results = p_widget.volumes # p_results are [port][dur] = {params} so running the same duration will # overwrite a previous run. unnest here so pi can keep a record unnested_results = {} for port, result in p_results.items(): unnested_results[port] = [] # result is [dur] = {params} for dur, inner_result in result.items(): inner_result['dur'] = dur unnested_results[port].append(inner_result) # send to pi self.node.send(to=pilot, key="CALIBRATE_RESULT", value = unnested_results) msgbox = QtWidgets.QMessageBox() msgbox.setText("Calibration results sent!") msgbox.exec_()
[docs] def test_bandwidth(self): """ Test bandwidth of Pilot connection with variable sized arrays as paylods See :class:`.gui.Bandwidth_Test` """ # turn off logging while we run prev_networking_loglevel = self.networking.logger.level prev_node_loglevel = self.node.logger.level self.networking.logger.setLevel(logging.ERROR) self.node.logger.setLevel(logging.ERROR) bandwidth_test = Bandwidth_Test(self.pilots) bandwidth_test.exec_() self.networking.logger.setLevel(prev_networking_loglevel) self.node.logger.setLevel(prev_node_loglevel)
[docs] def plot_psychometric(self): """ Select subject, step, and variables to plot a psychometric curve """ if not IMPORTED_VIZ: _ = pop_dialog("Vizualisation function couldn't be imported!", "error", VIZ_ERROR) return psychometric_dialog = Psychometric(self.subject_protocols) psychometric_dialog.exec_() # if user cancels, return if psychometric_dialog.result() != 1: return chart = viz.plot_psychometric(psychometric_dialog.plot_params) text, ok = QtGui.QInputDialog.getText(self, 'save plot?', 'what to call this thing') if ok: chart.save(text)
#chart.serve() #viz.plot_psychometric(self.subjects_protocols) #result = psychometric_dialog.exec_()
[docs] def manage_plugins(self): plugs = Plugins() plugs.exec_()
[docs] def stream_video(self): """ Open a window to stream videos from a connected pilot. Choose from connected pilots and configured :class:`~.hardware.cameras.Camera` objects (``prefs.json`` sent by Pilots in :meth:`.Pilot.handshake` ). Stream video, save to file. .. todo:: Configure camera parameters!!! """ video_dialog = Stream_Video(self.pilots)
[docs] def closeEvent(self, event): """ When Closing the Terminal Window, close any running subject objects, 'KILL' our networking object. Since the `:class:`.Net_Node` keeping us alive is a `daemon`, no need to explicitly kill it. """ # Save the window geometry, to be optionally restored next time self.settings.setValue("geometry", self.saveGeometry()) # TODO: Check if any subjects are currently running, pop dialog asking if we want to stop # Close all subjects files for m in self.subjects.values(): if m.running is True: m.stop_run() # Stop networking # send message to kill networking process self.node.send(key="KILL") event.accept()
# Create the QApplication and run it # Prefs were already loaded at the very top if __name__ == "__main__": app = QtWidgets.QApplication(sys.argv) #app.setGraphicsSystem("opengl") app.setStyle('GTK+') # Keeps some GTK errors at bay ex = Terminal() sys.exit(app.exec_())