import sys
import bisect
import types

from _pydev_bundle._pydev_saved_modules import threading
from _pydevd_bundle import pydevd_utils, pydevd_source_mapping
from _pydevd_bundle.pydevd_additional_thread_info import set_additional_thread_info
from _pydevd_bundle.pydevd_comm import (
    InternalGetThreadStack,
    internal_get_completions,
    InternalSetNextStatementThread,
    internal_reload_code,
    InternalGetVariable,
    InternalGetArray,
    InternalLoadFullValue,
    internal_get_description,
    internal_get_frame,
    internal_evaluate_expression,
    InternalConsoleExec,
    internal_get_variable_json,
    internal_change_variable,
    internal_change_variable_json,
    internal_evaluate_expression_json,
    internal_set_expression_json,
    internal_get_exception_details_json,
    internal_step_in_thread,
    internal_smart_step_into,
)
from _pydevd_bundle.pydevd_comm_constants import (
    CMD_THREAD_SUSPEND,
    file_system_encoding,
    CMD_STEP_INTO_MY_CODE,
    CMD_STOP_ON_START,
    CMD_SMART_STEP_INTO,
)
from _pydevd_bundle.pydevd_constants import (
    get_current_thread_id,
    set_protocol,
    get_protocol,
    HTTP_JSON_PROTOCOL,
    JSON_PROTOCOL,
    DebugInfoHolder,
    IS_WINDOWS,
    PYDEVD_USE_SYS_MONITORING,
)
from _pydevd_bundle.pydevd_net_command_factory_json import NetCommandFactoryJson
from _pydevd_bundle.pydevd_net_command_factory_xml import NetCommandFactory
import pydevd_file_utils
from _pydev_bundle import pydev_log
from _pydevd_bundle.pydevd_breakpoints import LineBreakpoint
from pydevd_tracing import get_exception_traceback_str
import os
import subprocess
import ctypes
from _pydevd_bundle.pydevd_collect_bytecode_info import code_to_bytecode_representation
import itertools
import linecache
from _pydevd_bundle.pydevd_utils import DAPGrouper, interrupt_main_thread
from _pydevd_bundle.pydevd_daemon_thread import run_as_pydevd_daemon_thread
from _pydevd_bundle.pydevd_thread_lifecycle import pydevd_find_thread_by_id, resume_threads
import tokenize
from _pydevd_sys_monitoring import pydevd_sys_monitoring

try:
    import dis
except ImportError:

    def _get_code_lines(code):
        raise NotImplementedError

else:

    def _get_code_lines(code):
        if not isinstance(code, types.CodeType):
            path = code
            with tokenize.open(path) as f:
                src = f.read()
            code = compile(src, path, "exec", 0, dont_inherit=True)
            return _get_code_lines(code)

        def iterate():
            # First, get all line starts for this code object. This does not include
            # bodies of nested class and function definitions, as they have their
            # own objects.
            for _, lineno in dis.findlinestarts(code):
                if lineno is not None:
                    yield lineno

            # For nested class and function definitions, their respective code objects
            # are constants referenced by this object.
            for const in code.co_consts:
                if isinstance(const, types.CodeType) and const.co_filename == code.co_filename:
                    for lineno in _get_code_lines(const):
                        yield lineno

        return iterate()


class PyDevdAPI(object):
    class VariablePresentation(object):
        def __init__(self, special="group", function="group", class_="group", protected="inline"):
            self._presentation = {
                DAPGrouper.SCOPE_SPECIAL_VARS: special,
                DAPGrouper.SCOPE_FUNCTION_VARS: function,
                DAPGrouper.SCOPE_CLASS_VARS: class_,
                DAPGrouper.SCOPE_PROTECTED_VARS: protected,
            }

        def get_presentation(self, scope):
            return self._presentation[scope]

    def run(self, py_db):
        py_db.ready_to_run = True

    def notify_initialize(self, py_db):
        py_db.on_initialize()

    def notify_configuration_done(self, py_db):
        py_db.on_configuration_done()

    def notify_disconnect(self, py_db):
        py_db.on_disconnect()

    def set_protocol(self, py_db, seq, protocol):
        set_protocol(protocol.strip())
        if get_protocol() in (HTTP_JSON_PROTOCOL, JSON_PROTOCOL):
            cmd_factory_class = NetCommandFactoryJson
        else:
            cmd_factory_class = NetCommandFactory

        if not isinstance(py_db.cmd_factory, cmd_factory_class):
            py_db.cmd_factory = cmd_factory_class()

        return py_db.cmd_factory.make_protocol_set_message(seq)

    def set_ide_os_and_breakpoints_by(self, py_db, seq, ide_os, breakpoints_by):
        """
        :param ide_os: 'WINDOWS' or 'UNIX'
        :param breakpoints_by: 'ID' or 'LINE'
        """
        if breakpoints_by == "ID":
            py_db._set_breakpoints_with_id = True
        else:
            py_db._set_breakpoints_with_id = False

        self.set_ide_os(ide_os)

        return py_db.cmd_factory.make_version_message(seq)

    def set_ide_os(self, ide_os):
        """
        :param ide_os: 'WINDOWS' or 'UNIX'
        """
        pydevd_file_utils.set_ide_os(ide_os)

    def set_gui_event_loop(self, py_db, gui_event_loop):
        py_db._gui_event_loop = gui_event_loop

    def send_error_message(self, py_db, msg):
        cmd = py_db.cmd_factory.make_warning_message("pydevd: %s\n" % (msg,))
        py_db.writer.add_command(cmd)

    def set_show_return_values(self, py_db, show_return_values):
        if show_return_values:
            py_db.show_return_values = True
        else:
            if py_db.show_return_values:
                # We should remove saved return values
                py_db.remove_return_values_flag = True
            py_db.show_return_values = False
        pydev_log.debug("Show return values: %s", py_db.show_return_values)

    def list_threads(self, py_db, seq):
        # Response is the command with the list of threads to be added to the writer thread.
        return py_db.cmd_factory.make_list_threads_message(py_db, seq)

    def request_suspend_thread(self, py_db, thread_id="*"):
        # Yes, thread suspend is done at this point, not through an internal command.
        threads = []
        suspend_all = thread_id.strip() == "*"
        if suspend_all:
            threads = pydevd_utils.get_non_pydevd_threads()

        elif thread_id.startswith("__frame__:"):
            sys.stderr.write("Can't suspend tasklet: %s\n" % (thread_id,))

        else:
            threads = [pydevd_find_thread_by_id(thread_id)]

        for t in threads:
            if t is None:
                continue
            py_db.set_suspend(
                t,
                CMD_THREAD_SUSPEND,
                suspend_other_threads=suspend_all,
                is_pause=True,
            )
            # Break here (even if it's suspend all) as py_db.set_suspend will
            # take care of suspending other threads.
            break

    def set_enable_thread_notifications(self, py_db, enable):
        """
        When disabled, no thread notifications (for creation/removal) will be
        issued until it's re-enabled.

        Note that when it's re-enabled, a creation notification will be sent for
        all existing threads even if it was previously sent (this is meant to
        be used on disconnect/reconnect).
        """
        py_db.set_enable_thread_notifications(enable)

    def request_disconnect(self, py_db, resume_threads):
        self.set_enable_thread_notifications(py_db, False)
        self.remove_all_breakpoints(py_db, "*")
        self.remove_all_exception_breakpoints(py_db)
        self.notify_disconnect(py_db)

        if resume_threads:
            self.request_resume_thread(thread_id="*")

    def request_resume_thread(self, thread_id):
        resume_threads(thread_id)

    def request_completions(self, py_db, seq, thread_id, frame_id, act_tok, line=-1, column=-1):
        py_db.post_method_as_internal_command(
            thread_id, internal_get_completions, seq, thread_id, frame_id, act_tok, line=line, column=column
        )

    def request_stack(self, py_db, seq, thread_id, fmt=None, timeout=0.5, start_frame=0, levels=0):
        # If it's already suspended, get it right away.
        internal_get_thread_stack = InternalGetThreadStack(
            seq, thread_id, py_db, set_additional_thread_info, fmt=fmt, timeout=timeout, start_frame=start_frame, levels=levels
        )
        if internal_get_thread_stack.can_be_executed_by(get_current_thread_id(threading.current_thread())):
            internal_get_thread_stack.do_it(py_db)
        else:
            py_db.post_internal_command(internal_get_thread_stack, "*")

    def request_exception_info_json(self, py_db, request, thread_id, thread, max_frames):
        py_db.post_method_as_internal_command(
            thread_id,
            internal_get_exception_details_json,
            request,
            thread_id,
            thread,
            max_frames,
            set_additional_thread_info=set_additional_thread_info,
            iter_visible_frames_info=py_db.cmd_factory._iter_visible_frames_info,
        )

    def request_step(self, py_db, thread_id, step_cmd_id):
        t = pydevd_find_thread_by_id(thread_id)
        if t:
            py_db.post_method_as_internal_command(
                thread_id,
                internal_step_in_thread,
                thread_id,
                step_cmd_id,
                set_additional_thread_info=set_additional_thread_info,
            )
        elif thread_id.startswith("__frame__:"):
            sys.stderr.write("Can't make tasklet step command: %s\n" % (thread_id,))

    def request_smart_step_into(self, py_db, seq, thread_id, offset, child_offset):
        t = pydevd_find_thread_by_id(thread_id)
        if t:
            py_db.post_method_as_internal_command(
                thread_id, internal_smart_step_into, thread_id, offset, child_offset, set_additional_thread_info=set_additional_thread_info
            )
        elif thread_id.startswith("__frame__:"):
            sys.stderr.write("Can't set next statement in tasklet: %s\n" % (thread_id,))

    def request_smart_step_into_by_func_name(self, py_db, seq, thread_id, line, func_name):
        # Same thing as set next, just with a different cmd id.
        self.request_set_next(py_db, seq, thread_id, CMD_SMART_STEP_INTO, None, line, func_name)

    def request_set_next(self, py_db, seq, thread_id, set_next_cmd_id, original_filename, line, func_name):
        """
        set_next_cmd_id may actually be one of:

        CMD_RUN_TO_LINE
        CMD_SET_NEXT_STATEMENT

        CMD_SMART_STEP_INTO -- note: request_smart_step_into is preferred if it's possible
                               to work with bytecode offset.

        :param Optional[str] original_filename:
            If available, the filename may be source translated, otherwise no translation will take
            place (the set next just needs the line afterwards as it executes locally, but for
            the Jupyter integration, the source mapping may change the actual lines and not only
            the filename).
        """
        t = pydevd_find_thread_by_id(thread_id)
        if t:
            if original_filename is not None:
                translated_filename = self.filename_to_server(original_filename)  # Apply user path mapping.
                pydev_log.debug("Set next (after path translation) in: %s line: %s", translated_filename, line)
                func_name = self.to_str(func_name)

                assert translated_filename.__class__ == str  # i.e.: bytes on py2 and str on py3
                assert func_name.__class__ == str  # i.e.: bytes on py2 and str on py3

                # Apply source mapping (i.e.: ipython).
                _source_mapped_filename, new_line, multi_mapping_applied = py_db.source_mapping.map_to_server(translated_filename, line)
                if multi_mapping_applied:
                    pydev_log.debug("Set next (after source mapping) in: %s line: %s", translated_filename, line)
                    line = new_line

            int_cmd = InternalSetNextStatementThread(thread_id, set_next_cmd_id, line, func_name, seq=seq)
            py_db.post_internal_command(int_cmd, thread_id)
        elif thread_id.startswith("__frame__:"):
            sys.stderr.write("Can't set next statement in tasklet: %s\n" % (thread_id,))

    def request_reload_code(self, py_db, seq, module_name, filename):
        """
        :param seq: if -1 no message will be sent back when the reload is done.

        Note: either module_name or filename may be None (but not both at the same time).
        """
        thread_id = "*"  # Any thread
        # Note: not going for the main thread because in this case it'd only do the load
        # when we stopped on a breakpoint.
        py_db.post_method_as_internal_command(thread_id, internal_reload_code, seq, module_name, filename)

    def request_change_variable(self, py_db, seq, thread_id, frame_id, scope, attr, value):
        """
        :param scope: 'FRAME' or 'GLOBAL'
        """
        py_db.post_method_as_internal_command(thread_id, internal_change_variable, seq, thread_id, frame_id, scope, attr, value)

    def request_get_variable(self, py_db, seq, thread_id, frame_id, scope, attrs):
        """
        :param scope: 'FRAME' or 'GLOBAL'
        """
        int_cmd = InternalGetVariable(seq, thread_id, frame_id, scope, attrs)
        py_db.post_internal_command(int_cmd, thread_id)

    def request_get_array(self, py_db, seq, roffset, coffset, rows, cols, fmt, thread_id, frame_id, scope, attrs):
        int_cmd = InternalGetArray(seq, roffset, coffset, rows, cols, fmt, thread_id, frame_id, scope, attrs)
        py_db.post_internal_command(int_cmd, thread_id)

    def request_load_full_value(self, py_db, seq, thread_id, frame_id, vars):
        int_cmd = InternalLoadFullValue(seq, thread_id, frame_id, vars)
        py_db.post_internal_command(int_cmd, thread_id)

    def request_get_description(self, py_db, seq, thread_id, frame_id, expression):
        py_db.post_method_as_internal_command(thread_id, internal_get_description, seq, thread_id, frame_id, expression)

    def request_get_frame(self, py_db, seq, thread_id, frame_id):
        py_db.post_method_as_internal_command(thread_id, internal_get_frame, seq, thread_id, frame_id)

    def to_str(self, s):
        """
        -- in py3 raises an error if it's not str already.
        """
        if s.__class__ != str:
            raise AssertionError("Expected to have str on Python 3. Found: %s (%s)" % (s, s.__class__))
        return s

    def filename_to_str(self, filename):
        """
        -- in py3 raises an error if it's not str already.
        """
        if filename.__class__ != str:
            raise AssertionError("Expected to have str on Python 3. Found: %s (%s)" % (filename, filename.__class__))
        return filename

    def filename_to_server(self, filename):
        filename = self.filename_to_str(filename)
        filename = pydevd_file_utils.map_file_to_server(filename)
        return filename

    class _DummyFrame(object):
        """
        Dummy frame to be used with PyDB.apply_files_filter (as we don't really have the
        related frame as breakpoints are added before execution).
        """

        class _DummyCode(object):
            def __init__(self, filename):
                self.co_firstlineno = 1
                self.co_filename = filename
                self.co_name = "invalid func name "

        def __init__(self, filename):
            self.f_code = self._DummyCode(filename)
            self.f_globals = {}

    ADD_BREAKPOINT_NO_ERROR = 0
    ADD_BREAKPOINT_FILE_NOT_FOUND = 1
    ADD_BREAKPOINT_FILE_EXCLUDED_BY_FILTERS = 2

    # This means that the breakpoint couldn't be fully validated (more runtime
    # information may be needed).
    ADD_BREAKPOINT_LAZY_VALIDATION = 3
    ADD_BREAKPOINT_INVALID_LINE = 4

    class _AddBreakpointResult(object):
        # :see: ADD_BREAKPOINT_NO_ERROR = 0
        # :see: ADD_BREAKPOINT_FILE_NOT_FOUND = 1
        # :see: ADD_BREAKPOINT_FILE_EXCLUDED_BY_FILTERS = 2
        # :see: ADD_BREAKPOINT_LAZY_VALIDATION = 3
        # :see: ADD_BREAKPOINT_INVALID_LINE = 4

        __slots__ = ["error_code", "breakpoint_id", "translated_filename", "translated_line", "original_line"]

        def __init__(self, breakpoint_id, translated_filename, translated_line, original_line):
            self.error_code = PyDevdAPI.ADD_BREAKPOINT_NO_ERROR
            self.breakpoint_id = breakpoint_id
            self.translated_filename = translated_filename
            self.translated_line = translated_line
            self.original_line = original_line

    def add_breakpoint(
        self,
        py_db,
        original_filename,
        breakpoint_type,
        breakpoint_id,
        line,
        condition,
        func_name,
        expression,
        suspend_policy,
        hit_condition,
        is_logpoint,
        adjust_line=False,
        on_changed_breakpoint_state=None,
    ):
        """
        :param str original_filename:
            Note: must be sent as it was received in the protocol. It may be translated in this
            function and its final value will be available in the returned _AddBreakpointResult.

        :param str breakpoint_type:
            One of: 'python-line', 'django-line', 'jinja2-line'.

        :param int breakpoint_id:

        :param int line:
            Note: it's possible that a new line was actually used. If that's the case its
            final value will be available in the returned _AddBreakpointResult.

        :param condition:
            Either None or the condition to activate the breakpoint.

        :param str func_name:
            If "None" (str), may hit in any context.
            Empty string will hit only top level.
            Any other value must match the scope of the method to be matched.

        :param str expression:
            None or the expression to be evaluated.

        :param suspend_policy:
            Either "NONE" (to suspend only the current thread when the breakpoint is hit) or
            "ALL" (to suspend all threads when a breakpoint is hit).

        :param str hit_condition:
            An expression where `@HIT@` will be replaced by the number of hits.
            i.e.: `@HIT@ == x` or `@HIT@ >= x`

        :param bool is_logpoint:
            If True and an expression is passed, pydevd will create an io message command with the
            result of the evaluation.

        :param bool adjust_line:
            If True, the breakpoint line should be adjusted if the current line doesn't really
            match an executable line (if possible).

        :param callable on_changed_breakpoint_state:
            This is called when something changed internally on the breakpoint after it was initially
            added (for instance, template file_to_line_to_breakpoints could be signaled as invalid initially and later
            when the related template is loaded, if the line is valid it could be marked as valid).

            The signature for the callback should be:
                on_changed_breakpoint_state(breakpoint_id: int, add_breakpoint_result: _AddBreakpointResult)

                Note that the add_breakpoint_result should not be modified by the callback (the
                implementation may internally reuse the same instance multiple times).

        :return _AddBreakpointResult:
        """
        assert original_filename.__class__ == str, "Expected str, found: %s" % (
            original_filename.__class__,
        )  # i.e.: bytes on py2 and str on py3

        original_filename_normalized = pydevd_file_utils.normcase_from_client(original_filename)

        pydev_log.debug("Request for breakpoint in: %s line: %s", original_filename, line)
        original_line = line
        # Parameters to reapply breakpoint.
        api_add_breakpoint_params = (
            original_filename,
            breakpoint_type,
            breakpoint_id,
            line,
            condition,
            func_name,
            expression,
            suspend_policy,
            hit_condition,
            is_logpoint,
        )

        translated_filename = self.filename_to_server(original_filename)  # Apply user path mapping.
        pydev_log.debug("Breakpoint (after path translation) in: %s line: %s", translated_filename, line)
        func_name = self.to_str(func_name)

        assert translated_filename.__class__ == str  # i.e.: bytes on py2 and str on py3
        assert func_name.__class__ == str  # i.e.: bytes on py2 and str on py3

        # Apply source mapping (i.e.: ipython).
        source_mapped_filename, new_line, multi_mapping_applied = py_db.source_mapping.map_to_server(translated_filename, line)

        if multi_mapping_applied:
            pydev_log.debug("Breakpoint (after source mapping) in: %s line: %s", source_mapped_filename, new_line)
            # Note that source mapping is internal and does not change the resulting filename nor line
            # (we want the outside world to see the line in the original file and not in the ipython
            # cell, otherwise the editor wouldn't be correct as the returned line is the line to
            # which the breakpoint will be moved in the editor).
            result = self._AddBreakpointResult(breakpoint_id, original_filename, line, original_line)

            # If a multi-mapping was applied, consider it the canonical / source mapped version (translated to ipython cell).
            translated_absolute_filename = source_mapped_filename
            canonical_normalized_filename = pydevd_file_utils.normcase(source_mapped_filename)
            line = new_line

        else:
            translated_absolute_filename = pydevd_file_utils.absolute_path(translated_filename)
            canonical_normalized_filename = pydevd_file_utils.canonical_normalized_path(translated_filename)

            if adjust_line and not translated_absolute_filename.startswith("<"):
                # Validate file_to_line_to_breakpoints and adjust their positions.
                try:
                    lines = sorted(_get_code_lines(translated_absolute_filename))
                except Exception:
                    pass
                else:
                    if line not in lines:
                        # Adjust to the first preceding valid line.
                        idx = bisect.bisect_left(lines, line)
                        if idx > 0:
                            line = lines[idx - 1]

            result = self._AddBreakpointResult(breakpoint_id, original_filename, line, original_line)

        py_db.api_received_breakpoints[(original_filename_normalized, breakpoint_id)] = (
            canonical_normalized_filename,
            api_add_breakpoint_params,
        )

        if not translated_absolute_filename.startswith("<"):
            # Note: if a mapping pointed to a file starting with '<', don't validate.

            if not pydevd_file_utils.exists(translated_absolute_filename):
                result.error_code = self.ADD_BREAKPOINT_FILE_NOT_FOUND
                return result

            if (
                py_db.is_files_filter_enabled
                and not py_db.get_require_module_for_filters()
                and py_db.apply_files_filter(self._DummyFrame(translated_absolute_filename), translated_absolute_filename, False)
            ):
                # Note that if `get_require_module_for_filters()` returns False, we don't do this check.
                # This is because we don't have the module name given a file at this point (in
                # runtime it's gotten from the frame.f_globals).
                # An option could be calculate it based on the filename and current sys.path,
                # but on some occasions that may be wrong (for instance with `__main__` or if
                # the user dynamically changes the PYTHONPATH).

                # Note: depending on the use-case, filters may be changed, so, keep on going and add the
                # breakpoint even with the error code.
                result.error_code = self.ADD_BREAKPOINT_FILE_EXCLUDED_BY_FILTERS

        if breakpoint_type == "python-line":
            added_breakpoint = LineBreakpoint(
                breakpoint_id, line, condition, func_name, expression, suspend_policy, hit_condition=hit_condition, is_logpoint=is_logpoint
            )

            file_to_line_to_breakpoints = py_db.breakpoints
            file_to_id_to_breakpoint = py_db.file_to_id_to_line_breakpoint
            supported_type = True

        else:
            add_plugin_breakpoint_result = None
            plugin = py_db.get_plugin_lazy_init()
            if plugin is not None:
                add_plugin_breakpoint_result = plugin.add_breakpoint(
                    "add_line_breakpoint",
                    py_db,
                    breakpoint_type,
                    canonical_normalized_filename,
                    breakpoint_id,
                    line,
                    condition,
                    expression,
                    func_name,
                    hit_condition=hit_condition,
                    is_logpoint=is_logpoint,
                    add_breakpoint_result=result,
                    on_changed_breakpoint_state=on_changed_breakpoint_state,
                )

            if add_plugin_breakpoint_result is not None:
                supported_type = True
                added_breakpoint, file_to_line_to_breakpoints = add_plugin_breakpoint_result
                file_to_id_to_breakpoint = py_db.file_to_id_to_plugin_breakpoint
            else:
                supported_type = False

        if not supported_type:
            raise NameError(breakpoint_type)

        pydev_log.debug("Added breakpoint:%s - line:%s - func_name:%s\n", canonical_normalized_filename, line, func_name)

        if canonical_normalized_filename in file_to_id_to_breakpoint:
            id_to_pybreakpoint = file_to_id_to_breakpoint[canonical_normalized_filename]
        else:
            id_to_pybreakpoint = file_to_id_to_breakpoint[canonical_normalized_filename] = {}

        id_to_pybreakpoint[breakpoint_id] = added_breakpoint
        py_db.consolidate_breakpoints(canonical_normalized_filename, id_to_pybreakpoint, file_to_line_to_breakpoints)
        if py_db.plugin is not None:
            py_db.has_plugin_line_breaks = py_db.plugin.has_line_breaks(py_db)
            py_db.plugin.after_breakpoints_consolidated(
                py_db, canonical_normalized_filename, id_to_pybreakpoint, file_to_line_to_breakpoints
            )

        py_db.on_breakpoints_changed()
        return result

    def reapply_breakpoints(self, py_db):
        """
        Reapplies all the received breakpoints as they were received by the API (so, new
        translations are applied).
        """
        pydev_log.debug("Reapplying breakpoints.")
        values = list(py_db.api_received_breakpoints.values())  # Create a copy with items to reapply.
        self.remove_all_breakpoints(py_db, "*")
        for val in values:
            _new_filename, api_add_breakpoint_params = val
            self.add_breakpoint(py_db, *api_add_breakpoint_params)

    def remove_all_breakpoints(self, py_db, received_filename):
        """
        Removes all the breakpoints from a given file or from all files if received_filename == '*'.

        :param str received_filename:
            Note: must be sent as it was received in the protocol. It may be translated in this
            function.
        """
        assert received_filename.__class__ == str  # i.e.: bytes on py2 and str on py3
        changed = False
        lst = [py_db.file_to_id_to_line_breakpoint, py_db.file_to_id_to_plugin_breakpoint, py_db.breakpoints]
        if hasattr(py_db, "django_breakpoints"):
            lst.append(py_db.django_breakpoints)

        if hasattr(py_db, "jinja2_breakpoints"):
            lst.append(py_db.jinja2_breakpoints)

        if received_filename == "*":
            py_db.api_received_breakpoints.clear()

            for file_to_id_to_breakpoint in lst:
                if file_to_id_to_breakpoint:
                    file_to_id_to_breakpoint.clear()
                    changed = True

        else:
            received_filename_normalized = pydevd_file_utils.normcase_from_client(received_filename)
            items = list(py_db.api_received_breakpoints.items())  # Create a copy to remove items.
            translated_filenames = []
            for key, val in items:
                original_filename_normalized, _breakpoint_id = key
                if original_filename_normalized == received_filename_normalized:
                    canonical_normalized_filename, _api_add_breakpoint_params = val
                    # Note: there can be actually 1:N mappings due to source mapping (i.e.: ipython).
                    translated_filenames.append(canonical_normalized_filename)
                    del py_db.api_received_breakpoints[key]

            for canonical_normalized_filename in translated_filenames:
                for file_to_id_to_breakpoint in lst:
                    if canonical_normalized_filename in file_to_id_to_breakpoint:
                        file_to_id_to_breakpoint.pop(canonical_normalized_filename, None)
                        changed = True

        if changed:
            py_db.on_breakpoints_changed(removed=True)

    def remove_breakpoint(self, py_db, received_filename, breakpoint_type, breakpoint_id):
        """
        :param str received_filename:
            Note: must be sent as it was received in the protocol. It may be translated in this
            function.

        :param str breakpoint_type:
            One of: 'python-line', 'django-line', 'jinja2-line'.

        :param int breakpoint_id:
        """
        received_filename_normalized = pydevd_file_utils.normcase_from_client(received_filename)
        for key, val in list(py_db.api_received_breakpoints.items()):
            original_filename_normalized, existing_breakpoint_id = key
            _new_filename, _api_add_breakpoint_params = val
            if received_filename_normalized == original_filename_normalized and existing_breakpoint_id == breakpoint_id:
                del py_db.api_received_breakpoints[key]
                break
        else:
            pydev_log.info("Did not find breakpoint to remove: %s (breakpoint id: %s)", received_filename, breakpoint_id)

        file_to_id_to_breakpoint = None
        received_filename = self.filename_to_server(received_filename)
        canonical_normalized_filename = pydevd_file_utils.canonical_normalized_path(received_filename)

        if breakpoint_type == "python-line":
            file_to_line_to_breakpoints = py_db.breakpoints
            file_to_id_to_breakpoint = py_db.file_to_id_to_line_breakpoint

        elif py_db.plugin is not None:
            result = py_db.plugin.get_breakpoints(py_db, breakpoint_type)
            if result is not None:
                file_to_id_to_breakpoint = py_db.file_to_id_to_plugin_breakpoint
                file_to_line_to_breakpoints = result

        if file_to_id_to_breakpoint is None:
            pydev_log.critical("Error removing breakpoint. Cannot handle breakpoint of type %s", breakpoint_type)

        else:
            try:
                id_to_pybreakpoint = file_to_id_to_breakpoint.get(canonical_normalized_filename, {})
                if DebugInfoHolder.DEBUG_TRACE_LEVEL >= 1:
                    existing = id_to_pybreakpoint[breakpoint_id]
                    pydev_log.info(
                        "Removed breakpoint:%s - line:%s - func_name:%s (id: %s)\n"
                        % (canonical_normalized_filename, existing.line, existing.func_name, breakpoint_id)
                    )

                del id_to_pybreakpoint[breakpoint_id]
                py_db.consolidate_breakpoints(canonical_normalized_filename, id_to_pybreakpoint, file_to_line_to_breakpoints)
                if py_db.plugin is not None:
                    py_db.has_plugin_line_breaks = py_db.plugin.has_line_breaks(py_db)
                    py_db.plugin.after_breakpoints_consolidated(
                        py_db, canonical_normalized_filename, id_to_pybreakpoint, file_to_line_to_breakpoints
                    )

            except KeyError:
                pydev_log.info(
                    "Error removing breakpoint: Breakpoint id not found: %s id: %s. Available ids: %s\n",
                    canonical_normalized_filename,
                    breakpoint_id,
                    list(id_to_pybreakpoint),
                )

        py_db.on_breakpoints_changed(removed=True)

    def set_function_breakpoints(self, py_db, function_breakpoints):
        function_breakpoint_name_to_breakpoint = {}
        for function_breakpoint in function_breakpoints:
            function_breakpoint_name_to_breakpoint[function_breakpoint.func_name] = function_breakpoint

        py_db.function_breakpoint_name_to_breakpoint = function_breakpoint_name_to_breakpoint
        py_db.on_breakpoints_changed()

    def request_exec_or_evaluate(self, py_db, seq, thread_id, frame_id, expression, is_exec, trim_if_too_big, attr_to_set_result):
        py_db.post_method_as_internal_command(
            thread_id, internal_evaluate_expression, seq, thread_id, frame_id, expression, is_exec, trim_if_too_big, attr_to_set_result
        )

    def request_exec_or_evaluate_json(self, py_db, request, thread_id):
        py_db.post_method_as_internal_command(thread_id, internal_evaluate_expression_json, request, thread_id)

    def request_set_expression_json(self, py_db, request, thread_id):
        py_db.post_method_as_internal_command(thread_id, internal_set_expression_json, request, thread_id)

    def request_console_exec(self, py_db, seq, thread_id, frame_id, expression):
        int_cmd = InternalConsoleExec(seq, thread_id, frame_id, expression)
        py_db.post_internal_command(int_cmd, thread_id)

    def request_load_source(self, py_db, seq, filename):
        """
        :param str filename:
            Note: must be sent as it was received in the protocol. It may be translated in this
            function.
        """
        try:
            filename = self.filename_to_server(filename)
            assert filename.__class__ == str  # i.e.: bytes on py2 and str on py3

            with tokenize.open(filename) as stream:
                source = stream.read()
            cmd = py_db.cmd_factory.make_load_source_message(seq, source)
        except:
            cmd = py_db.cmd_factory.make_error_message(seq, get_exception_traceback_str())

        py_db.writer.add_command(cmd)

    def get_decompiled_source_from_frame_id(self, py_db, frame_id):
        """
        :param py_db:
        :param frame_id:
        :throws Exception:
            If unable to get the frame in the currently paused frames or if some error happened
            when decompiling.
        """
        variable = py_db.suspended_frames_manager.get_variable(int(frame_id))
        frame = variable.value

        # Check if it's in the linecache first.
        lines = (linecache.getline(frame.f_code.co_filename, i) for i in itertools.count(1))
        lines = itertools.takewhile(bool, lines)  # empty lines are '\n', EOF is ''

        source = "".join(lines)
        if not source:
            source = code_to_bytecode_representation(frame.f_code)

        return source

    def request_load_source_from_frame_id(self, py_db, seq, frame_id):
        try:
            source = self.get_decompiled_source_from_frame_id(py_db, frame_id)
            cmd = py_db.cmd_factory.make_load_source_from_frame_id_message(seq, source)
        except:
            cmd = py_db.cmd_factory.make_error_message(seq, get_exception_traceback_str())

        py_db.writer.add_command(cmd)

    def add_python_exception_breakpoint(
        self,
        py_db,
        exception,
        condition,
        expression,
        notify_on_handled_exceptions,
        notify_on_unhandled_exceptions,
        notify_on_user_unhandled_exceptions,
        notify_on_first_raise_only,
        ignore_libraries,
    ):
        exception_breakpoint = py_db.add_break_on_exception(
            exception,
            condition=condition,
            expression=expression,
            notify_on_handled_exceptions=notify_on_handled_exceptions,
            notify_on_unhandled_exceptions=notify_on_unhandled_exceptions,
            notify_on_user_unhandled_exceptions=notify_on_user_unhandled_exceptions,
            notify_on_first_raise_only=notify_on_first_raise_only,
            ignore_libraries=ignore_libraries,
        )

        if exception_breakpoint is not None:
            py_db.on_breakpoints_changed()

    def add_plugins_exception_breakpoint(self, py_db, breakpoint_type, exception):
        supported_type = False
        plugin = py_db.get_plugin_lazy_init()
        if plugin is not None:
            supported_type = plugin.add_breakpoint("add_exception_breakpoint", py_db, breakpoint_type, exception)

        if supported_type:
            py_db.has_plugin_exception_breaks = py_db.plugin.has_exception_breaks(py_db)
            py_db.on_breakpoints_changed()
        else:
            raise NameError(breakpoint_type)

    def remove_python_exception_breakpoint(self, py_db, exception):
        try:
            cp = py_db.break_on_uncaught_exceptions.copy()
            cp.pop(exception, None)
            py_db.break_on_uncaught_exceptions = cp

            cp = py_db.break_on_caught_exceptions.copy()
            cp.pop(exception, None)
            py_db.break_on_caught_exceptions = cp

            cp = py_db.break_on_user_uncaught_exceptions.copy()
            cp.pop(exception, None)
            py_db.break_on_user_uncaught_exceptions = cp
        except:
            pydev_log.exception("Error while removing exception %s", sys.exc_info()[0])

        py_db.on_breakpoints_changed(removed=True)

    def remove_plugins_exception_breakpoint(self, py_db, exception_type, exception):
        # I.e.: no need to initialize lazy (if we didn't have it in the first place, we can't remove
        # anything from it anyways).
        plugin = py_db.plugin
        if plugin is None:
            return

        supported_type = plugin.remove_exception_breakpoint(py_db, exception_type, exception)

        if supported_type:
            py_db.has_plugin_exception_breaks = py_db.plugin.has_exception_breaks(py_db)
        else:
            pydev_log.info("No exception of type: %s was previously registered.", exception_type)

        py_db.on_breakpoints_changed(removed=True)

    def remove_all_exception_breakpoints(self, py_db):
        py_db.break_on_uncaught_exceptions = {}
        py_db.break_on_caught_exceptions = {}
        py_db.break_on_user_uncaught_exceptions = {}

        plugin = py_db.plugin
        if plugin is not None:
            plugin.remove_all_exception_breakpoints(py_db)
        py_db.on_breakpoints_changed(removed=True)

    def set_project_roots(self, py_db, project_roots):
        """
        :param str project_roots:
        """
        py_db.set_project_roots(project_roots)

    def set_stepping_resumes_all_threads(self, py_db, stepping_resumes_all_threads):
        py_db.stepping_resumes_all_threads = stepping_resumes_all_threads

    # Add it to the namespace so that it's available as PyDevdAPI.ExcludeFilter
    from _pydevd_bundle.pydevd_filtering import ExcludeFilter  # noqa

    def set_exclude_filters(self, py_db, exclude_filters):
        """
        :param list(PyDevdAPI.ExcludeFilter) exclude_filters:
        """
        py_db.set_exclude_filters(exclude_filters)

    def set_use_libraries_filter(self, py_db, use_libraries_filter):
        py_db.set_use_libraries_filter(use_libraries_filter)

    def request_get_variable_json(self, py_db, request, thread_id):
        """
        :param VariablesRequest request:
        """
        py_db.post_method_as_internal_command(thread_id, internal_get_variable_json, request)

    def request_change_variable_json(self, py_db, request, thread_id):
        """
        :param SetVariableRequest request:
        """
        py_db.post_method_as_internal_command(thread_id, internal_change_variable_json, request)

    def set_dont_trace_start_end_patterns(self, py_db, start_patterns, end_patterns):
        # Note: start/end patterns normalized internally.
        start_patterns = tuple(pydevd_file_utils.normcase(x) for x in start_patterns)
        end_patterns = tuple(pydevd_file_utils.normcase(x) for x in end_patterns)

        # After it's set the first time, we can still change it, but we need to reset the
        # related caches.
        reset_caches = False
        dont_trace_start_end_patterns_previously_set = py_db.dont_trace_external_files.__name__ == "custom_dont_trace_external_files"

        if not dont_trace_start_end_patterns_previously_set and not start_patterns and not end_patterns:
            # If it wasn't set previously and start and end patterns are empty we don't need to do anything.
            return

        if not py_db.is_cache_file_type_empty():
            # i.e.: custom function set in set_dont_trace_start_end_patterns.
            if dont_trace_start_end_patterns_previously_set:
                reset_caches = (
                    py_db.dont_trace_external_files.start_patterns != start_patterns
                    or py_db.dont_trace_external_files.end_patterns != end_patterns
                )

            else:
                reset_caches = True

        def custom_dont_trace_external_files(abs_path):
            normalized_abs_path = pydevd_file_utils.normcase(abs_path)
            return normalized_abs_path.startswith(start_patterns) or normalized_abs_path.endswith(end_patterns)

        custom_dont_trace_external_files.start_patterns = start_patterns
        custom_dont_trace_external_files.end_patterns = end_patterns
        py_db.dont_trace_external_files = custom_dont_trace_external_files

        if reset_caches:
            py_db.clear_dont_trace_start_end_patterns_caches()

    def stop_on_entry(self):
        main_thread = pydevd_utils.get_main_thread()
        if main_thread is None:
            pydev_log.critical("Could not find main thread while setting Stop on Entry.")
        else:
            info = set_additional_thread_info(main_thread)
            info.pydev_original_step_cmd = CMD_STOP_ON_START
            info.pydev_step_cmd = CMD_STEP_INTO_MY_CODE
            info.update_stepping_info()
            if PYDEVD_USE_SYS_MONITORING:
                pydevd_sys_monitoring.update_monitor_events(suspend_requested=True)
                pydevd_sys_monitoring.restart_events()

    def set_ignore_system_exit_codes(self, py_db, ignore_system_exit_codes):
        py_db.set_ignore_system_exit_codes(ignore_system_exit_codes)

    SourceMappingEntry = pydevd_source_mapping.SourceMappingEntry

    def set_source_mapping(self, py_db, source_filename, mapping):
        """
        :param str source_filename:
            The filename for the source mapping (bytes on py2 and str on py3).
            This filename will be made absolute in this function.

        :param list(SourceMappingEntry) mapping:
            A list with the source mapping entries to be applied to the given filename.

        :return str:
            An error message if it was not possible to set the mapping or an empty string if
            everything is ok.
        """
        source_filename = self.filename_to_server(source_filename)
        absolute_source_filename = pydevd_file_utils.absolute_path(source_filename)
        for map_entry in mapping:
            map_entry.source_filename = absolute_source_filename
        error_msg = py_db.source_mapping.set_source_mapping(absolute_source_filename, mapping)
        if error_msg:
            return error_msg

        self.reapply_breakpoints(py_db)
        return ""

    def set_variable_presentation(self, py_db, variable_presentation):
        assert isinstance(variable_presentation, self.VariablePresentation)
        py_db.variable_presentation = variable_presentation

    def get_ppid(self):
        """
        Provides the parent pid (even for older versions of Python on Windows).
        """
        ppid = None

        try:
            ppid = os.getppid()
        except AttributeError:
            pass

        if ppid is None and IS_WINDOWS:
            ppid = self._get_windows_ppid()

        return ppid

    def _get_windows_ppid(self):
        this_pid = os.getpid()
        for ppid, pid in _list_ppid_and_pid():
            if pid == this_pid:
                return ppid

        return None

    def _terminate_child_processes_windows(self, dont_terminate_child_pids):
        this_pid = os.getpid()
        for _ in range(50):  # Try this at most 50 times before giving up.
            # Note: we can't kill the process itself with taskkill, so, we
            # list immediate children, kill that tree and then exit this process.

            children_pids = []
            for ppid, pid in _list_ppid_and_pid():
                if ppid == this_pid:
                    if pid not in dont_terminate_child_pids:
                        children_pids.append(pid)

            if not children_pids:
                break
            else:
                for pid in children_pids:
                    self._call(["taskkill", "/F", "/PID", str(pid), "/T"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)

                del children_pids[:]

    def _terminate_child_processes_linux_and_mac(self, dont_terminate_child_pids):
        this_pid = os.getpid()

        def list_children_and_stop_forking(initial_pid, stop=True):
            children_pids = []
            if stop:
                # Ask to stop forking (shouldn't be called for this process, only subprocesses).
                self._call(["kill", "-STOP", str(initial_pid)], stdout=subprocess.PIPE, stderr=subprocess.PIPE)

            list_popen = self._popen(["pgrep", "-P", str(initial_pid)], stdout=subprocess.PIPE, stderr=subprocess.PIPE)

            if list_popen is not None:
                stdout, _ = list_popen.communicate()
                for line in stdout.splitlines():
                    line = line.decode("ascii").strip()
                    if line:
                        pid = str(line)
                        if pid in dont_terminate_child_pids:
                            continue
                        children_pids.append(pid)
                        # Recursively get children.
                        children_pids.extend(list_children_and_stop_forking(pid))
            return children_pids

        previously_found = set()

        for _ in range(50):  # Try this at most 50 times before giving up.
            children_pids = list_children_and_stop_forking(this_pid, stop=False)
            found_new = False

            for pid in children_pids:
                if pid not in previously_found:
                    found_new = True
                    previously_found.add(pid)
                    self._call(["kill", "-KILL", str(pid)], stdout=subprocess.PIPE, stderr=subprocess.PIPE)

            if not found_new:
                break

    def _popen(self, cmdline, **kwargs):
        try:
            return subprocess.Popen(cmdline, **kwargs)
        except:
            if DebugInfoHolder.DEBUG_TRACE_LEVEL >= 1:
                pydev_log.exception("Error running: %s" % (" ".join(cmdline)))
            return None

    def _call(self, cmdline, **kwargs):
        try:
            subprocess.check_call(cmdline, **kwargs)
        except:
            if DebugInfoHolder.DEBUG_TRACE_LEVEL >= 1:
                pydev_log.exception("Error running: %s" % (" ".join(cmdline)))

    def set_terminate_child_processes(self, py_db, terminate_child_processes):
        py_db.terminate_child_processes = terminate_child_processes

    def set_terminate_keyboard_interrupt(self, py_db, terminate_keyboard_interrupt):
        py_db.terminate_keyboard_interrupt = terminate_keyboard_interrupt

    def terminate_process(self, py_db):
        """
        Terminates the current process (and child processes if the option to also terminate
        child processes is enabled).
        """
        try:
            if py_db.terminate_child_processes:
                pydev_log.debug("Terminating child processes.")
                if IS_WINDOWS:
                    self._terminate_child_processes_windows(py_db.dont_terminate_child_pids)
                else:
                    self._terminate_child_processes_linux_and_mac(py_db.dont_terminate_child_pids)
        finally:
            pydev_log.debug("Exiting process (os._exit(0)).")
            os._exit(0)

    def _terminate_if_commands_processed(self, py_db):
        py_db.dispose_and_kill_all_pydevd_threads()
        self.terminate_process(py_db)

    def request_terminate_process(self, py_db):
        if py_db.terminate_keyboard_interrupt:
            if not py_db.keyboard_interrupt_requested:
                py_db.keyboard_interrupt_requested = True
                interrupt_main_thread()
                return

        # We mark with a terminate_requested to avoid that paused threads start running
        # (we should terminate as is without letting any paused thread run).
        py_db.terminate_requested = True
        run_as_pydevd_daemon_thread(py_db, self._terminate_if_commands_processed, py_db)

    def setup_auto_reload_watcher(self, py_db, enable_auto_reload, watch_dirs, poll_target_time, exclude_patterns, include_patterns):
        py_db.setup_auto_reload_watcher(enable_auto_reload, watch_dirs, poll_target_time, exclude_patterns, include_patterns)


def _list_ppid_and_pid():
    _TH32CS_SNAPPROCESS = 0x00000002

    class PROCESSENTRY32(ctypes.Structure):
        _fields_ = [
            ("dwSize", ctypes.c_uint32),
            ("cntUsage", ctypes.c_uint32),
            ("th32ProcessID", ctypes.c_uint32),
            ("th32DefaultHeapID", ctypes.c_size_t),
            ("th32ModuleID", ctypes.c_uint32),
            ("cntThreads", ctypes.c_uint32),
            ("th32ParentProcessID", ctypes.c_uint32),
            ("pcPriClassBase", ctypes.c_long),
            ("dwFlags", ctypes.c_uint32),
            ("szExeFile", ctypes.c_char * 260),
        ]

    kernel32 = ctypes.windll.kernel32
    snapshot = kernel32.CreateToolhelp32Snapshot(_TH32CS_SNAPPROCESS, 0)
    ppid_and_pids = []
    try:
        process_entry = PROCESSENTRY32()
        process_entry.dwSize = ctypes.sizeof(PROCESSENTRY32)
        if not kernel32.Process32First(ctypes.c_void_p(snapshot), ctypes.byref(process_entry)):
            pydev_log.critical("Process32First failed (getting process from CreateToolhelp32Snapshot).")
        else:
            while True:
                ppid_and_pids.append((process_entry.th32ParentProcessID, process_entry.th32ProcessID))
                if not kernel32.Process32Next(ctypes.c_void_p(snapshot), ctypes.byref(process_entry)):
                    break
    finally:
        kernel32.CloseHandle(snapshot)

    return ppid_and_pids