<commit_before>import zmq
import numpy as np
import random
import string
try:
    import blosc
    HAVE_BLOSC = True
except ImportError:
    HAVE_BLOSC = False

from .sharedarray import SharedArray
from .client import OutputStreamProxy

import weakref

default_stream = dict( protocol = 'tcp', interface = '127.0.0.1', port = '*',
                        transfermode = 'plaindata', streamtype = 'analogsignal',
                        dtype = 'float32', shape = (-1, 1), timeaxis = 0, 
                        compression ='', scale = None, offset = None, units = '',
                        sampling_rate = 1.,)


common_doc = """
    Parameters
    ----------
    protocol : 'tcp', 'udp', 'inproc' or 'inpc' (linux only)
        The type of protocol user for the zmq.PUB socket
    interface : str
        The bind adress for the zmq.PUB socket
    port : str
        The port for the zmq.PUB socket
    transfermode: 'plain_data', 'sharedarray', (not done 'shared_cuda_buffer' or 'share_opencl_buffer')
        The method used for data transfer:
            * 'plain_data': data are sent over a plain socket in two parts: (frame index, data).
            * 'sharedarray': data are stored in shared memory in a ring buffer and the current frame index is sent over the socket.
            * 'shared_cuda_buffer': data are stored in shared Cuda buffer and the current frame index is sent over the socket.
            * 'share_opencl_buffer': data are stored in shared OpenCL buffer and the current frame index is sent over the socket.
    streamtype: 'analogsignal', 'digitalsignal', 'event' or 'image/video'
        The type of data to be transferred.
    dtype: str ('float32','float64', [('r', 'uint16'), ('g', 'uint16'), , ('b', 'uint16')], ...)
        The numpy.dtype of the data buffer. It can be a composed dtype for event or images.
    shape: list
        The shape of each data frame. Unknown dim are -1 in case of variable chunk.
            * for image it is (-1, H, W), (n_frames, H, W), or (H, W).
            * for analogsignal it can be (n_samples, n_channels) or (-1, n_channels)
    timeaxis: int
        The index of the axis that represents time within a single data chunk, or
        -1 if the chunk lacks this axis (in this case, each chunk represents exactly one
        timepoint).
    compression: '', 'blosclz', 'blosc-lz4', 'mp4', 'h264'
        The compression for the data stream, the default is no compression ''.
    scale: float
        An optional scale factor + offset to apply to the data before it is sent over the stream.
        real_data = offset + scale * data
    offset:
        See scale.
    units: str
        Units of the stream. Mainly used for 'analogsignal'.
    sampling_rate: float or None
        Sampling rate of the stream in Hz.
    sampling_interval: float or None
        
    
    Parameters when using `transfermode` = 'sharedarray'
    -----
    sharedarray_shape: tuple
        Shape of the SharedArray
    ring_buffer_method: 'double' or 'single'
        Method for the ring buffer:
            *  'double' : there 2 ring buffer concatenated. This ensure continuous chunk whenever the sample position.
            * 'single': standart ring buffer
        Note that, The ring buffer is along `timeaxis` for each case. And in case, of 'double', concatenated axis is also `timeaxis`.
    shm_id : str or int (depending on platform)
        id of the SharedArray
        
    
    
"""

    

class OutputStream:
    """A OutputStream is a helper class to send data.
    """
    def __init__(self, spec = {}, node = None, name = None):
        self.configured = False
        self.spec = spec # this is a priori stream params, and must be change when Node.configure
        if node:
            self.node = weakref.ref(node)
        else:
            self.node = None
        self.name = name
    
    def configure(self, **kargs):
        """
        Configure the output stream.
        """+common_doc
        
        self.params = dict(default_stream)
        self.params.update(self.spec)
        self.params.update(kargs)
        if self.params['protocol'] in ('inproc', 'ipc'):
            pipename = u'pyacq_pipe_'+''.join(random.SystemRandom().choice(string.ascii_uppercase + string.digits) for _ in range(24))
            self.params['interface'] = pipename
            self.url = '{protocol}://{interface}'.format(**self.params)
        else:
            self.url = '{protocol}://{interface}:{port}'.format(**self.params)
        context = zmq.Context.instance()
        self.socket = context.socket(zmq.PUB)
        self.socket.bind(self.url)
        self.addr = self.socket.getsockopt(zmq.LAST_ENDPOINT).decode()
        self.port = self.addr.rpartition(':')[2]
        self.params['port'] = self.port
        
        if self.params['transfermode'] == 'plaindata':
            self.sender = PlainDataSender(self.socket, self.params)
        elif self.params['transfermode'] == 'sharedarray':
            self.sender = SharedArraySender(self.socket, self.params)

        self.configured = True
        if self.node and self.node():
            self.node().after_output_configure(self.name)

    def send(self, index, data):
        """
        Send the data chunk and its frame index.
        
        Parameters
        ----------
        index: int
            The absolut sample index. If the chunk is multiple sample then index is the last one (head).
        data: np.ndarray or bytes
            The chunk of data to be send
        """
        self.sender.send(index, data)

    def close(self):
        """
        Close the output.
        This close the socket and release the sharedarray if necessary.
        """
        self.sender.close()
        self.socket.close()
        del self.socket
        del self.sender

class InputStream:
    """InputStream is a helper class to receive data.
    """
    def __init__(self, spec = {}, node = None, name = None):
        self.configured = False
        self.spec = spec
        if node:
            self.node = weakref.ref(node)<commit_msg>explicit not None<commit_after>import zmq
import numpy as np
import random
import string
try:
    import blosc
    HAVE_BLOSC = True
except ImportError:
    HAVE_BLOSC = False

from .sharedarray import SharedArray
from .client import OutputStreamProxy

import weakref

default_stream = dict( protocol = 'tcp', interface = '127.0.0.1', port = '*',
                        transfermode = 'plaindata', streamtype = 'analogsignal',
                        dtype = 'float32', shape = (-1, 1), timeaxis = 0, 
                        compression ='', scale = None, offset = None, units = '',
                        sampling_rate = 1.,)


common_doc = """
    Parameters
    ----------
    protocol : 'tcp', 'udp', 'inproc' or 'inpc' (linux only)
        The type of protocol user for the zmq.PUB socket
    interface : str
        The bind adress for the zmq.PUB socket
    port : str
        The port for the zmq.PUB socket
    transfermode: 'plain_data', 'sharedarray', (not done 'shared_cuda_buffer' or 'share_opencl_buffer')
        The method used for data transfer:
            * 'plain_data': data are sent over a plain socket in two parts: (frame index, data).
            * 'sharedarray': data are stored in shared memory in a ring buffer and the current frame index is sent over the socket.
            * 'shared_cuda_buffer': data are stored in shared Cuda buffer and the current frame index is sent over the socket.
            * 'share_opencl_buffer': data are stored in shared OpenCL buffer and the current frame index is sent over the socket.
    streamtype: 'analogsignal', 'digitalsignal', 'event' or 'image/video'
        The type of data to be transferred.
    dtype: str ('float32','float64', [('r', 'uint16'), ('g', 'uint16'), , ('b', 'uint16')], ...)
        The numpy.dtype of the data buffer. It can be a composed dtype for event or images.
    shape: list
        The shape of each data frame. Unknown dim are -1 in case of variable chunk.
            * for image it is (-1, H, W), (n_frames, H, W), or (H, W).
            * for analogsignal it can be (n_samples, n_channels) or (-1, n_channels)
    timeaxis: int
        The index of the axis that represents time within a single data chunk, or
        -1 if the chunk lacks this axis (in this case, each chunk represents exactly one
        timepoint).
    compression: '', 'blosclz', 'blosc-lz4', 'mp4', 'h264'
        The compression for the data stream, the default is no compression ''.
    scale: float
        An optional scale factor + offset to apply to the data before it is sent over the stream.
        real_data = offset + scale * data
    offset:
        See scale.
    units: str
        Units of the stream. Mainly used for 'analogsignal'.
    sampling_rate: float or None
        Sampling rate of the stream in Hz.
    sampling_interval: float or None
        
    
    Parameters when using `transfermode` = 'sharedarray'
    -----
    sharedarray_shape: tuple
        Shape of the SharedArray
    ring_buffer_method: 'double' or 'single'
        Method for the ring buffer:
            *  'double' : there 2 ring buffer concatenated. This ensure continuous chunk whenever the sample position.
            * 'single': standart ring buffer
        Note that, The ring buffer is along `timeaxis` for each case. And in case, of 'double', concatenated axis is also `timeaxis`.
    shm_id : str or int (depending on platform)
        id of the SharedArray
        
    
    
"""

    

class OutputStream:
    """A OutputStream is a helper class to send data.
    """
    def __init__(self, spec = {}, node = None, name = None):
        self.configured = False
        self.spec = spec # this is a priori stream params, and must be change when Node.configure
        if node is not None:
            self.node = weakref.ref(node)
        else:
            self.node = None
        self.name = name
    
    def configure(self, **kargs):
        """
        Configure the output stream.
        """+common_doc
        
        self.params = dict(default_stream)
        self.params.update(self.spec)
        self.params.update(kargs)
        if self.params['protocol'] in ('inproc', 'ipc'):
            pipename = u'pyacq_pipe_'+''.join(random.SystemRandom().choice(string.ascii_uppercase + string.digits) for _ in range(24))
            self.params['interface'] = pipename
            self.url = '{protocol}://{interface}'.format(**self.params)
        else:
            self.url = '{protocol}://{interface}:{port}'.format(**self.params)
        context = zmq.Context.instance()
        self.socket = context.socket(zmq.PUB)
        self.socket.bind(self.url)
        self.addr = self.socket.getsockopt(zmq.LAST_ENDPOINT).decode()
        self.port = self.addr.rpartition(':')[2]
        self.params['port'] = self.port
        
        if self.params['transfermode'] == 'plaindata':
            self.sender = PlainDataSender(self.socket, self.params)
        elif self.params['transfermode'] == 'sharedarray':
            self.sender = SharedArraySender(self.socket, self.params)

        self.configured = True
        if self.node and self.node():
            self.node().after_output_configure(self.name)

    def send(self, index, data):
        """
        Send the data chunk and its frame index.
        
        Parameters
        ----------
        index: int
            The absolut sample index. If the chunk is multiple sample then index is the last one (head).
        data: np.ndarray or bytes
            The chunk of data to be send
        """
        self.sender.send(index, data)

    def close(self):
        """
        Close the output.
        This close the socket and release the sharedarray if necessary.
        """
        self.sender.close()
        self.socket.close()
        del self.socket
        del self.sender

class InputStream:
    """InputStream is a helper class to receive data.
    """
    def __init__(self, spec = {}, node = None, name = None):
        self.configured = False
        self.spec = spec
        if node is not None:
            self.node = weakref.ref(node)
{'fraction_non_alphanumeric': 0.1018564153113192, 'fraction_numerical': 0.005749958928864794, 'mean_word_length': 2.1980562122406093, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, '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': '1162272', 'n_tokens_mistral': 3402, 'n_tokens_neox': 3348, 'n_words': 1241}