1. Pypy
  2. Untitled project
  3. pypy

Commits

Brian Kearns  committed a0c9719

move buffer to rpython

  • Participants
  • Parent commits 086aa47
  • Branches refactor-buffer-api

Comments (0)

Files changed (18)

File pypy/interpreter/buffer.py

  • Ignore whitespace
-"""
-Buffer protocol support.
-"""
-from rpython.rlib.objectmodel import import_from_mixin
-
-
-class Buffer(object):
-    """Abstract base class for buffers."""
-    __slots__ = ['readonly']
-    _immutable_ = True
-
-    def getlength(self):
-        raise NotImplementedError
-
-    def as_str(self):
-        "Returns an interp-level string with the whole content of the buffer."
-        # May be overridden.
-        return self.getslice(0, self.getlength(), 1, self.getlength())
-
-    def getitem(self, index):
-        "Returns the index'th character in the buffer."
-        raise NotImplementedError   # Must be overriden.  No bounds checks.
-
-    def getslice(self, start, stop, step, size):
-        # May be overridden.  No bounds checks.
-        return ''.join([self.getitem(i) for i in range(start, stop, step)])
-
-    def setitem(self, index, char):
-        "Write a character into the buffer."
-        raise NotImplementedError   # Must be overriden.  No bounds checks.
-
-    def setslice(self, start, string):
-        # May be overridden.  No bounds checks.
-        for i in range(len(string)):
-            self.setitem(start + i, string[i])
-
-    def get_raw_address(self):
-        raise ValueError("no raw buffer")
-
-    def is_writable(self):
-        return not self.readonly
-
-
-class StringBuffer(Buffer):
-    __slots__ = ['value']
-    _immutable_ = True
-
-    def __init__(self, value):
-        self.value = value
-        self.readonly = True
-
-    def getlength(self):
-        return len(self.value)
-
-    def as_str(self):
-        return self.value
-
-    def getitem(self, index):
-        return self.value[index]
-
-    def getslice(self, start, stop, step, size):
-        if size == 0:
-            return ""
-        if step == 1:
-            assert 0 <= start <= stop
-            return self.value[start:stop]
-        return "".join([self.value[start + i*step] for i in xrange(size)])
-
-
-class SubBuffer(Buffer):
-    __slots__ = ['buffer', 'offset', 'size']
-    _immutable_ = True
-
-    def __init__(self, buffer, offset, size):
-        self.readonly = buffer.readonly
-        self.buffer = buffer
-        self.offset = offset
-        self.size = size
-
-    def getlength(self):
-        at_most = self.buffer.getlength() - self.offset
-        if 0 <= self.size <= at_most:
-            return self.size
-        elif at_most >= 0:
-            return at_most
-        else:
-            return 0
-
-    def getitem(self, index):
-        return self.buffer.getitem(self.offset + index)
-
-    def getslice(self, start, stop, step, size):
-        if start == stop:
-            return ''     # otherwise, adding self.offset might make them
-                          # out of bounds
-        return self.buffer.getslice(self.offset + start, self.offset + stop,
-                                    step, size)
-
-    def setitem(self, index, char):
-        self.buffer.setitem(self.offset + index, char)
-
-    def setslice(self, start, string):
-        if len(string) == 0:
-            return        # otherwise, adding self.offset might make 'start'
-                          # out of bounds
-        self.buffer.setslice(self.offset + start, string)

File pypy/interpreter/test/test_buffer.py

  • Ignore whitespace
-import py
-from rpython.tool.udir import udir
-
-testdir = udir.ensure('test_buffer', dir=1)
-
-
-class TestBuffer:
-    def test_buffer_w(self):
-        space = self.space
-        w_hello = space.wrap('hello world')
-        buf = space.buffer_w(w_hello, space.BUF_SIMPLE)
-        assert buf.getlength() == 11
-        assert buf.as_str() == 'hello world'
-        assert buf.getslice(1, 6, 1, 5) == 'ello '
-        assert space.buffer_w(space.newbuffer(buf), space.BUF_SIMPLE) is buf
-        assert space.bufferstr_w(w_hello) == 'hello world'
-        assert space.bufferstr_w(space.newbuffer(space.buffer_w(w_hello, space.BUF_SIMPLE))) == 'hello world'
-        space.raises_w(space.w_TypeError, space.buffer_w, space.wrap(5), space.BUF_SIMPLE)
-
-    def test_file_write(self):
-        space = self.space
-        w_buffer = space.newbuffer(space.buffer_w(space.wrap('hello world'), space.BUF_SIMPLE))
-        filename = str(testdir.join('test_file_write'))
-        space.appexec([w_buffer, space.wrap(filename)], """(buffer, filename):
-            f = open(filename, 'wb')
-            f.write(buffer)
-            f.close()
-        """)
-        f = open(filename, 'rb')
-        data = f.read()
-        f.close()
-        assert data == 'hello world'
-
-    def test_unicode(self):
-        space = self.space
-        s = space.bufferstr_w(space.wrap(u'hello'))
-        assert type(s) is str
-        assert s == 'hello'
-        space.raises_w(space.w_UnicodeEncodeError,
-                       space.bufferstr_w, space.wrap(u'\xe9'))
-
-
-# Note: some app-level tests for buffer are in objspace/std/test/test_memoryview.py.

File pypy/module/__pypy__/bytebuffer.py

View file
  • Ignore whitespace
 # A convenient read-write buffer.  Located here for want of a better place.
 #
 
-from pypy.interpreter.buffer import Buffer
+from rpython.rlib.buffer import Buffer
 from pypy.interpreter.gateway import unwrap_spec
 
 

File pypy/module/_cffi_backend/cbuffer.py

View file
  • Ignore whitespace
-from pypy.interpreter.buffer import Buffer
 from pypy.interpreter.error import oefmt
 from pypy.interpreter.gateway import unwrap_spec, interp2app
 from pypy.interpreter.typedef import TypeDef, make_weakref_descr
 from pypy.module._cffi_backend import cdataobj, ctypeptr, ctypearray
 from pypy.objspace.std.bufferobject import W_Buffer
 
+from rpython.rlib.buffer import Buffer
 from rpython.rtyper.annlowlevel import llstr
 from rpython.rtyper.lltypesystem import rffi
 from rpython.rtyper.lltypesystem.rstr import copy_string_to_raw

File pypy/module/_io/interp_bufferedio.py

View file
  • Ignore whitespace
 from pypy.interpreter.typedef import (
     TypeDef, GetSetProperty, generic_new_descr, interp_attrproperty_w)
 from pypy.interpreter.gateway import interp2app, unwrap_spec, WrappedDefault
-from pypy.interpreter.buffer import Buffer
+from rpython.rlib.buffer import Buffer
 from rpython.rlib.rstring import StringBuilder
 from rpython.rlib.rarithmetic import r_longlong, intmask
 from rpython.rlib import rposix

File pypy/module/_rawffi/buffer.py

View file
  • Ignore whitespace
-from pypy.interpreter.buffer import Buffer
+from rpython.rlib.buffer import Buffer
 
 # XXX not the most efficient implementation
 

File pypy/module/array/interp_array.py

View file
  • Ignore whitespace
 from __future__ import with_statement
 
 from rpython.rlib import jit
+from rpython.rlib.buffer import Buffer
 from rpython.rlib.objectmodel import keepalive_until_here
 from rpython.rlib.rarithmetic import ovfcheck, widen
 from rpython.rlib.unroll import unrolling_iterable
 from rpython.rtyper.lltypesystem.rstr import copy_string_to_raw
 
 from pypy.interpreter.baseobjspace import W_Root
-from pypy.interpreter.buffer import Buffer
 from pypy.interpreter.error import OperationError, oefmt
 from pypy.interpreter.gateway import (
     interp2app, interpindirect2app, unwrap_spec)

File pypy/module/cpyext/bufferobject.py

View file
  • Ignore whitespace
+from rpython.rlib.buffer import StringBuffer, SubBuffer
 from rpython.rtyper.lltypesystem import rffi, lltype
-from pypy.interpreter.buffer import StringBuffer, SubBuffer
 from pypy.interpreter.error import OperationError
 from pypy.module.cpyext.api import (
     cpython_api, Py_ssize_t, cpython_struct, bootstrap_function,

File pypy/module/cpyext/slotdefs.py

View file
  • Ignore whitespace
 from pypy.module.cpyext.pyerrors import PyErr_Occurred
 from pypy.module.cpyext.state import State
 from pypy.interpreter.error import OperationError, oefmt
-from pypy.interpreter.buffer import Buffer
 from pypy.interpreter.argument import Arguments
+from rpython.rlib.buffer import Buffer
 from rpython.rlib.unroll import unrolling_iterable
 from rpython.rlib.objectmodel import specialize
 from rpython.tool.sourcetools import func_renamer

File pypy/module/micronumpy/concrete.py

View file
  • Ignore whitespace
-from pypy.interpreter.buffer import Buffer
 from pypy.interpreter.error import OperationError, oefmt
 from rpython.rlib import jit
+from rpython.rlib.buffer import Buffer
 from rpython.rlib.debug import make_sure_not_resized
 from rpython.rlib.rawstorage import alloc_raw_storage, free_raw_storage, \
     raw_storage_getitem, raw_storage_setitem, RAW_STORAGE

File pypy/module/mmap/interp_mmap.py

View file
  • Ignore whitespace
 from pypy.interpreter.baseobjspace import W_Root
 from pypy.interpreter.typedef import TypeDef
 from pypy.interpreter.gateway import interp2app, unwrap_spec
-from pypy.interpreter.buffer import Buffer
 from rpython.rlib import rmmap, rarithmetic
+from rpython.rlib.buffer import Buffer
 from rpython.rlib.rmmap import RValueError, RTypeError, RMMapError
 
 if rmmap.HAVE_LARGEFILE_SUPPORT:

File pypy/objspace/fake/objspace.py

View file
  • Ignore whitespace
 from rpython.annotator.model import SomeInstance, s_None
 from pypy.interpreter import argument, gateway
 from pypy.interpreter.baseobjspace import W_Root, ObjSpace, SpaceCache
-from pypy.interpreter.buffer import Buffer
 from pypy.interpreter.typedef import TypeDef, GetSetProperty
 from pypy.objspace.std.stdtypedef import StdTypeDef
 from pypy.objspace.std.sliceobject import W_SliceObject
+from rpython.rlib.buffer import Buffer
 from rpython.rlib.objectmodel import instantiate, we_are_translated, specialize
 from rpython.rlib.nonconst import NonConstant
 from rpython.rlib.rarithmetic import r_uint, r_singlefloat

File pypy/objspace/std/bufferobject.py

View file
  • Ignore whitespace
 """
 import operator
 
-from pypy.interpreter import buffer
+from rpython.rlib.buffer import Buffer, StringBuffer, SubBuffer
 from pypy.interpreter.baseobjspace import W_Root
 from pypy.interpreter.error import OperationError
 from pypy.interpreter.gateway import interp2app, unwrap_spec
     """
 
     def __init__(self, buf):
-        assert isinstance(buf, buffer.Buffer)
+        assert isinstance(buf, Buffer)
         self.buf = buf
 
     def buffer_w(self, space, flags):
             builder = StringBuilder(len(unistr) * UNICODE_SIZE)
             for unich in unistr:
                 pack_unichar(unich, builder)
-            from pypy.interpreter.buffer import StringBuffer
             buf = StringBuffer(builder.build())
         else:
             buf = space.readbuf_w(w_object)
         if size < -1:
             raise OperationError(space.w_ValueError,
                                  space.wrap("size must be zero or positive"))
-        buf = buffer.SubBuffer(buf, offset, size)
+        buf = SubBuffer(buf, offset, size)
         return W_Buffer(buf)
 
     def descr_len(self, space):

File pypy/objspace/std/bytearrayobject.py

View file
  • Ignore whitespace
 
 from rpython.rlib.objectmodel import (
     import_from_mixin, newlist_hint, resizelist_hint)
+from rpython.rlib.buffer import Buffer
 from rpython.rlib.rstring import StringBuilder
 
 from pypy.interpreter.baseobjspace import W_Root
-from pypy.interpreter.buffer import Buffer
 from pypy.interpreter.error import OperationError, oefmt
 from pypy.interpreter.gateway import WrappedDefault, interp2app, unwrap_spec
 from pypy.interpreter.signature import Signature

File pypy/objspace/std/bytesobject.py

View file
  • Ignore whitespace
 from rpython.rlib.jit import we_are_jitted
 from rpython.rlib.objectmodel import (
     compute_hash, compute_unique_id, import_from_mixin)
+from rpython.rlib.buffer import StringBuffer
 from rpython.rlib.rstring import StringBuilder, replace
 
 from pypy.interpreter.baseobjspace import W_Root
-from pypy.interpreter.buffer import StringBuffer
 from pypy.interpreter.error import OperationError, oefmt
 from pypy.interpreter.gateway import (
     WrappedDefault, interp2app, interpindirect2app, unwrap_spec)

File pypy/objspace/std/memoryobject.py

View file
  • Ignore whitespace
 """
 import operator
 
-from pypy.interpreter import buffer
+from rpython.rlib.buffer import Buffer, SubBuffer
 from pypy.interpreter.baseobjspace import W_Root
 from pypy.interpreter.error import OperationError
 from pypy.interpreter.gateway import interp2app, unwrap_spec
     """
 
     def __init__(self, buf):
-        assert isinstance(buf, buffer.Buffer)
+        assert isinstance(buf, Buffer)
         self.buf = buf
 
     def buffer_w(self, space, flags):
         size = stop - start
         if size < 0:
             size = 0
-        buf = buffer.SubBuffer(self.buf, start, size)
+        buf = SubBuffer(self.buf, start, size)
         return W_MemoryView(buf)
 
     def descr_tobytes(self, space):

File rpython/rlib/buffer.py

View file
  • Ignore whitespace
+"""
+Buffer protocol support.
+"""
+from rpython.rlib.objectmodel import import_from_mixin
+
+
+class Buffer(object):
+    """Abstract base class for buffers."""
+    __slots__ = ['readonly']
+    _immutable_ = True
+
+    def getlength(self):
+        raise NotImplementedError
+
+    def as_str(self):
+        "Returns an interp-level string with the whole content of the buffer."
+        # May be overridden.
+        return self.getslice(0, self.getlength(), 1, self.getlength())
+
+    def getitem(self, index):
+        "Returns the index'th character in the buffer."
+        raise NotImplementedError   # Must be overriden.  No bounds checks.
+
+    def getslice(self, start, stop, step, size):
+        # May be overridden.  No bounds checks.
+        return ''.join([self.getitem(i) for i in range(start, stop, step)])
+
+    def setitem(self, index, char):
+        "Write a character into the buffer."
+        raise NotImplementedError   # Must be overriden.  No bounds checks.
+
+    def setslice(self, start, string):
+        # May be overridden.  No bounds checks.
+        for i in range(len(string)):
+            self.setitem(start + i, string[i])
+
+    def get_raw_address(self):
+        raise ValueError("no raw buffer")
+
+    def is_writable(self):
+        return not self.readonly
+
+
+class StringBuffer(Buffer):
+    __slots__ = ['value']
+    _immutable_ = True
+
+    def __init__(self, value):
+        self.value = value
+        self.readonly = True
+
+    def getlength(self):
+        return len(self.value)
+
+    def as_str(self):
+        return self.value
+
+    def getitem(self, index):
+        return self.value[index]
+
+    def getslice(self, start, stop, step, size):
+        if size == 0:
+            return ""
+        if step == 1:
+            assert 0 <= start <= stop
+            return self.value[start:stop]
+        return "".join([self.value[start + i*step] for i in xrange(size)])
+
+
+class SubBuffer(Buffer):
+    __slots__ = ['buffer', 'offset', 'size']
+    _immutable_ = True
+
+    def __init__(self, buffer, offset, size):
+        self.readonly = buffer.readonly
+        self.buffer = buffer
+        self.offset = offset
+        self.size = size
+
+    def getlength(self):
+        at_most = self.buffer.getlength() - self.offset
+        if 0 <= self.size <= at_most:
+            return self.size
+        elif at_most >= 0:
+            return at_most
+        else:
+            return 0
+
+    def getitem(self, index):
+        return self.buffer.getitem(self.offset + index)
+
+    def getslice(self, start, stop, step, size):
+        if start == stop:
+            return ''     # otherwise, adding self.offset might make them
+                          # out of bounds
+        return self.buffer.getslice(self.offset + start, self.offset + stop,
+                                    step, size)
+
+    def setitem(self, index, char):
+        self.buffer.setitem(self.offset + index, char)
+
+    def setslice(self, start, string):
+        if len(string) == 0:
+            return        # otherwise, adding self.offset might make 'start'
+                          # out of bounds
+        self.buffer.setslice(self.offset + start, string)

File rpython/rlib/test/test_buffer.py

View file
  • Ignore whitespace
+from rpython.rlib.buffer import *
+
+
+def test_string_buffer():
+    buf = StringBuffer('hello world')
+    assert buf.getitem(4) == 'o'
+    assert buf.getlength() == 11
+    assert buf.getslice(1, 6, 1, 5) == 'ello '
+    assert buf.as_str() == 'hello world'