import os import re import sys import glob import time import logging import asyncio import itertools from functools import partial, wraps from PyQt5 import QtWidgets, QtMultimedia, QtCore from PyQt5.QtGui import QPixmap, QIcon from PyQt5.QtCore import Qt, pyqtSlot, QUrl from PyQt5.QtWidgets import QFileDialog, QMessageBox, QApplication, QInputDialog, QLineEdit, QStatusBar, \ QSplashScreen, QProgressBar from quamash import QEventLoop # Importing gui form from server_gui import Ui_MainWindow from server import Server, Client, now import messaging_lib as messaging import config as cfg import copter_table_models as table from copter_table import CopterTableWidget from visual_land_dialog import VisualLandDialog from config_editor_models import ConfigDialog def multi_glob(*patterns): return itertools.chain.from_iterable(glob.iglob(pattern) for pattern in patterns) def b_partial(func, *args, **kwargs): # call argument blocker partial return lambda *a: func(*args, **kwargs) def confirmation_required(text="Are you sure?", label="Confirm operation?"): def inner(f): @wraps(f) def wrapper(*args, **kwargs): reply = QMessageBox.question( args[0], label, text, QMessageBox.Yes | QMessageBox.No, QMessageBox.No ) if reply == QMessageBox.Yes: logging.debug("Dialog accepted") return f(*args, **kwargs) logging.debug("Dialog declined") return wrapper return inner class ExitMsgbox(logging.Handler): def emit(self, record): loop.call_soon_threadsafe(self._emit, record) # TODO replace loop call def _emit(self, record): # window.close() QMessageBox.warning(None, "Critical error in {}: {}". format(record.name, record.threadName), record.msg) QApplication.quit() sys.exit(record.msg) class ServerQt(Server): def load_config(self): super().load_config() table.ModelChecks.battery_min = self.config.checks_battery_min table.ModelChecks.start_pos_delta_max = self.config.checks_start_pos_delta_max table.ModelChecks.time_delta_max = self.config.checks_time_delta_max # noinspection PyCallByClass,PyArgumentList class MainWindow(QtWidgets.QMainWindow): def __init__(self, server): super(MainWindow, self).__init__() self.server = server self.model = table.CopterDataModel() self.ui = Ui_MainWindow() self.ui.setupUi(self) self.init_ui() self.init_model() # self.statusBar = QStatusBar() # self.setStatusBar(self.statusBar) # self.statusBar.showMessage("Hey", 2000) self.register_callbacks() self.player = QtMultimedia.QMediaPlayer() def init_ui(self): # Connecting self.ui.check_button.clicked.connect(self.selfcheck_selected) self.ui.start_button.clicked.connect(self.send_start_time_selected) self.ui.pause_button.clicked.connect(self.pause_resume_selected) self.ui.land_all_button.clicked.connect(b_partial(Client.broadcast_message, "land")) self.ui.land_selected_button.clicked.connect(b_partial(self.send_to_selected, "land")) self.ui.disarm_all_button.clicked.connect(b_partial(Client.broadcast_message, "disarm")) self.ui.disarm_selected_button.clicked.connect(b_partial(self.send_to_selected, "disarm")) self.ui.visual_land_button.clicked.connect(self.visual_land) self.ui.emergency_land_button.clicked.connect(b_partial(self.send_to_selected, "emergency_land")) self.ui.leds_button.clicked.connect(b_partial(self.send_to_selected, "led_test")) self.ui.takeoff_button.clicked.connect(self.takeoff_selected) self.ui.flip_button.clicked.connect(self.flip_selected) self.ui.reboot_fcu.clicked.connect(b_partial(self.send_to_selected, "reboot_fcu")) self.ui.calibrate_gyro.clicked.connect(self.calibrate_gyro_selected) self.ui.calibrate_level.clicked.connect(self.calibrate_level_selected) self.ui.action_remove_row.triggered.connect(self.remove_selected) self.ui.action_send_animations.triggered.connect(self.send_animations) self.ui.action_send_calibrations.triggered.connect(self.send_calibrations) self.ui.action_send_configurations.triggered.connect(self.send_config) self.ui.action_send_Aruco_map.triggered.connect(self.send_aruco) self.ui.action_send_launch_file.triggered.connect(self.send_launch) self.ui.action_send_fcu_parameters.triggered.connect(self.send_fcu_parameters) self.ui.action_send_any_file.triggered.connect(self.send_any_file) self.ui.action_send_any_command.triggered.connect(self.send_any_command) self.ui.action_restart_clever.triggered.connect( b_partial(self.send_to_selected, "service_restart", {"name": "clever"})) self.ui.action_restart_clever_show.triggered.connect( b_partial(self.send_to_selected, "service_restart", {"name": "clever-show"})) self.ui.action_update_client_repo.triggered.connect(b_partial(self.send_to_selected, "update_repo")) self.ui.action_reboot_all.triggered.connect(b_partial(self.send_to_selected, "reboot_all")) self.ui.action_set_start_to_current_position.triggered.connect(b_partial(self.send_to_selected, "move_start")) self.ui.action_reset_start.triggered.connect(b_partial(self.send_to_selected, "reset_start")) self.ui.action_set_z_offset_to_ground.triggered.connect(b_partial(self.send_to_selected, "set_z_to_ground")) self.ui.action_reset_z_offset.triggered.connect(b_partial(self.send_to_selected, "reset_z_offset")) self.ui.action_restart_chrony.triggered.connect(b_partial(self.send_to_selected, "repair_chrony")) self.ui.action_select_music_file.triggered.connect(self.select_music_file) self.ui.action_play_music.triggered.connect(self.play_music) self.ui.action_stop_music.triggered.connect(self.stop_music) self.ui.action_select_all_rows.triggered.connect(self.model.select_all) self.init_table() # Set most safety-important buttons disabled self.ui.start_button.setEnabled(False) self.ui.takeoff_button.setEnabled(False) self.ui.flip_button.setEnabled(False) def init_table(self): # Remove standard table widget self.ui.horizontalLayout.removeWidget(self.ui.tableView) self.ui.tableView.close() # Init our custom widget self.ui.copter_table = CopterTableWidget(self.model) self.ui.copter_table.setObjectName("copter_table") # Insert to layout at right self.ui.horizontalLayout.insertWidget(0, self.ui.copter_table, 0) def init_model(self): # Connect model signals to UI self.model.selected_ready_signal.connect(self.ui.start_button.setEnabled) self.model.selected_takeoff_ready_signal.connect(self.ui.takeoff_button.setEnabled) self.model.selected_flip_ready_signal.connect(self.ui.flip_button.setEnabled) # Connect calibrating signal (testing) self.model.selected_calibrating_signal.connect(self.ui.check_button.setDisabled) self.model.selected_calibrating_signal.connect(self.ui.pause_button.setDisabled) self.model.selected_calibrating_signal.connect(self.ui.land_all_button.setDisabled) self.model.selected_calibrating_signal.connect(self.ui.land_selected_button.setDisabled) self.model.selected_calibrating_signal.connect(self.ui.disarm_selected_button.setDisabled) self.model.selected_calibrating_signal.connect(self.ui.disarm_all_button.setDisabled) self.model.selected_calibrating_signal.connect(self.ui.visual_land_button.setDisabled) self.model.selected_calibrating_signal.connect(self.ui.emergency_land_button.setDisabled) self.model.selected_calibrating_signal.connect(self.ui.leds_button.setDisabled) self.model.selected_calibrating_signal.connect(self.ui.reboot_fcu.setDisabled) self.model.selected_calibration_ready_signal.connect(self.ui.calibrate_gyro.setEnabled) self.model.selected_calibration_ready_signal.connect(self.ui.calibrate_level.setEnabled) def iterate_selected(self, f, *args, **kwargs): for copter in self.model.user_selected(): yield f(copter, *args, **kwargs) @pyqtSlot() def send_to_selected(self, command, command_args=None): return list(self.iterate_selected(lambda copter: copter.client.send_message(command, command_args))) def new_client_connected(self, client: Client): logging.debug("Added client {}".format(client)) self.ui.copter_table.add_client(copter_id=client.copter_id, client=client) def client_connection_changed(self, client: Client): logging.debug("Connection {} changed {}".format(client, client.connected)) row_data = self.model.get_row_by_attr("client", client) if row_data is None: logging.error("No row for client presented") return if self.server.config.table_remove_disconnected and (not client.connected): client.remove() self.ui.copter_table.remove_client_data(row_data) logging.debug("Removing from table") else: row_num = self.model.get_row_index(row_data) if row_num is not None: self.ui.copter_table.update_data(row_num, 0, client.connected, table.ModelStateRole) logging.debug("Client status updated") @pyqtSlot() def selfcheck_selected(self): for copter in self.model.user_selected(): copter.client.get_response("telemetry", self.update_table_data) @pyqtSlot(object, dict) def update_table_data(self, client, telems: dict): cols_dict = { "git_version": 1, "animation_id": 2, "battery": 3, "fcu_status": 4, "cal_status": 5, "mode": 6, "selfcheck": 7, "current_position": 8, "start_position": 9, "task": 10, "time": 11, } for key, value in telems.items(): col = cols_dict.get(key, None) if col is None: logging.error("No column {} present!".format(key)) continue row_data = self.model.get_row_by_attr("client", client) row_num = self.model.get_row_index(row_data) if row_num is not None: self.ui.copter_table.update_data(row_num, col, value, Qt.EditRole) @pyqtSlot() def remove_selected(self): for copter in self.model.user_selected(): copter.client.remove() if not self.server.config.table_remove_disconnected: self.ui.copter_table.remove_client_data(copter) logging.info("Client removed from table!") @pyqtSlot() @confirmation_required("This operation will takeoff selected copters with delay and start animation. Proceed?") def send_start_time_selected(self): time_now = server.time_now() dt = self.ui.start_delay_spin.value() logging.info('Wait {} seconds to start animation'.format(dt)) if self.ui.music_checkbox.isChecked(): music_dt = self.ui.music_delay_spin.value() asyncio.ensure_future(self.play_music_at_time(music_dt + time_now), loop=loop) logging.info('Wait {} seconds to play music'.format(music_dt)) # self.selfcheck_selected() for copter in filter(self.model.checks.all_checks, self.model.user_selected()): server.send_starttime(copter.client, dt + time_now) @pyqtSlot() def pause_resume_selected(self): if self.ui.pause_button.text() == 'Pause': self.send_to_selected("pause") self.ui.pause_button.setText('Resume') else: time_gap = 0.1 # TODO config? automatic delay detection? self.send_to_selected("resume", {"time": server.time_now() + time_gap}) self.ui.pause_button.setText('Pause') @pyqtSlot() @confirmation_required("This operation will takeoff copters immediately. Proceed?") def takeoff_selected(self): for copter in self.model.user_selected(): if self.model.checks.takeoff_checks(copter): if self.ui.z_checkbox.isChecked(): copter.client.send_message("takeoff_z", {"z": str(self.ui.z_spin.value())}) # todo int else: copter.client.send_message("takeoff") @pyqtSlot() @confirmation_required("This operation will flip(!!!) copters immediately. Proceed?") def flip_selected(self): for copter in self.model.user_selected(): if table.flip_checks(copter): copter.client.send_message("flip") @pyqtSlot() def calibrate_gyro_selected(self): # TODO merge commands for copter_data_row in self.model.user_selected(): client = copter_data_row.client # Update calibration status row = self.model.get_row_index(copter_data_row) col = 5 data = 'CALIBRATING' self.ui.copter_table.update_data(row, col, data, table.ModelDataRole) # Send request client.get_response("calibrate_gyro", self._get_calibration_info) @pyqtSlot() def calibrate_level_selected(self): for copter_data_row in self.model.user_selected(): client = copter_data_row.client # Update calibration status row = self.model.get_row_index(copter_data_row) col = 5 data = 'CALIBRATING' self.ui.copter_table.update_data(row, col, data, table.ModelDataRole) # Send request client.get_response("calibrate_level", self._get_calibration_info) def _get_calibration_info(self, client, value): col = 5 row_data = self.model.get_row_by_attr("client", client) row = self.model.get_row_index(row_data) if row is not None: data = str(value) self.ui.copter_table.update_data(row, col, data, table.ModelDataRole) def _send_files(self, files, copters=None, client_path="", client_filename="", match_id=False, callback=None): if copters is None: copters = self.model.user_selected() copters = list(copters) for num, file in enumerate(files): filepath, filename = os.path.split(file) logging.info("Preparing file for sending: {} {}".format(filepath, filename)) if match_id: name = os.path.splitext(filename)[0] to_send = [copter for copter in copters if re.fullmatch(name, copter.copter_id)] else: to_send = copters if not to_send: logging.error(f"No copters to send file {filename} to") continue logging.info(f"Sending file {filename} to clients: {to_send}") filename = client_filename.format(num, filename) or filename for copter in to_send: copter.client.send_file(file, os.path.join(client_path, filename)) if callback is not None: callback(copter) def send_files(self, prompt, ext_filter, copters=None, client_path="", client_filename="", match_id=False, onefile=False, callback=None): if onefile: file = QFileDialog.getOpenFileName(self, prompt, filter=ext_filter)[0] files = [file] if file else [] else: files = QFileDialog.getOpenFileNames(self, prompt, filter=ext_filter)[0] if not files: return self._send_files(files, copters, client_path, client_filename, match_id, callback) def send_directory_files(self, prompt, extensions=(), copters=None, client_path="", client_filename="", match_id=False, callback=None): path = QFileDialog.getExistingDirectory(self, prompt) if not path: return if extensions: patterns = [path + '/*' + ext for ext in extensions] else: patterns = [path+'/*.*'] files = multi_glob(*patterns) self._send_files(files, copters, client_path, client_filename, match_id, callback) @pyqtSlot() def send_any_file(self): file = QFileDialog.getOpenFileName(self, "Select any file")[0] if not file: return c_path, ok = QInputDialog.getText(self, "Enter path (and name) to send on client", "Destination:", QLineEdit.Normal, "/home/pi/") # TODO config? if not ok: return c_filepath, c_filename = os.path.split(c_path) files = [file] self._send_files(files, client_path=c_filepath, client_filename=c_filename) @pyqtSlot() def send_animations(self): self.send_directory_files("Select directory with animations", ('.csv', '.txt'), match_id=True, client_path="", client_filename="animation.csv") @pyqtSlot() def send_calibrations(self): self.send_directory_files("Select directory with calibrations", ('.yaml', ), match_id=True, client_path="/home/pi/catkin_ws/src/clever/clever/camera_info/", client_filename="calibration.yaml") # TODO callback to reload clever? # from os.path import expanduser # TODO on client # home = expanduser("~") -> "~catkin_ws/src/clever/clever/camera_info/" @pyqtSlot() def send_aruco(self): def callback(copter): copter.client.send_message("service_restart", {"name": "clever"}) self.send_files("Select aruco map configuration file", "Aruco map files (*.txt)", onefile=True, client_path="/home/pi/catkin_ws/src/clever/aruco_pose/map/", client_filename="animation_map.txt", callback=callback) @pyqtSlot() def send_launch(self): self.send_directory_files("Select directory with launch files", ('.launch', ), match_id=False, client_path='/home/pi/catkin_ws/src/clever/clever/launch/') # TODO clever restart callback? @pyqtSlot() def send_fcu_parameters(self): def request_callback(client, value): logging.info("Send parameters to {} success: {}".format(client.copter_id, value)) def callback(copter): copter.client.get_response("load_params", request_callback) self.send_files("Select px4 param file", "px4 params (*.params)", onefile=True, client_filename="temp.params", callback=callback) @pyqtSlot() def send_config(self): mode, ok = QInputDialog.getItem(self, "Select config sending mode", "Mode:", ("Modify", "Rewrite"), 0, False) if not ok or not mode: return path = QFileDialog.getOpenFileName(self, "Select configuration file", filter="Configs (*.ini *.txt *.cfg)")[0] if not path: return config = cfg.ConfigManager() config.load_only_config(path) data = config.full_dict logging.info(f"Loaded config from {path}") copters = self.model.user_selected() for copter in copters: copter.client.send_message("config", {"config": data, "mode": mode.lower()}) @pyqtSlot() def send_any_command(self): text, ok = QInputDialog.getText(self, "Enter command to send on copter", "Command:", QLineEdit.Normal, "") if ok and text: self.send_to_selected("execute", {"command": text}) @pyqtSlot() def select_music_file(self): path = QFileDialog.getOpenFileName(self, "Select music file", filter="Music files (*.mp3 *.wav)")[0] if not path: return media = QUrl.fromLocalFile(path) content = QtMultimedia.QMediaContent(media) self.player.setMedia(content) self.ui.action_select_music_file.setText(self.ui.action_select_music_file.text() + " (selected)") @pyqtSlot() def play_music(self): if self.player.mediaStatus() == QtMultimedia.QMediaPlayer.InvalidMedia: logging.info("Can't play media") return if self.player.mediaStatus() == QtMultimedia.QMediaPlayer.NoMedia: logging.info("No media file") return if self.player.state() == QtMultimedia.QMediaPlayer.StoppedState or \ self.player.state() == QtMultimedia.QMediaPlayer.PausedState: self.ui.action_play_music.setText("Pause music") self.player.play() else: self.ui.action_play_music.setText("Play music") self.player.pause() @pyqtSlot() def stop_music(self): if self.player.mediaStatus() == QtMultimedia.QMediaPlayer.InvalidMedia: logging.error("Can't stop media") return if self.player.mediaStatus() == QtMultimedia.QMediaPlayer.NoMedia: logging.error("No media file") return self.player.stop() @asyncio.coroutine def play_music_at_time(self, t): if self.player.mediaStatus() == QtMultimedia.QMediaPlayer.InvalidMedia: logging.error("Can't play media") return if self.player.mediaStatus() == QtMultimedia.QMediaPlayer.NoMedia: logging.error("No media file") return self.player.stop() yield from asyncio.sleep(t - time.time()) logging.info("Playing music") self.player.play() @pyqtSlot() def visual_land(self): dialog = VisualLandDialog(self.model) dialog.start() def register_callbacks(self): @messaging.message_callback("telemetry") def get_telem_data(client, value, **kwargs): self.update_table_data(client, value) def except_hook(cls, exception, traceback): sys.__excepthook__(cls, exception, traceback) def set_taskbar_icon(): import ctypes myappid = 'COEX.droneshow.droneserver' ctypes.windll.shell32.SetCurrentProcessExplicitAppUserModelID(myappid) if __name__ == "__main__": msgbox_handler = ExitMsgbox() msgbox_handler.setLevel(logging.CRITICAL) logging.basicConfig( level=logging.DEBUG, format="%(asctime)s [%(name)-7.7s] [%(threadName)-19.19s] [%(levelname)-7.7s] %(message)s", handlers=[ logging.FileHandler("server_logs/{}.log".format(now)), logging.StreamHandler(), msgbox_handler ]) sys.excepthook = except_hook # for debugging (exceptions traceback) app = QApplication(sys.argv) splash_pix = QPixmap('icons/coex_splash.jpg') splash = QSplashScreen(splash_pix) splash.setEnabled(False) splash.setWindowFlags(Qt.WindowStaysOnTopHint | Qt.FramelessWindowHint) progressBar = QProgressBar(splash) progressBar.setGeometry(25, splash_pix.height() - 80, splash_pix.width(), 35) splash.showMessage("Loading clever-show server"+"\n\n\n\n\n", int(Qt.AlignBottom | Qt.AlignCenter), Qt.white) app.processEvents() splash.show() # time.sleep(3) app_icon = QIcon() app_icon.addFile('icons/image.ico', QtCore.QSize(256, 256)) app.setWindowIcon(app_icon) if sys.platform == 'win32': set_taskbar_icon() loop = QEventLoop(app) asyncio.set_event_loop(loop) # app.exec_() with loop: server = ServerQt() window = MainWindow(server) Client.on_first_connect = window.new_client_connected Client.on_connect = window.client_connection_changed Client.on_disconnect = window.client_connection_changed server.start() window.show() splash.close() loop.run_forever() server.stop() sys.exit()