<commit_before>import struct
import array
import math
import time
import numpy as np
import re
import logging

from utils.utils import bitarray_to_array

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)-8s] (%(threadName)-10s) %(message)s")


class FEI4RegisterUtils(object):
    def __init__(self, device, readout, register):
        self.device = device
        self.readout = readout
        self.register = register
        self.command_memory_byte_offset = 16
        self.command_memory_byte_size = 2048 - self.command_memory_byte_offset  # 16 bytes of register data
        self.zero_cmd_length = 1
        self.zero_cmd = self.register.get_commands("zeros", length=self.zero_cmd_length)[0]
        self.zero_cmd_padded = self.zero_cmd.copy()
        self.zero_cmd_padded.fill()

    def add_commands(self, x, y):
            return x + self.zero_cmd + y  # FE needs a zero bits between commands

    def add_byte_padded_commands(self, x, y):
            x_fill = x.copy()
            x_fill.fill()
            y_fill = y.copy()
            y_fill.fill()
            return x_fill + self.zero_cmd_padded + y_fill  # FE needs a zero between commands

    def concatenate_commands(self, commands, byte_padding=False):
        if byte_padding:
            return reduce(self.add_byte_padded_commands, commands)
        else:
            return reduce(self.add_commands, commands)

    def send_commands(self, commands, repeat=1, wait_for_finish=True, concatenate=True, byte_padding=False, clear_memory=False):
        if concatenate:
            commands_iter = iter(commands)
            concatenated_cmd = commands_iter.next()
            for command in commands_iter:
                concatenated_cmd_tmp = self.concatenate_commands((concatenated_cmd, command), byte_padding=byte_padding)
                if concatenated_cmd_tmp.length() > self.command_memory_byte_size * 8:
                    self.send_command(command=concatenated_cmd, repeat=repeat, wait_for_finish=wait_for_finish, set_length=True, clear_memory=clear_memory)
                    concatenated_cmd = command
                else:
                    concatenated_cmd = concatenated_cmd_tmp
            # send remaining commands
            self.send_command(command=concatenated_cmd, repeat=repeat, wait_for_finish=wait_for_finish, set_length=True, clear_memory=clear_memory)
        else:
            max_length = 0
            self.set_hardware_repeat(repeat)
            for command in commands:
                max_length = max(command.length(), max_length)
                self.send_command(command=command, repeat=None, wait_for_finish=wait_for_finish, set_length=True, clear_memory=False)
            if clear_memory:
                self.clear_command_memory(length=max_length)

    def send_command(self, command, repeat=1, wait_for_finish=True, set_length=True, clear_memory=False):
        if repeat is not None:
            self.set_hardware_repeat(repeat)
        # write command into memory
        command_length = self.set_command(command, set_length=set_length)
        # sending command
        self.start_command()
        # wait for command to be finished
        if wait_for_finish:
            self.wait_for_command(length=command_length, repeat=repeat)
        # clear command memory
        if clear_memory:
            self.clear_command_memory(length=command_length)

    def clear_command_memory(self, length=None):
        self.set_command(self.register.get_commands("zeros", length=(self.command_memory_byte_size * 8) if length is None else length)[0], set_length=False)

    def set_command_length(self, lenght):
        bit_length_array = array.array('B', struct.pack('H', lenght))
        self.device.WriteExternal(address=0 + 3, data=bit_length_array)

    def set_command(self, command, set_length=True, byte_offset=0):
        command_length = command.length()
        # set command bit length
        if set_length:
            self.set_command_length(command_length)
        # set command
        data = bitarray_to_array(command)
        if self.command_memory_byte_size < len(data) + byte_offset:
            raise ValueError('Length of command or offset is too big')
        self.device.WriteExternal(address=0 + self.command_memory_byte_offset + byte_offset, data=data)
        return command_length

    def start_command(self):
        self.device.WriteExternal(address=0 + 1, data=(0, ))

    def set_hardware_repeat(self, repeat=1):
        if repeat is not None:
            repeat_array = array.array('B', struct.pack('I', repeat))
            self.device.WriteExternal(address=0 + 5, data=repeat_array)

    def wait_for_command(self, length=None, repeat=None):

        # print self.device.ReadExternal(address = 0+1, size = 1)[0]
        if length is not None:
            if repeat is None:
                repeat = 1
            # print 'sleeping'
            time.sleep((length + 500) * 0.000000025 * repeat)  # TODO: optimize wait time
        while not self.is_ready:
            pass

    @property
    def is_ready(self):
        return (self.device.ReadExternal(address=0 + 1, size=1)[0] & 0x01) == 1

    def global_reset(self):
        '''FEI4 Global Reset

        Special function to do a global reset on FEI4. Sequence of commands has to be like this, otherwise FEI4B will be left in weird state.
        '''
        logging.info('Global reset of FE')
        commands = []
        commands.extend(self.register.get_commands("confmode"))
        # vthin_altfine, vthin_altcoarse = self.register.get_global_register_value("Vthin_AltFine"), self.register.get_global_register_value("Vthin_AltCoarse")
        # self.register.set_global_register_value("Vthin_AltFine", 255)
        # self.register.set_global_register_value("Vthin_AltCoarse", 255)
        # commands.extend(self.register.get_commands("wrregister", name = ["Vthin_AltFine", "Vthin_AltCoarse"]))
        commands.extend(self.register.get_commands("globalreset"))
        self.send_commands(commands)
        time.sleep(0.1)
        commands = []
        commands.extend(self.register.get_commands("confmode"))
        # self.register.set_global_register_value("Vthin_AltFine", vthin_altfine)
        # self.register.set_global_register_value("Vthin_AltCoarse", vthin_altcoarse)
        # commands.extend(self.register.get_commands("wrregister", name = ["Vthin_AltFine", "Vthin_AltCoarse"]))
        commands.extend(self.register.get_commands("runmode"))
        self.send_commands(commands)

    def configure_all(self, same_mask_for_all_dc=False):
        self.configure_global()<commit_msg>MAINT: cleanup<commit_after>import struct
import array
import math
import time
import numpy as np
import re
import logging

from utils.utils import bitarray_to_array

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)-8s] (%(threadName)-10s) %(message)s")


class FEI4RegisterUtils(object):
    def __init__(self, device, readout, register):
        self.device = device
        self.readout = readout
        self.register = register
        self.command_memory_byte_offset = 16
        self.command_memory_byte_size = 2048 - self.command_memory_byte_offset  # 16 bytes of register data
        self.zero_cmd_length = 1
        self.zero_cmd = self.register.get_commands("zeros", length=self.zero_cmd_length)[0]
        self.zero_cmd_padded = self.zero_cmd.copy()
        self.zero_cmd_padded.fill()

    def add_commands(self, x, y):
            return x + self.zero_cmd + y  # FE needs a zero bits between commands

    def add_byte_padded_commands(self, x, y):
            x_fill = x.copy()
            x_fill.fill()
            y_fill = y.copy()
            y_fill.fill()
            return x_fill + self.zero_cmd_padded + y_fill  # FE needs a zero between commands

    def concatenate_commands(self, commands, byte_padding=False):
        if byte_padding:
            return reduce(self.add_byte_padded_commands, commands)
        else:
            return reduce(self.add_commands, commands)

    def send_commands(self, commands, repeat=1, wait_for_finish=True, concatenate=True, byte_padding=False, clear_memory=False):
        if concatenate:
            commands_iter = iter(commands)
            concatenated_cmd = commands_iter.next()
            for command in commands_iter:
                concatenated_cmd_tmp = self.concatenate_commands((concatenated_cmd, command), byte_padding=byte_padding)
                if concatenated_cmd_tmp.length() > self.command_memory_byte_size * 8:
                    self.send_command(command=concatenated_cmd, repeat=repeat, wait_for_finish=wait_for_finish, set_length=True, clear_memory=clear_memory)
                    concatenated_cmd = command
                else:
                    concatenated_cmd = concatenated_cmd_tmp
            # send remaining commands
            self.send_command(command=concatenated_cmd, repeat=repeat, wait_for_finish=wait_for_finish, set_length=True, clear_memory=clear_memory)
        else:
            max_length = 0
            self.set_hardware_repeat(repeat)
            for command in commands:
                max_length = max(command.length(), max_length)
                self.send_command(command=command, repeat=None, wait_for_finish=wait_for_finish, set_length=True, clear_memory=False)
            if clear_memory:
                self.clear_command_memory(length=max_length)

    def send_command(self, command, repeat=1, wait_for_finish=True, set_length=True, clear_memory=False):
        if repeat is not None:
            self.set_hardware_repeat(repeat)
        # write command into memory
        command_length = self.set_command(command, set_length=set_length)
        # sending command
        self.start_command()
        # wait for command to be finished
        if wait_for_finish:
            self.wait_for_command(length=command_length, repeat=repeat)
        # clear command memory
        if clear_memory:
            self.clear_command_memory(length=command_length)

    def clear_command_memory(self, length=None):
        self.set_command(self.register.get_commands("zeros", length=(self.command_memory_byte_size * 8) if length is None else length)[0], set_length=False)

    def set_command_length(self, lenght):
        bit_length_array = array.array('B', struct.pack('H', lenght))
        self.device.WriteExternal(address=0 + 3, data=bit_length_array)

    def set_command(self, command, set_length=True, byte_offset=0):
        command_length = command.length()
        # set command bit length
        if set_length:
            self.set_command_length(command_length)
        # set command
        data = bitarray_to_array(command)
        if self.command_memory_byte_size < len(data) + byte_offset:
            raise ValueError('Length of command or offset is too big')
        self.device.WriteExternal(address=0 + self.command_memory_byte_offset + byte_offset, data=data)
        return command_length

    def start_command(self):
        self.device.WriteExternal(address=0 + 1, data=(0, ))

    def set_hardware_repeat(self, repeat=1):
        if repeat is not None:
            repeat_array = array.array('B', struct.pack('I', repeat))
            self.device.WriteExternal(address=0 + 5, data=repeat_array)

    def wait_for_command(self, length=None, repeat=None):

        # print self.device.ReadExternal(address = 0+1, size = 1)[0]
        if length is not None:
            if repeat is None:
                repeat = 1
            # print 'sleeping'
            time.sleep((length + 500) * 0.000000025 * repeat)  # TODO: optimize wait time
        while not self.is_ready:
            pass

    @property
    def is_ready(self):
        return (self.device.ReadExternal(address=0 + 1, size=1)[0] & 0x01) == 1

    def global_reset(self):
        '''FEI4 Global Reset

        Special function to do a global reset on FEI4. Sequence of commands has to be like this, otherwise FEI4B will be left in weird state.
        '''
        logging.info('Global reset of FE')
        commands = []
        commands.extend(self.register.get_commands("confmode"))
#         vthin_altfine, vthin_altcoarse = self.register.get_global_register_value("Vthin_AltFine"), self.register.get_global_register_value("Vthin_AltCoarse")
#         self.register.set_global_register_value("Vthin_AltFine", 255)
#         self.register.set_global_register_value("Vthin_AltCoarse", 255)
#         commands.extend(self.register.get_commands("wrregister", name=["Vthin_AltFine", "Vthin_AltCoarse"]))
        commands.extend(self.register.get_commands("globalreset"))
        self.send_commands(commands)
        time.sleep(0.1)
        commands = []
        commands.extend(self.register.get_commands("confmode"))
#         self.register.set_global_register_value("Vthin_AltFine", vthin_altfine)
#         self.register.set_global_register_value("Vthin_AltCoarse", vthin_altcoarse)
#         commands.extend(self.register.get_commands("wrregister", name=["Vthin_AltFine", "Vthin_AltCoarse"]))
        commands.extend(self.register.get_commands("runmode"))
        self.send_commands(commands)

    def configure_all(self, same_mask_for_all_dc=False):
        self.configure_global()
{'fraction_non_alphanumeric': 0.09287387591626993, 'fraction_numerical': 0.01012619965238419, 'mean_word_length': 2.9981873111782478, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, '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': '1178399', 'n_tokens_mistral': 4034, 'n_tokens_neox': 3837, 'n_words': 790}