<commit_before>#!/usr/bin/env python
# -*- coding: utf-8 -*-
from __future__ import absolute_import
from __future__ import print_function
from __future__ import division
from __future__ import unicode_literals

import time
import glob
import threading

try:
    from . import Tools
    from . import pyserial
    from . import Messages
    from .Preferences import Preferences
except:
    from libs import Tools
    from libs import pyserial
    from libs import Messages
    from libs.Preferences import Preferences

if Tools.getOsName() == 'windows':
    if Tools.getPythonVersion() < 3:
        import _winreg as winreg
    else:
        import winreg


@Tools.singleton
class SerialListener(object):
    """
    Constantly checks if a port has been connected or disconnected
    """

    def __init__(self, func=None):
        self.func = func
        self.serial_list = []
        self.is_alive = False

    def start(self):
        if not self.is_alive:
            self.is_alive = True
            listener_thread = threading.Thread(target=self.update)
            listener_thread.start()

    def update(self):
        while self.is_alive:
            pre_serial_list = self.serial_list
            self.serial_list = listSerialPorts()
            if self.serial_list != pre_serial_list:
                if self.func:
                    self.func()
            time.sleep(1)

    def stop(self):
        self.is_alive = False


serials_in_use = []
serial_monitor_dict = {}


class SerialMonitor(object):
    """
    Handle the messages sended and received from/to the serial monitor
    """

    def __init__(self, serial_port, console=None):
        super(SerialMonitor, self).__init__()
        self.port = serial_port
        self.serial = pyserial.Serial()
        self.serial.port = serial_port
        self.queue = Messages.MessageQueue(console)
        self.Preferences = Preferences()
        self.is_alive = False

    def isRunning(self):
        return self.is_alive

    def start(self):
        self.queue.startPrint()
        if not self.is_alive:
            baudrate = self.Preferences.get('baudrate', 9600)
            self.serial.baudrate = baudrate
            if isSerialAvailable(self.port):
                self.serial.open()
                self.is_alive = True
                monitor_thread = threading.Thread(target=self.receive)
                monitor_thread.start()
            else:
                msg = 'Serial port {0} already in use. '
                msg += 'Try stopping any programs that may be using it.\\n'
                self.queue.put(msg, self.port)
                self.stop()

    def stop(self):
        self.is_alive = False
        self.queue.stopPrint()

    def receive(self):
        length_before = 0
        while self.is_alive:
            number = self.serial.inWaiting()
            if number > 0:
                in_text = self.serial.read(number)
                length_in_text = len(in_text)
                in_text = convertMode(in_text, length_before)
                self.queue.put(in_text)
                length_before += length_in_text
                length_before %= 16
            time.sleep(0.01)
        self.serial.close()

    def send(self, out_text):
        line_ending = self.Preferences.get('line_ending', '\n')
        out_text += line_ending
        self.queue.put('\\n[SENDED] {0}\\n', out_text)
        out_text = out_text.encode('utf-8', 'replace')
        self.serial.write(out_text)


def convertMode(in_text, str_len=0):
    """
    Convert a text in differents formats (ASCII,HEX)

    Arguments:
        in_text {string}
            Text to convert

    Keyword Arguments:
        str_len {int}
            leng of the in_text string (default: {0})

    Returns:
        [string] -- Converted string
    """
    text = u''
    display_mode = Preferences().get('display_mode', 'Text')
    if display_mode == 'Ascii':
        for character in in_text:
            text += chr(character)
    elif display_mode == 'Hex':
        for (index, character) in enumerate(in_text):
            text += u'%02X ' % character
            if (index + str_len + 1) % 8 == 0:
                text += '\t'
            if (index + str_len + 1) % 16 == 0:
                text += '\n'
    else:
        text = in_text.decode('utf-8', 'replace')
    return text


def isSerialAvailable(serial_port):<commit_msg>Removing CR and NULL tag in text mode<commit_after>#!/usr/bin/env python
# -*- coding: utf-8 -*-
from __future__ import absolute_import
from __future__ import print_function
from __future__ import division
from __future__ import unicode_literals

import time
import glob
import threading

try:
    from . import Tools
    from . import pyserial
    from . import Messages
    from .Preferences import Preferences
except:
    from libs import Tools
    from libs import pyserial
    from libs import Messages
    from libs.Preferences import Preferences

if Tools.getOsName() == 'windows':
    if Tools.getPythonVersion() < 3:
        import _winreg as winreg
    else:
        import winreg


@Tools.singleton
class SerialListener(object):
    """
    Constantly checks if a port has been connected or disconnected
    """

    def __init__(self, func=None):
        self.func = func
        self.serial_list = []
        self.is_alive = False

    def start(self):
        if not self.is_alive:
            self.is_alive = True
            listener_thread = threading.Thread(target=self.update)
            listener_thread.start()

    def update(self):
        while self.is_alive:
            pre_serial_list = self.serial_list
            self.serial_list = listSerialPorts()
            if self.serial_list != pre_serial_list:
                if self.func:
                    self.func()
            time.sleep(1)

    def stop(self):
        self.is_alive = False


serials_in_use = []
serial_monitor_dict = {}


class SerialMonitor(object):
    """
    Handle the messages sended and received from/to the serial monitor
    """

    def __init__(self, serial_port, console=None):
        super(SerialMonitor, self).__init__()
        self.port = serial_port
        self.serial = pyserial.Serial()
        self.serial.port = serial_port
        self.queue = Messages.MessageQueue(console)
        self.Preferences = Preferences()
        self.is_alive = False

    def isRunning(self):
        return self.is_alive

    def start(self):
        self.queue.startPrint()
        if not self.is_alive:
            baudrate = self.Preferences.get('baudrate', 9600)
            self.serial.baudrate = baudrate
            if isSerialAvailable(self.port):
                self.serial.open()
                self.is_alive = True
                monitor_thread = threading.Thread(target=self.receive)
                monitor_thread.start()
            else:
                msg = 'Serial port {0} already in use. '
                msg += 'Try stopping any programs that may be using it.\\n'
                self.queue.put(msg, self.port)
                self.stop()

    def stop(self):
        self.is_alive = False
        self.queue.stopPrint()

    def receive(self):
        length_before = 0
        while self.is_alive:
            number = self.serial.inWaiting()
            if number > 0:
                in_text = self.serial.read(number)
                length_in_text = len(in_text)
                in_text = convertMode(in_text, length_before)
                self.queue.put(in_text)
                length_before += length_in_text
                length_before %= 16
            time.sleep(0.01)
        self.serial.close()

    def send(self, out_text):
        line_ending = self.Preferences.get('line_ending', '\n')
        out_text += line_ending
        self.queue.put('\\n[SENDED] {0}\\n', out_text)
        out_text = out_text.encode('utf-8', 'replace')
        self.serial.write(out_text)


def convertMode(in_text, str_len=0):
    """
    Convert a text in differents formats (ASCII,HEX)

    Arguments:
        in_text {string}
            Text to convert

    Keyword Arguments:
        str_len {int}
            leng of the in_text string (default: {0})

    Returns:
        [string] -- Converted string
    """
    text = u''
    display_mode = Preferences().get('display_mode', 'Text')
    if display_mode == 'Ascii':
        for character in in_text:
            text += chr(character)
    elif display_mode == 'Hex':
        for (index, character) in enumerate(in_text):
            text += u'%02X ' % character
            if (index + str_len + 1) % 8 == 0:
                text += '\t'
            if (index + str_len + 1) % 16 == 0:
                text += '\n'
    else:
        text = in_text.decode('utf-8', 'replace')
        text = text.replace('\r', '').replace('NULL', '')
    return text


def isSerialAvailable(serial_port):
{'fraction_non_alphanumeric': 0.09871244635193133, 'fraction_numerical': 0.006550711542805511, 'mean_word_length': 1.9774714189643579, 'pattern_counts': {'":': 0, '<': 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': '601805', 'n_tokens_mistral': 2664, 'n_tokens_neox': 2548, 'n_words': 651}