KuttyPyVisual.py

'''
'''
# !/usr/bin/python3

import os, sys, time, re, traceback, platform
from typing import List

from PyQt5 import QtGui, QtCore, QtWidgets
from PyQt5.QtCore import QThread
from PyQt5.QtGui import QPixmap

sys.path.append(os.path.dirname(os.path.abspath(__file__)))

import KuttyPyLib
import socket

from utilities.templates import ui_layout_visual as layout
from utilities import REGISTERS
from functools import partial

import constants
import inspect

class AppWindow(QtWidgets.QMainWindow, layout.Ui_MainWindow):
    p = None
    logThis = QtCore.pyqtSignal(str)
    showStatusSignal = QtCore.pyqtSignal(str, bool)
    serverSignal = QtCore.pyqtSignal(str)
    addMPSignal = QtCore.pyqtSignal()
    delMPSignal = QtCore.pyqtSignal()
    queryMPSignal = QtCore.pyqtSignal()
    cameraReadySignal = QtCore.pyqtSignal()

    logThisPlain = QtCore.pyqtSignal(bytes)
    codeOutput = QtCore.pyqtSignal(str, str)
    serialGaugeSignal = QtCore.pyqtSignal(bytes)
    serialGaugeConvert = 'bytes'
    serialStream = b''

    def __init__(self, parent=None, **kwargs):
        super(AppWindow, self).__init__(parent)

        self.external = None
        self.last_query_time = time.time()
        self.local_ip = 'localhost'
        self.setupUi(self)

        self.compile_thread = None
        self.mp_thread = None

        self.defaultDirectory = path["examples"]
        self.serverActive = False
        self.VERSION = REGISTERS.VERSION_ATMEGA32  # This needs to be dynamically changed when hardware is connected

        self.logThisPlain.connect(self.appendLogPlain)  # Connect to the log window

        self.statusBar = self.statusBar()
        self.makeBottomMenu()

        global app

        self.initializeCommunications()


        self.startTime = time.time()
        self.timer = QtCore.QTimer()
        self.timer.timeout.connect(self.updateEverything)
        self.timer.start(20)

        # Auto-Detector
        self.shortlist = KuttyPyLib.getFreePorts()


    def addMP(self):
        from online.mp import HandLandmarkThread
        self.mpLabel.setVisible(True)
        self.mpLabel.resize(400, 300)

        self.last_query_time = time.time()
        if self.mp_thread is not None and self.mp_thread.isRunning():  #Available. ignore.
            self.mp_thread.ping()  # nudge
            pass
        else:
            self.mp_thread = HandLandmarkThread()
        self.mp_thread.setPriority(QThread.HighestPriority)
        self.mp_thread.change_pixmap_signal.connect(self.update_image)
        self.mp_thread.coordinates_signal.connect(self.compile_thread.updateCoords)
        self.mp_thread.dead_signal.connect(self.delMP)
        self.mp_thread.setCameraReadySignal(self.cameraReadySignal)
        self.mp_thread.start()

    def update_image(self, qt_image):
        """Update the QLabel with the new frame."""
        pixmap = QPixmap.fromImage(qt_image)
        self.mpLabel.setPixmap(pixmap)

    def closeEvent(self, event):
        """Ensure the thread is stopped when the dialog is closed."""
        event.ignore()
        print('closing...')

        if self.mp_thread is not None and self.mp_thread.isRunning:
            print('terminating camera...')
            self.mp_thread.terminate()
            self.mp_thread.wait()
            print('terminated.')
        event.accept()


    def delMP(self):
        print('closing MP window')
        if self.mp_thread is not None:
            self.mp_thread.running = False
        self.mpLabel.setVisible(False)

    def queryMP(self):
        if self.mp_thread is not None:
            self.mp_thread.ping()


    def makeBottomMenu(self):
        try:
            self.pushbutton.setParent(None)
        except:
            pass
        self.pushbutton = QtWidgets.QPushButton('Menu')
        self.pushbutton.setStyleSheet("color: #262;")
        menu = QtWidgets.QMenu()

        menu.addAction('PIP install mediapipe & CV', partial(self.showPipInstaller, 'mediapipe opencv-python-headless'))

        # Theme
        self.themeAction = QtWidgets.QWidgetAction(menu)
        themes = [a.split('.qss')[0] for a in os.listdir(path["themes"]) if '.qss' in a]
        self.themeBox = QtWidgets.QComboBox()
        self.themeBox.addItems(themes)
        self.themeBox.currentIndexChanged['QString'].connect(self.setTheme)
        self.themeAction.setDefaultWidget(self.themeBox)
        menu.addAction(self.themeAction)

        defaultTheme = "newtheme"
        self.themeBox.setCurrentIndex(themes.index(defaultTheme))
        self.setTheme(defaultTheme)

        self.pushbutton.setMenu(menu)

        # Compile thread
        self.compile_thread_button = QtWidgets.QPushButton('Visual Coding/Online Compiler')
        self.compile_thread_button.setStyleSheet("color: #262;")
        #self.compile_thread_button.clicked.connect(self.activateCompileServer)
        self.statusBar.addPermanentWidget(self.compile_thread_button)

        self.bottomLabel = QtWidgets.QLabel("Messages")
        self.activateCompileServer()

        # Menu button
        self.statusBar.addPermanentWidget(self.pushbutton)

    def activateCompileServer(self):
        if self.serverActive:  # Stop it
            '''
            if self.compile_thread is not None:
                self.compile_thread.terminate()
                self.compile_thread.wait()
                print('quit compile_thread')
                self.compile_thread_button.setParent(None)
                self.showStatus("Compiler Stopped ", False)
                return
            self.serverActive = False
            self.compile_thread_button.setText("Online Compiler")
            '''
            return
        else:
            s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            s.settimeout(0.1)  # Set a timeout to avoid blocking indefinitely
            try:
                s.connect(("8.8.8.8", 80))  # Connect to a public IP address
                self.local_ip = s.getsockname()[0]
            except:
                self.local_ip = 'localhost'

            from online.compile_server import create_server
            self.compile_thread = create_server(self.showStatusSignal, self.serverSignal, path, self.local_ip,
                                                self.p)
            self.compile_thread.setAddMPSignal(self.addMPSignal)
            self.compile_thread.setDelMPSignal(self.delMPSignal)
            self.compile_thread.setQueryMPSignal(self.queryMPSignal)
            self.compile_thread.setCameraReadySignal(self.cameraReadySignal)

            self.showStatusSignal.connect(self.showStatus)
            self.serverSignal.connect(self.showServerStatus)
            self.addMPSignal.connect(self.addMP)
            self.delMPSignal.connect(self.delMP)
            self.queryMPSignal.connect(self.queryMP)
            s.close()
            self.showStatus("Visual Coding and Compiler Active at " + self.local_ip + ":8888", False)
            self.compile_thread_button.setText(self.local_ip)
            self.serverActive = True

    def showPipInstaller(self, name):
        from utilities.pipinstaller import PipInstallDialog
        self.pipdialog = PipInstallDialog(name, self)
        self.pipdialog.show()


    def appendLog(self, txt):
        self.log.append(txt)

    def appendLogPlain(self, txt):
        self.log.moveCursor(QtGui.QTextCursor.End)
        self.log.insertPlainText(txt.decode('ascii'))

    def updateEverything(self):
        self.locateDevices()
        if not self.checkConnectionStatus(): return


    ##############################
    def setTheme(self, theme):
        self.setStyleSheet("")
        self.setStyleSheet(open(os.path.join(path["themes"], theme + ".qss"), "r").read())

    def initializeCommunications(self, port=False):
        if self.p:
            try:
                self.p.fd.close()
            except:
                pass
        if port:
            self.p = KuttyPyLib.connect(port=port)
        else:
            self.p = KuttyPyLib.connect(autoscan=True)

        if self.p.connected:
            self.VERSION = self.p.version
            #self.launchFirmwareButton.setChecked(False)
            self.setWindowTitle('KuttyPy Integrated Development Environment [{0:s}]'.format(self.p.portname))

        else:
            self.setWindowTitle('KuttyPy Integrated Development Environment [ Hardware not detected ]')


    def showServerStatus(self, msg):
        self.showStatus("Compiler: Error Launching Server (Restart app) ", True)
        QtWidgets.QMessageBox.warning(self, 'Server Error', msg)
        self.compile_thread_button.setText("Online Compiler")

    # self.statusBar.addPermanentWidget(self.compile_thread_button)

    def showStatus(self, msg, error=None):
        if error:
            self.statusBar.setStyleSheet("color:#633")
        else:
            self.statusBar.setStyleSheet("color:#333")
        self.statusBar.showMessage(msg)

    def locateDevices(self):
        try:
            L = KuttyPyLib.getFreePorts(self.p.portname)
        except Exception as e:
            print(e)
        total = len(L)
        menuChanged = False
        if L != self.shortlist:
            menuChanged = True
            if self.p.connected:
                if self.p.portname not in L:
                    self.setWindowTitle('Error : Device Disconnected')
                    QtWidgets.QMessageBox.warning(self, 'Connection Error',
                                                  'Device Disconnected. Please check the connections')
                    try:
                        self.p.close()
                    except:
                        pass
                    self.p.connected = False
                    self.setWindowTitle('KuttyPy Integrated Development Environment [ Hardware not detected ]')

            elif True in L.values():
                reply = QtWidgets.QMessageBox.question(self, 'Connection', 'Device Available. Connect?',
                                                       QtWidgets.QMessageBox.Yes, QtWidgets.QMessageBox.No)
                if reply == QtWidgets.QMessageBox.Yes:
                    self.initializeCommunications()

            # update the shortlist
            self.shortlist = L

    def checkConnectionStatus(self, dialog=False):
        if self.p.connected:
            return True
        else:
            if dialog: QtWidgets.QMessageBox.warning(self, 'Connection Error',
                                                     'Device not connected. Please connect a KuttyPy to the USB port')
            return False

def firstExistingPath(l):
    """
    Returns the first existing path taken from a list of
    possible paths.
    @param l a list of paths
    @return the first path which exists in the filesystem, or None
    """
    for p in l:
        if os.path.exists(p):
            return p
    return None


def common_paths():
    """
    Finds common paths
    @result a dictionary of common paths
    """
    path = {}
    curPath = os.path.dirname(os.path.realpath(__file__))
    path["current"] = curPath
    sharedPath = "/usr/share/kuttypy"
    path["translation"] = firstExistingPath(
        [os.path.join(p, "lang") for p in
         (curPath, sharedPath,)])
    path["utilities"] = firstExistingPath(
        [os.path.join(p, 'utilities') for p in
         (curPath, sharedPath,)])

    path["templates"] = firstExistingPath(
        [os.path.join(p, 'utilities', 'templates') for p in
         (curPath, sharedPath,)])

    path["themes"] = firstExistingPath(
        [os.path.join(p, 'utilities', 'themes') for p in
         (curPath, sharedPath,)])

    path["examples"] = firstExistingPath(
        [os.path.join(p, 'examples') for p in
         (curPath, sharedPath,)])

    path["kpy"] = firstExistingPath(
        [os.path.join(p, 'kpy') for p in
         (curPath, sharedPath,)])

    path["blockly"] = firstExistingPath(
        [os.path.join(p, 'online', 'static') for p in
         (curPath, sharedPath,)])

    path["editor"] = firstExistingPath(
        [os.path.join(p, 'editor') for p in
         (curPath, sharedPath,)])

    lang = str(QtCore.QLocale.system().name())
    shortLang = lang[:2]
    return path


def run():
    global path, app, myapp
    path = common_paths()
    print('QT Version', QtWidgets.__file__)
    app = QtWidgets.QApplication(sys.argv)
    myapp = AppWindow(app=app, path=path)
    myapp.show()
    r = app.exec_()
    if myapp.mp_thread is not None:
        myapp.mp_thread.stopRunning()

    if myapp.compile_thread is not None:
        myapp.compile_thread.stop_flask_app()
        myapp.compile_thread.terminate()
        if myapp.mp_thread is not None:
            myapp.mp_thread.running = False
            print('waiting to quit compile_thread')
        myapp.compile_thread.wait()

    app.deleteLater()
    sys.exit(r)


if __name__ == "__main__":
    run()