[Pymoul-svn] SF.net SVN: pymoul: [157] pymoul/trunk/src/moul/crypt

[<ti...@us...>]

Revision: 157
          http://pymoul.svn.sourceforge.net/pymoul/?rev=157&view=rev
Author:   tiran
Date:     2007-02-10 05:14:18 -0800 (Sat, 10 Feb 2007)

Log Message:
-----------
Updated binary file, added String16, String32 and UruString
Added unit tests

Modified Paths:
--------------
    pymoul/trunk/src/moul/crypt/binary.py

Added Paths:
-----------
    pymoul/trunk/src/moul/crypt/tests/test_binary.py

Property Changed:
----------------
    pymoul/trunk/src/moul/crypt/binary.py

Modified: pymoul/trunk/src/moul/crypt/binary.py
===================================================================
--- pymoul/trunk/src/moul/crypt/binary.py	2007-02-08 20:02:58 UTC (rev 156)
+++ pymoul/trunk/src/moul/crypt/binary.py	2007-02-10 13:14:18 UTC (rev 157)
@@ -18,14 +18,12 @@
 """Binary file helper
 """
 __author__ = "Christian Heimes"
-__version__ = "$Id: elf.py 122 2007-02-02 17:34:06Z tiran $"
-__revision__ = "$Revision: 122 $"
+__version__ = "$Id"
+__revision__ = "$Revision"
 
-from _struct import Struct
 from struct import pack
 from struct import unpack
 from struct import calcsize
-from tempfile import NamedTemporaryFile
 
 class BinaryFile(file):
     """Binary file
@@ -48,15 +46,22 @@
       - double: readDouble / writeDouble
       - packed data: readPacked(fmt) / writePacked(fmt, data)
       - quad (two int16): readQuad / writeQuad)
+      - NULL: read0 / write0
+      - null string: readString0 / writeString0 (size is string + NULL)
+      - uru string: readUruString(version), writeUruString
+      - string w/ 16bit size header: readString16(null terminated) / writeString16
+      - string w/ 32bit size header: readString32(null terminated) / writeString32
 
     For conveniance the class has a size() method
-    
-    The class is using some optimization tricks like using _struct.Struct
-    directly and binding functions to the local namespace of a method.
+
+    The class is using some optimization tricks like binding functions to the
+    local namespace of a method.
     """
     def __new__(cls, fname, mode='rb'):
         assert 'b' in mode
-        return file.__new__(cls, fname, mode)
+        self = file.__new__(cls, fname, mode)
+        self.NULL = '\x00'
+        return self
 
     def size(self):
         pos = self.tell()
@@ -66,93 +71,296 @@
         finally:
             self.seek(pos)
 
-    def readChar(self, __unpack=Struct('<c').unpack):
-        return __unpack(self.read(1))[0]
+    def readChar(self, _unpack=unpack):
+        return _unpack('<c', self.read(1))[0]
 
-    def readByte(self, __unpack=Struct('<B').unpack):
-        return __unpack(self.read(1))[0]
+    def readByte(self, _unpack=unpack):
+        return _unpack('<B',self.read(1))[0]
 
-    def readBool(self, __unpack=Struct('<B').unpack):
-        return bool(__unpack(self.read(1))[0])
+    def readBool(self, _unpack=unpack):
+        return bool(_unpack('<B', self.read(1))[0])
 
-    def read8(self, __unpack=Struct('<B').unpack):
-        return __unpack(self.read(1))[0]
+    def read8(self, _unpack=unpack):
+        return _unpack('<B', self.read(1))[0]
 
-    def read8s(self, __unpack=Struct('<b').unpack):
-        return __unpack(self.read(1))[0]
+    def read8s(self, _unpack=unpack):
+        return _unpack('<b', self.read(1))[0]
 
-    def read16(self, __unpack=Struct('<H').unpack):
-        return __unpack(self.read(2))[0]
+    def read16(self, _unpack=unpack):
+        return _unpack('<H', self.read(2))[0]
 
-    def read16s(self, __unpack=Struct('<h').unpack):
-        return __unpack(self.read(2))[0]
+    def read16s(self, _unpack=unpack):
+        return _unpack('<h', self.read(2))[0]
 
-    def read32(self, __unpack=Struct('<I').unpack):
-        return __unpack(self.read(4))[0]
+    def read32(self, _unpack=unpack):
+        return _unpack('<I', self.read(4))[0]
 
-    def read32s(self, __unpack=Struct('<i').unpack):
-        return __unpack(self.read(4))[0]
+    def read32s(self, _unpack=unpack):
+        return _unpack('<i', self.read(4))[0]
 
-    def read64(self, __unpack=Struct('<Q').unpack):
-        return __unpack(self.read(8))[0]
+    def read64(self, _unpack=unpack):
+        return _unpack('<Q', self.read(8))[0]
 
-    def read64s(self, __unpack=Struct('<q').unpack):
-        return __unpack(self.read(8))[0]
+    def read64s(self, _unpack=unpack):
+        return _unpack('<q',self.read(8))[0]
 
-    def readQuad(self, __unpack=Struct('<2I').unpack):
-        return __unpack(self.read(8))
+    def readQuad(self, _unpack=unpack):
+        return _unpack('<2I', self.read(8))
 
-    def readFloat(self, __unpack=Struct('<f').unpack):
-        return __unpack(self.read(4))[0]
+    def readFloat(self, _unpack=unpack):
+        return _unpack('<f', self.read(4))[0]
 
-    def readDouble(self, __unpack=Struct('<d').unpack):
-        return __unpack(self.read(8))[0]
+    def readDouble(self, _unpack=unpack):
+        return _unpack('<d', self.read(8))[0]
 
-    def readPacked(self, fmt):
+    def readPacked(self, fmt, _unpack=unpack):
         return unpack(fmt, self.read(calcsize(fmt)))
 
+    def read0(self):
+        null = self.read(1)
+        if null != self.NULL:
+            raise ValueError("%s != NULL at %i" % (null, self.tell()-1))
+        return null
+
+    def readUruString(self, version=5):
+        return UruString('', version=version).readfd(self)
+
+    def readString0(self, size):
+        s = self.read(size-1)
+        self.read0()
+        return s
+
+    def readString16(self, terminate=False):
+        return String16('', terminate=terminate).readfd(self)
+
+    def readString32(self, terminate=False):
+        return String32('', terminate=terminate).readfd(self)
+
     #write
-    def writeChar(self, data, __pack=Struct('<c').pack):
-        self.write(__pack(data))
+    def writeChar(self, data, _pack=pack):
+        self.write(_pack('<c', data))
 
-    def writeByte(self, data, __pack=Struct('<B').pack):
-        self.write(__pack(data))
+    def writeByte(self, data, _pack=pack):
+        self.write(_pack('<B', data))
 
-    def writeBool(self, data, __pack=Struct('<B').pack):
-        self.write(__pack(data))
+    def writeBool(self, data, _pack=pack):
+        self.write(_pack('<B', bool(data)))
 
-    def write8(self, data, __pack=Struct('<B').pack):
-        self.write(__pack(data))
+    def write8(self, data, _pack=pack):
+        self.write(_pack('<B', data))
 
-    def write8s(self, data, __pack=Struct('<b').pack):
-        self.write(__pack(data))
+    def write8s(self, data, _pack=pack):
+        self.write(_pack('<b', data))
 
-    def write16(self, data, __pack=Struct('<H').pack):
-        self.write(__pack(data))
+    def write16(self, data, _pack=pack):
+        self.write(_pack('<H', data))
 
-    def write16s(self, data, __pack=Struct('<h').pack):
-        self.write(__pack(data))
+    def write16s(self, data, _pack=pack):
+        self.write(_pack('<h', data))
 
-    def write32(self, data, __pack=Struct('<I').pack):
-        self.write(__pack(data))
+    def write32(self, data, _pack=pack):
+        self.write(_pack('<I', data))
 
-    def write32s(self, data, __pack=Struct('<i').pack):
-        self.write(__pack(data))
+    def write32s(self, data, _pack=pack):
+        self.write(_pack('<i', data))
 
-    def write64(self, data, __pack=Struct('<Q').pack):
-        self.write(__pack(data))
+    def write64(self, data, _pack=pack):
+        self.write(_pack('<Q', data))
 
-    def write64s(self, data, __pack=Struct('<q').pack):
-        self.write(__pack(data))
+    def write64s(self, data, _pack=pack):
+        self.write(_pack('<q', data))
 
-    def writeQuad(self, data, __pack=Struct('<2I').pack):
-        self.write(__pack(data))
+    def writeQuad(self, tupl, _pack=pack):
+        self.write(_pack('<2I', *tupl))
 
-    def writeFloat(self, data, __pack=Struct('<f').pack):
-        self.write(__pack(data))
+    def writeFloat(self, data, _pack=pack):
+        self.write(_pack('<f', data))
 
-    def writeDouble(self, data, __pack=Struct('<d').pack):
-        self.write(__pack(data))
+    def writeDouble(self, data, _pack=pack):
+        self.write(_pack('<d', data))
 
+    def write0(self):
+        self.write(self.NULL)
+
+    def writeString0(self, s):
+        self.write(s)
+        self.write0()
+
     def writePacked(self, data, fmt):
-        return self.write(pack(fmt, data))
+        self.write(pack(fmt, data))
+
+    def writeUruString(self, data, version=5):
+        UruString(data, version=version).writefd(self)
+
+    def writeString16(self, data, terminate=False):
+        String16(data, terminate=terminate).writefd(self)
+
+    def writeString32(self, data, terminate=False):
+        String32(data, terminate=terminate).writefd(self)
+
+class AbstractString(object):
+    """Abstract string class
+    """
+    def __init__(self, s=''):
+        self._data = s
+
+    def readfd(self, fd):
+        raise NotImplementedError
+
+    def writefd(self, fd):
+        raise NotImplementedError
+
+    def clear(self):
+        """Clear data
+        """
+        self._data = ''
+
+    def set(self, urustr):
+        """Replace current data with urustr
+        """
+        self._data = urustr
+
+    def __repr__(self):
+        """repr(self)
+        """
+        return ("<%s at %x (%i)" % (self.__class__.__name__, id(self),
+                len(self)))
+
+    def __len__(self):
+        """len(self)
+        """
+        return len(self._data)
+
+    def __cmp__(self, other):
+        if isinstance(other, AbstractString):
+            return cmp(self._data, other._data)
+        else:
+            return cmp(self._data, other)
+
+class UruString(AbstractString):
+    """Uru Safe String
+    
+    The algorithm is based on Alcug's Ustr. This version is optimized to
+    copy and convert as less data as possible.
+    
+    version 0 - normal str
+    version 1 - auto (normal/inverted)
+    version 5 - inverted
+    version 6 - myst5
+    """
+    MYST5KEY = [ord(s) for s in "mystnerd"]
+
+    def __init__(self, urustr='', version=1):
+        AbstractString.__init__(self, urustr)
+        self.version = version
+
+    def _setVersion(self, version):
+        assert version in (0, 1, 5, 6)
+        self._version = version
+
+    def _getVersion(self):
+        return self._version
+
+    version = property(_getVersion, _setVersion)
+
+    def setInverted(self, b):
+        """Set inverted flag
+        """
+        if b:
+            self.version=5
+        else:
+            self.version=0
+
+    def writefd(self, fd):
+        """Write uru string to a file
+        """
+        if self.version == 1:
+            raise RuntimeError("Version is not explicit")
+        size = len(self._data)
+        if size > 1024: # XXX: ???
+            raise ValueError("string is too long: %i" % size)
+        if self.version == 5:
+            size |= 0xF000
+            data = ''.join([chr(ord(d) ^ 0xff) for d in self._data])
+        elif self.version == 6:
+            data = ''.join([chr(ord(d) ^ self.MYST5KEY[i%8])
+                              for i, d in enumerate(self._data)])
+        else:
+            data = self._data
+        fd.write16(size)
+        fd.write(data)
+
+    def readfd(self, fd):
+        """Read uru string from a file
+        """
+        size = fd.read16()
+        if self.version == 1:
+            inv = (size & 0xF000) == 0xF000
+            self.version = 0 if not inv else 5
+        if self.version in (0, 5):
+            size = size & 0x0FFF
+        if size > 1024: # XXX: ???
+            raise ValueError("size '%i' > 1024 at position %s(%s)" % 
+                             (size, fd.tell(), repr(fd)))
+        if self.version == 5:
+            # XXX: testme
+            # read data as tuple of integeres
+            data = fd.readPacked("<%iI" % size)
+            # OR integers with 0xff and write their char equivalent to string
+            result = ''.join([chr(d ^ 0xff) for d in data])
+        elif self.version == 6:
+            data = fd.readPacked("<%iI" % size)
+            result = ''.join([chr(d ^ self.MYST5KEY[i%8])
+                              for i, d in enumerate(data)])
+        else:
+            result = fd.read(size)
+        self._data = result
+        return result
+
+class String32(AbstractString):
+    """String with 32 bit size header
+    """
+    def __init__(self, s='', terminate=False):
+        AbstractString.__init__(self, s)
+        self._terminate = bool(terminate)
+
+    def readfd(self, fd):
+        size = fd.read32()
+        if self._terminate:
+            self._data = fd.readString0(size)
+        else:
+            self._data = fd.read(size)
+        return self._data
+
+    def writefd(self, fd):
+        size = len(self)
+        if self._terminate:
+            fd.write32(size+1)
+            fd.writeString0(self._data)
+        else:
+            fd.write32(size)
+            fd.write(self._data)
+
+class String16(AbstractString):
+    """String with 16 bit size header
+    """
+    def __init__(self, s='', terminate=False):
+        AbstractString.__init__(self, s)
+        self._terminate = bool(terminate)
+
+    def readfd(self, fd):
+        size = fd.read16()
+        if self._terminate:
+            self._data = fd.readString0(size)
+        else:
+            self._data = fd.read(size)
+        return self._data
+
+    def writefd(self, fd):
+        size = len(self)
+        if self._terminate:
+            fd.write16(size+1)
+            fd.writeString0(self._data)
+        else:
+            fd.write16(size)
+            fd.write(self._data)


Property changes on: pymoul/trunk/src/moul/crypt/binary.py
___________________________________________________________________
Name: svn:keywords
   + Id Revision
Name: svn:eol-style
   + native

Added: pymoul/trunk/src/moul/crypt/tests/test_binary.py
===================================================================
--- pymoul/trunk/src/moul/crypt/tests/test_binary.py	                        (rev 0)
+++ pymoul/trunk/src/moul/crypt/tests/test_binary.py	2007-02-10 13:14:18 UTC (rev 157)
@@ -0,0 +1,148 @@
+# pyMoul - Python interface to Myst Online URU Live
+# Copyright (C) 2007 Christian Heimes <christian (at) cheimes (dot) de>
+
+# This program is free software; you can redistribute it and/or modify it
+# under the terms of the GNU General Public License as published by the
+# Free Software Foundation; either version 2 of the License, or (at your
+# option) any later version.
+# 
+# This program is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
+# more details.
+#
+# You should have received a copy of the GNU General Public License along
+# with this program; if not, write to the Free Software Foundation, Inc., 59
+# Temple Place, Suite 330, Boston, MA 02111-1307 USA
+#
+"""moul.crypt.binary unit tests
+"""
+__author__ = "Christian Heimes"
+__version__ = "$Id$"
+__revision__ = "$Revision$"
+
+import os
+import unittest
+from doctest import DocTestSuite
+from tempfile import mkstemp
+
+from moul.crypt.binary import BinaryFile
+
+class BinaryFileTest(unittest.TestCase):
+    def setUp(self):
+        self.tmpname = mkstemp()[1]
+        self.b = BinaryFile(self.tmpname, 'wb+')
+
+    def tearDown(self):
+        self.b.close()
+        os.unlink(self.tmpname)
+
+    def _testrw(self, name, data):
+        #import pdb; pdb.set_trace()
+        read = getattr(self.b, 'read%s' % name)
+        write = getattr(self.b, 'write%s' % name)
+        write(data)
+        self.b.seek(0)
+        fdata = read()
+        self.failUnlessEqual(data, fdata)
+
+    def test_char(self):
+        self._testrw('Char', 'a')
+
+    def test_byte(self):
+        self._testrw('Byte', 127)
+
+    def test_bool(self):
+        self._testrw('Bool', True)
+
+    def test_8(self):
+        self._testrw('8', 42)
+
+    def test_8s(self):
+        self._testrw('8s', -42)
+
+    def test_16(self):
+        self._testrw('16', 2**15)
+
+    def test_16s(self):
+        self._testrw('16s', -2**15)
+
+    def test_32(self):
+        self._testrw('32', 2*31)
+
+    def test_32s(self):
+        self._testrw('32s', -2*31)
+
+    def test_64(self):
+        self._testrw('64', 2*63)
+
+    def test_64s(self):
+        self._testrw('64s', -2*63)
+
+    def test_float(self):
+        data = -0.07
+        self.b.writeFloat(data)
+        self.b.seek(0)
+        self.failUnlessAlmostEqual(data, self.b.readFloat())
+
+    def test_double(self):
+        self._testrw('Double', -23*10e200)
+
+    def test_quad(self):
+        data = (23, 42)
+        self.b.writeQuad(data)
+        self.b.seek(0)
+        self.failUnlessEqual(data, self.b.readQuad())
+
+    def test_urustring(self):
+        # XXX: no test data
+        pass
+
+    def test_string0(self):
+        s = "a test string"
+        l = len(s)
+        self.b.writeString0(s)
+        self.b.seek(0)
+        self.failUnlessEqual(self.b.size(), l+1)
+        self.failUnlessEqual(self.b.readString0(l+1), s)
+        self.b.seek(0)
+        self.failUnlessEqual(self.b.read(), s+'\x00')
+
+    def test_string16(self):
+        s = "a test string"
+        l = len(s)
+        self.b.writeString16(s, terminate=False)
+        self.b.seek(0)
+        self.failUnlessEqual(self.b.size(), l+2)
+        self.failUnlessEqual(self.b.readString16(terminate=False), s)
+
+        self.b.seek(0)
+        self.b.truncate(0)
+        self.b.writeString16(s, terminate=True)
+        self.b.seek(0)
+        self.failUnlessEqual(self.b.size(), l+3)
+        self.failUnlessEqual(self.b.readString16(terminate=True), s)
+
+    def test_string32(self):
+        s = "a test string"
+        l = len(s)
+        self.b.writeString32(s, terminate=False)
+        self.b.seek(0)
+        self.failUnlessEqual(self.b.size(), l+4)
+        self.failUnlessEqual(self.b.readString32(terminate=False), s)
+
+        self.b.seek(0)
+        self.b.truncate(0)
+        self.b.writeString32(s, terminate=True)
+        self.b.seek(0)
+        self.failUnlessEqual(self.b.size(), l+5)
+        self.failUnlessEqual(self.b.readString32(terminate=True), s)
+
+
+def test_suite():
+    return unittest.TestSuite((
+        unittest.makeSuite(BinaryFileTest),
+        ))
+
+if __name__ == '__main__':
+    unittest.main(defaultTest="test_suite")


Property changes on: pymoul/trunk/src/moul/crypt/tests/test_binary.py
___________________________________________________________________
Name: svn:keywords
   + Id Revision
Name: svn:eol-style
   + native


This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.