Merge pull request #971 from cla7aye15I4nd/dev

Add connectivity peripheral's base class

Author: cla7aye15I4nd

examples/mcu/stm32f411_i2c_lcd.py

@@ -80,31 +80,32 @@ def render(self):
         print('LCD quit')
         pygame.quit()
 
-    def send(self, data):
-        self.buf.append(data)
-
-        if len(self.buf) == 4:
-            up = self.buf[0] & 0xf0
-            lo = self.buf[3] & 0xf0
-            cmd = up | (lo >> 4)
-
-            if self.buf[0] & 0x1:                
-                if self.cur_col < 16 and self.cur_row < 2:
-                    self.data[self.cur_row][self.cur_col] = cmd                
-                    self.cur_col += 1
-                self.pixels = LCD.make_screen(self.data)
-            
-            elif cmd == 0x1:
-                self.data = [[ord(' ') for _ in range(self.cols)] for _ in range(self.rows)]
-                self.pixels = LCD.make_screen(self.data)
-            
-            elif up == 0x80:
-                self.cur_row, self.cur_col = 0, lo >> 4
-
-            elif up == 0xc0:
-                self.cur_row, self.cur_col = 1, lo >> 4                
-
-            self.buf = []
+    def send(self, buf):
+        for data in buf:
+            self.buf.append(data)
+
+            if len(self.buf) == 4:
+                up = self.buf[0] & 0xf0
+                lo = self.buf[3] & 0xf0
+                cmd = up | (lo >> 4)
+
+                if self.buf[0] & 0x1:                
+                    if self.cur_col < 16 and self.cur_row < 2:
+                        self.data[self.cur_row][self.cur_col] = cmd                
+                        self.cur_col += 1
+                    self.pixels = LCD.make_screen(self.data)
+                
+                elif cmd == 0x1:
+                    self.data = [[ord(' ') for _ in range(self.cols)] for _ in range(self.rows)]
+                    self.pixels = LCD.make_screen(self.data)
+                
+                elif up == 0x80:
+                    self.cur_row, self.cur_col = 0, lo >> 4
+
+                elif up == 0xc0:
+                    self.cur_row, self.cur_col = 1, lo >> 4                
+
+                self.buf = []
 
     def run(self):
         threading.Thread(target=self.render).start()

qiling/hw/char/stm32f4xx_usart.py

@@ -6,9 +6,10 @@
 import ctypes
 
 from qiling.hw.peripheral import QlPeripheral
+from qiling.hw.connectivity import QlConnectivityPeripheral
 from qiling.hw.const.stm32f4xx_usart import USART_SR, USART_CR1
 
-class STM32F4xxUsart(QlPeripheral):
+class STM32F4xxUsart(QlConnectivityPeripheral):
     class Type(ctypes.Structure):
         """ the structure available in :
 			stm32f413xx.h
@@ -56,9 +57,6 @@ def __init__(self, ql, label, intn=None):
 
         self.intn = intn
 
-        self.recv_buf = bytearray()
-        self.send_buf = bytearray()
-
     @QlPeripheral.debug_info()
     def read(self, offset: int, size: int) -> int:
         buf = ctypes.create_string_buffer(size)
@@ -92,33 +90,15 @@ def transfer(self):
             data = self.usart.DR
 
             self.usart.SR |= USART_SR.TXE            
-            self.send_buf.append(data)
+            self.send_to_user(data)
 
         if not (self.usart.SR & USART_SR.RXNE): 
             # TXE bit must had been cleared
-            if self.recv_buf:
+            if self.has_input():
                 self.usart.SR |= USART_SR.RXNE
-                self.usart.DR = self.recv_buf.pop(0)        
-        
-
-    def send(self, data: bytes):
-        """ send user data into USART.
-
-        Args:
-            data (bytes): Input Data
-        """        
-        self.recv_buf += data
-
-    def recv(self) -> bytes:
-        """ receive data from USART.
-
-        Returns:
-            bytes: USART send buffer data
-        """
-        data = bytes(self.send_buf)
-        self.send_buf.clear()
-        return data
+                self.usart.DR = self.recv_from_user()
 
+    @QlConnectivityPeripheral.device_handler
     def step(self):
         self.transfer()
 

qiling/hw/connectivity.py

@@ -0,0 +1,95 @@
+#!/usr/bin/env python3
+# 
+# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
+#
+
+
+import ctypes
+import queue
+
+from qiling.core import Qiling
+from qiling.hw.peripheral import QlPeripheral
+
+
+class QlConnectivityPeripheral(QlPeripheral):
+    class Type(ctypes.Structure):
+        """ Define the reigister fields of peripheral.
+
+            Example:
+                fields_ = [
+                    ('SR'  , ctypes.c_uint32),
+                    ('DR'  , ctypes.c_uint32),
+                    ('BRR' , ctypes.c_uint32),
+                    ('CR1' , ctypes.c_uint32),
+                    ('CR2' , ctypes.c_uint32),
+                    ('CR3' , ctypes.c_uint32),
+                    ('GTPR', ctypes.c_uint32),
+                ]
+        """        
+        _fields_ = []
+
+    def __init__(self, ql: Qiling, label: str, limit:int = 1):
+        super().__init__(ql, label)
+        
+        self.rtube = queue.Queue()
+        self.wtube = queue.Queue()
+
+        self.limit = limit
+        self.device_list = []
+
+    def has_input(self):
+        return not self.rtube.empty()
+
+    def send(self, data: bytes):
+        """ Send data into the peripheral.
+            
+            Example:
+                ql.hw.usart1.send(b'hello')
+        """
+
+        for byte in bytearray(data):
+            self.rtube.put(byte)
+
+    def recv(self, numb:int = 4096) -> bytes:
+        """ Receive data from peripheral
+
+            Example:
+                data = ql.hw.i2c1.send()
+        """
+        data = bytearray()
+        while self.can_recv() and numb != 0:
+            data.append(self.wtube.get())
+            numb -= 1
+
+        return bytes(data) 
+
+    def can_recv(self):
+        return not self.wtube.empty()
+
+    def send_to_user(self, data: int):
+        """ send single byte to user
+        """
+        
+        self.wtube.put(data)
+
+    def recv_from_user(self) -> bytes:
+        """ Read single byte from user input
+        """
+        
+        return self.rtube.get()
+
+    def connect(self, device):
+        if len(self.device_list) < self.limit:
+            self.device_list.append(device)
+
+    @staticmethod
+    def device_handler(func):
+        def wrapper(self):            
+            if self.device_list and self.can_recv():
+                data = self.recv(numb=1)			
+                for device in self.device_list:
+                    device.send(data)
+
+            func(self)
+        
+        return wrapper

qiling/hw/i2c/stm32f4xx_i2c.py

@@ -6,10 +6,11 @@
 import ctypes
 
 from qiling.hw.peripheral import QlPeripheral
+from qiling.hw.connectivity import QlConnectivityPeripheral
 from qiling.hw.const.stm32f4xx_i2c import I2C_CR1, I2C_CR2, I2C_SR1, I2C_SR2, I2C_DR, I2C_OAR1, I2C_OAR2
 
 
-class STM32F4xxI2c(QlPeripheral):
+class STM32F4xxI2c(QlConnectivityPeripheral):
 	class Type(ctypes.Structure):
 		""" the structure is available in :
 			stm32f423xx.h
@@ -52,9 +53,6 @@ def __init__(self, ql, label, ev_intn=None, er_intn=None):
 		self.ev_intn = ev_intn # event interrupt
 		self.er_intn = er_intn # error interrupt
 
-		self.devices = []
-		self.current = None
-		
 		self.reset()
 
 	def reset(self):		
@@ -141,21 +139,21 @@ def generate_stop(self):
 
 		self.i2c.SR1 |= I2C_SR1.STOPF
 		self.i2c.SR1 &= ~I2C_SR1.ADDR
+		
 		self.set_slave_mode()
+		self.send_event_interrupt()
 
 	def send_address(self):
 		if self.i2c.DR == self.i2c.OAR1 >> 1:
-			for dev in self.devices:
-				if self.i2c.DR == dev.address:
-					self.current = dev
 
 			# TODO: send ACK
 			self.i2c.SR1 |= I2C_SR1.ADDR | I2C_SR1.TXE
 			self.send_event_interrupt()
 
 	def send_data(self):
 		self.i2c.SR1 |= I2C_SR1.BTF | I2C_SR1.TXE
-		self.current.send(self.i2c.DR)
+
+		self.send_to_user(self.i2c.DR)
 		self.send_event_interrupt()
 
 	## I2C Status register 2 (I2C_SR2)
@@ -189,16 +187,17 @@ def is_7bit_mode(self):
 	def fetch_device_address(self):
 		# dual addressing mode
 		if self.i2c.OAR2 & I2C_OAR2.ENDUAL:
-			self.i2c.OAR1 = self.devices[0].address << 1
-			self.i2c.OAR2 = I2C_OAR2.ENDUAL | (self.devices[1].address << 1)
+			self.i2c.OAR1 = self.device_list[0].address << 1
+			self.i2c.OAR2 = I2C_OAR2.ENDUAL | (self.device_list[1].address << 1)
 
 		# single device, 10-bit slave address
 		elif self.i2c.OAR1 & I2C_OAR1.ADDMODE: 
-			self.i2c.OAR1 = I2C_OAR1.ADDMODE | self.devices[0].address
+			self.i2c.OAR1 = I2C_OAR1.ADDMODE | self.device_list[0].address
 
 		# single device, 7-bit slave address
 		else:
-			self.i2c.OAR1 = self.devices[0].address << 1
+			self.i2c.OAR1 = self.device_list[0].address << 1
 
-	def connect(self, dev):
-		self.devices.append(dev)
+	@QlConnectivityPeripheral.device_handler
+	def step(self):
+		pass

qiling/hw/spi/stm32f4xx_spi.py

@@ -4,11 +4,13 @@
 #
 
 import ctypes
+
 from qiling.hw.peripheral import QlPeripheral
+from qiling.hw.connectivity import QlConnectivityPeripheral
 from qiling.hw.const.stm32f4xx_spi import SPI_CR1, SPI_CR2, SPI_SR, SPI_CRCPR, SPI_I2SCFGR, SPI_I2SPR
 
 
-class STM32F4xxSpi(QlPeripheral):
+class STM32F4xxSpi(QlConnectivityPeripheral):
     class Type(ctypes.Structure):
         """ the structure available in :
             stm32f413xx.h
@@ -99,10 +101,12 @@ def write(self, offset: int, size: int, value: int):
         ctypes.memmove(ctypes.addressof(self.spi) + offset, data, size)   
 
         if self.in_field(self.struct.DR, offset, size):
-            self.send_data()     
+            self.spi.SR |= SPI_SR.RXNE
+            self.send_to_user(self.spi.DR)
 
     def send_interrupt(self):
         self.ql.hw.nvic.set_pending(self.intn)
 
-    def send_data(self):
-        self.spi.SR |= SPI_SR.RXNE
+    @QlConnectivityPeripheral.device_handler
+    def step(self):
+        pass

qiling/hw/timer/stm32f4xx_rtc.py

@@ -3,8 +3,11 @@
 # Cross Platform and Multi Architecture Advanced Binary Emulation Framework
 #
 
+import time
 import ctypes
+
 from qiling.hw.peripheral import QlPeripheral
+from qiling.hw.const.stm32f4xx_rtc import RTC_TR, RTC_ISR
 
 
 class STM32F4xxRtc(QlPeripheral):
@@ -99,5 +102,27 @@ def read(self, offset: int, size: int) -> int:
 
     @QlPeripheral.debug_info()
     def write(self, offset: int, size: int, value: int):
+        if offset == self.struct.ISR.offset:
+            for bitmask in [
+                RTC_ISR.TAMP1F, 
+                RTC_ISR.TSOVF, 
+                RTC_ISR.TSF, 
+                RTC_ISR.WUTF, 
+                RTC_ISR.ALRBF, 
+                RTC_ISR.ALRAF, 
+                RTC_ISR.RSF
+            ]:
+                if value & bitmask == 0:
+                    self.rtc.ISR &= ~bitmask
+
+            self.rtc.ISR = (self.rtc.ISR & ~RTC_ISR.INIT) | (value & RTC_ISR.INIT)            
+            return
+
         data = (value).to_bytes(size, 'little')
-        ctypes.memmove(ctypes.addressof(self.rtc) + offset, data, size)
+        ctypes.memmove(ctypes.addressof(self.rtc) + offset, data, size)    
+
+    def step(self):
+        if self.rtc.ISR & RTC_ISR.INIT:
+            self.rtc.ISR |= RTC_ISR.INITF
+
+        self.rtc.ISR |= RTC_ISR.RSF

qiling/hw/utils/bcd.py

@@ -0,0 +1,16 @@
+#!/usr/bin/env python3
+# 
+# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
+#
+
+
+def byte2bcd(value):
+    bcdhigh = 0
+    while value >= 10:
+        bcdhigh += 1
+        value -= 10
+
+    return (bcdhigh << 4) | value
+
+def bcd2byte(value):
+    return ((value & 0xF0) >> 0x4) * 10 + (value & 0xf)