variable_server.py

"""
This module contains tools for communicating with a sim's variable
server, primarily via the Variable and VariableServer classes. See
https://nasa.github.io/trick/documentation/miscellaneous_trick_tools/Python-Variable-Server-Client.html for
a tutorial and examples.
"""

from collections import namedtuple
import os
import re
import socket
import struct
import threading
import time

# In Python 2, basestring is the parent for both str (ASCII) and unicode.
# In Python 3, str and unicode were unified into str, and basestring is gone.
try:
    basestring
except NameError:
    basestring = str

# In Python 3, itertools.izip became zip.
try:
    from itertools import izip as zip
except ImportError:
    pass

class VariableServerError(Exception):
    '''
    Variable Server communication I/O error.
    '''
    pass

class UnitsConversionError(VariableServerError):
    """
    Raised when a units conversion fails.

    Attributes
    ----------
    name : str
        The name of the variable for which the conversion failed.
    units : str
        The units to which the variable could not be converted.
    """
    def __init__(self, name, units):
        super(UnitsConversionError, self).__init__(
          '[{0}] cannot be converted to [{1}]'.format(name, units)
        )
        self.name = name
        self.units = units

class UnexpectedMessageError(VariableServerError):
    """
    Raised when a received message is not of the expected type.

    Attributes
    ----------
    expected_id : int
        The expected message indicator.
    actual_id : int
        The actual message indicator.
    """
    def __init__(self, expected_id, actual_id):
        super(UnexpectedMessageError, self).__init__(
          'Unexpected message received. Expected ID = {0}. Actual ID = {1}'
          .format(expected_id, actual_id)
        )
        self.expected_id = expected_id
        self.actual_id = actual_id

class ValueCountError(VariableServerError):
    """
    Raised when the number of received variable values does not match
    the expected count.

    Attributes
    ----------
    expected : int
        The expected count.
    actual : int
        The actual count.
    """
    def __init__(self, expected, actual):
        super(ValueCountError, self).__init__(
          'Number of values received ({0}) does not match expected ({1})'
          .format(actual, expected)
        )
        self.expected = expected
        self.actual = actual

def _create_enum(name, field_names, ordinal_values=True):
    """
    Create a namedtuple with automatic values.

    Parameters
    ----------
    name : str
        The name of the namedtuple.
    field_names : iterable of str
        The field names.
    ordinal_values : bool
        True to assign each field an ordinal number, starting at 0. This
        creates a classic number-based enum.
        False to use the field names themselves as the values for the
        fields. This creates a string-based enum.
    """
    return namedtuple(name, field_names)(
      *(range(len(field_names)) if ordinal_values else field_names)
    )

class Message(namedtuple('Message', ['indicator', 'data'])):
    """
    A message from the variable server.

    Attributes
    ----------
    indicator : int
        The indicator.
    data : str
        The rest of the message.
    """
    Indicator = _create_enum('Indicator', ['VAR_SEND', 'VAR_EXISTS'])

class Variable(object):
    """
    A variable whose value and units will be updated from the sim. You
    should not directly change any part of this class.

    Attributes
    ----------
    name : str
        The fully-qualified name.
    units : str
        The units. Use 'xx' to specify default units.

    Properties
    ----------
    value : type returned by the type_ parameter of __init__ (or None)
        The value.
    """

    def __init__(self, name, units=None, type_=str):
        """
        Create a new Variable.

        Parameters
        ----------
        name and units are as documented in this class's Attributes
        section.
        type_ : callable
            A callable that accepts one argument and is used to convert
            the stringified value from the variable server to the
            desired type. This can be a builtin like int or float or a
            custom function that does anything you like!
        """
        self.name = name
        self.units = units
        self._type = type_
        self._value = None

    @property
    def value(self):
        '''
        Get the converted value.
        '''
        return self._type(self._value)

    @value.setter
    def value(self, value):
        '''
        Set the value.
        '''
        self._value = value

    def __str__(self):
        return self.name

    def __repr__(self):
        return '{0} = {1}{2}'.format(
          self.name,
          self.value,
          ' {0}'.format(self.units) if self.units is not None else '')

def _connect_ipv4(hostname, port):
        # Prefer explicit IPv4; fall back to first AF_INET match
        for ai in socket.getaddrinfo(hostname, port, socket.AF_INET, socket.SOCK_STREAM):
            af, socktype, proto, _, sa = ai
            s = socket.socket(af, socktype, proto)
            try:
                s.connect(sa)
                return s
            except Exception:
                s.close()
        raise OSError(f"Could not connect to {hostname}:{port} over IPv4")

class VariableServer(object):
    """
    Send commands to and receive responses from a simulation's
    variable server.

    You must call close on instances of this class to release resources
    allocated during initialization.
    """

    Channel = _create_enum('Channel', ['ASYNC', 'SYNC', 'BOTH'], False)
    CopyMode = _create_enum('CopyMode', ['ASYNC', 'SCHEDULED', 'TOP_OF_FRAME'])

    
    
    def __init__(self, hostname, port):
        """
        Create a connection to the simulation variable server at
        host:port.

        Parameters
        ----------
        hostname : str
            The name of the machine that is running the simulation to
            which you want to connect.
        port : int
            The port on which the simulation's variable server is
            listening.
        """
        self._variables = []
        self._callbacks = {}
        self._error_callbacks = {}
        self._lock = threading.Lock()
        port = int(port)
        # self._synchronous_socket = socket.create_connection((hostname, port))
        # self._asynchronous_socket = socket.create_connection((hostname, port))
        self._synchronous_socket   = _connect_ipv4(hostname, port)
        self._asynchronous_socket  = _connect_ipv4(hostname, port)
        self._synchronous_file_interface = self._synchronous_socket.makefile()
        self._asynchronous_file_interface = self._asynchronous_socket.makefile()
        self._open = True
        self.pause(channel=self.Channel.SYNC)

        # Define a local function to be used by the sampling thread.
        def update_variables():
            '''
            Continuously update variables.
            '''
            while True:
                try:
                    values = self._read_values(False)
                except Exception as exception:
                    if self._open:
                        for function, args in self._error_callbacks.items():
                            function(*args[0], exception=exception, **args[1])
                    return
                # We must lock here to ensure that variables are not
                # removed while we are processing an update.
                with self._lock:
                    # If there are more values than variables, it must
                    # be that a variable was removed after this message
                    # was sent but before we processed it. Variables can
                    # be removed from any place within the list, so we
                    # don't know which "extra" values to ignore.
                    # We therefore discard the entire message.
                    #
                    # If there are fewer values than variables, it must
                    # be that a variable was added after this message
                    # was sent but before we processed it. Variables
                    # can only be appended to the list, so we know that
                    # any missing values are for variables at the end of
                    # the list. We can therefore still use all the
                    # values to update variables at the front of the
                    # list.
                    #
                    # We could still end up assigning values to the
                    # wrong variables if someone removed AND added
                    # variables after a message was sent but before we
                    # processed it, but the probability of someone
                    # doing that doesn't justify the work at this point.
                    # Besides, it would be corrected with the next
                    # message.
                    if len(values) <= len(self._variables):
                        for variable, value in zip(
                          self._variables, values):
                            variable.value, variable.units = \
                              _parse_value(value)

                        for function, args in self._callbacks.items():
                            function(*args[0], **args[1])

        # Start a thread that listens for data from the variable server,
        # updates the variables being sampled, and notifies listeners.
        # This thread can only be terminated by calling close. A Python
        # process cannot terminate while non-daemon threads are running,
        # so make this tread a deamon in case the user fails to call
        # close when finished with this instance.
        self._thread = threading.Thread(
          target=update_variables, name='Asynchronous Variable Sampler')
        self._thread.daemon = True
        self._thread.start()

    def __del__(self):
        """
        Call close in case the user forgot. Don't rely on this to clean
        up for you. You should really explicitly call close yourself.
        """
        try:
            self.close()
        except:
            pass

    def __enter__(self):
        return self

    def __exit__(self, *args):
        self.close()

    def get_value(self, name, units=None, type_=str):
        """
        Get the value of the named variable. If units are specified, the
        value is converted if possible. This function provides a simple
        interface and is most useful when you just want to get the value
        of one variable and don't require periodic sampling. For a more
        powerful but complex function, see get_values. To sample values
        periodically, see add_variables.

        Parameters
        ----------
        name, units, and type_ are as documented in Variable.

        Returns
        -------
        type returned by type_
            The result of calling type_ on the variable's value.

        Raises
        ------
        IOError
            If the remote endpoint has closed the connection.
        UnexpectedMessageError
            If the next message is not a set of variable values.
        ValueCountError
            If more than one value is received.
        UnitsConversionError
            If units are specified and the conversion fails.
        Additional errors may be raised by type_.

        Examples
        --------
        Get value (default type is string):
        >>> from variable_server import VariableServer
        >>> with VariableServer('localhost', 7000) as vs:
        ...     vs.get_value('ball.obj.state.input.position[0]')
        '5'

        Get value as an int:
        >>> from variable_server import VariableServer
        >>> with VariableServer('localhost', 7000) as vs:
        ...     vs.get_value('ball.obj.state.input.position[0]',
        ...     type_=int)
        5

        Get value as a float in millimeters:
        >>> from variable_server import VariableServer
        >>> with VariableServer('localhost', 7000) as vs:
        ...     vs.get_value('ball.obj.state.input.position[0]',
        ...     units='mm', type_=float)
        5000.0

        Convert value with custom function:
        >>> from variable_server import VariableServer
        >>> with VariableServer('localhost', 7000) as vs:
        ...     vs.get_value('ball.obj.state.input.position[0]',
        ...     type_=lambda x: int(x) * 2)
        10
        """

        # add the variable and poll its value
        self._var_add(name, units, self.Channel.SYNC)
        self._var_send()
        self._var_clear(self.Channel.SYNC)

        # make sure we only got one
        values = self._read_values()
        _assert_value_count(1, len(values))

        # check for units conversion
        value, actual_units = _parse_value(values[0])
        _assert_units_conversion(name, units, actual_units)

        return type_(value)

    def set_value(self, name, value, units=None):
        """
        Set the value of the named variable. If units are specified, the
        value is converted if possible. If the convserion fails, the
        value is unchanged, and no error is raised.

        Attributes
        ----------
        name : str
            The fully-qualified name.
        value : any
            The value. This should be of the type expected by the
            variable server.
        units : str
            The units.
        """
        self.send(
          'trick.var_set("{0}", {1}{2})'.format(
            name,
            '"{0}"'.format(value) if isinstance(value, basestring) else value,
            ', "{0}"'.format(units) if units is not None else ''))

    def get_values(self, *variables):
        """
        Get the values of variables. If units are specified, values are
        converted if possible. Each argument is also updated in place,
        so you can ignore the returned list of values if you were just
        going to store them anyway. The list is useful, for example, for
        inlining the results, eliminating the need to update and use
        the variables in separate statements.

        This function is more efficient than calling get_value
        for each variable, but has a steeper learning curve. It is most
        useful when you want to fetch multiple variables' values and
        units and don't require periodic sampling. To sample values
        periodically, see add_variables.

        Parameters
        ----------
        variables : zero or more Variables
            The variables for which to fetch values and units.

        Returns
        -------
        [any]
            A list of the requested values in order. The type of each
            element is that returned by the corresponding Variable's
            type_.

        Raises
        ------
        IOError
            If the remote endpoint has closed the connection.
        UnexpectedMessageError
            If the next message is not a set of variable values.
        ValueCountError
            If the number of received values does not match the number
            of variables.
        UnitsConversionError
            If units are specified and the conversion fails. In this
            case, variables before the error will have been updated,
            but variables after will not. The variable for which the
            error occurred will not be updated.
        Additional errors may be raised by each Variables's type_, but
        they do not prevent the rest of the Variables from being
        updated. Variables whose type_ conversion fails will raise an
        error every time their value property is accessed.

        Example
        -------
        >>> from variable_server import Variable, VariableServer
        >>> position = Variable('ball.obj.state.input.position[0]',
        ... type_=int)
        >>> mass = Variable('ball.obj.state.input.mass', units='g',
        ... type_=float)
        >>> with VariableServer('localhost', 7000) as vs:
        ...     vs.get_values(position, mass)
        [5, 10000.0]
        >>> position
        ball.obj.state.input.position[0] = 5 m
        >>> mass
        ball.obj.state.input.mass = 10000.0 g
        """

        # check for zero arguments
        if not variables:
            return []

        # add all the variables and poll their values
        for variable in variables:
            self._var_add(
              variable.name,
              variable.units if variable.units is not None else 'xx',
              self.Channel.SYNC)
        self._var_send()
        self._var_clear(self.Channel.SYNC)

        # make sure we got as many as expected
        values = self._read_values()
        _assert_value_count(len(variables), len(values))

        # update each Variable, checking units conversions
        for variable, entry in zip(variables, values):
            value, units = _parse_value(entry)
            if variable.units is not None:
                _assert_units_conversion(variable.name, variable.units, units)
            else:
                variable.units = units
            variable.value = value

        return [variable.value for variable in variables]

    def add_variables(self, *variables):
        """
        Immediately update and begin periodically sampling the given
        variables. This class retains references to all passed-in
        variables and updates their values and units when new values
        arrive from the variable server. All sampling is performed on a
        separate thread, but accessing a variable's fields is always
        safe. Adding variables which are already being sampled has no
        effect. To set the sampling period, see set_period. To register
        a function to be called whenever variables are updated, see
        register_callback.

        Parameters
        ----------
        variables : zero or more Variables
            The variables to begin sampling.

        Raises
        ------
        IOError
            If the remote endpoint has closed the connection.
        UnexpectedMessageError
            If the next message is not a set of variable values.
        ValueCountError
            If the number of received values does not match the number
            of variables.
        UnitsConversionError
            If units are specified and the conversion fails. In this
            case, variables before the error will have been updated,
            but variables after will not. The variable for which the
            error occurred will not be updated.
        Additional errors may be raised by each Variables's type_, but
        they do not prevent the rest of the Variables from being
        updated. Variables whose type_ conversion fails will raise an
        error every time their value property is accessed.

        If any error occurs, no variables are scheduled for sampling.
        """

        # remove existing variables
        variables = [variable for variable in variables
                     if variable not in self._variables]

        # check for type_ and units conversion errors
        self.get_values(*variables)

        for variable in variables:
            # No lock is needed here because:
            # - appending to the variables list while
            #   update_variables is executing does not invalidate
            #   the length check
            # - zip is bounded by the number of values, and the
            #   length check ensures there are at least as many
            #   variables as values
            self._variables.append(variable)
            self._var_add(
              variable.name,
              variable.units if variable.units is not None else 'xx')

    def remove_variables(self, *variables):
        """
        Stop sampling the given variables. Removing variables that are
        not being sampled has no effect.

        Parameters
        ----------
        variables : zero or more Variables
            The variables to stop sampling.
        """
        for variable in variables:
            if variable in self._variables:
                # Variables must not be removed while update_variables
                # is processing an update. See its comment for an
                # explanation.
                with self._lock:
                    self._variables.remove(variable)
                self._var_remove(variable.name)

    def remove_all_variables(self):
        """
        Stop sampling all variables. To merely suspend sampling,
        see pause.
        """
        self._var_clear()
        # No lock is needed here because:
        # - This assignment is atomic.
        # - If update_variables has already called zip, the list it is
        #   ierating over is unchanged as we're assigning a new list
        #   here instead of clearing the shared reference.
        # - If update_variables has not yet called zip, when it does,
        #   zip will return an empty generator, terminating the loop.
        self._variables = []

    def set_units(self, name, units):
        """
        Set the units in which the named variable is sampled. This only
        applies to Variables being periodically sampled.
        See add_variables.

        Parameters
        ----------
        name : str
            The variable's name.
        units : str
            The units to which to convert the sampled value.
        """
        self.send('trick.var_units("{0}", "{1}")'.format(name, units),
                  self.Channel.ASYNC)

    def set_period(self, period):
        """
        Set the sampling period (in seconds).

        Parameters
        ----------
        period : float
            The inverse of the rate (in Hz) at which you want to sample
            variable values.
        """
        self.send('trick.var_cycle({0})'.format(float(period)),
                  self.Channel.ASYNC)

    def register_callback(self, function, args=None, kwargs=None):
        """
        Call function whenever new variable values are sampled.
        Registering an aleady-registered function replaces its existing
        registration. The order in which functions are called is not
        specified. Functions are executed on the asynchronous sampling
        thread.

        Paramaters
        ----------
        function : callable
            The function to call.
        args : tuple
            The positional arguments to be passed to the function.
        kwargs : dict
            The keyword arguments to be passed to the function.
        """
        if args is None:
            args = []
        if kwargs is None:
            kwargs = {}
        self._callbacks[function] = args, kwargs

    def deregister_callback(self, function):
        """
        Do not call function whenever new variable values are sampled.
        Deregistering an unregistered function has no effect.

        Parameters
        ----------
        function : any
            A function previously added via register_callback.
        """
        self._callbacks.pop(function, None)

    def register_error_callback(self, function, args=None, kwargs=None):
        """
        Call function if an error occurs while sampling variable values.
        Registering an aleady-registered function replaces its existing
        registration. The order in which functions are called is not
        specified. Functions are executed on the asynchronous sampling
        thread.

        Paramaters
        ----------
        function : callable
            The function to call. It must accept a keyword argument
            named 'exception' which will contain the error.
        args : tuple
            The positional arguments to be passed to the function.
        kwargs : dict
            The keyword arguments to be passed to the function.
        """
        if args is None:
            args = []
        if kwargs is None:
            kwargs = {}
        self._error_callbacks[function] = args, kwargs

    def deregister_error_callback(self, function):
        """
        Do not call function if an error occurs while sampling variable
        values. Deregistering an unregistered function has no effect.

        Parameters
        ----------
        function : any
            A function previously added via register_error_callback.
        """
        self._error_callbacks.pop(function, None)


    def pause(self, pause=True, channel=Channel.ASYNC):
        """
        Pause or unpause sampling.

        Parameters
        ----------
        pause : bool
            True to pause sampling.
            False to resume sampling.
        channel : Channel
            The channel to affect. You should almost certainly leave
            this as the default.
        """
        self.send('trick.var_{0}pause()'.format('' if pause else 'un'),
                  channel)

    def set_debug(self, level, channel=Channel.BOTH):
        """
        Set the debugging level. This effects how much information the
        sim's variable server outputs to the sim's standard output
        stream. It is not a debugging option for this Python class and
        does not produce output on this process.

        Parameters
        ----------
        level : int
            The debugging level. 0 = no debugging.
        channel : Channel
            The channel to affect.
        """
        self.send('trick.var_debug({0})'.format(int(level)), channel)

    def set_tag(self, tag):
        """
        Set the identifier for this variable server client. The tag is
        used in log files to associate each message with its sender.

        Parameters
        ----------
        tag : str
            An identifier for this client.
        """
        for channel in [self.Channel.SYNC, self.Channel.ASYNC]:
            self.send(
              'trick.var_set_client_tag("{0}_{1}")'.format(tag, channel),
              channel)

    def set_copy_mode(self, mode=CopyMode.ASYNC, channel=Channel.BOTH):
        """
        Set the method by which variable values are copied.

        Parameters
        ----------
        mode : CopyMode
            ASYNC
            Values are copied by an independent thread. This has no
            effect on simulation execution, but values within a set may
            not all be from the same sim frame.

            SCHEDULED
            Values are copied as an automatic_last job in the sim's main
            thread. Values within a set are guaranteed to be from the
            same sim frame.

            TOP_OF_FRAME
            Values are copied at the top of each sim frame in the main
            thread. Values within a set are guaranteed to be from the
            same sim frame.

        channel : Channel
            The channel to affect.
        """
        self.send('trick.var_set_copy_mode({0})'.format(int(mode)), channel)

    def send_on_copy(self, enable=True):
        """
        Set the method by which variable values are sent.

        Parameters
        ----------
        enable : bool
            True to cause the variable server to send variable values
            immediately after copying them, using the same thread, which
            will impact simulation execution if the copy mode is not
            ASYNCHRONOUS.
            False to send values asynchronously on an independent
            thread.
        """
        self.send('trick.var_set_write_mode({0})'.format(bool(enable)),
                  self.Channel.ASYNC)

    def validate_addresses(self, validate=True, channel=Channel.BOTH):
        """
        Set whether or not addresses are validated.

        When bool(validate) is True, variable addresses will be
        validated against the memory manager before being read. Those
        not known to Trick are considered invalid and their values are
        reported as "BAD_REF". This prevents malformed variable
        requests, such as pointers with invalid offsets, from causing
        segmentation faults.

        Parameters
        ----------
        validate : bool
            The desired validation state.
        channel : Channel
            The channel to affect.
        """
        self.send('trick.var_validate_address({0})'.format(bool(validate)),
                  channel)

    def variable_exists(self, name):
        """
        Determine if name is known to the memory manager.

        Parameters
        ----------
        name : str
            The variable's name.

        Returns
        -------
        bool
            True if name has been registered with the memory manager.
            False if not.

        Raises
        ------
        IOError
            If the remote endpoint has closed the connection.
        UnexpectedMessageError
            If the next message on the synchronous channel is not a
            response to a var_exists inquiry.
        """
        self.send('trick.var_exists("{0}")'.format(name))
        message = self.readline()
        _assert_message_type(message, Message.Indicator.VAR_EXISTS)
        return message.data == '1'

    def freeze(self, freeze=True):
        """
        Freeze or unfreeze the sim.

        Parameters
        ----------
        freeze : bool
            True to freeze the sim.
            False to unfreeze it.
        """
        self.send('trick.exec_{0}()'
                  .format('freeze' if bool(freeze) else 'run'))

    def checkpoint(self, filename):
        """
        Dump a checkpoint.

        Parameters
        ----------
        filename : str
            The checkpoint file name. The checkpoint will be saved in the
            directory containing the sim's input file.
        """
        self.send('trick.checkpoint("' + filename + '")')

    def load_checkpoint(self, filename):
        """
        Load a checkpoint. The path is relative to the directory containing
        the sim's S_main executable.

        Parameters
        ----------
        filename : str
            The checkpoint file name.
        """
        self.send('trick.load_checkpoint("' + filename + '")')

    def enable_real_time(self, enable=True):
        """
        Toggle real-time execution.

        Parameters
        ----------
        enable : bool
            True so synchronize sim execution with the real-time clock.
            False to run as quickly as possible.
        """
        self.send('trick.real_time_{0}able()'
                  .format('en' if bool(enable) else 'dis'))

    def send(self, command, channel=Channel.SYNC):
        """
        Append a newline to command and send it, blocking until all data
        has been sent. Calling this directly is only necessary if you
        want to send a command for which there is not already a method
        in this class. You must not send a command on the asynchronous
        channel that produces a response, as the response will be
        consumed by the variable-sampling thread, which will likely
        cause an error. You must ensure the socket is left in a "clean"
        state before calling other methods of this class. If incoming
        data is left on either channel, it is likely to cause other
        methods to fail since they won't get what they're expecting when
        they go to read.

        Parameters
        ----------
        command : str
            The command to send.
        channel : Channel
            The channel on which to send. You should almost certainly
            leave this as the default.
        """
        command = '{0}\n'.format(command)

        for channel in {
          self.Channel.SYNC: [self._synchronous_socket],
          self.Channel.ASYNC: [self._asynchronous_socket],
          self.Channel.BOTH: [self._synchronous_socket,
                              self._asynchronous_socket]
        }[channel]:
            channel.sendall(command.encode())

    def readline(self, synchronous_channel=True):
        """
        Read a newline-terminated line, blocking if necessary. Calling
        this directly is only necessary if you have directly called
        send and expect a response from the variable server. The newline
        character is stripped.

        Parameters
        ----------
        synchronous_channel : bool
            True to read from the synchronous channel.
            False to read from the asynchronous channel.

        Returns
        -------
        Message
            The next available message.

        Raises
        ------
        IOError
            If the remote endpoint has closed the connection.
        """
        file_interface = (self._synchronous_file_interface
                          if synchronous_channel
                          else self._asynchronous_file_interface)
        line = file_interface.readline()
        if not line:
            raise IOError("The remote endpoint has closed the connection")
        line = line.rstrip(os.linesep).split('\t', 1)
        return Message(int(line[0]), line[1])

    def _var_add(self, name, units=None, channel=Channel.ASYNC):
        """
        Send a var_add command to the variable server.

        Parameters
        ----------
        name : str
            The variable's name.
        units : str
            The units in which to report the variable's value.
        channel : Channel
            The channel to affect.
        """
        self.send(
          'trick.var_add("{0}"{1})'
          .format(name, ', "{0}"'.format(units) if units is not None else ''),
          channel)

    def _var_remove(self, name):
        """
        Send a var_remove command to the variable server.

        Parameters
        ----------
        name : str
            The variable's name.
        channel : Channel
            The channel to affect.
        """
        self.send('trick.var_remove("{0}")'.format(name),
                  self.Channel.ASYNC)

    def _var_send(self, channel=Channel.SYNC):
        """
        Send a var_send command to the variable server.

        Parameters
        ----------
        channel : Channel
            The channel to affect.
        """
        self.send('trick.var_send()', channel)

    def _read_values(self, synchronous_channel=True):
        """
        Read a set of variable values.

        Parameters
        ----------
        synchronous_channel : bool
            True to read from the synchronous channel.
            False to read from the asynchronous channel.

        Returns
        -------
        [str]
            A list of values, which may include units.

        Raises
        ------
        IOError
            If the remote endpoint has closed the connection.
        UnexpectedMessageError
            If the next message is not a set of variable values.
        """
        message = self.readline(synchronous_channel)
        _assert_message_type(message, Message.Indicator.VAR_SEND)
        return message.data.split('\t')

    def _var_clear(self, channel=Channel.ASYNC):
        """
        Send a var_clear command to the variable server.

        Parameters
        ----------
        channel : Channel
            The channel to affect.
        """
        self.send('trick.var_clear()', channel)

    def close(self):
        """
        Close the connection and release resources. No methods can be
        called after this one. A new connection can be established only
        by creating a new instance.
        """
        # In Python 3, self._asynchronous_file_interface.close() hangs if
        # shutdown isn't called first. Something to do with self._thread
        # always blocking on a read from the asynchronous_socket. I suspect
        # Python 3's new io module.
        self._open = False
        self._synchronous_socket.shutdown(socket.SHUT_RDWR)
        self._asynchronous_socket.shutdown(socket.SHUT_RDWR)
        self._synchronous_file_interface.close()
        self._asynchronous_file_interface.close()
        self._synchronous_socket.close()
        self._asynchronous_socket.close()
        self._thread.join()

    def __str__(self):
        return 'VariableServer' + str(self._synchronous_socket.getpeername())

def _join_mcast(sock, group='224.3.14.15', port=9265, iface_ipv4='0.0.0.0'):
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    sock.bind(('', port))
    # Pick a concrete iface IPv4; e.g., your host LAN IP or 127.0.0.1 if locally looped
    mreq = struct.pack('=4s4s', socket.inet_aton(group), socket.inet_aton(iface_ipv4))
    sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq)
        
def find_simulation(host=None, port=None, user=None, pid=None,
                   version=None, sim_directory=None, s_main=None,
                   input_file=None, tag=None, timeout=None):
    """
    Listen for simulations on the multicast channel over which all sims broadcast
    their existence. Connect to the one that matches the provided arguments that
    are not None.

    If there are multiple matches, connect to the first one we happen to find.
    If all arguments are None, connect to the first sim we happen to find.
    Such matches will be non-deterministic.

    Parameters
    ----------
    host : str
        Host name of the machine on which the sim is running as reported by
        Trick.
    port : int
        Variable Server port.
    user : str
        Simulation process user.
    pid : int
        The sim's process ID.
    version : str
        Trick version.
    sim_directory : str
        SIM_* directory. If this starts with /, it will be considered an
        absolute path.
    s_main : str
        Filename of the S_main* executable. Not an absolute path.
    input_file : str
        Path to the input file relative to the simDirectory.
    tag : str
        Simulation tag.
    timeout : positive float or None
        How long to look for the sim before giving up. Pass None to wait
        indefinitely.

    Returns
    -------
    VariableServer
        A VariableServer connected to the sim matching the specified
        parameters.

    Raises
    ------
    socket.timeout
        If a timeout occurs.
    """
    clock = time.time()
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    

    _join_mcast(sock)
    
    # sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    # sock.bind(('', 9265))
    # sock.setsockopt(
    #   socket.IPPROTO_IP,
    #   socket.IP_ADD_MEMBERSHIP,
    #   struct.pack('=4sl', socket.inet_aton('224.3.14.15'), socket.INADDR_ANY))
    file_interface = sock.makefile()

    def candidate_matches(candidate):
        for parameter, candidate_parameter in zip(
          [host, port, user, pid, sim_directory,
           s_main, input_file, version, tag],
          candidate) :
            if parameter is not None and str(parameter) != candidate_parameter:
                return False
        return True

    # the socket will clean itself up when it's garbage-collected
    while True:
        if timeout is not None:
            timeout -= (time.time() - clock)
            if timeout < 0:
                raise socket.timeout
            clock = time.time()
            sock.settimeout(timeout)

        # 0: host
        # 1: port
        # 2: user
        # 3: pid
        # 4: SIM_*
        # 5: S_main*
        # 6: RUN_*
        # 7: version
        # 8: tag
        candidate = file_interface.readline().split('\t')[:9]

        if sim_directory is not None and not os.path.isabs(str(sim_directory)):
            candidate[4] = os.path.basename(candidate[4])
            print("candidates: ", candidate) 

        if candidate_matches(candidate):
            return VariableServer(candidate[0], candidate[1])

def _parse_value(text):
    """
    Parse a variable server value with optional units.

    Parameters
    ----------
    text : str
        The variable server text.

    Returns
    -------
    str, str (or None)
        A tuple containing:
            - The value.
            - The units, which may be None.
    """
    match = re.match(r"(?P<value>.*)(?: {(?P<units>.*)})", text)
    if match:
        return match.group('value'), match.group('units')
    return text, None

def _assert_message_type(message, indicator):
    """
    Raise an error if the message indicator does not match the expected
    indicator.

    Parameters
    ----------
    message : Message
        The message to test.
    indicator : Message.Indicator
        The expected indicator.

    Raises
    ------
    UnexpectedMessageError
        If the message indicator does not match the expected indicator.
    """
    if message.indicator != indicator:
        raise UnexpectedMessageError(indicator, message.indicator)

def _assert_units_conversion(name, expected_units, actual_units):
    """
    Raise an error if the actual units do not match the expected units.

    Parameters
    ----------
    name : str
        The name of the variable being tested.
    expected_units : str
        The expected units.
    actual_units : str
        The actual units.

    Raises
    ------
    UnitsConversionError
        If the actual units do not match the expected units.
    """
    if expected_units != actual_units and expected_units != 'xx':
        raise UnitsConversionError(name, expected_units)

def _assert_value_count(expected, actual):
    """
    Raise an error if actual does not match expected.

    Parameters
    ----------
    expected : any
        The expected value.
    actual : any
        The actual value.

    Raises
    ------
    ValueCountError
        If actual does not match expected.
    """
    if expected != actual:
        raise ValueCountError(expected, actual)

if __name__ == '__main__':
    import doctest
    doctest.testmod()