hex_writer.py

"""
Hex Viewer.

Licensed under MIT
Copyright (c) 2011-2020 Isaac Muse <isaacmuse@gmail.com>
"""
import sublime
import sublime_plugin
from os.path import dirname, exists
from . import hex_common as common
from .hex_checksum import Checksum, parse_view_data
import threading
import traceback
from io import StringIO
from .hex_notify import notify, error

USE_CHECKSUM_ON_SAVE = True
WRITE_GOOD = 0
WRITE_FAIL = 1

active_thread = None


class ThreadedWrite(threading.Thread):
    """Threaded write."""

    def __init__(self, data, file_name, fmt_callback=None, count=None):
        """Initialize."""

        self.data = data
        self.file_name = file_name
        self.chunk = 0
        self.chunks = len(data) if count is None else count
        self.abort = False
        self.fmt_callback = fmt_callback if fmt_callback is not None else self.format
        self.status = WRITE_GOOD
        threading.Thread.__init__(self)

    def format(self, data):  # noqa: A003
        """Format."""

        for x in data:
            yield x

    def run(self):
        """Run command."""

        try:
            with open(self.file_name, "wb") as f:
                for chunk in self.fmt_callback(self.data):
                    self.chunk += 1
                    if self.abort:
                        return
                    else:
                        f.write(chunk)
        except Exception:
            self.status = WRITE_FAIL
            print(str(traceback.format_exc()))


class HexWriterAbortCommand(sublime_plugin.WindowCommand):
    """Command to abort a write operation."""

    def run(self):
        """Run command."""

        if active_thread is not None and active_thread.is_alive():
            active_thread.abort = True

    def is_enabled(self):
        """Check if command is enabled."""

        return active_thread is not None and active_thread.is_alive()


class HexWriterCommand(sublime_plugin.WindowCommand):
    """Export hex view data to a file."""

    export_path = ""
    handshake = -1

    def is_enabled(self):
        """Check if command is enabled."""

        view = self.window.active_view()
        return (
            common.is_enabled() and
            view is not None and not view.settings().get("hex_viewer_fake", False) and
            not (active_thread is not None and active_thread.is_alive())
        )

    def export_panel(self):
        """Show the export panel."""

        self.window.show_input_panel(
            "Export To:",
            self.export_path,
            self.prepare_export,
            None,
            self.reset
        )

    def overwrite(self, value):
        """Handle the overwrite response."""

        if value.strip().lower() == "yes":
            self.export()
        else:
            self.export_path = self.view.settings().get("hex_viewer_file_name")
            self.export_panel()

    def prepare_export(self, file_path):
        """Prepare to export."""

        self.export_path = file_path
        if exists(dirname(file_path)):
            if exists(file_path):
                self.window.show_input_panel(
                    "Overwrite File? (yes | no):",
                    "no",
                    self.overwrite,
                    None,
                    self.reset
                )
            else:
                self.export()
        else:
            error("Directory does not exist!")
            self.export_path = self.view.settings().get("hex_viewer_file_name")
            self.export_panel()

    def reset_thread(self):
        """Rest the thread."""

        self.thread = None

    def finish_export(self):
        """Post export event."""

        if common.hv_settings("checksum_on_save", USE_CHECKSUM_ON_SAVE):
            hex_hash = Checksum()
            self.hex_buffer.seek(0)
            # Checksum will be threaded and will show the result when done
            sublime.set_timeout(lambda: sublime.status_message("Checksumming..."), 0)
            hex_hash.threaded_update(self.hex_buffer, parse_view_data, self.row)

        # Update the tab name
        self.view.set_name(common.basename(self.export_path) + ".hxv")
        # Update the internal path
        self.view.settings().set("hex_viewer_file_name", self.export_path)
        # Tie it to a real view if not already
        self.view.settings().set("hex_viewer_fake", False)
        # Clear the marked edits
        common.clear_edits(self.view)
        # Reset class
        self.reset()

    def export_thread(self):
        """Thread the export."""

        ratio = float(self.thread.chunk) / float(self.thread.chunks)
        percent = int(ratio * 10)
        leftover = 10 - percent
        message = "[" + "-" * percent + ">" + "-" * leftover + ("] %3d%%" % int(ratio * 100)) + " chunks written"
        sublime.status_message(message)
        if not self.thread.is_alive():
            if self.thread.abort is True:
                notify("Write aborted!")
                sublime.set_timeout(self.reset_thread, 500)
            else:
                status = self.thread.status
                self.reset_thread()
                if status == WRITE_GOOD:
                    sublime.set_timeout(self.finish_export, 500)
                else:
                    error("Failed to export to " + self.export_path)
        else:
            sublime.set_timeout(self.export_thread, 500)

    def export(self):
        """Export the data."""

        global active_thread
        self.view = self.window.active_view()
        if self.handshake != -1 and self.handshake == self.view.id():
            try:
                sublime.set_timeout(lambda: sublime.status_message("Writing..."), 0)
                self.row = self.view.rowcol(self.view.size())[0] + 1
                self.hex_buffer = StringIO(self.view.substr(sublime.Region(0, self.view.size())))
                self.thread = ThreadedWrite(self.hex_buffer, self.export_path, parse_view_data, self.row)
                self.thread.start()
                self.export_thread()
                active_thread = self.thread
            except Exception:
                print(str(traceback.format_exc()))
                error("Failed to export to " + self.export_path)
                self.reset()
                return

        else:
            error("Hex view is no longer in focus! File not saved.")
            self.reset()

    def reset(self):
        """Reset."""

        self.export_path = ""
        self.handshake = -1
        self.reset_thread()

    def run(self):
        """Run command."""

        if active_thread is not None and active_thread.is_alive():
            error("HexViewer is already exporting a file!\nPlease run the abort command to stop the current export.")
        else:
            self.view = self.window.active_view()

            # Identify view
            if self.handshake != -1 and self.handshake == self.view.id():
                self.reset()
            self.handshake = self.view.id()

            self.export_path = self.view.settings().get("hex_viewer_file_name")

            self.export_panel()