<commit_before>#  Copyright (c) 2012, <NAME>.
#  All rights reserved; available under the terms of the MIT License.
"""

playitagainsam.player: replay interactive terminal sessions
===========================================================

"""

import os
import sys
import time

import six

from playitagainsam.util import forkexec, get_default_terminal
from playitagainsam.util import forkexec_pty
from playitagainsam.util import get_pias_script, get_fd
from playitagainsam.coordinator import SocketCoordinator, proxy_to_coordinator

# XXX TODO: set the size of each terminal


class Player(SocketCoordinator):

    waypoint_chars = (six.b("\n"), six.b("\r"))

    def __init__(self, sock_path, eventlog, terminal=None, auto_type=False,
                 auto_waypoint=False, live_replay=False, replay_shell=None):
        super(Player, self).__init__(sock_path)
        self.eventlog = eventlog
        self.terminal = terminal or get_default_terminal()
        self.live_replay = live_replay
        self.replay_shell = replay_shell
        if not auto_type:
            self.auto_type = False
        else:
            self.auto_type = auto_type / 1000.0
        if not auto_waypoint:
            self.auto_waypoint = False
        else:
            self.auto_waypoint = auto_waypoint / 1000.0
        self.terminals = {}
        self.proc_fds = {}

    def run(self):
        event = self.eventlog.read_event()
        while event is not None:
            action = event["act"]
            term = event.get("term", None)
            data = event.get("data", None)

            # TODO (JC) -- possibly this should not be in the event process loop:
            # it should be event-driven by an asyncore.dispatcher.handle_read();
            # we would also ignore PAUSEs if (live-replay and not auto-type),
            # but for now it works well enough for the patch author's use cases.
            self._maybe_do_live_output(term)

            if action == "OPEN":
                self._do_open_terminal(term)
            elif action == "PAUSE":
                time.sleep(event["duration"])
            elif action == "READ":
                self._do_read(term, data)
            elif action == "WRITE":
                # when in --live-replay mode, eventlog sends no WRITE events,
                # so no need to check here whether we are on --live-replay or not
                self._do_write(term, data)
            if action == "CLOSE":
                self._do_close_terminal(term)

            event = self.eventlog.read_event()

    def cleanup(self):
        for term in self.terminals:
            view_sock, _, = self.terminals[term]
            view_sock.close()
        super(Player, self).cleanup()

    def _do_open_terminal(self, term):
        ready = self.wait_for_data([self.sock], 0.1)
        if self.sock not in ready:
            # XXX TODO: wait for a keypress from some existing terminal
            # to trigger the appearance of the terminal.
            # Specify options via the environment.
            # This allows us to spawn the joiner with no arguments,
            # so it will work with the "-e" option of terminal programs.
            env = {}
            env["PIAS_OPT_JOIN"] = "1"
            env["PIAS_OPT_COMMAND"] = "replay"
            env["PIAS_OPT_DATAFILE"] = self.eventlog.datafile
            env["PIAS_OPT_TERMINAL"] = self.terminal
            forkexec([self.terminal, "-e", get_pias_script()], env)
        view_sock, _ = self.sock.accept()
<commit_msg>Delay calling get_default_terminal() for as long as possible.<commit_after>#  Copyright (c) 2012, <NAME>.
#  All rights reserved; available under the terms of the MIT License.
"""

playitagainsam.player: replay interactive terminal sessions
===========================================================

"""

import os
import sys
import time

import six

from playitagainsam.util import forkexec, get_default_terminal
from playitagainsam.util import forkexec_pty
from playitagainsam.util import get_pias_script, get_fd
from playitagainsam.coordinator import SocketCoordinator, proxy_to_coordinator

# XXX TODO: set the size of each terminal


class Player(SocketCoordinator):

    waypoint_chars = (six.b("\n"), six.b("\r"))

    def __init__(self, sock_path, eventlog, terminal=None, auto_type=False,
                 auto_waypoint=False, live_replay=False, replay_shell=None):
        super(Player, self).__init__(sock_path)
        self.eventlog = eventlog
        self.terminal = terminal
        self.live_replay = live_replay
        self.replay_shell = replay_shell
        if not auto_type:
            self.auto_type = False
        else:
            self.auto_type = auto_type / 1000.0
        if not auto_waypoint:
            self.auto_waypoint = False
        else:
            self.auto_waypoint = auto_waypoint / 1000.0
        self.terminals = {}
        self.proc_fds = {}

    def run(self):
        event = self.eventlog.read_event()
        while event is not None:
            action = event["act"]
            term = event.get("term", None)
            data = event.get("data", None)

            # TODO (JC) -- possibly this should not be in the event process loop:
            # it should be event-driven by an asyncore.dispatcher.handle_read();
            # we would also ignore PAUSEs if (live-replay and not auto-type),
            # but for now it works well enough for the patch author's use cases.
            self._maybe_do_live_output(term)

            if action == "OPEN":
                self._do_open_terminal(term)
            elif action == "PAUSE":
                time.sleep(event["duration"])
            elif action == "READ":
                self._do_read(term, data)
            elif action == "WRITE":
                # when in --live-replay mode, eventlog sends no WRITE events,
                # so no need to check here whether we are on --live-replay or not
                self._do_write(term, data)
            if action == "CLOSE":
                self._do_close_terminal(term)

            event = self.eventlog.read_event()

    def cleanup(self):
        for term in self.terminals:
            view_sock, _, = self.terminals[term]
            view_sock.close()
        super(Player, self).cleanup()

    def _do_open_terminal(self, term):
        ready = self.wait_for_data([self.sock], 0.1)
        if self.sock not in ready:
            # XXX TODO: wait for a keypress from some existing terminal
            # to trigger the appearance of the terminal.
            # Specify options via the environment.
            # This allows us to spawn the joiner with no arguments,
            # so it will work with the "-e" option of terminal programs.
            env = {}
            env["PIAS_OPT_JOIN"] = "1"
            env["PIAS_OPT_COMMAND"] = "replay"
            env["PIAS_OPT_DATAFILE"] = self.eventlog.datafile
            env["PIAS_OPT_TERMINAL"] = self.terminal
            cmd = self.terminal or get_default_terminal()
            forkexec([cmd, "-e", get_pias_script()], env)
        view_sock, _ = self.sock.accept()

{'fraction_non_alphanumeric': 0.10597519729425028, 'fraction_numerical': 0.004791431792559188, 'mean_word_length': 2.205510388437218, 'pattern_counts': {'":': 10, '<': 5, '<?xml version=': 0, '>': 5, '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': '1040434', 'n_tokens_mistral': 2063, 'n_tokens_neox': 1977, 'n_words': 585}