CHIRP

#!/usr/bin/env python

# This script requires Python 3.

# $ pip install pyserial

# $ python gm30ctl.py

import io

import enum

import struct

import argparse

import typing as t

from pathlib import Path

from datetime import timedelta

from dataclasses import dataclass

import serial

##

# High level radio interface.

##

class RadioBootscreenMode(enum.Enum):

    LOGO = enum.auto()

    MESSAGE = enum.auto()

    VOLTAGE = enum.auto()

@dataclass

class Radio:

    bootscreen_mode: RadioBootscreenMode

    bootscreen_line1: str

    bootscreen_line2: str

class Protocol:

    """

    Radioddity GM-30 serial control and programming protocol.

    The protocol loosely follows a request/response flow with one-way or

    bi-directional acknowledgements. It relies on hardware flow control.

    Requests:

    - Command Requests

    - PSEARCH Request

    - PASSSTA Request (?)

    - SYSINFO Request (?)

    PSEARCH Request:

    - Send Bytes: 'PSEARCH' as ASCII

    - Read Ack:

      - 1x Byte: 0x06

    - Read Variable Length Response: Firmware variant name as ASCII

      - Known Variant: P13GMRS (US region GMRS firmware)

    Command Requests:

    - 1x Bytes: Request Type

    - 0x or 4x Bytes: Parameters

    Command Ack Request:

    - Type: 0x06

    - Parameters: N/A

    - Response: 0x06

    Command Read Request:

    - Radio must be in programming mode first.

    - Type: 0x52

    - Parameters:

      - 2x Bytes: Little-Endian Address

      - 1x Byte: 0x00 (?)

      - 1x Byte: Read Size

    - Response:

      - 1x Byte: 0x57

      - 2x Bytes: Little-Endian Address

      - 1x Byte: 0x00 (?)

      - 1x Byte: Read Size (Not Including 5x Byte Header)

      - Read Size x Bytes: Data

    - Requires Ack Request After

    Command Write Request:

    - Radio must be in programming mode first.

    - Type: 0x57

    - Parameters:

      - 2x Bytes: Little-Endian Address

      - 1x Byte: 0x00 (?)

      - 1x Byte: Write Size (Not Including 4x Byte Header)

      - Write Size x Bytes: Data

    - Response:

      - 1x Byte: 0x06

    """

    @staticmethod

    def open_port(device_path: Path) -> serial.Serial:

        return serial.Serial(

            port=str(device_path),

            baudrate=57600,

            bytesize=serial.EIGHTBITS,

            parity=serial.PARITY_NONE,

            stopbits=serial.STOPBITS_ONE,

            xonxoff=False,

            rtscts=True,

            dsrdtr=True,

            timeout=None)

    def __init__(self, port: serial.Serial, timeout = timedelta(seconds=1)):

        self.port = port

        self.timeout = timeout

    def _reset(self):

        # XXX: log warning if buffers are not empty

        # XXX: not sure if this is actually necessary or useful

        self.port.reset_input_buffer()

        self.port.reset_output_buffer()

    def _fixed_write(self, data: bytes) -> int:

        self.port.write(data)

        self.port.flush()

    def _variable_read(self, max_count: int) -> bytes:

        self.port.timeout = self.timeout.total_seconds()

        return self.port.read(max_count)

    def _fixed_read(self, expected_count: int) -> bytes:

        response = self._variable_read(expected_count)

        if not response:

            self._reset()

            raise RuntimeError("No response received")

        if len(response) != expected_count:

            self._reset()

            raise RuntimeError("Unexpected read size")

        return response

    def send_ack(self):

        self._fixed_write(bytes([0x06]))

    def receive_ack(self):

        response = self._fixed_read(1)

        if response != bytes([0x06]):

            raise RuntimeError("Failed to receive ACK")

    def read_memory(self, address: int, size: int) -> bytes:

        # Request: 0x52 ADDRx2 0x00 SIZEx1

        request = bytearray([0x52, 0x00, 0x00, 0x00, 0x00])

        struct.pack_into('<HxB', request, 1, address, size)

        self._fixed_write(request)

        # Response: 0x57 ADDRx2 0x00 SIZEx1 [DATAx1 .. DATAx1]

        response = self._fixed_read(5 + size)

        if response[0] != 0x57 or response[1:5] != request[1:5]:

            raise RuntimeError("Read memory response invalid header")

        # Sync

        self.send_ack()

        self.receive_ack()

        return response[5:]

    def write_memory(self, address: int, data: bytes):

        # Request: 0x57 ADDRx2 0x00 SIZEx1 [DATAx1 .. DATAx2]

        request = bytearray([0x57, 0x00, 0x00, 0x00, 0x00])

        struct.pack_into('<HxB', request, 1, address, len(bytes))

        self._fixed_write(request + data)

        # Sync

        self.receive_ack(port)

    def read_memory_bytes(self, addresses: t.List[int]) -> bytes:

        data = bytearray()

        for address in addresses:

            data += self.read_memory(address, 0x01)

        return bytes(data)

    def read_memory_range(self, address: int, size: int, chunk_size: int = 0x40) -> bytes:

        data = bytearray()

        read_address = address

        read_bytes_remaining = size

        while read_bytes_remaining > 0:

            read_size = min(chunk_size, read_bytes_remaining)

            data += self.read_memory(read_address, read_size)

            read_bytes_remaining -= read_size

            read_address += read_size

        return bytes(data)

    def write_memory_range(self, address: int, data: bytes, chunk_size: int = 0x40):

        write_counter = 0

        while write_counter < len(data):

            write_size = min(chunk_size, len(data) - write_counter)

            write_data = data[write_counter:write_counter + write_size]

            write_memory(port, address + write_counter, write_data)

            write_counter += write_size

    def query_firmware_variant(self) -> str:

        # Request

        self._fixed_write(b'PSEARCH')

        self.receive_ack()

        # Response: firmware variant name

        # Known variants: P13GMRS

        response = self._variable_read(16)

        if not response:

            self._reset()

            raise RuntimeError("Firmware variant query did not receive a response")

        return response.decode()

##

# Protocol functionality not fully understood yet.

##

def unknown_passsta(protocol):

    protocol._fixed_write(b'PASSSTA')

    response = protocol._fixed_read(3)

    assert response[:1].decode() == 'P'

    assert response[1] == 0x00

    assert response[2] == 0x00

def unknown_sysinfo(protocol):

    protocol._fixed_write(b'SYSINFO')

    protocol.receive_ack()

def common_init(

    protocol,

    query_unknown_passsta = True,

    query_unknown_sysinfo = True

):

    # querying for use later on when entering programming mode

    # XXX not required to enter read/write mode

    fw_variant = protocol.query_firmware_variant()

    assert fw_variant == 'P13GMRS'

    # XXX checking whether a password is set?

    # XXX not required to enter programming mode

    if query_unknown_passsta:

        unknown_passsta(protocol)

    # XXX required to enter programming mode

    unknown_sysinfo(protocol)

    # XXX some kind of timestamp query or checksum calculation?

    # XXX seems to change based on the contents of radio memory

    # XXX does not seem to change over time on it's own

    # XXX requires sysinfo command to be sent first

    # XXX not required to enter programming mode

    if query_unknown_sysinfo:

        # XXX

        protocol._fixed_write(bytes([0x56, 0x00, 0x00, 0x0A, 0x0D]))

        response = protocol._fixed_read(3)

        assert response == bytes([0x56, 0x0D, 0x0A])

        response = protocol._fixed_read(10)

        print(response.hex(' '))

        protocol.send_ack()

        protocol.receive_ack()

        # XXX

        protocol._fixed_write(bytes([0x56, 0x00, 0x10, 0x0A, 0x0D]))

        response = protocol._fixed_read(3)

        assert response == bytes([0x56, 0x0D, 0x0A])

        response = protocol._fixed_read(10)

        print(response.hex(' '))

        protocol.send_ack()

        protocol.receive_ack()

        # XXX

        protocol._fixed_write(bytes([0x56, 0x00, 0x20, 0x0A, 0x0D]))

        response = protocol._fixed_read(3)

        assert response == bytes([0x56, 0x0D, 0x0A])

        response = protocol._fixed_read(10)

        print(response.hex(' '))

        protocol.send_ack()

        protocol.receive_ack()

        # XXX seems to be different variant then three queries above?

        protocol._fixed_write(bytes([0x56, 0x00, 0x00, 0x00, 0x0A]))

        response = protocol._fixed_read(3)

        print(response.hex(' '))

        response = protocol._fixed_read(3)

        print(response.hex(' '))

        response = protocol._fixed_read(5)

        print(response.hex(' '))

        protocol.send_ack()

        protocol.receive_ack()

    # XXX: this seems to set a timeout where if no further commands are

    # received within a certain window the radio will reset

    # required to enter programming mode

    protocol._fixed_write(bytes([0xFF, 0xFF, 0xFF, 0xFF, 0x0C]))

    protocol._fixed_write(fw_variant.encode()) # XXX: b'P13GMRS'

    protocol.receive_ack()

    # XXX required to enter programming mode

    protocol._fixed_write(bytes([0x02]))

    response = protocol._fixed_read(8)

    assert response == bytes([0xFF] * 8)

    protocol.send_ack()

    protocol.receive_ack()

##

# Console Interface

##

def read(device_path: Path):

    with Protocol.open_port(device_path) as port:

        protocol = Protocol(port)

        common_init(protocol)

        # XXX no idea what this data is or why it's read

        data = protocol.read_memory_bytes([

            0x1FFF,

            0x2FFF,

            0x3FFF,

            0x4FFF,

            0x5FFF,

            0x6FFF,

            0x7FFF,

            0x8FFF,

            0x9FFF,

            0xAFFF,

            0xBFFF,

            0xCFFF,

            0xDFFF,

            0xEFFF,

            0xFFFF])

        with open('region_1FFF.bin', 'wb') as output_file:

            output_file.write(data)

        # XXX data file 0x2000, ???

        data = protocol.read_memory_range(0x1000, 0xFC0)

        with open('region_1000.bin', 'wb') as output_file:

            output_file.write(data)

        # XXX data file 0x3000 with some differences, ???

        data = protocol.read_memory_range(0xF000, 0xFC0)

        with open('region_F000.bin', 'wb') as output_file:

            output_file.write(data)

        # XXX data file 0x4000, channel names?

        data = protocol.read_memory_range(0x3000, 0xFC0)

        with open('region_3000.bin', 'wb') as output_file:

            output_file.write(data)

        # XXX data file 0x5000, general settings?

        data = protocol.read_memory_range(0xB000, 0xFC0)

        with open('region_B000.bin', 'wb') as output_file:

            output_file.write(data)

        # XXX data file 0x6000, ???

        data = protocol.read_memory_range(0xD000, 0xFC0)

        with open('region_D000.bin', 'wb') as output_file:

            output_file.write(data)

def write(device_path: Path):

    with Protocol.open_port(device_path) as port:

        protocol = Protocol(port)

        common_init(protocol)

        # XXX no idea what this data is or why it's read

        protocol.read_memory_bytes([

            0x1FFF,

            0x2FFF,

            0x3FFF,

            0x4FFF,

            0x5FFF,

            0x6FFF,

            0x7FFF,

            0x8FFF,

            0x9FFF,

            0xAFFF,

            0xBFFF,

            0xCFFF,

            0xDFFF,

            0xEFFF,

            0xFFFF])

        # XXX data file 0x3000, ???

        with open('region_7000.bin', 'rb') as input_file:

            data = input_file.read()

        assert len(data) == 0xFC0

        protocol.write_memory_range(0x7000, data)

        # XXX data file 0x4000, channel names?

        with open('region_3000.bin', 'rb') as input_file:

            data = input_file.read()

        assert len(data) == 0xFC0

        protocol.write_memory_range(0x3000, data)

        # XXX data file offset 0x5000, general settings?

        with open('region_4000.bin', 'rb') as input_file:

            data = input_file.read()

        assert len(data) == 0xFC0

        protocol.write_memory_range(0x4000, data)

        # XXX data file offset 0x6000, ???

        with open('region_D000.bin', 'rb') as input_file:

            data = input_file.read()

        assert len(data) == 0xFC0

        protocol.write_memory_range(0xD000, data)

def main():

    parser = argparse.ArgumentParser()

    parser.add_argument(

        '-d', '--device',

        type=Path,

        default='/dev/ttyUSB0')

    subparsers = parser.add_subparsers()

    parser_read = subparsers.add_parser('read', help="read radio memory")

    parser_read.set_defaults(run=read)

    parser_write = subparsers.add_parser('write', help="write radio memory")

    parser_write.set_defaults(run=write)

    args = parser.parse_args()

    args.run(device_path=args.device)

if __name__ == '__main__':

    main()