st-ten-1/src/components/remote_api.py

import json
import os
import shlex
import subprocess
import tarfile
import traceback

import cv2
from bottle import post, request, response, route, run
from PyQt5.QtCore import QObject, pyqtSignal, pyqtSlot
from PyQt5.QtWidgets import QMessageBox

from .component import Component


def aupdate_available_msg():
    QMessageBox.warning(
        None,
        "Aggiornamento disponibile",
        "È disponibile un agiornamento per il banco.\nRiavviare il programma appena possibile.",
    )


class RemoteAPI(Component):
    api_cmd = pyqtSignal(str)

    def __init__(self, config=None, name=None, period=1, lazy=True, paused=False, main=None, threaded=True):
        super().__init__(config=config, name=name, period=period, lazy=lazy, paused=paused, threaded=threaded)
        self.main = main

    @pyqtSlot()
    def start(self):
        @route("/")
        def root():
            return "Usage: /api/<cmd>"

        @route("/api/identify")
        def identify():
            return f"{self.config.system_id!r} as {self.config.machine_id!r}"

        # @route("/api/profile")
        # def profile():
        #     response.content_type = "application/json"
        #     return json.dumps(self.parent.watcher.watches)

        # @route("/api/get_det")
        # def get_det():
        #     response.content_type = "application/json"
        #     return json.dumps(self.parent.processed_detections)

        # @route("/api/get_img")
        # def get_img():
        #     type = request.query.type or None
        #     if type in ["original", "o", "frame", "f"]:
        #         img = self.bench.inputs["frame"].last_frame
        #     elif type in ["gui", "g"]:
        #         img = self.bench.inputs["frame"].last_frame_scaled
        #     elif type in ["terminals", "t"]:
        #         img = self.bench.inputs["terminals"].last_frame
        #     elif type in ["wires", "w"]:
        #         img = self.bench.inputs["wires"].last_frame
        #     elif type in ["drawn", "d"]:
        #         img = self.bench.inputs["renderer"].last_drawn
        #     elif type in ["zoomed", "z"]:
        #         img = self.bench.inputs["renderer"].last_zoomed
        #     else:
        #         img = self.bench.inputs["frame"].last_frame_scaled
        #     response.content_type = "image/jpeg"
        #     _, img = cv2.imencode(".jpeg", cv2.cvtColor(img, cv2.COLOR_RGB2BGR))
        #     return img.tobytes()

        # @route("/api/get_plc")
        # def get_plc():
        #     response.content_type = "application/json"
        #     return json.dumps(self.parent.plc.get_all_vars())

        # @route("/api/set_model")
        # @post("/api/set_model")
        # def set_model():
        #     name = request.query.name or None
        #     if name is None:
        #         response.status = 400
        #         return "missing model 'name' parameter"
        #     try:
        #         self.parent.vision_test.load_model(name)  # very unsafe
        #         return "ok"
        #     except Exception:
        #         e = traceback.format_exc()
        #         self.log.exception(e)
        #         response.status = 400
        #         return e

        # @route("/api/new_model")
        # @post("/api/new_model")
        # def new_model():
        #     file = request.files.get("model") or None
        #     if file is None:
        #         response.status = 400
        #         return "missing 'model' parameter"
        #     try:
        #         with tarfile.open(mode="r", fileobj=file.file) as file:
        #             file.extractall("neural_networks")
        #         return "ok"
        #     except Exception:
        #         e = traceback.format_exc()
        #         self.log.exception(e)
        #         response.status = 400
        #         return e

        @route("/api/git")
        @post("/api/git")
        def git():
            try:
                parameters = request.forms.get("command", "")
                cmd_str = f"git {parameters}"
                cmd = shlex.split(cmd_str)  # very important
                r = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
                response.content_type = "application/json"
                return json.dumps({
                    "command": cmd_str,
                    "result": r.stdout.decode("utf-8"),
                    "return_code": r.returncode,
                })
            except Exception:
                e = traceback.format_exc()
                self.log.exception(e)
                response.status = 400
                return e

        @route("/api/auto_update")
        def auto_update():
            try:
                r = subprocess.run(shlex.split("git diff --exit-code"), stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
                if r.returncode != 0:
                    response.status = 409
                    response.content_type = "text/plain"
                    return r.stdout.decode("utf-8")
                r = subprocess.run(shlex.split("git pull"), stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
                if r.returncode != 0:
                    response.status = 409
                    response.content_type = "text/plain"
                    return r.stdout.decode("utf-8")
                self.main.main_window.do.emit({"f": aupdate_available_msg})
                return "ok"
            except Exception:
                e = traceback.format_exc()
                self.log.exception(e)
                response.status = 400
                return e

        super().start()
        run(host="0.0.0.0", port=8080)

    def update_data(self, new_data):
        self.current_data = new_data