grmon_ahbuart.py

#!/usr/bin/env python3

import os, termios
import logging
from colorlog import ColoredFormatter
import signal
import sys
import enum

### Настройка логгирования
READ  = 21
WRITE = 22
MARK  = 23
logging.addLevelName(READ,  'READ')
logging.addLevelName(WRITE, 'WRITE')
logging.addLevelName(MARK,  'MARK')

mark_record = logging.LogRecord("", MARK, "/", 0, 80*"-", None, None)

LOGFORMAT = "%(asctime)s | %(log_color)s%(levelname)-8s%(reset)s | - %(message)s"

logger = logging.getLogger("mypty")
logger.setLevel(logging.DEBUG)

ch = logging.StreamHandler()
ch.setLevel(logging.INFO)

formatter = ColoredFormatter(
    LOGFORMAT,
    log_colors={'READ': 'yellow', 'WRITE': 'cyan', 'MARK': 'red'})
ch.setFormatter(formatter)
logger.addHandler(ch)

def handler(signum, frame):
    ch.emit(mark_record)
    
signal.signal(signal.SIGALRM, handler)

class TimeoutHandler(logging.Handler):
    def emit(self, record):
        signal.alarm(1)

th = TimeoutHandler()
logger.addHandler(th)

### Открываем PTY, переключаем их в "сырой" режим и создаем симлинки
mst_sim,   slv_sim   = os.openpty()
mst_grmon, slv_grmon = os.openpty()

tp = termios.tcgetattr(mst_sim)
tp[0] &= ~(
    termios.IGNBRK | termios.BRKINT |
    termios.PARMRK | termios.ISTRIP |
    termios.INLCR  | termios.IGNCR  |
    termios.ICRNL  | termios.IXON)
tp[1] &= ~termios.OPOST
tp[2] &= ~(termios.CSIZE | termios.PARENB)
tp[2] |= termios.CS8
tp[3] &= ~(termios.ECHO   | termios.ECHONL |
           termios.ICANON | termios.ISIG   |
           termios.IEXTEN)
termios.tcsetattr(mst_sim,   termios.TCSANOW, tp)
termios.tcsetattr(mst_grmon, termios.TCSANOW, tp)

if os.path.exists("slave_sim"):
    os.unlink("slave_sim")
if os.path.exists("slave_grmon"):
    os.unlink("slave_grmon")

os.symlink(os.ttyname(slv_sim),   "slave_sim")
os.symlink(os.ttyname(slv_grmon), "slave_grmon")


### Конечный автомат для анализа протокола AHBUART
class FSM(enum.Enum):
    IDLE = 0
    RA0 = 1
    RA1 = 2
    RA2 = 3
    RA3 = 4
    RD0 = 5
    RD1 = 6
    RD2 = 7
    RD3 = 8
    WA0 = 9
    WA1 = 10
    WA2 = 11
    WA3 = 12
    WD0 = 13
    WD1 = 14
    WD2 = 15
    WD3 = 16

state = FSM.IDLE

while True:
    if ((state == FSM.RD0) or (state == FSM.RD1) or
        (state == FSM.RD2) or (state == FSM.RD3)):
        dd = os.read(mst_sim, 1)
        os.write(mst_grmon, dd)
    else:
        dd = os.read(mst_grmon, 1)
        os.write(mst_sim, dd)

    assert (len(dd) == 1)
    d = dd[0]
    logger.debug(f"{d:02x} {state}")

    if state == FSM.IDLE:
        if (d & 0xc0) == 0xc0:
            state = FSM.WA0
            xlen = (d & 0x3f) + 1
            address = 0x00000000
        elif (d & 0xc0) == 0x80:
            state = FSM.RA0
            xlen = (d & 0x3f) + 1
            address = 0x00000000
    elif state == FSM.RA0:
        address |= (d << 24)
        state = FSM.RA1
    elif state == FSM.RA1:
        address |= (d << 16)
        state = FSM.RA2
    elif state == FSM.RA2:
        address |= (d << 8)
        state = FSM.RA3
    elif state == FSM.RA3:
        address |= d
        datav = []
        data = 0x00000000
        state = FSM.RD0
    elif state == FSM.RD0:
        data |= (d << 24)
        state = FSM.RD1
    elif state == FSM.RD1:
        data |= (d << 16)
        state = FSM.RD2
    elif state == FSM.RD2:
        data |= (d << 8)
        state = FSM.RD3
    elif state == FSM.RD3:
        data |= d
        datav.append(data)
        data = 0
        xlen -= 1
        if (xlen == 0):
            s = ", ".join(["0x{:08x}".format(x) for x in datav])
            logger.log(READ, f"[{address:08x}] = [{s}]")
            state = FSM.IDLE
        else:
            state = FSM.RD0
    elif state == FSM.WA0:
        address |= (d << 24)
        state = FSM.WA1
    elif state == FSM.WA1:
        address |= (d << 16)
        state = FSM.WA2
    elif state == FSM.WA2:
        address |= (d << 8)
        state = FSM.WA3
    elif state == FSM.WA3:
        address |= d
        state = FSM.WD0
        n = xlen
        datav = []
        data = 0x00000000
    elif state == FSM.WD0:
        data |= (d << 24)
        state = FSM.WD1
    elif state == FSM.WD1:
        data |= (d << 16)
        state = FSM.WD2
    elif state == FSM.WD2:
        data |= (d << 8)
        state = FSM.WD3
    elif state == FSM.WD3:
        data |= d
        datav.append(data)
        xlen -= 1
        if (xlen == 0):
            state = FSM.IDLE
            s = ", ".join(["0x{:08x}".format(x) for x in datav])
            logger.log(WRITE, f"[{address:08x}] = [{s}]")
        else:
            data = 0
            state = FSM.WD0