### Navigation

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

# Source code for json_link.xchg.raw_xchg

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

    Classes:
      MasterXchg
      SlaveXchg

    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.4.1b","2013-04-29"  #22:20   BryanP   BUGFIX: receive timeout when connect not ready. Move receive to _BaseXchg
    # [Older, in reverse order]
    # version      date       time     who     Comments
    # -------   ----------    ----- ---------- ---------------------------------------
    #  0.4.0b   2013-04-28    22:20   BryanP   RENAMED to raw_xchg & methods to Xchg* and Errors to RawXchg*Error
    #  0.3.1a   2013-04-27    02:00   BryanP   Throw exception when can't send. Added codes to Exceptions, Add "Raw" prefix to verbose msg
    #  0.3.0a   2013-04-25    16:23   BryanP   Use Exceptions for error handling. Renamed Link* to Xchg*
    #  0.2.0a   2013-04-25    03:30   BryanP   Works! Added Master TCP timeouts
    #  0.1.0a   2013-04-24    21:43   BryanP   Extracted from json_test.py v1.6.1a

    #Standard Library imports
    import socket
    import timeit
    import sys
    import time

    #===============================================================================
    # Module level constants
    #===============================================================================
    UDP = socket.SOCK_DGRAM
    TCP = socket.SOCK_STREAM

    HEAD_VER = '0'
    MSG_TYPE = 'JSON'
    MSG_VER = '1.0'

    #===============================================================================
    # Error (Exception) Class definitions
    #===============================================================================
    class RawXchgError(IOError):
        """Base class for exceptions in the raw_xchg module"""
        code = 1000

    class RawXchgTimeoutError(RawXchgError):
        """Timeout waiting for reply"""
        code = RawXchgError.code + 1

    class RawXchgSetupError(RawXchgError):
        """Problems setting up the communication (e.g. opening the socket)"""
        code = RawXchgError.code + 2

    class RawXchgHeaderError(RawXchgError):
        """Malformed or incorrect header"""
        code = RawXchgError.code + 3

    class RawXchgNoDataError(RawXchgError):
        """No data read"""
        code = RawXchgError.code + 4

    class RawXchgCantSendError(RawXchgError):
        """Unable to send packet"""
        code = RawXchgError.code + 5

    #===============================================================================
    # _BaseXchg (module internal)
    #===============================================================================
    class _BaseXchg(object):
        """ Base class for UDP/TCP link subclasses"""

        #socket and settings
        _sock = None

        PROTOCOL = None
        HOST_ADDR = None
        PORT = None
        TIMEOUT = None
        BUF_SIZE = 4906

        response_code_out = None     #override in subclass

        #Options
        opt_verbose = True
        opt_header = True

        ignore_head_err = False
        ignore_no_data_err = True

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

        def __init__(self, host_addr, port, protocol,
                     opt_verbose=True, opt_header=True, timeout=None,
                     buf_size=4906):
            """Constructor"""
            self.HOST_ADDR = host_addr
            self.PORT = port
            self.PROTOCOL = protocol

            self.opt_verbose = opt_verbose
            self.opt_header = opt_header
            self.TIMEOUT = max(timeout,0)
            self.BUF_SIZE = buf_size

        def isReady(self):
            """Returns True if we have assigned the socket. Does not check socket connectivity"""
            return self._sock is not None

        #-------------------
        # wrapPacket
        #-------------------
        # From: https://sourceforge.net/apps/mediawiki/gridlab-d/index.php?title=JSON_link_protocol
        #
        # Bytes     Contents
        # --------- --------------------------------------------------------------------------------
        # 00-01     Header version (currently "0", space padded)
        # 02-05     Offset to message from byte 00 (currently always 32, max is 999, space padded)
        # 06-13     Size of message (varies, max is 9999999, left justified, space padded), count does not include string termination character
        # 14-19     Type of message (currently "JSON", left justified, space padded)
        # 20-23     Version of type (currently "1.0", left justified, space padded)
        # 24-25     Connection status (0=closed, 1-9=keep-alive timeout, space padded)
        # 26-29     Response code (0 for requests, must be 200 for responses, left justified, space padded)
        # 30-31     Reserved (space padded)
        # 32-       Message content (current JSON 1.0 format)
        #
        #  Byte Number          111111111122222222223333333333444444444455555555556666666666
        #             0123456789 123456789 123456789 123456789 123456789 123456789 123456789
        # Master Ex: '0 32  27       JSON 1.0 1 0     {"method":"example","id":1}'
        # Slave Ex:  '0 32  27       JSON 1.0 1 200   {"result":"example","id":1}'
        #  Message Length                                      111111111122222222223333333333444444444455555555556666666666
        #                                             123456789 123456789 123456789 123456789 123456789 123456789 123456789
        def wrapPacket(self, msg, keep_alive=9):
            """ Add GridLAB UDP/TCP header and return complete data packet to send"""
            msg_size = len(msg)

            #Build potentially variable portion of header
            #            0123456- 234-892123 456-  9312
            head_str =        "%-7d %5s %3s %d %-3s   "%(msg_size, MSG_TYPE, MSG_VER, keep_alive, self.response_code_out)
            #Measure current size and add 5bytes for the header ver & msg offset
            head_len = len(head_str) + 6
            assert head_len==32, "outgoing packet header length should be 32 not %d"%head_len

            #Return the complete Packet String
            return "%1s %-3d %s%s"%(HEAD_VER, head_len, head_str, msg)

        #-------------------
        # unwrapPacket
        #-------------------
        def unwrapPacket(self, packet_str):
            """ Remove GridLAB UDP/TCP header, check validity and return Packet JSON content"""
            body = None
            reopen_conn = False

            if len(packet_str) == 0:
                raise RawXchgNoDataError("No Packet Data")

            if packet_str[0] != HEAD_VER:
                raise RawXchgHeaderError("Header: unsupported version must be %s not %s"%(HEAD_VER, packet_str[0]))

            try:
                #Extract header size (data offset) and split header and body
                head_len = int(packet_str[2:5])
                head = packet_str[0:head_len]
                body = packet_str[head_len:]

                #Attempt to extract whitespace delimited header
                _head_ver, _offset, msg_size, msg_type, msg_ver, conn_status, response = head.split()
            except:
                raise RawXchgHeaderError('Header: Invalid Format-must have 7 whitespace delimited fields')

            #Verify/Parse header contents
            # Note: already checked head_ver & offset is redundant b/c we used the same
            # same data to split head and body. (At least assuming reasonable space
            # delimiting
            if int(msg_size) != len(body):
                raise RawXchgHeaderError('Header: message length mismatch (%s vs %d actual)'%(msg_size, len(body)))
            if msg_type != MSG_TYPE:
                raise RawXchgHeaderError('Header: Unsupported message type (%s). Must be %s'%(msg_type, MSG_TYPE))
            if msg_ver != MSG_VER:
                raise RawXchgHeaderError('Header: Unsupported message version (%s). Must be %s'%(msg_ver, MSG_VER))
            if response != self.response_code_in:
                raise RawXchgHeaderError('Header: Invalid request (%s), expected %s'%(response, self.response_code_in))

            #Extract connection re-open flag
            if int(conn_status) <= 0:
                reopen_conn = True

            return body, reopen_conn

        def close(self):
            """gracefully close the current socket"""
            try:
                self._sock.shutdown(socket.SHUT_RDWR)
            except socket.error:
                pass
            try:
                self._sock.close()
            except socket.error:
                pass
            self._sock = None

        def receive(self):
            """Receive data and process. May need to be proceeded by some additional setup"""
            if self._sock is None:
                self.setupXchg()

            msg_in = None

            #Listen for data with timeout
            end_time = timeit.default_timer() + self.TIMEOUT
            not_run_once = True
            while timeit.default_timer() < end_time or not_run_once:
                not_run_once = False
                try:
                    msg_in, self._remote_addr = self._sock.recvfrom(self.BUF_SIZE)
                except socket.error:
                    time.sleep(self.TIMEOUT/1000)
                    continue
                if msg_in:
                    break

            if not msg_in:
                raise RawXchgTimeoutError('Receive timeout')

            msg_in=msg_in.decode('utf-8')

            if self.opt_verbose:
                print("  Raw RX: %s"%(msg_in))

            if self.opt_header:
                try:
                    msg_in, reopen_conn = self.unwrapPacket(msg_in)
                except RawXchgHeaderError:
                    err = sys.exc_info()[1]   #Use instead of "as err" For compatibility with python < 2.6
                    if self.ignore_head_err:
                        print('WARNING: %s'%err)
                        msg_in = None
                    else:
                        raise err
                except RawXchgNoDataError:
                    err = sys.exc_info()[1]   #Use instead of "as err" For compatibility with python < 2.6
                    if self.ignore_no_data_err:
                        print('WARNING: %s'%err)
                        msg_in = None
                    else:
                        raise err
                else:
                    if reopen_conn:
                        self.close()

            return msg_in

    #===============================================================================
    # MasterXchg
    #===============================================================================
    class MasterXchg(_BaseXchg):
        """ Implements a low-level GridLAB-D (JSON) UDP/TCP Master/client link"""

        response_code_out = '0'
        response_code_in = '200'

        def setupXchg(self):
            if not self.isReady():
                self._sock = socket.socket(family=socket.AF_INET, type=self.PROTOCOL)

                try:
                    self._sock.connect((self.HOST_ADDR, self.PORT))
                except Exception:
                    print("ERROR: Can't Connect to Slave (Are you using the right packet type?)")
                    self._sock=None
                    return
                self._sock.settimeout(self.TIMEOUT)

        def send(self, msg_out):
            """Wrap message with a header and send out"""
            if not self.isReady():
                self.setupXchg()

            if self.opt_header:
                msg_out = self.wrapPacket(msg_out)
            try:
                self._sock.sendall(msg_out.encode('utf-8'))
            except IOError:
                if self.opt_verbose:
                    print("Unable to Send Packet")
                self.close()
                raise RawXchgCantSendError("Unable to Send Packet")

            if self.opt_verbose:
                print("  Raw TX: %s"%(msg_out))

    #===============================================================================
    # SlaveXchg
    #===============================================================================
    class SlaveXchg(_BaseXchg):
        """ Implements a low_level GridLAB-D (JSON) UDP/TCP Slave/server link"""

        response_code_out = '200'
        response_code_in = '0'

        _reopen_conn = False
        _remote_addr = None
        _connected = False
        _conn = None

        def close(self):
            """Expanded close to handle our state flags"""
            if self.PROTOCOL == UDP:
                _BaseXchg.close(self)
            else:
                self._reopen_conn = False
                self._connected = False
                if self._conn is not None:
                    try:
                        self._conn.shutdown(socket.SHUT_RDWR)
                    except socket.error:
                        pass
                    try:
                        self._conn.close()
                    except IOError:
                        pass
                    self._conn = None

        def setupXchg(self):
            """Setup (Internet) socket for Slave."""
            if not self.isReady():
                self._sock = socket.socket(socket.AF_INET, self.PROTOCOL)
                try:
                    self._sock.bind((self.HOST_ADDR, self.PORT))
                except OSError:
                    self.close()

                if self._sock is None:
                    raise RawXchgSetupError('could not bind to socket')

                if self.PROTOCOL == TCP:
                    self._sock.listen(1)
                    self._conn, self._remote_addr = self._sock.accept()  #Updates our socket for connection
                    self._connected = True
                    self._conn.settimeout(self.TIMEOUT)
                    print('Connected by', self._remote_addr)

                self._sock.settimeout(self.TIMEOUT)

        def receive(self):
            """(Slave) wait to receive a packet, decode the header, and return data"""
            if self._sock is None:
                self.setupXchg()

            if self.PROTOCOL == TCP:
                self.setupXchg()

            return _BaseXchg.receive(self)

        def send(self, msg_out):
            """Wrap message with a header and send out"""

            if self.opt_header:
                msg_out = self.wrapPacket(msg_out)

            if self.PROTOCOL == TCP:
                self._conn.send(msg_out.encode('utf-8'))
            else:
                self._sock.sendto(msg_out.encode('utf-8'), self._remote_addr)

            if self.opt_verbose:
                print("  Raw TX: %s"%(msg_out))

            if self._reopen_conn:
                self.close()

### Navigation

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