### Navigation

- index
- modules |
- JSON Link 0.9.2a documentation »
- Module code »
- json_link.json_link »

# Source code for json_link.json_link.json_link

    """
    A python module for GridLAB-D data (JSON) link handshaking (over UDP and TCP)

    Classes:
        MasterLink
        SlaveLink

    Implemented Methods:

    Notes:
    * Should work with Python 2.6+ and 3.x. Tested with 2.7.2 and 3.3
    * Currently only supports IPv4
    * The SLAVE will handle connections from any number of masters, BUT since the
    code is only single threaded each processing (including delay) are
    blocking. Hence, the SLAVE may miss packets if it is busy processing
    (or waiting to send) a different reply.

    @author: Bryan Palmintier, NREL 2013
    """
    # Enable use of Python 3.0 print statement in older (2.x) versions, ignored in
    # newer (3.x) versions
    # Must occur at top of file
    from __future__ import print_function

    __author__ = "Bryan Palmintier"
    __copyright__ = "Copyright (c) 2013 Bryan Palmintier"
    __license__ = """
    @copyright: Copyright (c) 2013, Bryan Palmintier
    @license: BSD 3-clause --
    All rights reserved.

    Redistribution and use in source and binary forms, with or without modification,
    are permitted provided that the following conditions are met:
     1) Redistributions of source code must retain the above copyright notice, this
        list of conditions and the following disclaimer.
     2) Redistributions in binary form must reproduce the above copyright notice,
        this list of conditions and the following disclaimer in the documentation
        and/or other materials provided with the distribution.
     3) Neither the name of the National Renewable Energy Lab nor the names of its
        contributors may be used to endorse or promote products derived from this
        software without specific prior written permission.

    THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
    ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
    WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
    DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR
    ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
    (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
    LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
    ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
    (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
    SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE."""

    ## ===== History ====
    #  [Current]             version     date       time     who     Comments
    #                        -------  ----------    ----- ---------- -------------
    __version__, __date__ = "0.9.2b","2013-04-29"  #22:20   BryanP   Move __init__ to _BaseLink. Master retry Init until slave ready
    # [Older, in reverse order]
    # version      date       time     who     Comments
    # -------   ----------    ----- ---------- ---------------------------------------
    #  0.9.1b   2013-04-28    22:20   BryanP   Corrected slave timing & start/term sequencing. Rename gld_tcp_udp to raw_xchg with *Xchg*
    #  0.9.0b   2013-04-27    02:00   BryanP   Full State Machine Functionality
    #  0.1.0a   2013-04-25    17:20   BryanP   Initial Code

    #Standard Library imports
    import sys
    import json
    import timeit
    import time
    import datetime as dt
    import random

    #local imports
    from . import raw_xchg

    #===============================================================================
    # Module level constants
    #===============================================================================

    #===============================================================================
    # Error (Exception) Class definitions
    #===============================================================================
    class JsonLinkError(IOError):
        """Base class for exceptions in the gld_json_link module"""
        code = 2000

    class JsonLinkTimeoutError(JsonLinkError):
        """Timeout waiting for reply"""
        code = JsonLinkError.code + 1

    class JsonLinkNoDataError(JsonLinkError):
        """Expected data is empty"""
        code = JsonLinkError.code + 2

    class JsonLinkBadWrapperError(JsonLinkError):
        """Malformed wrapper"""
        code = JsonLinkError.code + 3

    class JsonLinkRemoteError(JsonLinkError):
        """received a (reasonably) well formed error packet from the remote machine"""
        code = JsonLinkError.code + 4

    class JsonLinkOldMsgError(JsonLinkError):
        """Received a message with id older than a previously processed message"""
        code = JsonLinkError.code + 5

    class JsonLinkActionMismatchError(JsonLinkError):
        """Incorrect handshaking order"""
        code = JsonLinkError.code + 6

    #===============================================================================
    # _BaseLink (module internal)
    #===============================================================================
    class _BaseLink(object):
        """ Base class for JSON link subclasses"""

        #internal state machine
        state = 'beginState'

        in_schema = None
        out_schema = None

        #handshaking
        msg_num = 0
        remote_msg_num = 0

        response_str_out = None     #override in subclass
        response_str_in = None     #override in subclass
        retries = 0

        #low level data exchange object
        _xchg = None

        #Options
        opt_verbose = True

        ignore_old = False
        ignore_dup = False

        #Track link statistics
        num_err = 0
        num_tx = 0
        num_rx = 0

        #Internal process (Run) timing information
        _run_end_time = None
        run_timestep = 1.0
        run_delay = 0.1
        run_max = -1
        run_count = 0

        def __init__(self, run_timestep=1, run_delay=0.1, run_max=-1, **Xchg_args):
            """Constructor"""
            self.run_timestep = run_timestep
            self.run_delay = run_delay
            self.run_max = run_max

            self._setupOurXchg(**Xchg_args)

        def _setupOurXchg(self, **Xchg_args):
            """Must be implemented by subclasses to configure their raw exchange layer"""
            pass

        def go(self):
            """Method to beginState link execution"""
            pass

        def wrapJson(self, action, data, data_name='data'):
            """encapsulate data (typically a nested data dictionary) in side a JSON link wrapper"""

            #build message
            msg = {self.response_str_out:action}
            if data is not None:
                msg[data_name] = data
            msg['id'] = self.msg_num

            #convert to JSON string
            return json.dumps(msg)

        def unwrapJson(self, action, msg, data_name='data', ignore_msg_num=False):
            """check JSON link wrapper and decode data (typically a nested data dictionary)"""
            if not msg:
                raise JsonLinkNoDataError('No Data received')

            msg = json.loads(msg)

            if type(msg) is not dict:
                raise JsonLinkBadWrapperError('JSON Message is not a dictionary object')

            if 'error' in msg:
                err_msg = ''
                if 'params' in msg['error']:
                    err_msg += msg['error']['params']
                if 'code' in msg['error']:
                    err_msg += " [code=%s]"%msg['error']['code']
                if err_msg == '':
                    err_msg = 'Remote Error Received, no message param or code'
                raise JsonLinkRemoteError(err_msg)

            if ignore_msg_num:
                if not 'id' in msg:
                    raise JsonLinkBadWrapperError('Missing id field in JSON message')

                # Process id (sequence number)
                msg_id = int(msg['id'])
                if msg_id < self.remote_msg_num:
                    if self.ignore_old:
                        if self.opt_verbose:
                            print('WARNING: Ignoring packet with old message id %d (expected %d)'%(msg_id, self.remote_msg_num + 1))
                        return None
                    else:
                        raise JsonLinkOldMsgError('A newer JSON message has been received (id=%d expected=%d)'%(msg_id, self.remote_msg_num + 1))
                elif msg_id == self.remote_msg_num:
                    if self.ignore_dup:
                        if self.opt_verbose:
                            print('WARNING: Ignoring duplicate message (id=%d)'%(msg_id))
                        return None
                    else:
                        raise JsonLinkOldMsgError('Duplicate message (id=%d)'%(msg_id))
                self.remote_msg_num = msg_id

            #check action/response_str
            if self.response_str_in not in msg:
                raise JsonLinkBadWrapperError('Wrong type of JSON message (missing %s)'%(self.response_str_in))

            if msg[self.response_str_in] != action:
                return (msg[self.response_str_in], None)

            if data_name is None:
                return (action, None)

            if data_name not in msg:
                raise JsonLinkBadWrapperError('JSON Message missing data payload (%s)'%(data_name))

            return (action, msg[data_name])

        def tx(self, msg, increment=True):
            """Transmit preformatted data message"""
            self._xchg.send(msg)

            if increment:
                self.msg_num += 1

        def rx(self):
            """Wait for and receive data"""
            msg = self._xchg.receive()

            return msg

        def handleAltAction(self, action, msg, data_name='data', out_data=None, prompt=None):
            _actual_action, data = self.unwrapJson(action, msg, data_name, ignore_msg_num=True)
            if self.opt_verbose:
                if prompt is None:
                    print('Rx %s: %s'%(action.upper(), data))
                else:
                    print('  %s%s'%(prompt,data))

            msg = self.wrapJson(action, out_data, data_name)
            self.tx(msg)
            if self.opt_verbose:
                print('Tx %s: %s'%(action.upper(), out_data))

            return data

        def sendMsg(self, action, data, data_name='data',
                    raw_json=False, raw_packet=False):
            """Helper function to send general messages"""
            if raw_packet:
                self._xchg.send(data)
                return True
            elif raw_json:
                self.tx(data)
            else:
                msg = self.wrapJson(action, data, data_name)
                if self.opt_verbose:
                    print('Tx %s %s: %s'%(action.upper(), data_name, data))
                try:
                    self.tx(msg)
                except raw_xchg.RawXchgCantSendError:
                    return False
            return True

    #------- Simple Message functions. Designed to be overloaded for fault injection
        def sendInit(self, data=None):
            """Send INIT message"""
            return self.sendMsg('init', data, 'params')

        def sendInSchema(self, data=None):
            """Send INPUT message"""
            return self.sendMsg('input', data, 'schema')

        def sendOutSchema(self, data=None):
            """Send OUTPUT message"""
            return self.sendMsg('output', data, 'schema')

        def sendStart(self, data=None):
            """Send START message"""
            return self.sendMsg('start', data)

        def sendSync(self, data=None):
            """Send SYNC message"""
            return self.sendMsg('sync', data)

        def sendTerm(self, data=None):
            """Send TERM message"""
            return self.sendMsg('term', data)

        def runWaitForTimeStep(self):
            while timeit.default_timer() < self._run_end_time:
                time.sleep(self.run_timestep/1000)

            #Find next time period end.
            self._run_end_time += self.run_timestep

            #Ensure that the next time end is in the future and warn if missed step
            while timeit.default_timer() > self._run_end_time:
                if self.opt_verbose:
                    print('Oops: that run step was too long. Skipping a step to resync')
                self._run_end_time += self.run_timestep

        def endState(self, old_state):
            """endState state"""
            if self.opt_verbose:
                print('\nEND\n')
                exit()

    #===============================================================================
    # MasterLink
    #===============================================================================
    class MasterLink(_BaseLink):
        """ Implements a GridLAB-D JSON Link Master/client state machine"""

        response_str_out = 'method'
        response_str_in = 'result'

        description_dict = {'application':"Generic_JSON_Link_Master",
                            'version':__version__,
                            'modelname':sys.argv[0]
                            }
        in_schema = {"clock": "timestamp global.clock",
                     "y0": "double test_0.y",
                     "y1": "double test_1.y"
                     }
        out_schema = {"clock": "timestamp global.clock",
                      "x0": "random test_0.x",
                      "x1": "random test_1.y"
                      }

        send_in_schema = True
        send_out_schema = True

        remote_data = None
        local_data = None

        _run_send = True

        INIT_RECHECK_DELAY = 1 #Additional time (beyond timeout) to wait before resending Init

        def _setupOurXchg(self, **Xchg_args):
            """Setup the raw MASTER data exchange"""
            self._xchg = raw_xchg.MasterXchg(**Xchg_args)

        def go(self):
            """Begin JSON link Master state machine execution"""
            old_state = self.state
            while True:
                if self.state == 'beginState':
                    new_state = self.beginState(old_state)
                elif self.state == 'waitInitReplyState':
                    new_state = self.waitInitReplyState(old_state)
                elif self.state == 'decideInSchema':
                    new_state = self.decideInSchema(old_state)
                elif self.state == 'waitInSchemaReplyState':
                    new_state = self.waitInSchemaReplyState(old_state)
                elif self.state == 'decideOutSchema':
                    new_state = self.decideOutSchema(old_state)
                elif self.state == 'waitOutSchemaReplyState':
                    new_state = self.waitOutSchemaReplyState(old_state)
                elif self.state == 'waitStartReplyState':
                    new_state = self.waitStartReplyState(old_state)
                elif self.state == 'runState':
                    new_state = self.runState(old_state)
                elif self.state == 'waitTermReplyState':
                    new_state = self.waitTermReplyState(old_state)
                elif self.state == 'endState':
                    new_state = self.endState(old_state)
                else:
                    raise JsonLinkError('Invalid state %s'%self.state)

                old_state = self.state
                if new_state is not None:
                    self.state = new_state

        def beginState(self, old_state):
            """beginState state"""
            if self.opt_verbose:
                print('BEGIN')
            self._xchg.setupXchg()

            self.msg_num += 1

            if self.sendInit(self.description_dict):
                return 'waitInitReplyState'
            else:
                return self.state

        def waitInitReplyState(self, old_state):
            """waitInitReplyState state"""
            if self.opt_verbose and old_state != 'waitInitReplyState':
                print('\nSTATE: waitInitReplyState...')

            msg = None

            #Keep trying to send the Init message until we get a response
            while msg is None:
                try:
                    msg = self.rx()
                except raw_xchg.RawXchgError:
                    if self.opt_verbose:
                        print('\nNo Response. Retrying INIT. Has the slave been started?')
                    time.sleep(self.INIT_RECHECK_DELAY)
                    if not self.sendInit(self.description_dict):
                        continue

            try:
                actual_action, remote_info = self.unwrapJson('init', msg, 'params')
            except (JsonLinkError, raw_xchg.RawXchgError) as err:
                print('ERROR: %s [code=%d]'%(err.args[0], err.code))
                return 'endState'

            if actual_action == 'error':
                self.sendTerm({})
                return 'endState'
            elif actual_action != 'init':
                self.sendInit(self.description_dict)
                return None

            if self.opt_verbose:
                print("--> Linked with REMOTE: %s\n"%remote_info)

            return 'decideInSchema'

        def decideInSchema(self, old_state):
            """decideInSchema state"""
            if self.opt_verbose and old_state != 'decideInSchema':
                print('\nSTATE: decideInSchema...')

            if self.send_in_schema:
                if self.sendInSchema(self.in_schema):
                    if self.opt_verbose:
                        print('--> INPUT Schema sent: %s\n'%self.in_schema)
                    return 'waitInSchemaReplyState'
                else:
                    return self.state
            else:
                return 'decideOutSchemaState'

        def waitInSchemaReplyState(self, old_state):
            """waitInSchemaReplyState state"""
            if self.opt_verbose and old_state != 'waitInSchemaReplyState':
                print('\nSTATE: waitInSchemaReplyState...')

            msg = self.rx()

            if msg is None:
                self.sendTerm({})
                return 'endState'

            try:
                actual_action, _data_in = self.unwrapJson('input', msg, None)
            except (JsonLinkError, raw_xchg.RawXchgError) as err:
                print('ERROR: %s [code=%d]'%(err.args[0], err.code))
                return 'endState'

            if actual_action == 'error':
                self.sendTerm({})
                return 'endState'
            elif actual_action != 'input':
                self.sendInSchema(self.in_schema)
                return None

            if self.opt_verbose:
                print("--> INPUT schema Accepted\n")

            return 'decideOutSchema'

        def decideOutSchema(self, old_state):
            """decideOutSchema state"""
            if self.opt_verbose and old_state != 'decideOutSchema':
                print('\nSTATE: decideOutSchema...')

            if self.send_out_schema:
                if self.sendOutSchema(self.out_schema):
                    if self.opt_verbose:
                        print('--> OUTPUT Schema sent: %s\n'%self.out_schema)
                    return 'waitOutSchemaReplyState'
                else:
                    return self.state
            else:
                return 'waitStartReply'

        def waitOutSchemaReplyState(self, old_state):
            """waitOutSchemaReplyState state"""
            if self.opt_verbose and old_state != 'waitOutSchemaReplyState':
                print('\nSTATE: waitOutSchemaReplyState...')

            msg = self.rx()

            if msg is None:
                self.sendTerm({})
                return 'endState'

            try:
                actual_action, _data_in = self.unwrapJson('output', msg, None)
            except (JsonLinkError, raw_xchg.RawXchgError) as err:
                print('ERROR: %s [code=%d]'%(err.args[0], err.code))
                return 'endState'

            if actual_action == 'error':
                self.sendTerm({})
                return 'endState'
            elif actual_action != 'output':
                self.sendOutSchema(self.in_schema)
                return None

            if self.opt_verbose:
                print("--> OUTPUT schema Accepted\n")

            if self.sendStart(dummyData(self.out_schema)):
                return 'waitStartReplyState'
            else:
                return None

        def waitStartReplyState(self, old_state):
            """waitStartReplyState state"""
            if self.opt_verbose and old_state != 'waitStartReplyState':
                print('\nSTATE: waitStartReplyState...')

            msg = self.rx()

            if msg is None:
                self.sendTerm({})
                return 'endState'

            try:
                actual_action, data_in = self.unwrapJson('start', msg, 'data')
            except (JsonLinkError, raw_xchg.RawXchgError) as err:
                print('ERROR: %s [code=%d]'%(err.args[0], err.code))
                return 'endState'

            if actual_action == 'error':
                self.sendTerm({})
                return 'endState'
            elif actual_action != 'start':
                self.sendStart(dummyData(self.out_schema))
                return None

            self.remote_data = data_in
            if self.opt_verbose:
                print("--> Link STARTed. Remote Data=%s\n"%data_in)

            return 'runState'

        def runState(self, old_state):
            """Run internal process and sync with remote"""
            if self.opt_verbose and old_state != 'runState':
                print('\nSTATE: Run')
            self.state = 'runDoMaster'
            self._run_end_time = timeit.default_timer() + self.run_timestep

            while True: #Note iteration count handled in runDoMaster
                if self.state == 'runWaitSyncReplyState':
                    new_state = self.runWaitSyncReplyState(old_state)
                elif self.state == 'runDoMaster':
                    new_state = self.runDoMaster(old_state)
                else:
                    return new_state

                old_state = self.state
                if new_state is not None:
                    self.state = new_state

            #Should never get here
            raise JsonLinkError('INVALID STATE: Final runDoMaster should switch to runWaitTermReply')

        def runWaitSyncReplyState(self, old_state):
            """(Master) wait for remote to reply with sync data"""
            msg = None

            # Note: this while loop attempts to work around some common errors as
            # long as there is time left. It is NOT the main timing loop, which can
            # be found in runDoMaster
            while timeit.default_timer() < self._run_end_time:
                try:
                    msg = self.rx()
                except raw_xchg.RawXchgError:
                    #TODO: add a few other specific error
                    continue

                try:
                    actual_action, new_remote_data = self.unwrapJson('sync', msg, 'data')
                except JsonLinkRemoteError as err:
                    print('Rx ERROR: %s'%(err))
                    return 'endState'
                except JsonLinkNoDataError:
                    continue

                if actual_action != 'sync':
                    # handle alternative message types
                    if actual_action == 'error':
                        return 'endState'
                    else:
                        print('Rx %s packet: Ignoring'%(actual_action.upper()))
                        continue

                if self.opt_verbose:
                    print('Rx SYNC: %s'%new_remote_data)
                self.remote_data = new_remote_data
                break

            return 'runDoMaster'

        def runDoMaster(self, old_state):
            """Internal Process for master. This version creates a simple set of 2 data points"""

            self.run_count += 1

            if self.opt_verbose:
                print('Remote Data: %s'%self.remote_data)
                print('Run (%d)...\n'%self.run_count)

            #Do internal process, including simulated processing delay
            time.sleep(self.run_delay)
            my_data = dummyData(self.out_schema)
            self.local_data = my_data

            if self.opt_verbose:
                print('Local (master) Data: %s'%my_data)

            self.runWaitForTimeStep()

            #Monitor number of runs and transition appropriately
            if self.run_count != self.run_max:
                self.sendSync(my_data)
                return 'runWaitSyncReplyState'
            else:
                if self.opt_verbose:
                    print('\nRun: Max Iteration Count (%d)'%(self.run_count))
                self.sendTerm(my_data)
                return 'waitTermReplyState'

        def waitTermReplyState(self, old_state):
            """waitTermReplyState state"""
            if self.opt_verbose and old_state != 'waitTermReplyState':
                print('\nSTATE: waitTermReplyState...')

            msg = self.rx()

            try:
                _actual_action, data_in = self.unwrapJson('term', msg, 'data')
            #TODO: Catch receipt of error packets
            except (JsonLinkError, raw_xchg.RawXchgError) as err:
                print('ERROR: %s [code=%d]'%(err.args[0], err.code))
                return 'endState'

            self.remote_data = data_in
            if self.opt_verbose:
                print("--> Link TERMinated. Final Remote Data=%s"%data_in)

            return 'endState'

    #===============================================================================
    # SlaveLink
    #===============================================================================
    class SlaveLink(_BaseLink):
        """ Implements a GridLAB-D JSON Link Slave/server state machine"""

        response_str_out = 'result'
        response_str_in = 'method'

        remote_data = None

        _run_send = True

        in_schema = None
        out_schema = None

        #Passed via the init link message, must include 'remote'
        description_dict = {'remote':"Generic GridLAB-D Slave Link", 'version':__version__}

        def _setupOurXchg(self, **Xchg_args):
            """Setup the raw Slave data exchange"""
            self._xchg = raw_xchg.SlaveXchg(**Xchg_args)

        def go(self):
            """Begin JSON link Slave state machine execution"""
            old_state = self.state
            while True:
                if self.state == 'beginState':
                    new_state = self.beginState(old_state)
                elif self.state == 'preInitState':
                    new_state = self.preInitState(old_state)
                elif self.state == 'waitStartState':
                    new_state = self.waitStartState(old_state)
                elif self.state == 'runState':
                    new_state = self.runState(old_state)
                elif self.state == 'endState':
                    new_state = self.endState(old_state)
                else:
                    raise JsonLinkError('Invalid state %s'%self.state)

                old_state = self.state
                if new_state is not None:
                    self.state = new_state

        def sendError(self, action, code=None, err_str=''):
            """Helper function to send error messages"""
            if self.opt_verbose:
                print('ERROR: %s [code=%d]'%(err_str,code))

            err_dict = {'code':code, 'params':err_str}
            err_msg = self.wrapJson(action, err_dict, 'error')

            if self.opt_verbose:
                print('Tx ERROR: %s'%err_dict)
            self.tx(err_msg)
            return True

        def rxIgnoreErr(self):
            """Robust message receive that ignores timeouts and other errors"""
            msg = None

            while not msg:
                try:
                    msg = self.rx()
                except raw_xchg.RawXchgError:
                    continue
            return msg

        def beginState(self, old_state):
            """beginState state"""
            if self.opt_verbose:
                print('BEGIN')
            self._xchg.setupXchg()

            self.msg_num += 1

            return 'preInitState'

        def preInitState(self, old_state):
            """preInitState state"""
            if self.opt_verbose and old_state != 'preInitState':
                print('\nSTATE: Pre-Init...')

            msg = self.rxIgnoreErr()

            try:
                actual_action, remote_info = self.unwrapJson('init', msg, 'params')
            except JsonLinkOldMsgError as err:
                self.sendError('init', err.code, err.args[0])
                return 'endState'

            if actual_action != 'init':
                self.sendError('init', JsonLinkActionMismatchError.code, 'Sequence Error, expecting init')
                return 'endState'

            if self.opt_verbose:
                print('Rx INIT remote info: %s'%remote_info)

            self.sendInit(self.description_dict)

            return 'waitStartState'

        def waitStartState(self, old_state):
            """waitStartState state"""
            if self.opt_verbose and old_state != 'waitStartState':
                print('\nSTATE: Wait for Start (waitStartState)...')

            msg = self.rxIgnoreErr()

            try:
                actual_action, data = self.unwrapJson('start', msg, 'data')
            except (JsonLinkError, raw_xchg.RawXchgError) as err:
                self.sendError('start', code=err.code, err_str=err.args[0])
                return 'endState'

            if actual_action != 'start':
                # handle alternative message types
                if actual_action == 'init':
                    self.handleAltAction('init', msg, data_name='params',
                                         out_data=self.description_dict)
                    return None
                elif actual_action == 'input':
                    self.in_schema = self.handleAltAction('input', msg, data_name='schema')
                    if self.opt_verbose:
                        print('--> Storing INPUT SCHEMA: %s\n'%self.in_schema)
                    return None
                elif actual_action == 'output':
                    self.out_schema = self.handleAltAction('output', msg, data_name='schema')
                    if self.opt_verbose:
                        print('--> Storing OUTPUT SCHEMA: %s\n'%self.out_schema)
                    return None
                elif actual_action == 'term':
                    self.handleAltAction('term', msg)
                    return 'endState'
                else:
                    self.sendError('start', JsonLinkActionMismatchError.code,
                                   err_str='Sequence Error, not expecting %s'%actual_action)
                    return 'endState'

            #Setup our local run time interval
            self._run_end_time = timeit.default_timer() + self.run_timestep

            #Store a copy of the remote data
            self.remote_data = data
            if self.opt_verbose:
                print('Rx START: %s'%data)

            #Respond to start message with initial conditions
            self.sendStart(dummyData(self.out_schema))

            if self.opt_verbose:
                print('--> Link STARTed: %s\n'%data)

            return 'runState'

        def runState(self, old_state):
            """Run internal process and sync with remote"""
            if self.opt_verbose and old_state != 'runState':
                print('\nSTATE: Run')
            self.state = 'runDoSlave'
            self._run_send = False

            while self.run_count < self.run_max or self.run_max < 0:
                if self.state == 'runWaitSyncState':
                    new_state = self.runWaitSyncState(old_state)
                elif self.state == 'runDoSlave':
                    new_state = self.runDoSlave(old_state)
                else:
                    return new_state

                old_state = self.state
                if new_state is not None:
                    self.state = new_state

            return 'endState'

        def runWaitSyncState(self, old_state):
            """sync data with remote"""
            msg = None
            self._run_send = False
            retry = 0
            MAX_RETRY=1

            print()

            while retry < MAX_RETRY:
                retry +=1
                try:
                    msg = self.rx()
                except raw_xchg.RawXchgError as err:
                    continue

                try:
                    actual_action, data = self.unwrapJson('sync', msg, 'data')
                except JsonLinkError as err:
                        self.sendError('sync', err.code, err.args[0])
                        return 'endState'

                if actual_action != 'sync':
                    # handle alternative message types
                    if actual_action == 'term':
                        self.handleAltAction('term', msg, out_data=dummyData(self.out_schema))
                        return 'endState'
                    else:
                        self.sendError(actual_action, JsonLinkActionMismatchError.code,
                                       err_str='Sequence Error, not expecting %s'%actual_action)
                        return 'endState'

                #Store our data
                self.remote_data = data
                self._run_send = True
                if self.opt_verbose:
                    print('Rx SYNC: %s'%data)

            return 'runDoSlave'

        def runDoSlave(self, old_state):
            """Internal Process for slave. This version simply echos the current data"""

            self.run_count += 1

            if self.opt_verbose:
                print('Remote Data: %s'%self.remote_data)
                print('Run (%d)...'%self.run_count)

            time.sleep(self.run_delay)
            my_data = dummyData(self.in_schema)

            if self.opt_verbose:
                print('Local (slave) Data: %s'%my_data)

            if self._run_send:
                self.sendSync(my_data)

            self.runWaitForTimeStep()

            return 'runWaitSyncState'

    #===============================================================================
    # Related Functions
    #===============================================================================
    def dummyData(schema):
        """Build up dummy json data package based on the provided schema"""
        data = dict()
        if type(schema) is dict:
            for field in schema:
                f_type = schema[field].split()[0]
                if f_type ==  'timestamp':
                    data[field] = dt.datetime.now().isoformat(' ')
                else:
                    data[field] = random.random()
        return data

### Navigation

- index
- modules |
- JSON Link 0.9.2a documentation »
- Module code »
- json_link.json_link »