Buildings_Utilities_IO_Python27_Examples_KalmanFilter.py sometimes fails

[mwetter]

In the unit tests, Buildings.Utilities.IO.Python27.Examples.KalmanFilter sometimes fails.

This can be reproduced with the Buildings library, branch issue1578_jmodelica_python and commit e60e5f5.

To reproduce, copy the Buildings library to a directory such as /tmp/tmp-Buildings-1-PtpREc,
set

export PYTHONPATH=`pwd`/Buildings/Resources/Python-Sources

and run

/tmp/tmp-Buildings-1-PtpREc$ jm_ipython.sh Buildings_Utilities_IO_Python27_Examples_KalmanFilter.py

with the file below:

###############################################################
# Script to test the model Buildings.Utilities.IO.Python27.Examples.KalmanFilter
# with JModelica.
# This script will create a json file that contains translation
# and simulation information.
###############################################################

##############################################################
# Capture stdout of a Python function.
# From https://stackoverflow.com/questions/24277488/in-python-how-to-capture-the-stdout-from-a-c-shared-library-to-a-variable/29834357

class OutputGrabber(object):
    """
    Class used to grab standard output or another stream.
    """
    escape_char = "\b"

    def __init__(self, stream=None, threaded=False):
        import sys
        import os
        self.origstream = stream
        self.threaded = threaded
        if self.origstream is None:
            self.origstream = sys.stdout
        self.origstreamfd = self.origstream.fileno()
        self.capturedtext = ""
        # Create a pipe so the stream can be captured:
        self.pipe_out, self.pipe_in = os.pipe()

    def __enter__(self):
        self.start()
        return self

    def __exit__(self, type, value, traceback):
        self.stop()

    def start(self):
        """
        Start capturing the stream data.
        """
        import os
        self.capturedtext = ""
        # Save a copy of the stream:
        self.streamfd = os.dup(self.origstreamfd)
        # Replace the original stream with our write pipe:
        os.dup2(self.pipe_in, self.origstreamfd)
        if self.threaded:
            # Start thread that will read the stream:
            self.workerThread = threading.Thread(target=self.readOutput)
            self.workerThread.start()
            # Make sure that the thread is running and os.read() has executed:
            time.sleep(0.01)

    def stop(self):
        """
        Stop capturing the stream data and save the text in `capturedtext`.
        """
        import os
        # Print the escape character to make the readOutput method stop:
        self.origstream.write(self.escape_char)
        # Flush the stream to make sure all our data goes in before
        # the escape character:
        self.origstream.flush()
        if self.threaded:
            # wait until the thread finishes so we are sure that
            # we have until the last character:
            self.workerThread.join()
        else:
            self.readOutput()
        # Close the pipe:
        os.close(self.pipe_in)
        os.close(self.pipe_out)
        # Restore the original stream:
        os.dup2(self.streamfd, self.origstreamfd)
        # Close the duplicate stream:
        os.close(self.streamfd)

    def readOutput(self):
        """
        Read the stream data (one byte at a time)
        and save the text in `capturedtext`.
        """
        import os
        while True:
            char = os.read(self.pipe_out, 1)
            if not char or self.escape_char in char:
                break
            self.capturedtext += char
##############################################################

def process_with_timeout(target, timeout):
    import multiprocessing
    import time
    import copy
    import sys
    import os

    manager = multiprocessing.Manager()
    return_dict = manager.dict()
    print("Target: {}".format(target))
    p = multiprocessing.Process(target=target, args=(0, return_dict))
    sys.stdout = open(str(os.getpid()) + ".out", "a", buffering=0)
    sys.stderr = open(str(os.getpid()) + "_error.out", "a", buffering=0)

    p.daemon = True
    start = time.time()
    p.start()
    p.join()
    while time.time() - start <= timeout:
        if p.is_alive():
            time.sleep(0.1)
        else:
            time.sleep(0.1)
            try:
                print("Exit code = {}".format(p.exitcode))
                cpu_time = time.time()-start
                ret = copy.copy(return_dict[0])
                ret.update({'cpu_time': cpu_time})
                return ret
            except KeyError as e:
                ret = {'success': False,
                       'exception': "{}".format(len(return_dict)) }
#                ret = {'successXXXXXX': False,
#                       'exception': '{}: {}  --- {}  ---- {}'.format(type(e).__name__, e.message, type(return_dict), type(return_dict)),
#                       'type(return_dict)': '{}'.format(type(return_dict)),
#                       'type(return_dict[0])': '{}'.format(type(return_dict[0]))}
                return ret
    else:
        # Entered if there was a timeout
        p.terminate()
        p.join()
        raise RuntimeError("Process timeout")

def _translate(proc_num, return_dict):
  from pymodelica import compile_fmu

  try:
    # Grab the stdoutput
    out = OutputGrabber()
    out.start()
    fmu_name = compile_fmu("Buildings.Utilities.IO.Python27.Examples.KalmanFilter",
                           version="2.0",
                           compiler_log_level='warning',
                           compiler_options = {"generate_html_diagnostics" : False})
    out.stop()
    # The standard output is returned as a list, with each line being an element
    return_dict[proc_num] = {'success': True, 'fmu_name': str(fmu_name), 'stdout': out.capturedtext.split('\n')}

  except Exception as e:
    return_dict[proc_num] = {'success': False,
                             'exception': '{}: {}'.format(type(e).__name__, e.message)}
  return

def _simulate(proc_num, return_dict):
  from pyfmi import load_fmu

  if not True:
    return_dict[proc_num] = {'success': False,
                             'message': 'No simulation requested.'}
    return return_dict

  # Simulate the model
#

  try:
    fmu_name = "Buildings.Utilities.IO.Python27.Examples.KalmanFilter".replace(".", "_") + ".fmu"
    mod = load_fmu(fmu_name)
    x_nominal = mod.nominal_continuous_states

    opts = mod.simulate_options() #Retrieve the default options
    opts['logging'] = False
    opts['solver'] = 'CVode'
    opts['ncp'] = 500

    rtol = 1e-6

    if len(x_nominal) > 0:
      atol = rtol*x_nominal
    else:
      atol = rtol

    if opts['solver'].lower() == 'cvode':
      # Set user-specified tolerance if it is smaller than the tolerance in the .mo file
      opts['CVode_options'] = {
        'external_event_detection': False,
        'maxh': (mod.get_default_experiment_stop_time()-mod.get_default_experiment_stop_time())/float(opts['ncp']),
        'iter': 'Newton',
        'discr': 'BDF',
        'rtol': rtol,
        'atol': atol,
        'store_event_points': True
    }

    opts['filter'] = [u'kalFil.yR[[]1[]]', u'kalFil.uR[[]1[]]', u'sine.y']
    res = mod.simulate(options=opts)
    start_time = res['time'][0]
    final_time = res['time'][-1]
    return_dict[proc_num] = {'success': True, 'start_time': start_time, 'final_time': final_time}

  except Exception as e:
    return_dict[proc_num] = {'success': False,
                             'exception': '{}: {}'.format(type(e).__name__, e.message)}
  return return_dict

def run():
    import os
    import timeit
    import json
    import traceback
    import sys

    import pymodelica
    # Increase memory
    pymodelica.environ['JVM_ARGS'] = '-Xmx4096m'

    time_out = 1200
    model = "Buildings.Utilities.IO.Python27.Examples.KalmanFilter"
    result = {"model": model,
              "translation": {"success": False},
              "simulation": {"success": False}}

    # Compile model
    log_file = "{}_run.json".format(model.replace(".", "_"))
    try:
        os.remove(log_file)
    except OSError:
        pass

    try:
        ret_dic = process_with_timeout(target=_translate, timeout=time_out)
        result["translation"] = ret_dic

    except Exception as e:
        result["translation"]["exception"] = "{}: {}".format(type(e).__name__, e.message)
        result["translation"]["traceback"] = traceback.format_exc()

    # Load model if translation was successful
    if result["translation"]["success"]:
        try:
            ret_dic = process_with_timeout(target=_simulate, timeout=time_out)
            result["simulation"] = ret_dic

        except Exception as e:
            result["simulation"]["exception"] = "{}: {}".format(type(e).__name__, e.message)
            result["simulation"]["traceback"] = traceback.format_exc()

    with open(log_file, "w") as log:
            log.write("{}\n".format(json.dumps(result, indent=4, sort_keys=False)) )

if __name__=="__main__":
    run()

[AntoineGautier]

Hmm... I cannot explain the behavior I get. By simply adding some logging instructions I cannot reproduce the issue anymore. Even if the logging level is so that actually no logging message is output. However the output of OutputGrabber is now an empty string in the JSON log file. Can you try on your system with the version of the script attached?
Buildings_Utilities_IO_Python27_Examples_KalmanFilter.py.txt

[mwetter]

@AntoineGautier : I run it 100 times as

set -e
for i in {0..100}
do
  echo "******************************** $i"
  jm_ipython.sh Buildings_Utilities_IO_Python27_Examples_KalmanFilter.py
done

and there were no failures.

[AntoineGautier]

So I have no further explanation. I have documented the tests that have been done and the simple patch at the beginning of the script attached. I have also tidied up the code a little bit e.g. timeout implementation.
Note: There is still a if not True statement at the beginning of _simulate that I left there, not knowing what it was meant for.
(I can try and set up a proper debugging framework with Visual Studio and the Linux C tools but it will take some time, especially as the program runs in docker...)

Buildings_Utilities_IO_Python27_Examples_KalmanFilter.py.txt

[mwetter]

@AntoineGautier : Thanks for the analysis. I will add

logging.basicConfig(
    level=0,
    format='\n[%(levelname)s] (%(processName)-10s) %(message)s\n',
)

in BuildingsPy and run more tests to see if this avoids the problem.