<commit_before># -*- coding: utf-8 -*-
"""
    pyvisa-py.tcpip
    ~~~~~~~~~~~~~~~~

    TCPIP Session implementation using Python Standard library.


    :copyright: 2014 by PyVISA-py Authors, see AUTHORS for more details.
    :license: MIT, see LICENSE for more details.
"""

import random

from pyvisa import constants, attributes

from .sessions import Session, UnknownAttribute
from .protocols import vxi11


StatusCode = constants.StatusCode
SUCCESS = StatusCode.success


@Session.register(constants.InterfaceType.tcpip, 'INSTR')
class TCPIPSession(Session):
    """A TCPIP Session that uses the network standard library to do the low level communication.
    """

    lock_timeout = 1000
    timeout = 1000
    client_id = None

    link = None
    max_recv_size = 1024

    @staticmethod
    def list_resources():
        # TODO: is there a way to get this?
        return []

    def after_parsing(self):
        # TODO: board_number not handled
        # TODO: lan_device_name not handled
        self.interface = vxi11.CoreClient(self.parsed['host_address'])

        self.lock_timeout = 10000
        self.timeout = 10000
        self.client_id = random.getrandbits(31)

        (error, link,
         abort_port,
         max_recv_size) = self.interface.create_link(self.client_id, 0, self.lock_timeout,
                                                     self.parsed['lan_device_name'])

        if error:
            raise Exception("error creating link: %d" % error)

        self.link = link
        self.max_recv_size = min(max_recv_size, 2 ** 30) # 1GB

        for name in 'SEND_END_EN,TERMCHAR,TERMCHAR_EN'.split(','):
            attribute = getattr(constants, 'VI_ATTR_' + name)
            self.attrs[attribute] = attributes.AttributesByID[attribute].default

    def close(self):
        self.interface.destroy_link(self.link)
        self.interface.close()
        self.link = None
        self.interface = None

    def read(self, count):
        """Reads data from device or interface synchronously.

        Corresponds to viRead function of the VISA library.

        :param count: Number of bytes to be read.
        :return: data read, return value of the library call.
        :rtype: bytes, VISAStatus
        """
        if count < self.max_recv_size:
            chunk_length = count
        else:
            chunk_length = self.max_recv_size

        flags = 0
        reason = 0

        if self.get_attribute(constants.VI_ATTR_TERMCHAR_EN)[0]:
            term_char, _ = self.get_attribute(constants.VI_ATTR_TERMCHAR)
        else:
            term_char = 0

        if self.term_char:
            flags = vxi11.OP_FLAG_TERMCHAR_SET
            term_char = str(self.term_char).encode('utf-8')[0]

        read_data = b''

        end_reason = vxi11.RX_END | vxi11.RX_CHR

        read_fun = self.interface.device_read

        while reason & end_reason == 0:
            error, reason, data = read_fun(self.link, chunk_length, self.timeout,
                                           self.lock_timeout, flags, term_char)

            if error:
                return read_data, StatusCode.error_io

            read_data += data
            count -= len(data)

            if count <= 0:
                break

            chunk_length = min(count, chunk_length)

        return read_data, SUCCESS

    def write(self, data):<commit_msg>fixed self.term_char access to term_char in TCPIPSession<commit_after># -*- coding: utf-8 -*-
"""
    pyvisa-py.tcpip
    ~~~~~~~~~~~~~~~~

    TCPIP Session implementation using Python Standard library.


    :copyright: 2014 by PyVISA-py Authors, see AUTHORS for more details.
    :license: MIT, see LICENSE for more details.
"""

import random

from pyvisa import constants, attributes

from .sessions import Session, UnknownAttribute
from .protocols import vxi11


StatusCode = constants.StatusCode
SUCCESS = StatusCode.success


@Session.register(constants.InterfaceType.tcpip, 'INSTR')
class TCPIPSession(Session):
    """A TCPIP Session that uses the network standard library to do the low level communication.
    """

    lock_timeout = 1000
    timeout = 1000
    client_id = None

    link = None
    max_recv_size = 1024

    @staticmethod
    def list_resources():
        # TODO: is there a way to get this?
        return []

    def after_parsing(self):
        # TODO: board_number not handled
        # TODO: lan_device_name not handled
        self.interface = vxi11.CoreClient(self.parsed['host_address'])

        self.lock_timeout = 10000
        self.timeout = 10000
        self.client_id = random.getrandbits(31)

        (error, link,
         abort_port,
         max_recv_size) = self.interface.create_link(self.client_id, 0, self.lock_timeout,
                                                     self.parsed['lan_device_name'])

        if error:
            raise Exception("error creating link: %d" % error)

        self.link = link
        self.max_recv_size = min(max_recv_size, 2 ** 30) # 1GB

        for name in 'SEND_END_EN,TERMCHAR,TERMCHAR_EN'.split(','):
            attribute = getattr(constants, 'VI_ATTR_' + name)
            self.attrs[attribute] = attributes.AttributesByID[attribute].default

    def close(self):
        self.interface.destroy_link(self.link)
        self.interface.close()
        self.link = None
        self.interface = None

    def read(self, count):
        """Reads data from device or interface synchronously.

        Corresponds to viRead function of the VISA library.

        :param count: Number of bytes to be read.
        :return: data read, return value of the library call.
        :rtype: bytes, VISAStatus
        """
        if count < self.max_recv_size:
            chunk_length = count
        else:
            chunk_length = self.max_recv_size

        flags = 0
        reason = 0

        if self.get_attribute(constants.VI_ATTR_TERMCHAR_EN)[0]:
            term_char, _ = self.get_attribute(constants.VI_ATTR_TERMCHAR)
        else:
            term_char = 0

        if term_char:
            flags = vxi11.OP_FLAG_TERMCHAR_SET
            term_char = str(self.term_char).encode('utf-8')[0]

        read_data = b''

        end_reason = vxi11.RX_END | vxi11.RX_CHR

        read_fun = self.interface.device_read

        while reason & end_reason == 0:
            error, reason, data = read_fun(self.link, chunk_length, self.timeout,
                                           self.lock_timeout, flags, term_char)

            if error:
                return read_data, StatusCode.error_io

            read_data += data
            count -= len(data)

            if count <= 0:
                break

            chunk_length = min(count, chunk_length)

        return read_data, SUCCESS

    def write(self, data):
{'fraction_non_alphanumeric': 0.08917010005885816, 'fraction_numerical': 0.015303119482048263, 'mean_word_length': 2.113605130554283, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1024707', 'n_tokens_mistral': 2041, 'n_tokens_neox': 1900, 'n_words': 511}