<commit_before>from serial import SerialException
from multiprocessing import Process, Pipe, Event
from collections import namedtuple
from time import sleep, time
from struct import pack, unpack
import numpy as np
import logging
import signal

from .. import config as c
from .tamp_serial import *

logger = logging.getLogger('tamproxy.controller')

class PacketController(Process):

    # Constants
    WINDOW_SIZE =           c.host.window_size
    ENABLE_TIMEOUT =        c.host.enable_timeout
    INITIAL_TIMEOUT =       c.host.initial_timeout
    SRTT_ALPHA =            c.host.srtt_alpha
    RTTDEV_ALPHA =          c.host.rttdev_alpha
    SRTT_GAIN =             c.host.srtt_gain
    RTTDEV_GAIN =           c.host.rttdev_gain
    SERIAL_RETRIES =        c.host.serial_retries
    SERIAL_RETRY_TIMEOUT =  c.host.serial_retry_timeout
    PACK_FORMAT =           c.host.pack_format
    UNPACK_FORMAT =         c.host.unpack_format
    RESET_MSG =             c.host.reset_msg
    START_BYTE =            c.packet.start_byte
    DEVICELIST_CODE =       c.devices.devicelist.code
    HELLO_PAYLOAD =         c.devices.devicelist.hello_code

    compare_pids = lambda self, a,b: np.int16(b - a)

    def __init__(self):
        self._stop = Event()
        self.tserial = None
        self.packet_parser = None
        self.pipe_inside, self.pipe_outside = Pipe()
        self.continuous_requests = set()
        self.weighted_tdma_list = []
        self.reset()
        super(PacketController, self).__init__()

    def reset(self, clear_pipe=False):
        self.rtt_smoothed = None
        self.rtt_deviation = None
        self.timeout = self.INITIAL_TIMEOUT
        self.en_route = dict()          # pid -> packet, time_sent
        self.receiving_buffer = dict()  # pid -> sent_packet, payload
        self.next_send_pid = np.uint16(0)
        self.next_recv_pid = np.uint16(0)
        self.weighted_tdma_pos = 0
        self.packets_received = 0
        if clear_pipe:
            while self.pipe_inside.poll():
                self.pipe_inside.recv()

    def connect(self):
        self.reset(True)
        raw_hello_packet = self.encode_raw_packet(0, self.DEVICELIST_CODE,
                                              self.HELLO_PAYLOAD)
        hello_response_length = len(raw_hello_packet) - 1
        self.tserial = TAMPSerial(raw_hello_packet, hello_response_length)
        self.packet_parser = PacketParser(self.tserial)
        self.pipe_inside.send((None, self.RESET_MSG))

    def encode_raw_packet(self, pid, dest, payload):
        pack_format = self.PACK_FORMAT.format(len(payload))
        length = len(payload) + 5
        return pack(pack_format, self.START_BYTE, pid, length, dest, payload)

    def get_new_packet_to_send(self):
        # process new requests
        while self.pipe_inside.poll():
            packet_request = PacketRequest(*self.pipe_inside.recv())
            if packet_request.is_continuous:
                if not packet_request.remove_continuous:
                    self.continuous_requests.add(packet_request[:2])
                    self.weighted_tdma_list += (
                        [packet_request[:2]] * packet_request.weight)
                else: 
                    self.continuous_requests.discard(packet_request[:2])
            else:
                return packet_request[:2]

        # resend exiting continuous_requests
        while self.weighted_tdma_list:
            key = self.weighted_tdma_list[self.weighted_tdma_pos]
            if key in self.continuous_requests:
                self.weighted_tdma_pos += 1
                self.weighted_tdma_pos %= len(self.weighted_tdma_list)
                return key
            else:
                # this item was removed
                self.weighted_tdma_list.pop(self.weighted_tdma_pos)
                if self.weighted_tdma_list:
                    self.weighted_tdma_pos += 1
                    self.weighted_tdma_pos %= len(self.weighted_tdma_list)

        # nothing else to do
        return None

    def slide_window(self):
        if self.ENABLE_TIMEOUT:
            # resend all en_route packets that have timed out
            for pid in self.en_route:
                packet, time_sent = self.en_route[pid]
                dt = time() - time_sent
                if dt > self.timeout:
                    logger.debug("dt={}, timeout={}, rtt_smoothed={}, rtt_deviation={}"
                        .format(self.timeout, self.rtt_smoothed, self.rtt_deviation)
                    )
                    self.en_route[pid] = (packet, time())
                    self.transmit(pid, *packet[:2])
                    return

        # transmit the next packet
        if len(self.en_route) < self.WINDOW_SIZE:
            packet = self.get_new_packet_to_send()
            if packet:
                new_pid = self.next_send_pid
                self.en_route[new_pid] = (packet, time())
                self.transmit(new_pid, *packet)
                self.next_send_pid += np.uint16(1)

    def decode_raw_packet(self, raw_packet):
        payload_length = len(raw_packet) - 4
        unpack_format = self.UNPACK_FORMAT.format(payload_length)
        start_byte, pid, length, payload = unpack(unpack_format, raw_packet)
        return (np.uint16(pid), payload)

    def transmit(self, pid, dest, payload):
        if self.tserial.out_waiting > 50:
            self.tserial.flush()
        raw_packet = self.encode_raw_packet(pid, dest, payload)
        self.tserial.write(raw_packet)

    def receive(self):
        raw_packets = self.packet_parser.receive()
        packets = [self.decode_raw_packet(p) for p in raw_packets]
        for pid, payload in packets:
            if payload != self.HELLO_PAYLOAD:
                self.process_packet(pid, payload)
        return len(packets)

    def process_packet(self, pid, payload):
        self.packets_received += 1
        if pid not in self.en_route:
            logger.warn("retransmitted packet received")
            return
        sent_packet, time_sent = self.en_route.pop(pid)
        if self.ENABLE_TIMEOUT:
            self.update_timeout(time_sent)
            if self.next_recv_pid == pid:<commit_msg>shorter and more consistent timeout logging<commit_after>from serial import SerialException
from multiprocessing import Process, Pipe, Event
from collections import namedtuple
from time import sleep, time
from struct import pack, unpack
import numpy as np
import logging
import signal

from .. import config as c
from .tamp_serial import *

logger = logging.getLogger('tamproxy.controller')

class PacketController(Process):

    # Constants
    WINDOW_SIZE =           c.host.window_size
    ENABLE_TIMEOUT =        c.host.enable_timeout
    INITIAL_TIMEOUT =       c.host.initial_timeout
    SRTT_ALPHA =            c.host.srtt_alpha
    RTTDEV_ALPHA =          c.host.rttdev_alpha
    SRTT_GAIN =             c.host.srtt_gain
    RTTDEV_GAIN =           c.host.rttdev_gain
    SERIAL_RETRIES =        c.host.serial_retries
    SERIAL_RETRY_TIMEOUT =  c.host.serial_retry_timeout
    PACK_FORMAT =           c.host.pack_format
    UNPACK_FORMAT =         c.host.unpack_format
    RESET_MSG =             c.host.reset_msg
    START_BYTE =            c.packet.start_byte
    DEVICELIST_CODE =       c.devices.devicelist.code
    HELLO_PAYLOAD =         c.devices.devicelist.hello_code

    compare_pids = lambda self, a,b: np.int16(b - a)

    def __init__(self):
        self._stop = Event()
        self.tserial = None
        self.packet_parser = None
        self.pipe_inside, self.pipe_outside = Pipe()
        self.continuous_requests = set()
        self.weighted_tdma_list = []
        self.reset()
        super(PacketController, self).__init__()

    def reset(self, clear_pipe=False):
        self.rtt_smoothed = None
        self.rtt_deviation = None
        self.timeout = self.INITIAL_TIMEOUT
        self.en_route = dict()          # pid -> packet, time_sent
        self.receiving_buffer = dict()  # pid -> sent_packet, payload
        self.next_send_pid = np.uint16(0)
        self.next_recv_pid = np.uint16(0)
        self.weighted_tdma_pos = 0
        self.packets_received = 0
        if clear_pipe:
            while self.pipe_inside.poll():
                self.pipe_inside.recv()

    def connect(self):
        self.reset(True)
        raw_hello_packet = self.encode_raw_packet(0, self.DEVICELIST_CODE,
                                              self.HELLO_PAYLOAD)
        hello_response_length = len(raw_hello_packet) - 1
        self.tserial = TAMPSerial(raw_hello_packet, hello_response_length)
        self.packet_parser = PacketParser(self.tserial)
        self.pipe_inside.send((None, self.RESET_MSG))

    def encode_raw_packet(self, pid, dest, payload):
        pack_format = self.PACK_FORMAT.format(len(payload))
        length = len(payload) + 5
        return pack(pack_format, self.START_BYTE, pid, length, dest, payload)

    def get_new_packet_to_send(self):
        # process new requests
        while self.pipe_inside.poll():
            packet_request = PacketRequest(*self.pipe_inside.recv())
            if packet_request.is_continuous:
                if not packet_request.remove_continuous:
                    self.continuous_requests.add(packet_request[:2])
                    self.weighted_tdma_list += (
                        [packet_request[:2]] * packet_request.weight)
                else: 
                    self.continuous_requests.discard(packet_request[:2])
            else:
                return packet_request[:2]

        # resend exiting continuous_requests
        while self.weighted_tdma_list:
            key = self.weighted_tdma_list[self.weighted_tdma_pos]
            if key in self.continuous_requests:
                self.weighted_tdma_pos += 1
                self.weighted_tdma_pos %= len(self.weighted_tdma_list)
                return key
            else:
                # this item was removed
                self.weighted_tdma_list.pop(self.weighted_tdma_pos)
                if self.weighted_tdma_list:
                    self.weighted_tdma_pos += 1
                    self.weighted_tdma_pos %= len(self.weighted_tdma_list)

        # nothing else to do
        return None

    def slide_window(self):
        if self.ENABLE_TIMEOUT:
            # resend all en_route packets that have timed out
            for pid in self.en_route:
                packet, time_sent = self.en_route[pid]
                dt = time() - time_sent
                if dt > self.timeout:
                    logger.debug("Packet timed out, will retransmit: "
                        "pid={}, packet={}, timeout={}"
                        .format(pid, packet, self.timeout)
                    )
                    self.en_route[pid] = (packet, time())
                    self.transmit(pid, *packet[:2])
                    return

        # transmit the next packet
        if len(self.en_route) < self.WINDOW_SIZE:
            packet = self.get_new_packet_to_send()
            if packet:
                new_pid = self.next_send_pid
                self.en_route[new_pid] = (packet, time())
                self.transmit(new_pid, *packet)
                self.next_send_pid += np.uint16(1)

    def decode_raw_packet(self, raw_packet):
        payload_length = len(raw_packet) - 4
        unpack_format = self.UNPACK_FORMAT.format(payload_length)
        start_byte, pid, length, payload = unpack(unpack_format, raw_packet)
        return (np.uint16(pid), payload)

    def transmit(self, pid, dest, payload):
        if self.tserial.out_waiting > 50:
            self.tserial.flush()
        raw_packet = self.encode_raw_packet(pid, dest, payload)
        self.tserial.write(raw_packet)

    def receive(self):
        raw_packets = self.packet_parser.receive()
        packets = [self.decode_raw_packet(p) for p in raw_packets]
        for pid, payload in packets:
            if payload != self.HELLO_PAYLOAD:
                self.process_packet(pid, payload)
        return len(packets)

    def process_packet(self, pid, payload):
        self.packets_received += 1
        if pid not in self.en_route:
            logger.debug("Retransmitted packet received: pid={}, payload={}"
                            .format(pid,payload))
            return
        sent_packet, time_sent = self.en_route.pop(pid)
        if self.ENABLE_TIMEOUT:
            self.update_timeout(time_sent)
            if self.next_recv_pid == pid:
{'fraction_non_alphanumeric': 0.08383329333653308, 'fraction_numerical': 0.004639628829693624, 'mean_word_length': 1.9988006716238906, 'pattern_counts': {'":': 0, '<': 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': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4804607', 'n_tokens_mistral': 3832, 'n_tokens_neox': 3620, 'n_words': 725}