OctoTray/octotray

#!/usr/bin/env python3

# OctoTray Linux Qt System Tray OctoPrint client
#
# depends on:
# - python-pyqt5
# - curl
# - xdg-open
#
# see also:
# https://doc.qt.io/qt-5/qtwidgets-widgets-imageviewer-example.html
# https://stackoverflow.com/a/22618496

import json
import subprocess
import sys
import os
import threading
import time
from PyQt5 import QtWidgets, QtGui, QtCore, QtNetwork
from PyQt5.QtWidgets import QSystemTrayIcon, QAction, QMenu, QMessageBox, QWidget, QLabel, QVBoxLayout, QHBoxLayout, QDesktopWidget, QSizePolicy, QSlider, QLayout
from PyQt5.QtGui import QIcon, QPixmap, QImageReader
from PyQt5.QtCore import QCoreApplication, QSettings, QUrl, QTimer, QSize, Qt

class AspectRatioPixmapLabel(QLabel):
    def __init__(self, *args, **kwargs):
        super(AspectRatioPixmapLabel, self).__init__(*args, **kwargs)
        self.setMinimumSize(1, 1)
        self.setScaledContents(False)
        self.pix = QPixmap(0, 0)

    def setPixmap(self, p):
        self.pix = p
        super(AspectRatioPixmapLabel, self).setPixmap(self.scaledPixmap())

    def heightForWidth(self, width):
        if self.pix.isNull():
            return self.height()
        else:
            return (self.pix.height() * width) / self.pix.width()

    def sizeHint(self):
        w = self.width()
        return QSize(int(w), int(self.heightForWidth(w)))

    def scaledPixmap(self):
        return self.pix.scaled(self.size(), Qt.KeepAspectRatio, Qt.SmoothTransformation)

    def resizeEvent(self, e):
        if not self.pix.isNull():
            super(AspectRatioPixmapLabel, self).setPixmap(self.scaledPixmap())

class CamWindow(QWidget):
    reloadDelayDefault = 1000 # in ms
    addSize = 100
    reloadOn = True

    def __init__(self, parent, name, icon, app, manager, host, *args, **kwargs):
        super(CamWindow, self).__init__(*args, **kwargs)
        self.url = "http://" + host + ":8080/?action=snapshot"

        self.app = app
        self.parent = parent
        self.manager = manager
        self.manager.finished.connect(self.handleResponse)

        self.setWindowTitle(name + " Webcam Stream")
        self.setWindowIcon(icon)

        box = QVBoxLayout()
        self.setLayout(box)

        label = QLabel(self.url)
        box.addWidget(label, 0)
        box.setAlignment(label, Qt.AlignHCenter)

        self.img = AspectRatioPixmapLabel()
        self.img.setPixmap(QPixmap(640, 480))
        box.addWidget(self.img, 1)

        slide = QHBoxLayout()
        box.addLayout(slide, 0)

        self.slider = QSlider(Qt.Horizontal)
        self.slider.setMinimum(0)
        self.slider.setMaximum(2000)
        self.slider.setTickInterval(100)
        self.slider.setPageStep(100)
        self.slider.setSingleStep(100)
        self.slider.setTickPosition(QSlider.TicksBelow)
        self.slider.setValue(self.reloadDelayDefault)
        self.slider.valueChanged.connect(self.sliderChanged)
        slide.addWidget(self.slider, 1)

        self.slideLabel = QLabel(str(self.reloadDelayDefault) + "ms")
        slide.addWidget(self.slideLabel, 0)

        size = self.size()
        size.setHeight(size.height() + self.addSize)
        self.resize(size)

        self.loadImage()

    def sliderChanged(self):
        self.slideLabel.setText(str(self.slider.value()) + "ms")

    def closeEvent(self, event):
        self.reloadOn = False
        self.url = ""
        self.parent.removeWebcamWindow(self)

    def scheduleLoad(self):
        if self.reloadOn:
            QTimer.singleShot(self.slider.value(), self.loadImage)

    def loadImage(self):
        url = QUrl(self.url)
        request = QtNetwork.QNetworkRequest(url)
        self.manager.get(request)

    def handleResponse(self, reply):
        if reply.url().url() == self.url:
            if reply.error() == QtNetwork.QNetworkReply.NoError:
                reader = QImageReader(reply)
                reader.setAutoTransform(True)
                image = reader.read()
                if image != None:
                    if image.colorSpace().isValid():
                        image.convertToColorSpace(QColorSpace.SRgb)
                    self.img.setPixmap(QPixmap.fromImage(image))
                    self.scheduleLoad()
                else:
                    print("Error decoding image: " + reader.errorString())
            else:
                print("Error loading image: " + reply.errorString())

class OctoTray():
    name = "OctoTray"
    vendor = "xythobuz"
    version = "0.2"

    iconPath = "/usr/share/pixmaps/"
    iconName = "octotray_icon.png"

    # 0=host, 1=key
    # (2=method, 3=menu, 4...=actions)
    printers = [
        [ "PRINTER_HOST_HERE", "PRINTER_API_KEY_HERE" ]
    ]
    
    statesWithWarning = [
        "Printing", "Pausing", "Paused"
    ]

    camWindows = []

    def __init__(self):
        self.app = QtWidgets.QApplication(sys.argv)
        QCoreApplication.setApplicationName(self.name)

        if not QSystemTrayIcon.isSystemTrayAvailable():
            print("System Tray is not available on this platform!")
            sys.exit(0)

        self.manager = QtNetwork.QNetworkAccessManager()

        self.menu = QMenu()

        for p in self.printers:
            method = self.getMethod(p[0], p[1])
            print("Printer " + p[0] + " has method " + method)
            p.append(method)
            if method == "unknown":
                continue

            menu = QMenu(self.getName(p[0], p[1]))
            p.append(menu)
            self.menu.addMenu(menu)

            action = QAction("Turn on")
            action.triggered.connect(lambda chk, x=p: self.printerOnAction(x))
            p.append(action)
            menu.addAction(action)

            action = QAction("Turn off")
            action.triggered.connect(lambda chk, x=p: self.printerOffAction(x))
            p.append(action)
            menu.addAction(action)

            action = QAction("Get Status")
            action.triggered.connect(lambda chk, x=p: self.printerStatusAction(x))
            p.append(action)
            menu.addAction(action)

            action = QAction("Show Webcam")
            action.triggered.connect(lambda chk, x=p: self.printerWebcamAction(x))
            p.append(action)
            menu.addAction(action)

            action = QAction("Open Web UI")
            action.triggered.connect(lambda chk, x=p: self.printerWebAction(x))
            p.append(action)
            menu.addAction(action)

        self.quitAction = QAction("&Quit")
        self.quitAction.triggered.connect(self.exit)
        self.menu.addAction(self.quitAction)

        iconPathName = ""
        if os.path.isfile(self.iconName):
            iconPathName = self.iconName
        elif os.path.isfile(self.iconPath + self.iconName):
            iconPathName = self.iconPath + self.iconName
        else:
            print("no icon found")

        self.icon = QIcon()
        if iconPathName != "":
            pic = QPixmap(32, 32)
            pic.load(iconPathName)
            self.icon = QIcon(pic)

        trayIcon = QSystemTrayIcon(self.icon)
        trayIcon.setToolTip(self.name + " " + self.version)
        trayIcon.setContextMenu(self.menu)
        trayIcon.setVisible(True)

        sys.exit(self.app.exec_())

    def openBrowser(self, url):
        os.system("xdg-open http://" + url)

    def showDialog(self, title, text1, text2, question, warning):
        msg = QMessageBox()

        if warning:
            msg.setIcon(QMessageBox.Warning)
        else:
            msg.setIcon(QMessageBox.Information)

        msg.setWindowTitle(title)
        msg.setText(text1)

        if text2 is not None:
            msg.setInformativeText(text2)

        if question:
            msg.setStandardButtons(QMessageBox.Yes | QMessageBox.No)
        else:
            msg.setStandardButtons(QMessageBox.Ok)

        retval = msg.exec_()
        if retval == QMessageBox.Yes:
            return True
        return False

    def sendRequest(self, host, headers, path, content = None):
        cmdline = 'curl -s'
        for h in headers:
            cmdline += " -H \"" + h + "\""
        if content == None:
            cmdline += " -X GET"
        else:
            cmdline += " -X POST"
            cmdline += " -d '" + content + "'"
        cmdline += " http://" + host + "/api/" + path
        r = subprocess.run(cmdline, shell=True, capture_output=True, timeout=10, text=True)
        return r.stdout

    def sendPostRequest(self, host, key, path, content):
        headers = [ "Content-Type: application/json",
                   "X-Api-Key: " + key ]
        return self.sendRequest(host, headers, path, content)

    def sendGetRequest(self, host, key, path):
        headers = [ "X-Api-Key: " + key ]
        return self.sendRequest(host, headers, path)

    def getState(self, host, key):
        r = self.sendGetRequest(host, key, "job")
        try:
            rd = json.loads(r)
            if "state" in rd:
                return rd["state"]
        except json.JSONDecodeError:
            pass
        return "Unknown"

    def getProgress(self, host, key):
        r = self.sendGetRequest(host, key, "job")
        try:
            rd = json.loads(r)
            if "progress" in rd:
                return rd["progress"]
        except json.JSONDecodeError:
            pass
        return "Unknown"

    def getName(self, host, key):
        r = self.sendGetRequest(host, key, "printerprofiles")
        try:
            rd = json.loads(r)
            if "profiles" in rd:
                p = next(iter(rd["profiles"]))
                if "name" in rd["profiles"][p]:
                    return rd["profiles"][p]["name"]
        except json.JSONDecodeError:
            pass
        return host

    def getMethod(self, host, key):
        r = self.sendGetRequest(host, key, "plugin/psucontrol")
        try:
            rd = json.loads(r)
            if "isPSUOn" in rd:
                return "psucontrol"
        except json.JSONDecodeError:
            pass

        r = self.sendGetRequest(host, key, "system/commands/custom")
        try:
            rd = json.loads(r)
            for c in rd:
                if "action" in c:
                    if (c["action"] == "all off") or (c["action"] == "all on"):
                        return "system"
        except json.JSONDecodeError:
            pass

        return "unknown"

    def setPSUControl(self, host, key, state):
        cmd = "turnPSUOff"
        if state:
            cmd = "turnPSUOn"
        return self.sendPostRequest(host, key, "plugin/psucontrol", '{ "command":"' + cmd + '" }')

    def setSystemCommand(self, host, key, state):
        cmd = "all%20off"
        if state:
            cmd = "all%20on"
        return self.sendPostRequest(host, key, "system/commands/custom/" + cmd, '')

    def setPrinter(self, host, key, method, state):
        if method == "psucontrol":
            return self.setPSUControl(host, key, state)
        elif method == "system":
            return self.setSystemCommand(host, key, state)
        else:
            return "error"

    def exit(self):
        QCoreApplication.quit()

    def printerOnAction(self, item):
        self.setPrinter(item[0], item[1], item[2], True)

    def printerOffAction(self, item):
        state = self.getState(item[0], item[1])
        if state in self.statesWithWarning:
            if self.showDialog("OctoTray Warning", "The printer seems to be running currently!", "Do you really want to turn it off?", True, True) == True:
                self.setPrinter(item[0], item[1], item[2], False)
        else:
            self.setPrinter(item[0], item[1], item[2], False)

    def printerWebAction(self, item):
        self.openBrowser(item[0])

    def printerStatusAction(self, item):
        progress = self.getProgress(item[0], item[1])
        s = ""
        warning = False
        if ("completion" in progress) and ("printTime" in progress) and ("printTimeLeft" in progress) and (progress["completion"] != None) and (progress["printTime"] != None) and (progress["printTimeLeft"] != None):
            s = "%.1f%% Completion\n" % progress["completion"]
            s += "Printing since " + time.strftime("%H:%M:%S", time.gmtime(progress["printTime"])) + "\n"
            s += time.strftime("%H:%M:%S", time.gmtime(progress["printTimeLeft"])) + " left\n"
        elif ("completion" in progress) and ("printTime" in progress) and ("printTimeLeft" in progress):
            s = "No job is currently running"
        else:
            s = "Could not read printer status!"
            warning = True
        self.showDialog("OctoTray Status", s, None, False, warning)

    def printerWebcamAction(self, item):
        window = CamWindow(self, self.name, self.icon, self.app, self.manager, item[0])
        self.camWindows.append(window)

        screenGeometry = QDesktopWidget().screenGeometry()
        width = screenGeometry.width()
        height = screenGeometry.height()
        x = (width - window.width()) / 2
        y = (height - window.height()) / 2
        window.setGeometry(int(x), int(y), int(window.width()), int(window.height()))

        window.show()
        window.activateWindow()

    def removeWebcamWindow(self, window):
        self.camWindows.remove(window)

if __name__ == "__main__":
    tray = OctoTray()