<commit_before># -*- coding: utf-8 -*-
"""
Utilities for working with the AnyBody Console applicaiton.

Created on Fri Oct 19 21:14:59 2012
@author: Morten
"""

import os
import io
import sys
import time
import copy
import types
import ctypes
import shelve
import atexit
import pathlib
import logging
import warnings
import collections
from pathlib import Path
from subprocess import Popen, TimeoutExpired
from tempfile import NamedTemporaryFile
from threading import Thread, RLock
from queue import Queue

from typing import Generator, List

import numpy as np
from tqdm.auto import tqdm

from .tools import (
    ON_WINDOWS,
    make_hash,
    AnyPyProcessOutputList,
    parse_anybodycon_output,
    getsubdirs,
    get_anybodycon_path,
    BELOW_NORMAL_PRIORITY_CLASS,
    AnyPyProcessOutput,
    run_from_ipython,
    get_ncpu,
    winepath,
    silentremove,
    case_preserving_replace,
)
from .macroutils import AnyMacro, MacroCommand

logger = logging.getLogger("abt.anypytools")

_thread_lock = RLock()
_KILLED_BY_ANYPYTOOLS = 10
_TIMEDOUT_BY_ANYPYTOOLS = 11
_NO_LICENSES_AVAILABLE = -22
_UNABLE_TO_ACQUIRE_LICENSE = 234  # May indicate wrong password


class _SubProcessContainer(object):
    """Class to hold a record of process pids from Popen.

    Properties
    ----------
    stop_all: boolean
        If set to True all process held by the object will be automatically
        killed

    Methods
    -------
    add(pid):
        Add process id to the record of process

    remove(pid):
        Remove process id from the record

    """

    def __init__(self):
        self._pids = set()
        self._stop_all = False

    def add(self, pid):
        with _thread_lock:
            self._pids.add(pid)
        if self.stop_all:
            self._kill_running_processes()

    def remove(self, pid):
        with _thread_lock:
            try:
                self._pids.remove(pid)
            except KeyError:
                pass

    @property
    def stop_all(self):
        return self._stop_all

    @stop_all.setter
    def stop_all(self, value):
        if value:
            self._stop_all = True
            self._kill_running_processes()
        else:
            self._stop_all = False

    def _kill_running_processes(self):
        """Clean up and shut down any running processes."""
        # Kill any rouge processes that are still running.
        with _thread_lock:
            killed = []
            for pid in self._pids:
                try:
                    os.kill(pid, _KILLED_BY_ANYPYTOOLS)
                    killed.append(str(pid))
                except Exception:
                    pass
            self._pids.clear()


_subprocess_container = _SubProcessContainer()
atexit.register(_subprocess_container._kill_running_processes)


def execute_anybodycon(
    macro,
    logfile=None,
    anybodycon_path=None,
    timeout=3600,
    keep_macrofile=False,
    env=None,
    priority=BELOW_NORMAL_PRIORITY_CLASS,
    debug_mode=0,
    folder=None,
):
    """Launch a single AnyBodyConsole applicaiton.

    This is a low level function to start a AnyBody Console process
    with a given list macros.

    Parameters
    ----------
    macro : list of str
        List of macros strings to pass to the AnyBody Console Application
    logfile : file like object, optional
        An open file like object to write to pipe the output of AnyBody
        into. (Defaults to None, in which case it will use sys.stdout)
    anybodycon_path : str, optional
        Path to the AnyBodyConsole application. Default to None, in which
        case the default installed AnyBody installation will be looked up
        in the Windows registry.
    timeout : int, optional
        Timeout before the process is killed autmotically. Defaults to
        3600 seconds (1 hour).
    keep_macrofile : bool, optional
        Set to True to prevent the temporary macro file from beeing deleted.
        (Defaults to False)
    env: dict
        Environment varaibles which are passed to the started AnyBody console
        application.
    priority : int, optional
        The priority of the subprocesses. This can be on of the following:
        ``anypytools.IDLE_PRIORITY_CLASS``, ``anypytools.BELOW_NORMAL_PRIORITY_CLASS``,
        ``anypytools.NORMAL_PRIORITY_CLASS``, ``anypytools.HIGH_PRIORITY_CLASS``
        Default is BELOW_NORMAL_PRIORITY_CLASS.
    debug_mode : int
        The AMS debug mode to use. Defaults to 0 which is disabled. 1 correspond to
        crashdump enabled
    folder : the folder in which AnyBody is executed

    Returns
    -------
    int
        The return code from the AnyBody Console application.

    """
    if folder is None:
        folder = os.getcwd()

    try:
        macrofile_path = Path(folder).resolve() / (Path(logfile.name).stem + ".anymcr")
    except AttributeError:
        macrofile_path = Path("macrofile.anymcr")

    macrofile_cleanup = [macrofile_path]

    if logfile is None:
        logfile = sys.stdout

    if anybodycon_path is None:
        anybodycon_path = Path(get_anybodycon_path())

    if macro and macro[-1] != "exit":
        macro.append("exit")

    if not os.path.isfile(anybodycon_path):
        raise IOError(f"Can not find anybodycon.exe: {anybodycon_path}")

    with open(macrofile_path, "w+b") as fh:
        fh.write("\n".join(macro).encode("UTF-8"))
        fh.flush()

    if ON_WINDOWS:
        # Don't display the Windows GPF dialog if the invoked program dies.
        # See comp.os.ms-windows.programmer.win32
        # How to suppress crash notification dialog?, Jan 14,2004 -
        # <NAME>'s response [1]
        SEM_NOGPFAULTERRORBOX = 0x0002  # From MSDN
        ctypes.windll.kernel32.SetErrorMode(SEM_NOGPFAULTERRORBOX)
        subprocess_flags = 0x8000000  # win32con.CREATE_NO_WINDOW?
        subprocess_flags |= priority
        extra_kwargs = {"creationflags": subprocess_flags}

        anybodycmd = [
            str(anybodycon_path.resolve()),
            "--macro=",
            str(macrofile_path),
            "/deb",
            str(debug_mode),
            "/ni",
        ]
        proc = Popen(
            anybodycmd,
            stdout=logfile,
            stderr=logfile,
            env=env,
            cwd=folder,
            **extra_kwargs,
        )

    else:
        if os.environ.get("WINE_REDIRECT_OUTPUT", 0):
            cmd = [
                "wine",
                str(anybodycon_path.resolve()),
                "--macro=",
                winepath(macrofile_path, "--windows"),
                "/deb",
                str(debug_mode),
                "/ni",
            ]
            proc = Popen(
                cmd,
                env=env,
                cwd=folder,
                close_fds=False,
                stdout=logfile,
                stderr=logfile,
            )
        else:
            # ON Linux/Wine we use a bat file to redirect the output into a file on wine/windows
            # side. This prevents a bug with AnyBody starts it's builtin python.
            anybodycmd = (
                f'@call "{winepath(anybodycon_path.resolve(), "--windows")}"'
                f' -m "{winepath(macrofile_path, "--windows")}"'
                f" -deb {str(debug_mode)}"
                " -ni"
                f' >> "{winepath(str(logfile.name), "--windows")}" 2>&1\n'
                r"@exit /b %ERRORLEVEL%"
            )
            # Wine can have problems with arbitrary names. Create simple uniqe name for the file
            hash_id = abs(hash(logfile.name)) % (10 ** 8)
            batfile = macrofile_path.with_name(f"wine_{hash_id}.bat")
            batfile.write_text(anybodycmd)
            macrofile_cleanup.append(batfile)

            cmd = ["wine", "cmd", "/c", str(batfile) + r"& exit /b %ERRORLEVEL%"]
            proc = Popen(
                cmd,
                env=env,
                cwd=folder,
            )

    _subprocess_container.add(proc.pid)
    try:
        proc.wait(timeout=timeout)
        retcode = ctypes.c_int32(proc.returncode).value
    except TimeoutExpired:
        proc.kill()
        proc.communicate()
        retcode = _TIMEDOUT_BY_ANYPYTOOLS
    except KeyboardInterrupt:
        proc.terminate()
        proc.communicate()
        retcode = _KILLED_BY_ANYPYTOOLS
        raise
    finally:
        _subprocess_container.remove(proc.pid)

    if retcode == _TIMEDOUT_BY_ANYPYTOOLS:<commit_msg>black formating<commit_after># -*- coding: utf-8 -*-
"""
Utilities for working with the AnyBody Console applicaiton.

Created on Fri Oct 19 21:14:59 2012
@author: Morten
"""

import os
import io
import sys
import time
import copy
import types
import ctypes
import shelve
import atexit
import pathlib
import logging
import warnings
import collections
from pathlib import Path
from subprocess import Popen, TimeoutExpired
from tempfile import NamedTemporaryFile
from threading import Thread, RLock
from queue import Queue

from typing import Generator, List

import numpy as np
from tqdm.auto import tqdm

from .tools import (
    ON_WINDOWS,
    make_hash,
    AnyPyProcessOutputList,
    parse_anybodycon_output,
    getsubdirs,
    get_anybodycon_path,
    BELOW_NORMAL_PRIORITY_CLASS,
    AnyPyProcessOutput,
    run_from_ipython,
    get_ncpu,
    winepath,
    silentremove,
    case_preserving_replace,
)
from .macroutils import AnyMacro, MacroCommand

logger = logging.getLogger("abt.anypytools")

_thread_lock = RLock()
_KILLED_BY_ANYPYTOOLS = 10
_TIMEDOUT_BY_ANYPYTOOLS = 11
_NO_LICENSES_AVAILABLE = -22
_UNABLE_TO_ACQUIRE_LICENSE = 234  # May indicate wrong password


class _SubProcessContainer(object):
    """Class to hold a record of process pids from Popen.

    Properties
    ----------
    stop_all: boolean
        If set to True all process held by the object will be automatically
        killed

    Methods
    -------
    add(pid):
        Add process id to the record of process

    remove(pid):
        Remove process id from the record

    """

    def __init__(self):
        self._pids = set()
        self._stop_all = False

    def add(self, pid):
        with _thread_lock:
            self._pids.add(pid)
        if self.stop_all:
            self._kill_running_processes()

    def remove(self, pid):
        with _thread_lock:
            try:
                self._pids.remove(pid)
            except KeyError:
                pass

    @property
    def stop_all(self):
        return self._stop_all

    @stop_all.setter
    def stop_all(self, value):
        if value:
            self._stop_all = True
            self._kill_running_processes()
        else:
            self._stop_all = False

    def _kill_running_processes(self):
        """Clean up and shut down any running processes."""
        # Kill any rouge processes that are still running.
        with _thread_lock:
            killed = []
            for pid in self._pids:
                try:
                    os.kill(pid, _KILLED_BY_ANYPYTOOLS)
                    killed.append(str(pid))
                except Exception:
                    pass
            self._pids.clear()


_subprocess_container = _SubProcessContainer()
atexit.register(_subprocess_container._kill_running_processes)


def execute_anybodycon(
    macro,
    logfile=None,
    anybodycon_path=None,
    timeout=3600,
    keep_macrofile=False,
    env=None,
    priority=BELOW_NORMAL_PRIORITY_CLASS,
    debug_mode=0,
    folder=None,
):
    """Launch a single AnyBodyConsole applicaiton.

    This is a low level function to start a AnyBody Console process
    with a given list macros.

    Parameters
    ----------
    macro : list of str
        List of macros strings to pass to the AnyBody Console Application
    logfile : file like object, optional
        An open file like object to write to pipe the output of AnyBody
        into. (Defaults to None, in which case it will use sys.stdout)
    anybodycon_path : str, optional
        Path to the AnyBodyConsole application. Default to None, in which
        case the default installed AnyBody installation will be looked up
        in the Windows registry.
    timeout : int, optional
        Timeout before the process is killed autmotically. Defaults to
        3600 seconds (1 hour).
    keep_macrofile : bool, optional
        Set to True to prevent the temporary macro file from beeing deleted.
        (Defaults to False)
    env: dict
        Environment varaibles which are passed to the started AnyBody console
        application.
    priority : int, optional
        The priority of the subprocesses. This can be on of the following:
        ``anypytools.IDLE_PRIORITY_CLASS``, ``anypytools.BELOW_NORMAL_PRIORITY_CLASS``,
        ``anypytools.NORMAL_PRIORITY_CLASS``, ``anypytools.HIGH_PRIORITY_CLASS``
        Default is BELOW_NORMAL_PRIORITY_CLASS.
    debug_mode : int
        The AMS debug mode to use. Defaults to 0 which is disabled. 1 correspond to
        crashdump enabled
    folder : the folder in which AnyBody is executed

    Returns
    -------
    int
        The return code from the AnyBody Console application.

    """
    if folder is None:
        folder = os.getcwd()

    try:
        macrofile_path = Path(folder).resolve() / (Path(logfile.name).stem + ".anymcr")
    except AttributeError:
        macrofile_path = Path("macrofile.anymcr")

    macrofile_cleanup = [macrofile_path]

    if logfile is None:
        logfile = sys.stdout

    if anybodycon_path is None:
        anybodycon_path = Path(get_anybodycon_path())

    if macro and macro[-1] != "exit":
        macro.append("exit")

    if not os.path.isfile(anybodycon_path):
        raise IOError(f"Can not find anybodycon.exe: {anybodycon_path}")

    with open(macrofile_path, "w+b") as fh:
        fh.write("\n".join(macro).encode("UTF-8"))
        fh.flush()

    if ON_WINDOWS:
        # Don't display the Windows GPF dialog if the invoked program dies.
        # See comp.os.ms-windows.programmer.win32
        # How to suppress crash notification dialog?, Jan 14,2004 -
        # <NAME>'s response [1]
        SEM_NOGPFAULTERRORBOX = 0x0002  # From MSDN
        ctypes.windll.kernel32.SetErrorMode(SEM_NOGPFAULTERRORBOX)
        subprocess_flags = 0x8000000  # win32con.CREATE_NO_WINDOW?
        subprocess_flags |= priority
        extra_kwargs = {"creationflags": subprocess_flags}

        anybodycmd = [
            str(anybodycon_path.resolve()),
            "--macro=",
            str(macrofile_path),
            "/deb",
            str(debug_mode),
            "/ni",
        ]
        proc = Popen(
            anybodycmd,
            stdout=logfile,
            stderr=logfile,
            env=env,
            cwd=folder,
            **extra_kwargs,
        )

    else:
        if os.environ.get("WINE_REDIRECT_OUTPUT", 0):
            cmd = [
                "wine",
                str(anybodycon_path.resolve()),
                "--macro=",
                winepath(macrofile_path, "--windows"),
                "/deb",
                str(debug_mode),
                "/ni",
            ]
            proc = Popen(
                cmd,
                env=env,
                cwd=folder,
                close_fds=False,
                stdout=logfile,
                stderr=logfile,
            )
        else:
            # ON Linux/Wine we use a bat file to redirect the output into a file on wine/windows
            # side. This prevents a bug with AnyBody starts it's builtin python.
            anybodycmd = (
                f'@call "{winepath(anybodycon_path.resolve(), "--windows")}"'
                f' -m "{winepath(macrofile_path, "--windows")}"'
                f" -deb {str(debug_mode)}"
                " -ni"
                f' >> "{winepath(str(logfile.name), "--windows")}" 2>&1\n'
                r"@exit /b %ERRORLEVEL%"
            )
            # Wine can have problems with arbitrary names. Create simple uniqe name for the file
            hash_id = abs(hash(logfile.name)) % (10**8)
            batfile = macrofile_path.with_name(f"wine_{hash_id}.bat")
            batfile.write_text(anybodycmd)
            macrofile_cleanup.append(batfile)

            cmd = ["wine", "cmd", "/c", str(batfile) + r"& exit /b %ERRORLEVEL%"]
            proc = Popen(
                cmd,
                env=env,
                cwd=folder,
            )

    _subprocess_container.add(proc.pid)
    try:
        proc.wait(timeout=timeout)
        retcode = ctypes.c_int32(proc.returncode).value
    except TimeoutExpired:
        proc.kill()
        proc.communicate()
        retcode = _TIMEDOUT_BY_ANYPYTOOLS
    except KeyboardInterrupt:
        proc.terminate()
        proc.communicate()
        retcode = _KILLED_BY_ANYPYTOOLS
        raise
    finally:
        _subprocess_container.remove(proc.pid)

    if retcode == _TIMEDOUT_BY_ANYPYTOOLS:
{'fraction_non_alphanumeric': 0.08374851720047449, 'fraction_numerical': 0.00830367734282325, 'mean_word_length': 2.0611837327523603, 'pattern_counts': {'":': 4, '<': 5, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '316764', 'n_tokens_mistral': 5118, 'n_tokens_neox': 4880, 'n_words': 1369}