HDQ decoder for sigrok

__init__.py

# Copyright (c) 2024 JJTech <jjtech@jjtech.dev>
#
# SPDX-License-Identifier: GPL-2.0-or-later

"""
HDQ (High-speed Data Queue) is a simple protocol used for Texas Instruments
'Gas Gauge' battery fuel gauges. It is used to read and write data to the fuel
gauge's memory.
Similarly to the '1-wire' protocol, it is a single-wire protocol that uses a
single data line to communicate with the fuel gauge.

Protocol documentation: https://www.ti.com/lit/pdf/slua408
"""
import sigrokdecode as srd # type: ignore

from .pd import Decoder

pd.py

# Copyright (c) 2024 JJTech <jjtech@jjtech.dev>
# Copyright (c) 2024 Nicolás Alvarez <nicolas.alvarez@gmail.com>
#
# SPDX-License-Identifier: GPL-2.0-or-later

import sigrokdecode as srd # type: ignore

class Decoder(srd.Decoder):
    api_version = 3
    id = 'hdq'
    name = 'HDQ'
    longname = 'HDQ (High-speed Data Queue)'
    desc = 'Texas Instruments Gas Gauge battery fuel gauge protocol'
    license = 'mit'
    inputs = ['logic']
    outputs = ['hdq']
    tags = ['Embedded/industrial']
    channels = (
        {'id': 'data', 'name': 'Data', 'desc': 'Data line'},
    )
    options = ()
    annotations = (
        ('bit', 'Bit'),
        ('addr', 'Address'),
        ('data', 'Data'),
        ('rw', 'R/W Bit'),
        ('write', 'Write'),
        ('read', 'Read'),
    )
    annotation_rows = (
        ('bits', 'Bits', (0,)),
        ('values', 'Values', (1,2,3)),
        ('txs', 'TXs', (4,5)),
    )

    def __init__(self):
        print("INIT")
        self.reset()

    def reset(self):
        print("RESET")
        self.sample_rate = None

        self.previous_sample_num = 0

    def metadata(self, key, value):
        print("METADATA")
        if key == srd.SRD_CONF_SAMPLERATE:
            self.sample_rate = value

    def start(self):
        print("START")
        self.out_ann = self.register(srd.OUTPUT_ANN)

    def decode_txs(self, bits):
        while True:
            tx = self.decode_tx(bits)
            if tx:
                yield tx

    def decode_tx(self, bits):
        # Parse address
        addr = 0
        field_start = 0
        for bitcount in range(7):
            bit, start, end = next(bits)
            if bit == "reset": return

            if bitcount == 0:
                field_start = start

            addr |= (bit << bitcount)

        tx_start = field_start
        self.put(field_start, end, self.out_ann, [1, [
            f'Address: #0x{addr:02x}',
            f'Addr: #0x{addr:02x}',
            f'#0x{addr:02x}',
            f'0x{addr:x}'
        ]])

        # Parse R/W bit
        bit, start, end = next(bits)
        if bit == "reset": return

        if bit:
            label = ['Write', 'W']
        else:
            label = ['Read', 'R']
        dir = bit
        self.put(start, end, self.out_ann, [3, label])

        # Parse data
        data = 0
        field_start = 0
        for bitcount in range(8):
            bit, start, end = next(bits)
            if bit == "reset": return

            if bitcount == 0:
                field_start = start

            data |= (bit << bitcount)

        tx_end = end
        self.put(field_start, end, self.out_ann, [2, [
            f'Data {"to write" if dir else "read"}: 0x{data:x}',
            f'Data {"W" if dir else "R"}: 0x{data:x}',
            f'Data: 0x{data:x}',
            f'D: 0x{data:x}',
            f'0x{data:x}',
        ]])

        return (dir, addr, data, tx_start, tx_end)

    def decode_edges(self):
        while True:
            self.wait({0: 'f'})
            fall_sample = self.samplenum
            self.wait({0: 'r'})
            raise_sample = self.samplenum
            yield (fall_sample, raise_sample)

    def decode_bits(self, edges):
        """
        Measures the timings between edges and decodes them into bits.
        Yields tuples with ("reset"|"0"|"1", start, end).
        """
        MICRO = 1_000_000
        MIN_BREAK   = self.sample_rate * 190 // MICRO
        MIN_ZERO    = self.sample_rate * 86 // MICRO
        MAX_ZERO    = self.sample_rate * 145 // MICRO
        MIN_ONE     = self.sample_rate * 17 // MICRO
        MAX_ONE     = self.sample_rate * 50 // MICRO

        for fall_sample, raise_sample in edges:
            t = raise_sample - fall_sample

            ann_start = fall_sample
            ann_end = raise_sample

            if t > MIN_BREAK:
                self.put(ann_start, ann_end, self.out_ann, [0, ['Reset']])
                yield ("reset", ann_start, ann_end)
            elif MIN_ZERO <= t <= MAX_ZERO:
                self.put(ann_start, ann_end, self.out_ann, [0, ['0']])
                yield (0, ann_start, ann_end)
            elif MIN_ONE <= t <= MAX_ONE:
                self.put(ann_start, ann_end, self.out_ann, [0, ['1']])
                yield (1, ann_start, ann_end)

    def decode(self):
        print("DECODE")
        assert self.sample_rate is not None

        for tx in self.decode_txs(self.decode_bits(self.decode_edges())):
            dir, addr, data, tx_start, tx_end = tx
            if dir:
                self.put(tx_start, tx_end, self.out_ann, [4, [
                    f"Write addr 0x{addr:02x} data 0x{data:02x}",
                    f"Write 0x{addr:02x} data 0x{data:02x}",
                    f"W 0x{addr:02x} data 0x{data:02x}",
                ]])
            else:
                self.put(tx_start, tx_end, self.out_ann, [5, [
                    f"Read addr 0x{addr:02x} data 0x{data:02x}",
                    f"Read 0x{addr:02x} data 0x{data:02x}",
                    f"R 0x{addr:02x} data 0x{data:02x}",
                ]])