py_gen/templates/generic_util.py

:: # Copyright 2013, Big Switch Networks, Inc.
:: #
:: # LoxiGen is licensed under the Eclipse Public License, version 1.0 (EPL), with
:: # the following special exception:
:: #
:: # LOXI Exception
:: #
:: # As a special exception to the terms of the EPL, you may distribute libraries
:: # generated by LoxiGen (LoxiGen Libraries) under the terms of your choice, provided
:: # that copyright and licensing notices generated by LoxiGen are not altered or removed
:: # from the LoxiGen Libraries and the notice provided below is (i) included in
:: # the LoxiGen Libraries, if distributed in source code form and (ii) included in any
:: # documentation for the LoxiGen Libraries, if distributed in binary form.
:: #
:: # Notice: "Copyright 2013, Big Switch Networks, Inc. This library was generated by the LoxiGen Compiler."
:: #
:: # You may not use this file except in compliance with the EPL or LOXI Exception. You may obtain
:: # a copy of the EPL at:
:: #
:: # http://www.eclipse.org/legal/epl-v10.html
:: #
:: # Unless required by applicable law or agreed to in writing, software
:: # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
:: # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
:: # EPL for the specific language governing permissions and limitations
:: # under the EPL.
::
:: include('_copyright.py')
"""
Utility functions independent of the protocol version
"""

:: include('_autogen.py')

import loxi
import struct

def unpack_array(deserializer, element_size, buf):
    """
    Deserialize an array of fixed length elements.
    The deserializer function should take a buffer and return the new object.
    """
    if len(buf) % element_size != 0: raise loxi.ProtocolError("invalid array length")
    n = len(buf) / element_size
    return [deserializer(buffer(buf, i*element_size, element_size)) for i in range(n)]

def unpack_list(deserializer, length_fmt, buf, extra_len=0):
    """
    Deserialize a list of variable-length entries.
    'length_fmt' is a struct format string with exactly one non-padding format
    character that returns the length of the given element, minus extra_len.
    The deserializer function should take a buffer and return the new object.
    """
    entries = []
    offset = 0
    length_struct = struct.Struct(length_fmt)
    n = len(buf)
    while offset < n:
        if offset + length_struct.size > len(buf): raise loxi.ProtocolError("entry header overruns list length")
        length, = length_struct.unpack_from(buf, offset)
        length += extra_len
        if length < length_struct.size: raise loxi.ProtocolError("entry length is less than the header length")
        if offset + length > len(buf): raise loxi.ProtocolError("entry length overruns list length")
        entries.append(deserializer(buffer(buf, offset, length)))
        offset += length
    return entries

class OFReader(object):
    """
    Cursor over a read-only buffer

    OpenFlow messages are best thought of as a sequence of elements of
    variable size, rather than a C-style struct with fixed offsets and
    known field lengths. This class supports efficiently reading
    fields sequentially and is intended to be used recursively by the
    parsers of child objects which will implicitly update the offset.
    """
    def __init__(self, buf):
        self.buf = buf
        self.offset = 0

    def read(self, fmt):
        st = struct.Struct(fmt)
        if self.offset + st.size > len(self.buf):
            raise loxi.ProtocolError("Buffer too short")
        result = st.unpack_from(self.buf, self.offset)
        self.offset += st.size
        return result

    def read_all(self):
        buf = buffer(self.buf, self.offset)
        self.offset += len(buf)
        return str(buf)

    def peek(self, fmt):
        st = struct.Struct(fmt)
        if self.offset + st.size > len(self.buf):
            raise loxi.ProtocolError("Buffer too short")
        result = st.unpack_from(self.buf, self.offset)
        return result

    def skip(self, length):
        if self.offset + length > len(self.buf):
            raise loxi.ProtocolError("Buffer too short")
        self.offset += length

    def is_empty(self):
        return self.offset == len(self.buf)

    # Used when parsing variable length objects which have external length
    # fields (e.g. the actions list in an OF 1.0 packet-out message).
    def slice(self, length):
        if self.offset + length > len(self.buf):
            raise loxi.ProtocolError("Buffer too short")
        buf = OFReader(buffer(self.buf, self.offset, length))
        self.offset += length
        return buf