Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# coding=utf-8
from __future__ import absolute_import
from subprocess import check_output
from subprocess import call, Popen, PIPE
from octoprint.settings import settings, valid_boolean_trues
import flask
import octoprint.plugin
import os
import re
class UsbcontrolPlugin(octoprint.plugin.SettingsPlugin,
octoprint.plugin.AssetPlugin,
octoprint.plugin.SimpleApiPlugin,
octoprint.plugin.StartupPlugin,
octoprint.plugin.TemplatePlugin):
def get_settings_defaults(self):
s = settings()
return dict(
usb2 = True,
usb3 = True if s.get(["plugins", "usbcontrol", "isRaspi3Bplus"]) else False,
usb4 = True if s.get(["plugins", "usbcontrol", "isRaspi3Bplus"]) else False,
usb5 = True if s.get(["plugins", "usbcontrol", "isRaspi3Bplus"]) else False,
all = True,
init = False,
isRaspi2B = False,
isRaspi3B = False,
isRaspi3Bplus = False,
cpuRevision = 'unknown',
from octoprint.util.comm import strip_comment
from octoprint.printer import PrinterInterface
import flask
import time
from threading import Thread
__plugin_pythoncompat__ = ">=2.7,<4"
class PreheatError(Exception):
def __init__(self, message):
super(PreheatError, self).__init__(message)
self.message = message
class PreheatAPIPlugin(octoprint.plugin.TemplatePlugin,
octoprint.plugin.SimpleApiPlugin,
octoprint.plugin.AssetPlugin,
octoprint.plugin.SettingsPlugin):
def get_settings_defaults(self):
return dict(enable_tool = True,
enable_bed = True,
enable_chamber = True,
fallback_tool = 0,
fallback_bed = 0,
fallback_chamber = 0,
wait_for_bed = False,
on_start_send_gcode = False,
on_start_send_gcode_command = "M117 Preheating... ; Update LCD",
on_complete_show_popup = False,
on_conplete_send_gcode = False,
on_conplete_send_gcode_command = "M117 Preheat complete. ; Update LCD\nM300 S660 P200 ; Beep",
# This is a basic skeleton for your plugin's __init__.py. You probably want to adjust the class name of your plugin
# 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"
currentLayer = str(self._currentLayerCount)
else:
currentLayer = str(matched.group(groupIndex))
line = line + LAYER_MESSAGE_PREFIX + currentLayer + "\r\n"
return line
class DisplaylayerprogressPlugin(
octoprint.plugin.StartupPlugin,
octoprint.plugin.SettingsPlugin,
octoprint.plugin.AssetPlugin,
octoprint.plugin.TemplatePlugin,
# my stuff
octoprint.plugin.EventHandlerPlugin,
octoprint.plugin.ProgressPlugin,
octoprint.plugin.SimpleApiPlugin,
octoprint.plugin.BlueprintPlugin
):
# VAR
_tempCurrentHeightFromFile = 0.0
_tempCurrentTotalHeight = 0.0
_currentLayerCount = 0
_layerTotalCount = NOT_PRESENT
_currentLayer = NOT_PRESENT
_progress = str(0)
_currentHeight = NOT_PRESENT
_totalHeight = NOT_PRESENT
_totalHeightWithExtrusion = NOT_PRESENT # DEPRECATED will be skiped in next release
_totalHeightFromExpression = NOT_PRESENT
_feedrate = NOT_PRESENT
_feedrateG0 = NOT_PRESENT
_feedrateG1 = NOT_PRESENT
'Rainbow Cycle': rainbow_cycle,
'Theater Chase Rainbow': theater_chase_rainbow,
'Pulse': pulse,
'Knight Rider': knight_rider,
'Plasma': plasma,
}
class RGBStatusPlugin(
plugin.AssetPlugin,
plugin.StartupPlugin,
plugin.ProgressPlugin,
plugin.EventHandlerPlugin,
plugin.SettingsPlugin,
plugin.TemplatePlugin,
plugin.ShutdownPlugin,
plugin.SimpleApiPlugin,
plugin.WizardPlugin):
api_errors = []
def is_wizard_required(self):
return any([not value for key, value in self.get_wizard_details().items()])
def get_wizard_version(self):
return 3
def get_wizard_details(self):
return {
'adduser_done': self.adduser_done(),
'spi_enabled': self.spi_enabled(),
'buffer_increased': self.buffer_increased(),
'frequency_set': self.frequency_set(),
# coding=utf-8
from __future__ import absolute_import
import octoprint.plugin
from octoprint.server import user_permission
import docker
class youtubelive(octoprint.plugin.StartupPlugin,
octoprint.plugin.TemplatePlugin,
octoprint.plugin.AssetPlugin,
octoprint.plugin.SettingsPlugin,
octoprint.plugin.SimpleApiPlugin,
octoprint.plugin.EventHandlerPlugin):
def __init__(self):
self.client = docker.from_env()
self.container = None
##~~ StartupPlugin
def on_after_startup(self):
self._logger.info("OctoPrint-YouTubeLive loaded! Checking stream status.")
try:
self.container = self.client.containers.get('YouTubeLive')
self._logger.info("%s is streaming " % self.container.name)
self._plugin_manager.send_plugin_message(self._identifier, dict(status=True,streaming=True))
except Exception, e:
self._logger.error(str(e))
from .job_notifications import JobNotifications
from .bed_notifications import BedNotifications
from .tools_notifications import ToolsNotifications
from .mmu import MMUAssistance
from .paused_for_user import PausedForUser
from .palette2 import Palette2Notifications
# Plugin that stores APNS tokens reported from iOS devices to know which iOS devices to alert
# when print is done or other relevant events
class OctopodPlugin(octoprint.plugin.SettingsPlugin,
octoprint.plugin.AssetPlugin,
octoprint.plugin.TemplatePlugin,
octoprint.plugin.StartupPlugin,
octoprint.plugin.SimpleApiPlugin,
octoprint.plugin.EventHandlerPlugin):
def __init__(self):
super(OctopodPlugin, self).__init__()
self._logger = logging.getLogger("octoprint.plugins.octopod")
self._checkTempTimer = None
self._job_notifications = JobNotifications(self._logger)
self._tool_notifications = ToolsNotifications(self._logger)
self._bed_notifications = BedNotifications(self._logger)
self._mmu_assitance = MMUAssistance(self._logger)
self._paused_for_user = PausedForUser(self._logger)
self._palette2 = Palette2Notifications(self._logger)
# StartupPlugin mixin
def on_after_startup(self):
import logging
from flask import jsonify
import os.path
from os import linesep
import datetime
import mmap
import re
import contextlib
from shutil import copyfile
import os
import io
class MultiColorsPlugin(octoprint.plugin.AssetPlugin,
octoprint.plugin.SimpleApiPlugin,
octoprint.plugin.TemplatePlugin,
octoprint.plugin.SettingsPlugin):
def initialize(self):
#self._logger.setLevel(logging.DEBUG)
self.gcode_file = os.path.join(self.get_plugin_data_folder(),"gcode.txt")
self.regex_file = os.path.join(self.get_plugin_data_folder(),"regex.txt")
self._logger.info("MultiColors init")
def get_template_configs(self):
return [
dict(type="tab", template="multi_colors_tab.jinja2", custom_bindings=True)
]
def get_assets(self):
return dict(
from octoprint.events import Events
# import sys
import time
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
# coding=utf-8
from __future__ import absolute_import
from subprocess import check_output
from subprocess import call
from octoprint.settings import settings, valid_boolean_trues
import octoprint.plugin
import os
class GitfilesPlugin(octoprint.plugin.SettingsPlugin,
octoprint.plugin.AssetPlugin,
octoprint.plugin.SimpleApiPlugin,
octoprint.plugin.TemplatePlugin):
def get_settings_defaults(self):
return dict(url="https://github.com/YourUserID/YourRepository.git", path="gitfiles")
def get_template_vars(self):
return dict(url=self._settings.get(["url"]), path=self._settings.get(["path"]))
def get_template_configs(self):
return [dict(type="settings", custom_bindings=False)]
def get_assets(self):
return dict(
js=["js/gitfiles.js"],
css=["css/gitfiles.css"],
less=["less/gitfiles.less"]