A script generated as part of SUPPORT-90 to convert list-mode data files to JSONL.

support-90.py

""" SPDX-License-Identifier: Apache-2.0 """
import argparse
import json
import logging
import os
import struct
import sys

import dolosse.constants.data as dcd
from dolosse.hardware.xia.pixie16.list_mode_data_mask import ListModeDataMask
import dolosse.hardware.xia.pixie16.list_mode_data_decoder as lmd

"""
Copyright 2022 XIA LLC, All rights reserved.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""

""" support-90.py

A small list-mode data converter generated as part of SUPPORT-90. The converter is memory efficient
so that it's not necessary to load all of the data into memory during the conversion process.

The process works under the following assumptions:
   
   1. The Header Length and Event Length are equal to 4.
   2. The data file only contains list-mode data records with no metadata.  
   
NOTE: This script is not recommended for regular production use. It's provided as an example only.
"""

logging.basicConfig(stream=sys.stdout, datefmt="%Y-%m-%dT%H:%M:%S.000Z", level=logging.INFO,
                    format='%(asctime)s - %(levelname)s - %(message)s')


def convert_to_jsonl(args, filename):
    """
    Performs the conversion from encoded list-mode data to JSONL.
    :param args: The arguments from the command line.
    :param filename: The output file name based on the input file name.
    :return: None
    """
    mask = ListModeDataMask(args.freq, args.rev)
    out = open(filename, "w")
    logging.info(f"Processing {args.file}")

    if os.path.exists(args.file):
        file_size = os.path.getsize(args.file)
        if file_size:
            logging.info(
                f"Sending {args.file} with size of {file_size} bytes for processing.")
            try:
                logging.info(f"Started to decode {args.file}")
                with open(args.file, 'rb') as raw:
                    while raw.tell() != file_size:
                        decoded_data = lmd.decode_word_zero(
                            struct.unpack("I", raw.read(dcd.WORD))[0], mask)

                        if decoded_data['header_length'] != 4 and decoded_data[
                            'event_length'] != 4:
                            raise RuntimeError(
                                "This script only supports 4 word list-mode data.")

                        decoded_data.update(
                            {'event_time_low': struct.unpack("I", raw.read(dcd.WORD))[0]})
                        decoded_data.update(
                            lmd.decode_word_two(struct.unpack("I", raw.read(dcd.WORD))[0],
                                                mask))
                        decoded_data.update(
                            lmd.decode_word_three(struct.unpack("I", raw.read(dcd.WORD))[0],
                                                  mask))
                        out.write(json.dumps(decoded_data))
                        out.write("\n")
                logging.info("Finished decoding file.")
            except Exception as ex:
                logging.error(ex)
        else:
            logging.warning(f'{args.file} is empty!!')
    else:
        logging.warning(f'{args.file} does not exist!!')
    out.close()


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Converts list-mode data files to JSONL')
    parser.add_argument('-f', '--file', type=str, dest="file",
                        required=True, help="The list of filenames that we want to convert.")
    parser.add_argument('--freq', type=int, dest='freq', required=True,
                        help="The sampling frequency used to collect list-mode data. Ex. 250")
    parser.add_argument('--rev', type=int, dest='rev', required=True,
                        help="The firmware used to collect list-mode data. Ex. 30474")
    args = parser.parse_args()
    try:
        base_filename, ext = os.path.splitext(args.file)
        filename = f'{base_filename}.jsonl'
        if ext == '.bin' or ext == '.lmd':
            convert_to_jsonl(args, filename)
        else:
            raise RuntimeError("This process only works on .bin or .lmd data files.")
    except KeyboardInterrupt:
        logging.info("Received keyboard interrupt. Stopping execution.")
    except Exception as e:
        logging.error(repr(e))
    finally:
        logging.info("See you, space cowboy.")