UART test

Author: noname

.idea/vcs.xml

@@ -0,0 +1,71 @@
+from _socket import inet_ntoa
+
+from proto.proto_py import uart_pb2
+from enum import Enum
+
+
+RX_BUFFER = bytearray()
+TX_COMPLETE = True
+TX_SPACE = 64
+SEQ_NUM = 0
+
+
+class UartId(Enum):
+    UART1 = 0
+    UART2 = 1
+
+
+def encode_uart_config_msg(uart_id: UartId, baud_rate: int) -> bytes:
+    global SEQ_NUM
+    SEQ_NUM += 1
+    msg = uart_pb2.UartMsg()
+
+    if uart_id == UartId.UART1:
+        msg.cfg_msg.id = uart_pb2.UartId.UART1
+    else:
+        msg.cfg_msg.id = uart_pb2.UartId.UART2
+
+    msg.cfg_msg.baudrate = baud_rate
+    msg.sequence_number = SEQ_NUM
+
+    return msg.SerializeToString()
+
+
+def encode_uart_data_msg(uart_id: UartId, data: bytes) -> bytes:
+    global SEQ_NUM
+    SEQ_NUM += 1
+    msg = uart_pb2.UartMsg()
+
+    if uart_id == UartId.UART1:
+        msg.data_msg.id = uart_pb2.UartId.UART1
+    else:
+        msg.data_msg.id = uart_pb2.UartId.UART2
+
+    msg.data_msg.data = data
+    msg.sequence_number = SEQ_NUM
+
+    return msg.SerializeToString()
+
+
+def decode_uart_msg(data: bytes):
+    global SEQ_NUM
+    msg = uart_pb2.UartMsg()
+    msg.ParseFromString(data)
+
+    inner_msg = msg.WhichOneof("msg")
+    # print("Inner msg:", inner_msg)
+    if inner_msg == "data_msg":
+        global RX_BUFFER
+        RX_BUFFER += msg.data_msg.data
+    elif inner_msg == "status_msg":
+        global TX_COMPLETE, TX_SPACE
+        if msg.sequence_number == SEQ_NUM:
+            TX_COMPLETE = msg.status_msg.tx_complete
+            TX_SPACE = msg.status_msg.tx_space
+            print("Status: tx_complete {}, tx_space {}, tx_overflow {}, seq_num {}"
+                  .format(msg.status_msg.tx_complete, msg.status_msg.tx_space,
+                          msg.status_msg.tx_overflow, msg.sequence_number))
+            if msg.status_msg.tx_overflow:
+                raise Exception("Tx overflow!")
+    else:
+        print("Rejected status msg! --------------------------------------->")

proto

@@ -3,53 +3,71 @@
 """A simple python script template.
 """
 
-import serial
-import tf.TinyFrame as TF
 import sys
+import enum
 import time
+import string
+import random
+import serial
+import tiny_frame
+import proto_msg as pm
 
-TYPE_DEFAULT = 0x00
-TYPE_UART = 0x01
-TYPE_PERIPHERAL = 0x02
-TYPE_CENTRAL = 0x03
+TX_LOOPS = 100000
+MIN_TX_SIZE = 1
+MAX_DATA_SIZE = 64
+TX_HISTORY = bytearray()
+TX_COUNTER = 0
 
 
-def fallback_listener(tf, msg):
-    print("Fallback listener")
-    print(msg.data)
+def uart_send(tf):
+    global TX_COUNTER, TX_HISTORY
+    if pm.TX_SPACE < MIN_TX_SIZE:
+        return
 
+    tx_data = "This is a test. With little data ({}).".format(TX_COUNTER)
+    size = random.randint(MIN_TX_SIZE, min(MAX_DATA_SIZE, pm.TX_SPACE))
+    tx_data = ''.join(random.choice(string.ascii_letters + string.digits) for _ in range(size))
 
-def uart_listener(tf, msg):
-    print("Uart listener")
-    print(msg.data)
+    if pm.TX_SPACE >= len(tx_data) and TX_COUNTER < TX_LOOPS:
+        print("Data: '{}'".format(tx_data))
+        tx_data = bytes(tx_data, 'ascii')
+        msg = pm.encode_uart_data_msg(pm.UartId.UART1, tx_data)
+        print("Send msg (id={}, len={}, tx={}, seq={}) : {}".format(TX_COUNTER, len(msg), len(tx_data), pm.SEQ_NUM, msg))
+
+        tf.send(tiny_frame.TfMsgType.TYPE_UART.value, msg, 0)
+        pm.TX_SPACE -= len(tx_data)
+        TX_COUNTER += 1
+        TX_HISTORY += tx_data
 
 
 def main(arguments):
+    global TX_HISTORY
     print("TinyFrame test sender")
 
     with serial.Serial('COM4', 115200, timeout=1) as ser:
-        tf = TF.TinyFrame()
-        tf.SOF_BYTE = 0x01
-        tf.ID_BYTES = 1
-        tf.LEN_BYTES = 1
-        tf.TYPE_BYTES = 1
-        tf.CKSUM_TYPE = 'xor'
-        tf.write = ser.write
-
-        # Add listeners
-        tf.add_fallback_listener(fallback_listener)
-        tf.add_type_listener(TYPE_UART, uart_listener)
-
-        # Send a frame
-        tf.send(TYPE_UART, b"This are more than 64 bytes. This verifies that the frame is only processed when its complete.", 0)
-        time.sleep(1)
+        tf = tiny_frame.tf_init(ser.write)
 
         while True:
+            uart_send(tf)
             if ser.in_waiting > 0:
                 # Read the incoming data
-                data = ser.read(ser.in_waiting)
+                rx_data = ser.read(ser.in_waiting)
                 # print(data)
-                tf.accept(data)
+                tf.accept(rx_data)
+
+            rx_len = len(pm.RX_BUFFER)
+            if rx_len > 0:
+                if pm.RX_BUFFER[:rx_len] == TX_HISTORY[:rx_len]:
+                    print("Loop ok for {} bytes: '{}'".format(rx_len, pm.RX_BUFFER.decode("ascii")))
+                    TX_HISTORY = TX_HISTORY[rx_len:]
+                    pm.RX_BUFFER.clear()
+                    if len(TX_HISTORY) == 0 and TX_COUNTER >= TX_LOOPS:
+                        break
+                else:
+                    print("Data missmatch detected!")
+                    print(TX_HISTORY[:rx_len])
+                    print(pm.RX_BUFFER)
+                    raise Exception("Data missmatch!")
 
 
 if __name__ == '__main__':

proto_msg.py

@@ -0,0 +1,101 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+env/
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+.hypothesis/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# pyenv
+.python-version
+
+# celery beat schedule file
+celerybeat-schedule
+
+# SageMath parsed files
+*.sage.py
+
+# dotenv
+.env
+
+# virtualenv
+.venv
+venv/
+ENV/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/