<commit_before># -*- coding: utf-8 -*-
from __future__ import absolute_import, division, print_function, unicode_literals

__license__ = 'GNU Affero General Public License http://www.gnu.org/licenses/agpl.html'
__copyright__ = "Copyright (C) 2018 The OctoPrint Project - Released under terms of the AGPLv3 License"

import octoprint.plugin

from octoprint.events import Events
from octoprint.server import user_permission

from .checks.firmware_unsafe import FirmwareUnsafeChecks

import flask
from flask_babel import gettext

import textwrap

TERMINAL_SAFETY_WARNING = """
!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
{message}

Learn more at https://faq.octoprint.org/warning-{warning_type}
!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!

"""

SAFETY_CHECKS = {
	"firmware-unsafe": FirmwareUnsafeChecks.as_dict()
}

class PrinterSafetyCheckPlugin(octoprint.plugin.AssetPlugin,
                               octoprint.plugin.EventHandlerPlugin,
                               octoprint.plugin.SimpleApiPlugin,
                               octoprint.plugin.TemplatePlugin):

	# noinspection PyMissingConstructor
	def __init__(self):
		self._warnings = dict()
		self._scan_received = True

	##~~ TemplatePlugin API

	def get_template_configs(self):
		return [
			dict(type="sidebar",
			     name=gettext("Printer Safety Warning"),
			     data_bind="visible: printerState.isOperational() && loginState.isAdmin() && warnings().length > 0",
			     icon="exclamation-triangle",
			     styles_wrapper=["display: none"])
		]

	##~~ AssetPlugin API

	def get_assets(self):
		return dict(js=("js/printer_safety_check.js",),
		            clientjs=("clientjs/printer_safety_check.js",),
		            css=("css/printer_safety_check.css",),
		            less=("less/printer_safety_check.less",))

	##~~ EventHandlerPlugin API

	def on_event(self, event, payload):
		if event == Events.DISCONNECTED:
			self._reset_warnings()
			self._reset_state()
			self._reset_checks()

	##~~ SimpleApiPlugin API

	def on_api_get(self, request):
		if not user_permission.can():
			return flask.make_response("Insufficient rights", 403)
		return flask.jsonify(self._warnings)

	##~~ GCODE received hook handler

	def on_gcode_received(self, comm_instance, line, *args, **kwargs):
		if self._scan_received:
			self._run_checks("received", line)
		return line

	##~~ Firmware info hook handler

	def on_firmware_info_received(self, comm_instance, firmware_name, firmware_data):
		self._run_checks("m115", firmware_name, firmware_data)
		self._scan_received = False

	##~~ Firmware capability hook handler

	def on_firmware_cap_received(self, comm_instance, cap, enabled, all_caps):
		self._run_checks("cap", cap, enabled)

	##~~ Helpers

	def _run_checks(self, check_type, *args, **kwargs):
		changes = False

		for warning_type, check_data in SAFETY_CHECKS.items():
			checks = check_data.get("checks")
			message = check_data.get("message")
			if not checks or not message:
				continue

			for check in checks:
				if not check.active:
					# skip non active checks
					continue

				method = getattr(check, check_type, None)
				if not callable(method):
					# skip uncallable checks
					continue<commit_msg>🐛 printer safety: Fix for unicode issues<commit_after># -*- coding: utf-8 -*-
from __future__ import absolute_import, division, print_function, unicode_literals

__license__ = 'GNU Affero General Public License http://www.gnu.org/licenses/agpl.html'
__copyright__ = "Copyright (C) 2018 The OctoPrint Project - Released under terms of the AGPLv3 License"

import octoprint.plugin

from octoprint.events import Events
from octoprint.server import user_permission
from octoprint.util import to_unicode

from .checks.firmware_unsafe import FirmwareUnsafeChecks

import flask
from flask_babel import gettext

import textwrap

TERMINAL_SAFETY_WARNING = """
!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
{message}

Learn more at https://faq.octoprint.org/warning-{warning_type}
!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!

"""

SAFETY_CHECKS = {
	"firmware-unsafe": FirmwareUnsafeChecks.as_dict()
}

class PrinterSafetyCheckPlugin(octoprint.plugin.AssetPlugin,
                               octoprint.plugin.EventHandlerPlugin,
                               octoprint.plugin.SimpleApiPlugin,
                               octoprint.plugin.TemplatePlugin):

	# noinspection PyMissingConstructor
	def __init__(self):
		self._warnings = dict()
		self._scan_received = True

	##~~ TemplatePlugin API

	def get_template_configs(self):
		return [
			dict(type="sidebar",
			     name=gettext("Printer Safety Warning"),
			     data_bind="visible: printerState.isOperational() && loginState.isAdmin() && warnings().length > 0",
			     icon="exclamation-triangle",
			     styles_wrapper=["display: none"])
		]

	##~~ AssetPlugin API

	def get_assets(self):
		return dict(js=("js/printer_safety_check.js",),
		            clientjs=("clientjs/printer_safety_check.js",),
		            css=("css/printer_safety_check.css",),
		            less=("less/printer_safety_check.less",))

	##~~ EventHandlerPlugin API

	def on_event(self, event, payload):
		if event == Events.DISCONNECTED:
			self._reset_warnings()
			self._reset_state()
			self._reset_checks()

	##~~ SimpleApiPlugin API

	def on_api_get(self, request):
		if not user_permission.can():
			return flask.make_response("Insufficient rights", 403)
		return flask.jsonify(self._warnings)

	##~~ GCODE received hook handler

	def on_gcode_received(self, comm_instance, line, *args, **kwargs):
		if self._scan_received:
			self._run_checks("received", to_unicode(line, errors="replace"))
		return line

	##~~ Firmware info hook handler

	def on_firmware_info_received(self, comm_instance, firmware_name, firmware_data):
		self._run_checks("m115",
		                 to_unicode(firmware_name, errors="replace"),
		                 dict((to_unicode(key, errors="replace"), to_unicode(value, errors="replace"))
		                      for key, value in firmware_data.items()))
		self._scan_received = False

	##~~ Firmware capability hook handler

	def on_firmware_cap_received(self, comm_instance, cap, enabled, all_caps):
		self._run_checks("cap",
		                 to_unicode(cap, errors="replace"),
		                 enabled)

	##~~ Helpers

	def _run_checks(self, check_type, *args, **kwargs):
		changes = False

		for warning_type, check_data in SAFETY_CHECKS.items():
			checks = check_data.get("checks")
			message = check_data.get("message")
			if not checks or not message:
				continue

			for check in checks:
				if not check.active:
					# skip non active checks
					continue

				method = getattr(check, check_type, None)
				if not callable(method):
					# skip uncallable checks
					continue
{'fraction_non_alphanumeric': 0.14647641234711706, 'fraction_numerical': 0.0037856726849155504, 'mean_word_length': 3.955988455988456, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3951498', 'n_tokens_mistral': 2407, 'n_tokens_neox': 2033, 'n_words': 454}