Add "readwire" interface to *some* Graphene Types.

graphenebase/account.py

@@ -331,6 +331,14 @@ def __bytes__(self):
         """ Returns the raw public key (has length 33)"""
         return bytes(self._pk)
 
+    @staticmethod
+    def fromBytes(d, prefix="GPH"):
+        _pk = hexlify(d[:33]).decode('ascii')
+
+        k = PublicKey(_pk)
+
+        return k, d[33:]
+
     def __lt__(self, other):
         """ For sorting of public keys (due to graphene),
             we actually sort according to addresses

graphenebase/types.py

@@ -33,6 +33,22 @@ def varintdecode(data):  # pragma: no cover
     return result
 
 
+def varintdecode2(data):
+    """ Varint decoding (with length counting)
+    """
+    nbytes = 0
+    shift = 0
+    result = 0
+    for c in data:
+        b = c # ord(c) not needed, `data` is bytes
+        result |= ((b & 0x7f) << shift)
+        nbytes += 1
+        if not (b & 0x80):
+            break
+        shift += 7
+    return result, nbytes
+
+
 def variable_buffer(s):
     """ Encode variable length buffer
     """
@@ -88,6 +104,11 @@ def __bytes__(self):
     def __str__(self):
         return '%d' % self.data
 
+    @staticmethod
+    def fromBytes(d):
+       val = struct.unpack("<I", d[:4]) [0]
+       return Uint32( val ), d[4:]
+
 
 class Uint64():
     def __init__(self, d):
@@ -121,6 +142,12 @@ def __bytes__(self):
     def __str__(self):
         return '%d' % self.data
 
+    @staticmethod
+    def fromBytes(d):
+       val = struct.unpack("<q", d[:8]) [0]
+       d = d[8:]
+       return Int64( val ), d
+
 
 class String():
     def __init__(self, d):
@@ -136,6 +163,12 @@ def __bytes__(self):
     def __str__(self):
         return '%s' % str(self.data)
 
+    @staticmethod
+    def fromBytes(d):
+        vallen, lenlen = varintdecode2(d)
+        d = d[lenlen:]
+        val = d[:vallen]
+        return String(val), d[vallen:]
 
 class Bytes():
     def __init__(self, d):
@@ -148,6 +181,42 @@ def __bytes__(self):
     def __str__(self):
         return str(self.data)
 
+    def __json__(self):
+        return str(self)
+
+    @staticmethod
+    def fromBytes(d):
+        vallen, lenlen = varintdecode2(d)
+        d = d[lenlen:]
+        val = d[:vallen]
+        return Bytes(val), d[vallen:]
+
+
+class Fixed_Bytes():
+    def __init__(self, d, length=None):
+        if isinstance(d, str):
+            d = unhexlify(bytes(d, 'utf-8'))
+        self.data = d
+        if length:
+            self.length = length
+        else:
+            self.length = len(self.data)
+
+    def __bytes__(self):
+        # TODO: constrain to self.length
+        return self.data
+
+    def __str__(self):
+        return str(hexlify(self.data), 'utf-8')
+
+    def __json__(self):
+        return str(self)
+
+    @staticmethod
+    def fromBytes(d, vallen):
+        val = d[:vallen]
+        return Fixed_Bytes(val, vallen), d[vallen:]
+
 
 class Void():
     def __init__(self):
@@ -247,6 +316,13 @@ def isempty(self):
             return True
         return not bool(bytes(self.data))
 
+    @staticmethod
+    def fromBytes(d, stype, **skwargs):
+        b = d[0] #int(unhexlify(d[0:2]))
+        if not b:
+            return Optional(None), d[1:]
+        v, d = stype.fromBytes(d[1:], **skwargs)
+        return Optional(v), d
 
 class Static_variant():
     def __init__(self, d, type_id):
@@ -332,6 +408,11 @@ def __bytes__(self):
     def __str__(self):
         return self.Id
 
+    @staticmethod
+    def fromBytes(d, prefix="1.2."):
+        val, vallen = varintdecode2(d)
+        return ObjectId(prefix + str(val)), d[vallen:]
+
 
 class FullObjectId():
     """ Encodes object ids - serializes to a full object id

tests/test_bytes.py

@@ -0,0 +1,74 @@
+import pytest
+import unittest
+from binascii import hexlify, unhexlify
+from graphenebase.account import PrivateKey, PublicKey
+from graphenebase.types import (
+    Uint32, Int64, String, Bytes,
+    Fixed_Bytes,
+    Optional, ObjectId
+)
+
+
+wif = "5J4KCbg1G3my9b9hCaQXnHSm6vrwW9xQTJS6ZciW2Kek7cCkCEk"
+pub_key = str(PrivateKey(wif).pubkey)
+
+class Testcases(unittest.TestCase):
+
+    def test_read_pubkey(self):
+        w = bytes( PublicKey(pub_key) )
+        r, b = PublicKey.fromBytes(w)
+        self.assertEqual(str(r), pub_key)
+
+    def test_read_uint32(self):
+        v = 256
+        w = bytes( Uint32( v ) )
+        r, b = Uint32.fromBytes(w)
+        self.assertEqual(r.data, v)
+
+    def test_read_int64(self):
+        v = -23
+        w = bytes( Int64( v ) )
+        r, b = Int64.fromBytes(w)
+        self.assertEqual(r.data, v)
+
+    def test_read_string(self):
+        v = "hello world"
+        w = bytes( String( v ) )
+        r, b = String.fromBytes(w)
+        self.assertEqual(r.data.decode('utf-8'), v)
+
+    def test_read_bytes(self):
+        v = "040810172A"
+        w = bytes( Bytes( v ) )
+        r, b = Bytes.fromBytes(w)
+        self.assertEqual(r.data, unhexlify(v))
+
+    def test_read_fixedbytes(self):
+        v = "0408"
+        w = bytes( Fixed_Bytes( v, 4 ) )
+        r, b = Fixed_Bytes.fromBytes(w, 4)
+        self.assertEqual(r.data, unhexlify(v))
+
+    def test_read_objectid(self):
+        v = "1.2.0"
+        w = bytes( ObjectId( v ) )
+        r, b = ObjectId.fromBytes(w)
+        self.assertEqual(r.Id, v)
+
+    def test_read_optional_none(self):
+        v = None
+        w = bytes( Optional( None ) )
+        r, b = Optional.fromBytes(w, String)
+        self.assertEqual(r.data, v)
+
+    def test_read_optional_string(self):
+        v = "hello world"
+        w = bytes( Optional( String( v ) ) )
+        r, b = Optional.fromBytes(w, String)
+        r = r.data
+        self.assertEqual(r.data.decode('utf-8'), v)
+
+
+
+if __name__ == '__main__':
+    unittest.main()