133 lines
4 KiB
Python
133 lines
4 KiB
Python
# coding=utf-8
|
|
from __future__ import absolute_import
|
|
import threading
|
|
import subprocess
|
|
import requests
|
|
import logging
|
|
from requests.auth import HTTPBasicAuth
|
|
from pathlib import Path
|
|
import base64
|
|
import octoprint.plugin
|
|
from octoprint.events import Events
|
|
|
|
|
|
class FactotumPlugin(
|
|
octoprint.plugin.StartupPlugin,
|
|
octoprint.plugin.SettingsPlugin,
|
|
octoprint.plugin.TemplatePlugin,
|
|
octoprint.plugin.EventHandlerPlugin,
|
|
):
|
|
|
|
def __init__(self):
|
|
self.signalUrl = ""
|
|
self.signalNumber = ""
|
|
self.signalRecipient = ""
|
|
self.httpBasicAuth = False
|
|
self.basicAuthUsername = ""
|
|
self.basicAuthPassword = ""
|
|
|
|
def get_settings_defaults(self):
|
|
return {
|
|
"signalUrl": "",
|
|
"signalNumber": "",
|
|
"signalRecipient": "",
|
|
"httpBasicAuth": False,
|
|
"basicAuthUsername": "",
|
|
"basicAuthPassword": "",
|
|
}
|
|
|
|
def load_settings(self):
|
|
self.signalUrl = self._settings.get(["signalUrl"])
|
|
self.signalNumber = self._settings.get(["signalNumber"])
|
|
self.signalRecipient = self._settings.get(["signalRecipient"])
|
|
self.httpBasicAuth = self._settings.get_boolean(["httpBasicAuth"])
|
|
self.basicAuthUsername = self._settings.get(["basicAuthUsername"])
|
|
self.basicAuthPassword = self._settings.get(["basicAuthPassword"])
|
|
|
|
def on_after_startup(self):
|
|
self.load_settings()
|
|
|
|
def get_template_configs(self):
|
|
return [
|
|
dict(type="settings", custom_bindings=False),
|
|
]
|
|
|
|
def get_update_information(self):
|
|
return {
|
|
"Factotum": {
|
|
"displayName": "Factotum Plugin",
|
|
"displayVersion": self._plugin_version,
|
|
"type": "stable_branch",
|
|
"branch": "main",
|
|
"name": "Main",
|
|
"current": self._plugin_version,
|
|
"pip": "https://git.walbeck.it/mwalbeck/OctoPrint-Factotum/archive/{target_version}.zip",
|
|
}
|
|
}
|
|
|
|
def on_event(self, event, payload):
|
|
if event == Events.PLUGIN_OCTOLAPSE_MOVIE_DONE:
|
|
thread = threading.Thread(target=self.process, args=(payload["movie"],payload["gcode"],))
|
|
thread.start()
|
|
|
|
def process(self, movie, gcode):
|
|
self._logger.info("Generate GIF based on timelapse")
|
|
|
|
movie_path = Path(movie)
|
|
gif_path = movie_path.with_suffix(".gif")
|
|
|
|
# ffmpeg -i timelapse.mp4 -vf scale=640:-1 -pix_fmt rgb8 output_new.gif
|
|
subprocess.run(
|
|
[
|
|
"ffmpeg",
|
|
"-loglevel",
|
|
"warning",
|
|
"-itsscale",
|
|
"0.5",
|
|
"-i",
|
|
str(movie_path),
|
|
"-vf",
|
|
"scale=640:-1",
|
|
"-pix_fmt",
|
|
"rgb8",
|
|
str(gif_path),
|
|
]
|
|
)
|
|
|
|
self._logger.info("GIF done")
|
|
|
|
self._logger.info("base64 encoding GIF")
|
|
with open(gif_path, "rb") as gif:
|
|
encoded_gif = base64.b64encode(gif.read())
|
|
|
|
payload = {
|
|
"base64_attachments": [
|
|
f"data:image/gif;filename={gif_path.name};base64,{encoded_gif.decode('utf-8')}"
|
|
],
|
|
"number": self.signalNumber,
|
|
"recipients": [self.signalRecipient],
|
|
"message": gcode
|
|
}
|
|
|
|
if self.httpBasicAuth:
|
|
auth = HTTPBasicAuth(self.basicAuthUsername, self.basicAuthPassword)
|
|
else:
|
|
auth = None
|
|
|
|
self._logger.info("Sending GIF")
|
|
r = requests.post(self.signalUrl + "/v2/send", json=payload, auth=auth)
|
|
self._logger.info(r.status_code)
|
|
|
|
|
|
__plugin_name__ = "Factotum"
|
|
__plugin_pythoncompat__ = ">=3,<4"
|
|
|
|
|
|
def __plugin_load__():
|
|
global __plugin_implementation__
|
|
__plugin_implementation__ = FactotumPlugin()
|
|
|
|
global __plugin_hooks__
|
|
__plugin_hooks__ = {
|
|
"octoprint.plugin.softwareupdate.check_config": __plugin_implementation__.get_update_information
|
|
}
|