mapped_struct.py 文件源码

python
阅读 19 收藏 0 点赞 0 评论 0

项目:sharedbuffers 作者: jampp 项目源码 文件源码
def get(self, key, default = None):
        if not isinstance(key, (int, long)):
            return default
        if key < 0 or key > self.dtypemax:
            return default
        hkey = key
        startpos = self._search_hkey(hkey)
        nitems = self.index_elements
        if 0 <= startpos < nitems:
            buf = self._buf
            dtype = self._dtype
            if cython.compiled:
                #lint:disable
                buf = self._likebuf
                PyObject_GetBuffer(buf, cython.address(pybuf), PyBUF_SIMPLE)
                try:
                    if dtype is npuint64:
                        PyObject_GetBuffer(self.index, cython.address(indexbuf), PyBUF_STRIDED_RO)
                        try:
                            if ( indexbuf.strides == cython.NULL 
                                    or indexbuf.ndim < 2
                                    or indexbuf.len < nitems * indexbuf.strides[0] ):
                                raise ValueError("Invalid buffer state")
                            stride0 = indexbuf.strides[0]
                            stride1 = indexbuf.strides[1]
                            pindex = cython.cast(cython.p_char, indexbuf.buf) + startpos * stride0
                            pindexend = cython.cast(cython.p_char, indexbuf.buf) + indexbuf.len - stride0 + 1
                            if pindex < pindexend and cython.cast(cython.p_ulonglong, pindex)[0] == hkey:
                                return cython.cast(cython.p_ulonglong, pindex + stride1)[0]
                        finally:
                            PyBuffer_Release(cython.address(indexbuf))
                    elif dtype is npuint32:
                        PyObject_GetBuffer(self.index, cython.address(indexbuf), PyBUF_STRIDED_RO)
                        try:
                            if ( indexbuf.strides == cython.NULL 
                                    or indexbuf.ndim < 2
                                    or indexbuf.len < nitems * indexbuf.strides[0] ):
                                raise ValueError("Invalid buffer state")
                            stride0 = indexbuf.strides[0]
                            stride1 = indexbuf.strides[1]
                            pindex = cython.cast(cython.p_char, indexbuf.buf) + startpos * stride0
                            pindexend = cython.cast(cython.p_char, indexbuf.buf) + indexbuf.len - stride0 + 1
                            if pindex < pindexend and cython.cast(cython.p_uint, pindex)[0] == hkey:
                                return cython.cast(cython.p_uint, pindex + stride1)[0]
                        finally:
                            PyBuffer_Release(cython.address(indexbuf))
                    else:
                        index = self.index
                        if startpos < nitems and index[startpos,0] == hkey:
                            return index[startpos,1]
                finally:
                    PyBuffer_Release(cython.address(pybuf))
                #lint:enable
            else:
                index = self.index
                if startpos < nitems and index[startpos,0] == hkey:
                    return index[startpos,1]
        return default
评论列表
文章目录


问题


面经


文章

微信
公众号

扫码关注公众号