Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
import octoprint.server.util.flask
from octoprint.server import admin_permission, NO_CONTENT
from octoprint.events import Events
# import the flash methods
from octoprint_firmwareupdater.methods import avrdude
from octoprint_firmwareupdater.methods import bossac
from octoprint_firmwareupdater.methods import dfuprog
from octoprint_firmwareupdater.methods import lpc1768
from octoprint_firmwareupdater.methods import stm32flash
class FirmwareupdaterPlugin(octoprint.plugin.BlueprintPlugin,
octoprint.plugin.TemplatePlugin,
octoprint.plugin.AssetPlugin,
octoprint.plugin.SettingsPlugin,
octoprint.plugin.EventHandlerPlugin):
def __init__(self):
self._flash_thread = None
self._flash_prechecks = dict()
self._flash_methods = dict()
self._console_logger = None
def initialize(self):
# TODO: make method configurable via new plugin hook "octoprint.plugin.firmwareupdater.flash_methods",
# also include prechecks
self._flash_prechecks = dict(avrdude=avrdude._check_avrdude, bossac=bossac._check_bossac, dfuprogrammer=dfuprog._check_dfuprog, lpc1768=lpc1768._check_lpc1768, stm32flash=stm32flash._check_stm32flash)
self._flash_methods = dict(avrdude=avrdude._flash_avrdude, bossac=bossac._flash_bossac, dfuprogrammer=dfuprog._flash_dfuprog, lpc1768=lpc1768._flash_lpc1768, stm32flash=stm32flash._flash_stm32flash)
console_logging_handler = logging.handlers.RotatingFileHandler(self._settings.get_plugin_logfile_path(postfix="console"), maxBytes=2*1024*1024)
# coding=utf-8
from __future__ import absolute_import
import octoprint.plugin
import os
import json
import requests
class SlackPlugin(octoprint.plugin.SettingsPlugin,
octoprint.plugin.TemplatePlugin,
octoprint.plugin.EventHandlerPlugin):
## SettingsPlugin
def get_settings_defaults(self):
return dict(
webhook_url="",
print_events=dict(
PrintStarted=dict(
Enabled=True,
Message="A new print has started! :muscle:",
Fallback="Print started! Filename: {filename}",
Color="good",
),
PrintFailed=dict(
Enabled=True,
Message="Oh no! The print has failed... :rage2:",
import sys
import glob
import os
from datetime import datetime
from datetime import timedelta
import octoprint.util
import requests
import inspect
import threading
import json
import copy
class EnclosurePlugin(octoprint.plugin.StartupPlugin, octoprint.plugin.TemplatePlugin, octoprint.plugin.SettingsPlugin,
octoprint.plugin.AssetPlugin, octoprint.plugin.BlueprintPlugin,
octoprint.plugin.EventHandlerPlugin):
rpi_outputs = []
rpi_inputs = []
waiting_temperature = []
rpi_outputs_not_changed = []
notifications = []
pwm_instances = []
event_queue = []
temp_hum_control_status = []
temperature_sensor_data = []
last_filament_end_detected = []
print_complete = False
development_mode = False
dummy_value = 30.0
dummy_delta = 0.5
def start_timer(self):
# as well as the plugin mixins it's subclassing from. This is really just a basic skeleton to get you started,
# defining your plugin as a template plugin, settings and asset plugin. Feel free to add or remove mixins
# as necessary.
#
# Take a look at the documentation on what other plugin mixins are available.
import octoprint.plugin
from octoprint.server import user_permission
import docker
class WebcamStreamerPlugin(octoprint.plugin.StartupPlugin,
octoprint.plugin.TemplatePlugin,
octoprint.plugin.AssetPlugin,
octoprint.plugin.SettingsPlugin,
octoprint.plugin.SimpleApiPlugin,
octoprint.plugin.EventHandlerPlugin):
def __init__(self):
# Docker connection and container object
self.client = None
self.image = None
self.container = None
self.frame_rate_default = 5
self.ffmpeg_cmd_default = (
"ffmpeg -re -f mjpeg -framerate 5 -i {webcam_url} " # Video input
"-ar 44100 -ac 2 -acodec pcm_s16le -f s16le -ac 2 -i /dev/zero " # Audio input
"-acodec aac -ab 128k " # Audio output
"-vcodec h264 -pix_fmt yuv420p -framerate {frame_rate} -g {gop_size} -strict experimental -filter:v {filter} " # Video output
"-f flv {stream_url}") # Output stream
self.docker_image_default = "adilinden/rpi-ffmpeg:latest"
self.docker_container_default = "WebStreamer"
import math
import os
import subprocess
import octoprint.plugin
import re
import logging
import json
import flask
class BetterGrblSupportPlugin(octoprint.plugin.SettingsPlugin,
octoprint.plugin.SimpleApiPlugin,
octoprint.plugin.AssetPlugin,
octoprint.plugin.TemplatePlugin,
octoprint.plugin.StartupPlugin,
octoprint.plugin.EventHandlerPlugin):
def __init__(self):
self.hideTempTab = True
self.hideControlTab = True
self.hideGCodeTab = True
self.customControls = False
self.helloCommand = "M5"
self.statusCommand = "?$G"
self.dwellCommand = "G4 P0"
self.positionCommand = "?"
self.suppressM114 = True
self.suppressM400 = True
self.suppressM105 = True
self.suppressM115 = True
self.suppressM110 = True
self.disablePolling = True
# coding=utf-8
from __future__ import absolute_import
import octoprint.plugin
import octoprint.events
import time
import threading
from .piglow import PiGlow
class OctoGlowPlugin(octoprint.plugin.EventHandlerPlugin,
octoprint.plugin.ProgressPlugin,
octoprint.plugin.StartupPlugin):
"""
Plugin for animating the LEDs on a PiGlow board based on OctoPrint events.
"""
def __init__(self):
self._lock = threading.Lock()
self._currentAnimation = None
self._printProgress = 0
def on_after_startup(self):
"""
Callback for just after launch of OctoPrint.
"""
self._piglow = PiGlow()
return command, data, None
def create_ws_token(public_key = None, api_key = None):
from itsdangerous import URLSafeTimedSerializer
s = URLSafeTimedSerializer(api_key)
return s.dumps({ 'public_key': public_key })
class AstroprintPlugin(octoprint.plugin.SettingsPlugin,
octoprint.plugin.AssetPlugin,
octoprint.plugin.StartupPlugin,
octoprint.plugin.TemplatePlugin,
octoprint.plugin.BlueprintPlugin,
octoprint.plugin.EventHandlerPlugin,
octoprint.printer.PrinterCallback):
##~~ SettingsPlugin mixin
def initialize(self):
self.user = {}
self.designs = None
self.db = None
self.astroprintCloud = None
self.cameraManager = None
self.materialCounter= None
self._printerListener = None
def logOutHandler(sender, **kwargs):
self.onLogout()
import datetime
import numpy as np
import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot as plt
import matplotlib.image as mpimg
import re
import octoprint.plugin
import octoprint.printer
class PrusameshmapPlugin(octoprint.plugin.SettingsPlugin,
octoprint.plugin.AssetPlugin,
octoprint.plugin.TemplatePlugin,
octoprint.plugin.StartupPlugin,
octoprint.plugin.EventHandlerPlugin):
##~~ SettingsPlugin mixin
def get_settings_defaults(self):
return dict(
do_level_gcode = 'G28 W ; home all without mesh bed level\nG80 ; mesh bed leveling\nG81 ; check mesh leveling results',
matplotlib_heatmap_theme = 'viridis'
)
##~~ AssetPlugin mixin
def get_assets(self):
# Define your plugin's asset files to automatically include in the
# core UI here.
return dict(
js=["js/PrusaMeshMap.js"],
import re
import subprocess
import copy
import netifaces
import pytz
SLACKER_TIMEOUT = 60
class OctoslackPlugin(
octoprint.plugin.SettingsPlugin,
octoprint.plugin.AssetPlugin,
octoprint.plugin.StartupPlugin,
octoprint.plugin.ShutdownPlugin,
octoprint.plugin.ProgressPlugin,
octoprint.plugin.EventHandlerPlugin,
octoprint.plugin.TemplatePlugin,
):
##TODO FEATURE - generate an animated gif of the print - easy enough if we can find a python ib to create the gif (images2gif is buggy & moviepy, imageio, and and visvis which rely on numpy haven't worked out as I never neven let numpy try to finish installing after 5/10 minutes on my RasPi3)
##TODO FEATURE - add the timelapse gallery for cancelled/failed/completed as a single image
##TODO FEATURE - Add support for Imgur image title + description
##TODO FEATURE - Optionally upload timelapse video to youtube & send a Slack message when the upload is complete
##TODO FEATURE - Define a third set of messages for each event to allow sending M117 commands to the printer
##TODO ENHANCEMENT - The progress event fires on gcode uploads and triggers Octoslack events. That needs to be fixed.
##TODO ENHANCEMENT - Consider extending the progress snapshot minimum interval beyond Slack to other providers
##~~ SettingsPlugin mixin
def get_settings_defaults(self):
return {
"connection_method": "APITOKEN", ##APITOKEN or WEBHOOK