meta data for this page
  •  

Homework 02, message coding \ CSPA

Complete library system messages codings started on exercises exer_codec.

Code to point

In python package exer_codec

codec.py

from exer_codec.interfaces import *
import struct
import logging
log = logging.getLogger(__name__)
from util.common import *

class list_items(list_items_pdu):
  def __init__(self):
    super(list_items,self).__init__()
    
  def encode(self):
    """ Encode the fields into data buffer.
    Return data buffer."""
    self.msgtype = 1
    self.msgsize = struct.calcsize("HH") 
    buf = str() ## Python 2 data buffer
    buf += struct.pack("HH", self.msgtype, self.msgsize)
    log.debug(buf_debug(buf))
    return buf    
  
  def decode(self,buf):
    """ Decode the buffer into message fields. """
    log.debug(buf_debug(buf))    
    offset = 0
    ## REMEMBER TUPLE RETURN !
    self.msgtype = struct.unpack("H",buf[offset:offset + 2])[0]
    offset += 2
    self.msgsize = struct.unpack("H",buf[offset:offset + 2])[0]

interfaces.py

'''
Created on 30.1.2012

@author: hevi
'''

## PDU definitions

class Message_pdu(object):
  def __init__(self):
    self.msgtype = -1
    self.msgsize = -1
    
class list_items_pdu(Message_pdu):
  def __init__(self):
    super(list_items_pdu,self).__init__()
    
class list_info_pdu(Message_pdu):
  def __init__(self):
    super(list_info_pdu,self).__init__()
    self.itemid = -1

class receive_item_pdu(Message_pdu):
  def __init__(self):
    super(receive_item_pdu,self).__init__()
    self.itemid = -1
    self.userid = -1

class list_item_reply_pdu(Message_pdu):
  def __init__(self):
    super(list_item_reply_pdu,self).__init__()
    self.itemids = list() # list of itemid
    
class BOOK_info(object):
  def __init__(self):
    self.itemstatus = -1
    self.language = None
    self.title = None
    self.authors = list() # list of str
    self.ISBN = None
    self.year = -1
    self.classification = None
    
class JOURNAL_info(object):
  def __init__(self):
    self.itemstatus = -1
    self.language = None
    self.name = None
    self.year = -1
    self.issue = -1
    self.ISSN = None
    
    
class item_info_reply_pdu(Message_pdu):
  def __init__(self):
    super(item_info_reply_pdu,self).__init__()
    self.itemid = -1
    self.itemtype = -1
    self.subblock = None # either Book_info or JOURNAL_info object

class OperationSuccess_pdu(Message_pdu):    
  def __init__(self): 
    super(OperationSuccess_pdu,self).__init__()
    self.reason = None

test_codec.py

'''
Created on 30.1.2012

@author: hevi
'''
import unittest
from exer_codec.codec import *
import logging
log = logging.getLogger(__name__)


class Test(unittest.TestCase):

  def test_list_items_01(self):
    msg_send = list_items()
    buf = msg_send.encode()
    msg_recv = list_items()
    msg_recv.decode(buf)
    ## checking
    log.debug(str(msg_send.msgtype))
    log.debug(str(msg_recv.msgtype))
    assert msg_send.msgtype == msg_recv.msgtype
    assert msg_send.msgsize == msg_recv.msgsize


if __name__ == "__main__":
    #import sys;sys.argv = ['', 'Test.testName']
    logging.basicConfig(level=logging.DEBUG)
    unittest.main()