reamp2csv.py

"""
*********************************************************************************
***                                                                           ***
***  Reamp exporter                                                           ***
***                                                                           ***
***  (c) DL2SBA Dietmar Krause 2026                                           ***
***                                                                           ***
*********************************************************************************
"""

import argparse
import struct
import csv
import locale
import os
import sys
import logging
from datetime import timezone
from datetime import datetime
from pathlib import Path

PARM_TIME_RELATIVE = "relative"
PARM_TIME_TIMESTAMP = "timestamp"
PARM_TIME_MICROS = "unix"
DOUBLE_BYTES = 8

LOGGING_CHUNK = 10000


def process_reamp_data(
    parm_input_path,
    parm_output_path,
    parm_delimiter,
    parm_encoding,
    parm_time,
):
    """
    *********************************************************************************
    ***                                                                           ***
    ***  Exporter                                                                 ***
    ***                                                                           ***
    *********************************************************************************
    """
    with open(parm_input_path, "rb") as in_file:
        #
        logging.info(
            f"inputfile size.....{os.fstat(in_file.fileno()).st_size:n} bytes",
        )

        #   process header section
        binary_content = in_file.read(0x200)

        #   extract header fields
        file_version = struct.unpack("<h", binary_content[0:2])[0]
        file_header_size = struct.unpack("<h", binary_content[2:4])[0]
        file_channel_count = struct.unpack("B", binary_content[8:9])[0]
        file_channel_map = struct.unpack("B", binary_content[9:10])[0]
        file_sample_time = (
            struct.unpack("<h", binary_content[10:12])[0] / 10
        )  # in ms ticks
        file_type = struct.unpack("B", binary_content[8:9])[0]
        file_start_ts = struct.unpack("<Q", binary_content[0x1F8:0x200])[0] / 1000

        logging.info("file_version.......%s", file_version)
        logging.info("file_header_size...%s", file_header_size)
        logging.info("file_channel_count.%d", file_channel_count)
        logging.info("file_channel_map...%s", file_channel_map)
        logging.info("file_sample_time...%s ms", file_sample_time)
        logging.info("file_type......... %d", file_type)
        logging.info("file_start_ts......%s s", file_start_ts)
        logging.info(
            "                   %s",
            datetime.fromtimestamp(file_start_ts, tz=timezone.utc),
        )

        #   check file version
        if file_version != 4:
            raise ValueError("Fileversion not supported")

        #   process data records
        #   create output file
        out_file = open(parm_output_path, "w", newline="", encoding=parm_encoding)
        writer = csv.writer(out_file, delimiter=parm_delimiter)

        # write csv header
        writer.writerow(["time", "channel-0", "channel-1", "channel-2"])

        #   detector for end of file
        #   increments are in ms
        previous_increment = -1
        increment = 0
        bytes_per_channel = (file_channel_count + 1) * DOUBLE_BYTES
        timestamp_from_increment = ""
        num_samples = 0
        #
        #   read the data lines
        while True:
            value_array = []
            #   read all data of a sample
            sample_raw = in_file.read(bytes_per_channel)
            #   read DOUBLE_BYTES byte double value
            double_increment = struct.unpack("<d", sample_raw[0:DOUBLE_BYTES])[0]
            #   convert to ms
            increment = double_increment
            # print(increment)
            if increment == previous_increment:
                break

            previous_increment = increment
            timestamp_from_increment = file_start_ts + increment

            # interpret time column depending on parm_time
            if parm_time == PARM_TIME_RELATIVE:
                value_array.append(locale.format_string("%f", increment))
            elif parm_time == PARM_TIME_TIMESTAMP:
                #   timestamp of current sample in ms
                ziel_datum = datetime.fromtimestamp(
                    timestamp_from_increment, tz=timezone.utc
                )
                #   force that always microseconds are printed
                zeit_string = ziel_datum.isoformat(timespec="microseconds")
                value_array.append(zeit_string)
            else:
                #   timestamp of current sample in s
                value_array.append(timestamp_from_increment)

            # process each channel in a row
            for chan_no in range(file_channel_count):
                start_idx = chan_no * DOUBLE_BYTES + DOUBLE_BYTES
                stop_idx = start_idx + DOUBLE_BYTES
                chan_value = struct.unpack("<d", sample_raw[start_idx:stop_idx])[0]
                chan_value_locale = locale.format_string("%f", chan_value)
                value_array.append(chan_value_locale)

            writer.writerow(value_array)

            # one more sample written
            num_samples += 1

            # cyclic status update
            if num_samples % LOGGING_CHUNK == 0:
                logging.info("samples written....%d", num_samples)

        logging.info("samples written....%d", num_samples)
        logging.info("fileLastTS.........%s", timestamp_from_increment)
        logging.info(
            f"outputfile size....{os.fstat(out_file.fileno()).st_size:n} bytes",
        )
        logging.info(
            "                   %s",
            datetime.fromtimestamp(timestamp_from_increment, tz=timezone.utc),
        )
        logging.info("... bye ...")


def main():
    """
    *********************************************************************************
    ***                                                                           ***
    ***  Main                                                                     ***
    ***                                                                           ***
    *********************************************************************************
    """
    # default level is warning, so that the INFO messages are suppressed by default
    logging.basicConfig(
        level=logging.WARNING, format="%(asctime)s - %(levelname)s - %(message)s"
    )

    # init Parser
    parser = argparse.ArgumentParser(
        description="Converts Reamp-Datafile into CSV-Format. "
        "In Excel use the following cell format for English version `DD.MM.YYYY hh:mm:ss,000`. "
        "For German version `TT.MM.JJJJ hh:mm:ss,000`",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )

    # mandatory parameter input filename
    parser.add_argument("input_file", help="filename of *.reamp datafile")

    # optional delimeter
    parser.add_argument(
        "-d",
        "--delimiter",
        choices=[",", ";"],
        default=";",
        help="Value delimiter in CSV-file",
    )
    # optional output encoding
    parser.add_argument(
        "-e",
        "--encoding",
        default="utf-8",
        help="Encoding of output file. For details check https://docs.python.org/3/library/codecs.html",
    )
    # optional locale
    parser.add_argument(
        "-l",
        "--locale",
        default="EN_US",
        help="Locale used for number formatting",
    )
    # optional output file name
    parser.add_argument(
        "-o",
        "--output",
        help="Name of the output file. If ommitted, 'inputfile>.csv' is used",
    )
    # optional mode of time column
    parser.add_argument(
        "-t",
        "--time",
        choices=[PARM_TIME_MICROS, PARM_TIME_RELATIVE, PARM_TIME_TIMESTAMP],
        default=PARM_TIME_RELATIVE,
        help=f"Time column in CSV: "
        f"{PARM_TIME_MICROS} gives µs since 1900-01-01. "
        f"{PARM_TIME_RELATIVE} gives increasing µs since start of measurement. "
        f"{PARM_TIME_TIMESTAMP} gives full timestamp in ISO format with µs resolution",
    )
    # optional verbose
    parser.add_argument(
        "-v", "--verbose", action="store_true", help="Verbose log output"
    )

    # parse args
    args = parser.parse_args()

    input_file_name = Path(args.input_file)
    output_file_name = (
        args.output if args.output else input_file_name.with_suffix(".csv")
    )
    locale_used = args.locale
    delimiter_used = args.delimiter
    encoding_used = args.encoding
    time_used = args.time

    if args.verbose:
        # CRITICAL	50
        # ERROR	40
        # WARNING	30
        # INFO	20
        # DEBUG	10
        logging.getLogger().setLevel(20)

    logging.info("*******************************************************************")
    logging.info("***                                                             ***")
    logging.info("***  Reamp exporter                                             ***")
    logging.info("***                                                             ***")
    logging.info("***    Repository  https://github.com/dl2sba/Reamp2CSV          ***")
    logging.info("***    Homepage    https://dl2sba.com                           ***")
    logging.info("***                                                             ***")
    logging.info("***  (c) DL2SBA Dietmar Krause 2026                             ***")
    logging.info("***                                                             ***")
    logging.info("*******************************************************************")
    logging.info("input filename.....%s", input_file_name)
    logging.info("output filename....%s", output_file_name)
    logging.info("locale used........%s", locale_used)
    logging.info("data delimiter.....%s", delimiter_used)
    logging.info("output encoding....%s", encoding_used)
    logging.info("time column........%s", time_used)

    # pylint: disable=broad-exception-caught
    try:
        locale.setlocale(locale.LC_ALL, locale_used)

        process_reamp_data(
            input_file_name, output_file_name, delimiter_used, encoding_used, time_used
        )

    except Exception as e:
        logging.error("An unexpected error occured: [%s]", e)


# *********************************************************************************
# ***                                                                           ***
# ***  Launcher                                                                 ***
# ***                                                                           ***
# *********************************************************************************
if __name__ == "__main__":
    main()