<commit_before># coding=utf-8
from __future__ import absolute_import

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

import logging
import os

import octoprint.plugin
import octoprint.util

from octoprint.events import eventManager, Events
from octoprint.settings import settings

from .destinations import FileDestinations
from .analysis import QueueEntry, AnalysisQueue
from .storage import LocalFileStorage
from .util import AbstractFileWrapper, StreamWrapper, DiskFileWrapper

from collections import namedtuple

ContentTypeMapping = namedtuple("ContentTypeMapping", "extensions, content_type")
ContentTypeDetector = namedtuple("ContentTypeDetector", "extensions, detector")

extensions = dict(
)

def full_extension_tree():
	result = dict(
		# extensions for 3d model files
		model=dict(
			stl=ContentTypeMapping(["stl"], "application/sla")
		),
		# extensions for printable machine code
		machinecode=dict(
			gcode=ContentTypeMapping(["gcode", "gco", "g"], "text/plain")
		)
	)

	extension_tree_hooks = octoprint.plugin.plugin_manager().get_hooks("octoprint.filemanager.extension_tree")
	for name, hook in extension_tree_hooks.items():
		try:
			hook_result = hook()
			if hook_result is None or not isinstance(hook_result, dict):
				continue
			result = octoprint.util.dict_merge(result, hook_result)
		except:
			logging.getLogger(__name__).exception("Exception while retrieving additional extension tree entries from hook {name}".format(name=name))

	return result

def get_extensions(type, subtree=None):
	if not subtree:
		subtree = full_extension_tree()

	for key, value in subtree.items():
		if key == type:
			return get_all_extensions(subtree=value)
		elif isinstance(value, dict):
			sub_extensions = get_extensions(type, subtree=value)
			if sub_extensions:
				return sub_extensions

	return None

def get_all_extensions(subtree=None):
	if not subtree:
		subtree = full_extension_tree()

	result = []
	if isinstance(subtree, dict):
		for key, value in subtree.items():
			if isinstance(value, dict):
				result += get_all_extensions(value)
			elif isinstance(value, (ContentTypeMapping, ContentTypeDetector)):
				result += value.extensions
			elif isinstance(value, (list, tuple)):
				result += value
	elif isinstance(subtree, (ContentTypeMapping, ContentTypeDetector)):
		result = subtree.extensions
	elif isinstance(subtree, (list, tuple)):
		result = subtree
	return result

def get_path_for_extension(extension, subtree=None):
	if not subtree:
		subtree = full_extension_tree()

	for key, value in subtree.items():
		if isinstance(value, (ContentTypeMapping, ContentTypeDetector)) and extension in value.extensions:
			return [key]
		elif isinstance(value, (list, tuple)) and extension in value:
			return [key]
		elif isinstance(value, dict):
			path = get_path_for_extension(extension, subtree=value)
			if path:
				return [key] + path

	return None

def get_content_type_mapping_for_extension(extension, subtree=None):
	if not subtree:
		subtree = full_extension_tree()

	for key, value in subtree.items():
		content_extension_matches = isinstance(value, (ContentTypeMapping, ContentTypeDetector)) and extension in value. extensions
		list_extension_matches = isinstance(value, (list, tuple)) and extension in value

		if content_extension_matches or list_extension_matches:
			return value
		elif isinstance(value, dict):
			result = get_content_type_mapping_for_extension(extension, subtree=value)
			if result is not None:
				return result

	return None

def valid_extension(extension, type=None):
	if not type:
		return extension in get_all_extensions()
	else:
		extensions = get_extensions(type)
		if extensions:
			return extension in extensions

def valid_file_type(filename, type=None):
	_, extension = os.path.splitext(filename)
	extension = extension[1:].lower()
	return valid_extension(extension, type=type)

def get_file_type(filename):
	_, extension = os.path.splitext(filename)
	extension = extension[1:].lower()
	return get_path_for_extension(extension)

def get_mime_type(filename):
	_, extension = os.path.splitext(filename)
	extension = extension[1:].lower()
	mapping = get_content_type_mapping_for_extension(extension)
	if mapping:
		if isinstance(mapping, ContentTypeMapping) and mapping.content_type is not None:
			return mapping.content_type
		elif isinstance(mapping, ContentTypeDetector) and callable(mapping.detector):
			result = mapping.detector(filename)
			if result is not None:
				return result
	return "application/octet-stream"


class NoSuchStorage(Exception):
	pass


class FileManager(object):
	def __init__(self, analysis_queue, slicing_manager, printer_profile_manager, initial_storage_managers=None):
		self._logger = logging.getLogger(__name__)
		self._analysis_queue = analysis_queue
		self._analysis_queue.register_finish_callback(self._on_analysis_finished)

		self._storage_managers = dict()
		if initial_storage_managers:
			self._storage_managers.update(initial_storage_managers)

		self._slicing_manager = slicing_manager
		self._printer_profile_manager = printer_profile_manager

		import threading
		self._slicing_jobs = dict()
		self._slicing_jobs_mutex = threading.Lock()

		self._slicing_progress_callbacks = []
		self._last_slicing_progress = None

		self._progress_plugins = []
		self._preprocessor_hooks = dict()

		import octoprint.settings
		self._recovery_file = os.path.join(octoprint.settings.settings().getBaseFolder("data"), "print_recovery_data.yaml")

	def initialize(self):
		self.reload_plugins()

		def worker():
			self._logger.info("Adding backlog items from all storage types to analysis queue...".format(**locals()))
			for storage_type, storage_manager in self._storage_managers.items():
				self._determine_analysis_backlog(storage_type, storage_manager)

		import threading
		thread = threading.Thread(target=worker)
		thread.daemon = True
		thread.start()

	def reload_plugins(self):
		self._progress_plugins = octoprint.plugin.plugin_manager().get_implementations(octoprint.plugin.ProgressPlugin)
		self._preprocessor_hooks = octoprint.plugin.plugin_manager().get_hooks("octoprint.filemanager.preprocessor")

	def register_slicingprogress_callback(self, callback):
		self._slicing_progress_callbacks.append(callback)

	def unregister_slicingprogress_callback(self, callback):
		try:
			self._slicing_progress_callbacks.remove(callback)
		except ValueError:
			# callback was not registered
			pass

	def _determine_analysis_backlog(self, storage_type, storage_manager):
		counter = 0
		for entry, path, printer_profile in storage_manager.analysis_backlog:
			file_type = get_file_type(path)[-1]

			# we'll use the default printer profile for the backlog since we don't know better
			queue_entry = QueueEntry(entry, file_type, storage_type, path, self._printer_profile_manager.get_default())
			if self._analysis_queue.enqueue(queue_entry, high_priority=False):
				counter += 1
		self._logger.info("Added {counter} items from storage type \"{storage_type}\" to analysis queue".format(**locals()))

	def add_storage(self, storage_type, storage_manager):
		self._storage_managers[storage_type] = storage_manager
		self._determine_analysis_backlog(storage_type, storage_manager)

	def remove_storage(self, type):
		if not type in self._storage_managers:
			return
		del self._storage_managers[type]

	@property
	def slicing_enabled(self):
		return self._slicing_manager.slicing_enabled

	@property
	def registered_slicers(self):
		return self._slicing_manager.registered_slicers

	@property
	def default_slicer(self):
		return self._slicing_manager.default_slicer

	def slice(self, slicer_name, source_location, source_path, dest_location, dest_path, resolution=None, nozzle_size=None,
	          position=None, profile=None, printer_profile_id=None, overrides=None, callback=None, callback_args=None):
		absolute_source_path = self.path_on_disk(source_location, source_path)

		def stlProcessed(source_location, source_path, tmp_path, dest_location, dest_path, start_time, printer_profile_id, callback, callback_args, _error=None, _cancelled=False, _analysis=None):
			try:
				if _error:
					eventManager().fire(Events.SLICING_FAILED, {"stl": source_path, "gcode": dest_path, "reason": _error})
				elif _cancelled:
					eventManager().fire(Events.SLICING_CANCELLED, {"stl": source_path, "gcode": dest_path})
				else:
					source_meta = self.get_metadata(source_location, source_path)
					hash = source_meta["hash"]

					import io
					links = [("model", dict(name=source_path))]
					_, stl_name = self.split_path(source_location, source_path)
					file_obj = StreamWrapper(os.path.basename(dest_path),
					                         io.BytesIO(u";Generated from {stl_name} {hash}\n".format(**locals()).encode("ascii", "replace")),
					                         io.FileIO(tmp_path, "rb"))

					printer_profile = self._printer_profile_manager.get(printer_profile_id)

					self.add_file(dest_location, dest_path, file_obj, links=links, allow_overwrite=True, printer_profile=printer_profile, analysis=_analysis)

					# If the BVC analyser was defined, overrides the Cura time estimation
					analyser = settings().get(['gcodeAnalysis', 'analyser'])
					if analyser is not None and analyser == 'BVC':
						import octoprint.util.bvc_gcoder as bvcGcoder
						gcoder_result = bvcGcoder.analyse(self.path_on_disk(dest_location, dest_path))
						if 'estimated_duration' in gcoder_result:
							_analysis["estimatedPrintTime"] = gcoder_result['estimated_duration']
							self._add_analysis_result(dest_location, dest_path, _analysis)

					end_time = time.time()
					eventManager().fire(Events.SLICING_DONE, {"stl": source_path, "gcode": dest_path, "time": end_time - start_time})

					if callback is not None:
						if callback_args is None:
							callback_args = ()
						callback(*callback_args)
			finally:
				os.remove(tmp_path)

				source_job_key = (source_location, source_path)
				dest_job_key = (dest_location, dest_path)

				with self._slicing_jobs_mutex:
					if source_job_key in self._slicing_jobs:
						del self._slicing_jobs[source_job_key]
					if dest_job_key in self._slicing_jobs:
						del self._slicing_jobs[dest_job_key]

		slicer = self._slicing_manager.get_slicer(slicer_name)
<commit_msg>BEEWEB-205 New custom GCode estimator from previous BEESOFT 3.15 - replace Cura estimation<commit_after># coding=utf-8
from __future__ import absolute_import

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

import logging
import os

import octoprint.plugin
import octoprint.util

from octoprint.events import eventManager, Events
from octoprint.settings import settings

from .destinations import FileDestinations
from .analysis import QueueEntry, AnalysisQueue
from .storage import LocalFileStorage
from .util import AbstractFileWrapper, StreamWrapper, DiskFileWrapper

from collections import namedtuple

ContentTypeMapping = namedtuple("ContentTypeMapping", "extensions, content_type")
ContentTypeDetector = namedtuple("ContentTypeDetector", "extensions, detector")

extensions = dict(
)

def full_extension_tree():
	result = dict(
		# extensions for 3d model files
		model=dict(
			stl=ContentTypeMapping(["stl"], "application/sla")
		),
		# extensions for printable machine code
		machinecode=dict(
			gcode=ContentTypeMapping(["gcode", "gco", "g"], "text/plain")
		)
	)

	extension_tree_hooks = octoprint.plugin.plugin_manager().get_hooks("octoprint.filemanager.extension_tree")
	for name, hook in extension_tree_hooks.items():
		try:
			hook_result = hook()
			if hook_result is None or not isinstance(hook_result, dict):
				continue
			result = octoprint.util.dict_merge(result, hook_result)
		except:
			logging.getLogger(__name__).exception("Exception while retrieving additional extension tree entries from hook {name}".format(name=name))

	return result

def get_extensions(type, subtree=None):
	if not subtree:
		subtree = full_extension_tree()

	for key, value in subtree.items():
		if key == type:
			return get_all_extensions(subtree=value)
		elif isinstance(value, dict):
			sub_extensions = get_extensions(type, subtree=value)
			if sub_extensions:
				return sub_extensions

	return None

def get_all_extensions(subtree=None):
	if not subtree:
		subtree = full_extension_tree()

	result = []
	if isinstance(subtree, dict):
		for key, value in subtree.items():
			if isinstance(value, dict):
				result += get_all_extensions(value)
			elif isinstance(value, (ContentTypeMapping, ContentTypeDetector)):
				result += value.extensions
			elif isinstance(value, (list, tuple)):
				result += value
	elif isinstance(subtree, (ContentTypeMapping, ContentTypeDetector)):
		result = subtree.extensions
	elif isinstance(subtree, (list, tuple)):
		result = subtree
	return result

def get_path_for_extension(extension, subtree=None):
	if not subtree:
		subtree = full_extension_tree()

	for key, value in subtree.items():
		if isinstance(value, (ContentTypeMapping, ContentTypeDetector)) and extension in value.extensions:
			return [key]
		elif isinstance(value, (list, tuple)) and extension in value:
			return [key]
		elif isinstance(value, dict):
			path = get_path_for_extension(extension, subtree=value)
			if path:
				return [key] + path

	return None

def get_content_type_mapping_for_extension(extension, subtree=None):
	if not subtree:
		subtree = full_extension_tree()

	for key, value in subtree.items():
		content_extension_matches = isinstance(value, (ContentTypeMapping, ContentTypeDetector)) and extension in value. extensions
		list_extension_matches = isinstance(value, (list, tuple)) and extension in value

		if content_extension_matches or list_extension_matches:
			return value
		elif isinstance(value, dict):
			result = get_content_type_mapping_for_extension(extension, subtree=value)
			if result is not None:
				return result

	return None

def valid_extension(extension, type=None):
	if not type:
		return extension in get_all_extensions()
	else:
		extensions = get_extensions(type)
		if extensions:
			return extension in extensions

def valid_file_type(filename, type=None):
	_, extension = os.path.splitext(filename)
	extension = extension[1:].lower()
	return valid_extension(extension, type=type)

def get_file_type(filename):
	_, extension = os.path.splitext(filename)
	extension = extension[1:].lower()
	return get_path_for_extension(extension)

def get_mime_type(filename):
	_, extension = os.path.splitext(filename)
	extension = extension[1:].lower()
	mapping = get_content_type_mapping_for_extension(extension)
	if mapping:
		if isinstance(mapping, ContentTypeMapping) and mapping.content_type is not None:
			return mapping.content_type
		elif isinstance(mapping, ContentTypeDetector) and callable(mapping.detector):
			result = mapping.detector(filename)
			if result is not None:
				return result
	return "application/octet-stream"


class NoSuchStorage(Exception):
	pass


class FileManager(object):
	def __init__(self, analysis_queue, slicing_manager, printer_profile_manager, initial_storage_managers=None):
		self._logger = logging.getLogger(__name__)
		self._analysis_queue = analysis_queue
		self._analysis_queue.register_finish_callback(self._on_analysis_finished)

		self._storage_managers = dict()
		if initial_storage_managers:
			self._storage_managers.update(initial_storage_managers)

		self._slicing_manager = slicing_manager
		self._printer_profile_manager = printer_profile_manager

		import threading
		self._slicing_jobs = dict()
		self._slicing_jobs_mutex = threading.Lock()

		self._slicing_progress_callbacks = []
		self._last_slicing_progress = None

		self._progress_plugins = []
		self._preprocessor_hooks = dict()

		import octoprint.settings
		self._recovery_file = os.path.join(octoprint.settings.settings().getBaseFolder("data"), "print_recovery_data.yaml")

	def initialize(self):
		self.reload_plugins()

		def worker():
			self._logger.info("Adding backlog items from all storage types to analysis queue...".format(**locals()))
			for storage_type, storage_manager in self._storage_managers.items():
				self._determine_analysis_backlog(storage_type, storage_manager)

		import threading
		thread = threading.Thread(target=worker)
		thread.daemon = True
		thread.start()

	def reload_plugins(self):
		self._progress_plugins = octoprint.plugin.plugin_manager().get_implementations(octoprint.plugin.ProgressPlugin)
		self._preprocessor_hooks = octoprint.plugin.plugin_manager().get_hooks("octoprint.filemanager.preprocessor")

	def register_slicingprogress_callback(self, callback):
		self._slicing_progress_callbacks.append(callback)

	def unregister_slicingprogress_callback(self, callback):
		try:
			self._slicing_progress_callbacks.remove(callback)
		except ValueError:
			# callback was not registered
			pass

	def _determine_analysis_backlog(self, storage_type, storage_manager):
		counter = 0
		for entry, path, printer_profile in storage_manager.analysis_backlog:
			file_type = get_file_type(path)[-1]

			# we'll use the default printer profile for the backlog since we don't know better
			queue_entry = QueueEntry(entry, file_type, storage_type, path, self._printer_profile_manager.get_default())
			if self._analysis_queue.enqueue(queue_entry, high_priority=False):
				counter += 1
		self._logger.info("Added {counter} items from storage type \"{storage_type}\" to analysis queue".format(**locals()))

	def add_storage(self, storage_type, storage_manager):
		self._storage_managers[storage_type] = storage_manager
		self._determine_analysis_backlog(storage_type, storage_manager)

	def remove_storage(self, type):
		if not type in self._storage_managers:
			return
		del self._storage_managers[type]

	@property
	def slicing_enabled(self):
		return self._slicing_manager.slicing_enabled

	@property
	def registered_slicers(self):
		return self._slicing_manager.registered_slicers

	@property
	def default_slicer(self):
		return self._slicing_manager.default_slicer

	def slice(self, slicer_name, source_location, source_path, dest_location, dest_path, resolution=None, nozzle_size=None,
	          position=None, profile=None, printer_profile_id=None, overrides=None, callback=None, callback_args=None):
		absolute_source_path = self.path_on_disk(source_location, source_path)

		def stlProcessed(source_location, source_path, tmp_path, dest_location, dest_path, start_time, printer_profile_id, callback, callback_args, _error=None, _cancelled=False, _analysis=None):
			try:
				if _error:
					eventManager().fire(Events.SLICING_FAILED, {"stl": source_path, "gcode": dest_path, "reason": _error})
				elif _cancelled:
					eventManager().fire(Events.SLICING_CANCELLED, {"stl": source_path, "gcode": dest_path})
				else:
					source_meta = self.get_metadata(source_location, source_path)
					hash = source_meta["hash"]

					import io
					links = [("model", dict(name=source_path))]
					_, stl_name = self.split_path(source_location, source_path)
					file_obj = StreamWrapper(os.path.basename(dest_path),
					                         io.BytesIO(u";Generated from {stl_name} {hash}\n".format(**locals()).encode("ascii", "replace")),
					                         io.FileIO(tmp_path, "rb"))

					printer_profile = self._printer_profile_manager.get(printer_profile_id)

					self.add_file(dest_location, dest_path, file_obj, links=links, allow_overwrite=True, printer_profile=printer_profile, analysis=_analysis)

					# If the BVC analyser was defined, overrides the Cura time estimation
					analyser = settings().get(['gcodeAnalysis', 'analyser'])
					try:
						if analyser is not None and analyser == 'BVC':
							import octoprint.util.bvc_gcoder as bvcGcoder
							gcoder_result = bvcGcoder.analyse(self.path_on_disk(dest_location, dest_path))
							if 'estimated_duration' in gcoder_result:
								_analysis["estimatedPrintTime"] = gcoder_result['estimated_duration']
								self._add_analysis_result(dest_location, dest_path, _analysis)
					except Exception as ex:
						self._logger.error(ex)

					end_time = time.time()
					eventManager().fire(Events.SLICING_DONE, {"stl": source_path, "gcode": dest_path, "time": end_time - start_time})

					if callback is not None:
						if callback_args is None:
							callback_args = ()
						callback(*callback_args)
			finally:
				os.remove(tmp_path)

				source_job_key = (source_location, source_path)
				dest_job_key = (dest_location, dest_path)

				with self._slicing_jobs_mutex:
					if source_job_key in self._slicing_jobs:
						del self._slicing_jobs[source_job_key]
					if dest_job_key in self._slicing_jobs:
						del self._slicing_jobs[dest_job_key]

		slicer = self._slicing_manager.get_slicer(slicer_name)

{'fraction_non_alphanumeric': 0.09415073652095152, 'fraction_numerical': 0.0015254802879344042, 'mean_word_length': 5.356969696969697, 'pattern_counts': {'":': 16, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, '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': '5815620', 'n_tokens_mistral': 7240, 'n_tokens_neox': 6437, 'n_words': 1463}