adapter_frontendclass.py

Go to the documentation of this file.

 
 
from ctypes import sizeof, c_uint32
from io import StringIO
from enum import Enum
from math import trunc
 
from .adapter_backendfunctions import *
 
 
class DataReaderWriterInitException(Exception):
    pass
 
 
 
class DataReaderWriterException(Exception):
    pass
 
 
 
class MemoryBlock(Enum):
    
    MEMORY_BLOCK_0 = 0,
    
    MEMORY_BLOCK_1 = 1,
    
    MEMORY_BLOCK_2 = 2
 
 
 
class DataReaderWriter:
 
    
    def __init__(self, blockNumber: MemoryBlock):
        self._initialized = False
        self._blockNumber = blockNumber
        self._dataHandle = Handle()
        self._memoryBlockByteSize = 0
 
    
 
    
    def __enter__(self):
        block_name = self._GetBlockNameForBlockNumber(self._blockNumber)
        errorCode = ddr_openmemoryblock(block_name, self._dataHandle)
        if errorCode == DDR_ErrorCode.DDR_ErrorCode_Success:
            value = ValueHandle()
            errorCode = ddr_getmemorysize(self._dataHandle, value)
            if errorCode == DDR_ErrorCode.DDR_ErrorCode_Success:
                self._memoryBlockByteSize = value.value * sizeof(c_uint32)
                self._initialized = True
            else:
                msg = self._ConstructErrorMessage(errorCode, "Getting size of memory block")
                raise DataReaderWriterInitException(msg)
        else:
            msg = self._ConstructErrorMessage(errorCode, "Initializing data reader/writer")
            raise DataReaderWriterInitException(msg)
        return self
 
 
    
    def __exit__(self, *args):
        if self._initialized:
            ddr_closememoryblock(self._dataHandle)
            self._initialized = False
 
 
    
    @property
    def MemoryBlockByteSize(self):
        if not self._initialized:
            msg = self._ConstructErrorMessage(DDR_ErrorCode.DDR_ErrorCode_Block_Not_Opened,
                                             "Memory block not opened so cannot retrieve memory block size")
            raise DataReaderWriterInitException(msg)
        return self._memoryBlockByteSize
 
 
    
    def _GetBlockNameForBlockNumber(self, block_number: MemoryBlock):
        block_name = "unknown"
 
        if block_number == MemoryBlock.MEMORY_BLOCK_0:
            block_name = BLOCK_NAME_0
        elif block_number == MemoryBlock.MEMORY_BLOCK_1:
            block_name = BLOCK_NAME_1
        elif block_number == MemoryBlock.MEMORY_BLOCK_2:
            block_name = BLOCK_NAME_2
        return block_name
 
 
    
    def _GetErrorMessage(self, errorCode: DDR_ErrorCode) -> str:
        message = "";
 
        match errorCode:
            case DDR_ErrorCode.DDR_ErrorCode_Success:
                message = "Operation succeeded";
 
            case DDR_ErrorCode.DDR_ErrorCode_Block_Already_Opened:
                message = "Memory block is already open and cannot be opened again";
 
            case DDR_ErrorCode.DDR_ErrorCode_Block_Not_Opened:
                message = "Memory block is closed and cannot be accessed";
 
            case DDR_ErrorCode.DDR_ErrorCode_Invalid_Block_Name:
                message = "The given name is not a recognized memory block name";
 
            case DDR_ErrorCode.DDR_ErrorCode_Invalid_Handle:
                message = "The handle argument does not correspond to a valid open memory block";
 
            case DDR_ErrorCode.DDR_ErrorCode_Invalid_Offset:
                message = "The given offset is out of bounds";
 
            case DDR_ErrorCode.DDR_ErrorCode_Null_Argument:
                message = "The block name pointer or return handle pointer argument is NULL";
 
            case _ :
                message = f"Unrecognized error code: {errorCode}"
 
        return message
 
    
    def _ConstructErrorMessage(self, errorCode: int, operation : str) -> str:
        msg = self._GetErrorMessage(errorCode)
        return "{0}: {1}".format(operation, msg)
 
 
    
    def Read(self, byteOffset: int, maxBytes : int) -> bytearray:
        if not self._initialized:
            raise DataReaderWriterInitException("Data reader/writer is not initialized.  Unable to read.");
 
        return_data = bytearray()
        if maxBytes > 0:
            chunkOffset = byteOffset // sizeof(c_uint32)
            value = ValueHandle()
            bufferIndex = 0
            errorCode = ddr_getdatachunk(self._dataHandle, chunkOffset, value)
            if errorCode == DDR_ErrorCode.DDR_ErrorCode_Success:
                byteOffsetInChunk = byteOffset % sizeof(c_uint32)
                while bufferIndex < maxBytes:
                    return_data.append(value.value & 0xff)
                    bufferIndex += 1
                    value.value >>= 8
                    byteOffsetInChunk += 1
                    if byteOffsetInChunk == sizeof(c_uint32):
                        chunkOffset += 1
                        if chunkOffset >= DDR_MAX_OFFSET:
                            break
                        byteOffsetInChunk = 0
                        errorCode = ddr_getdatachunk(self._dataHandle, chunkOffset, value)
                        if errorCode != DDR_ErrorCode.DDR_ErrorCode_Success:
                            msg = self._ConstructErrorMessage(errorCode, "Reading memory")
                            raise DataReaderWriterException(msg)
            else:
                msg = self._ConstructErrorMessage(errorCode, "Reading memory")
                raise DataReaderWriterException(msg)
 
        return return_data
 
 
    
    def Write(self, byteOffset: int, data : bytearray, maxBytes: int) -> None:
        if not self._initialized:
            raise DataReaderWriterInitException("Data reader/writer is not initialized.  Unable to write.");
        if maxBytes > 0:
            errorCode = DDR_ErrorCode.DDR_ErrorCode_Success
            chunkOffset = byteOffset // sizeof(c_uint32)
            value = ValueHandle()
            byteOffsetInChunk = int(byteOffset % sizeof(c_uint32))
            bufferIndex = 0
            byteMask = 0xff << (byteOffsetInChunk * 8)
            if byteOffsetInChunk != 0:
                errorCode = ddr_getdatachunk(self._dataHandle, chunkOffset, value)
            if errorCode == DDR_ErrorCode.DDR_ErrorCode_Success:
                while bufferIndex < maxBytes:
                    value.value = value.value & (byteMask ^ 0xffffffff)
                    value.value = value.value | (data[bufferIndex] << (byteOffsetInChunk * 8))
                    bufferIndex += 1
                    byteMask <<= 8
                    byteOffsetInChunk += 1
                    if byteOffsetInChunk == sizeof(c_uint32):
                        errorCode = ddr_setdatachunk(self._dataHandle, chunkOffset, value.value)
                        if errorCode == DDR_ErrorCode.DDR_ErrorCode_Success:
                            byteMask = 0xff
                            byteOffsetInChunk = 0
                            chunkOffset += 1
                            if chunkOffset >= DDR_MAX_OFFSET:
                                break
                            errorCode = ddr_getdatachunk(self._dataHandle, chunkOffset, value)
                            if errorCode != DDR_ErrorCode.DDR_ErrorCode_Success:
                                msg = self._ConstructErrorMessage(errorCode, "Reading memory")
                                raise DataReaderWriterException(msg)
                        else:
                            msg = self._ConstructErrorMessage(errorCode, "Writing memory")
                            raise DataReaderWriterException(msg)
                if errorCode == DDR_ErrorCode.DDR_ErrorCode_Success:
                    if byteOffsetInChunk != 0:
                        errorCode = ddr_setdatachunk(self._dataHandle, chunkOffset, value.value)
 
                if errorCode != DDR_ErrorCode.DDR_ErrorCode_Success:
                    msg = self._ConstructErrorMessage(errorCode, "Writing memory")
                    raise DataReaderWriterException(msg)
            else:
                msg = self._ConstructErrorMessage(errorCode, "Reading memory in preparation to writing memory")
                raise DataReaderWriterException(msg)
 
 
    
    def BufferToString(self, data: bytearray, maxBytes : int, indent: int) -> str:
        output = StringIO()
        indent_spaces = ' ' * indent
 
        if maxBytes != 0:
            byteCount = maxBytes
            if byteCount > len(data):
                byteCount = len(data)
            bytesPerRow = 32
            for row in range(0, maxBytes, bytesPerRow):
                output.write("{}{:04x} --".format(indent_spaces, row))
                for col in range(0, bytesPerRow):
                    if (col + row) >= maxBytes:
                        break
                    if col > 0:
                        output.write(" ")
                    dataIndex = row + col
                    output.write("{:02x}".format(data[dataIndex]))
                output.write("\n")
 
        return output.getvalue()