Source code for dispatch.process_dcmsend_result

import argparse
import json
import sys
from pathlib import Path
from typing import Dict
import daiquiri
from common import config

logger = config.get_logger()


def _parse_header(header) -> Dict:
    result = {}
    for line in header:
        if line.startswith("Communication Peer"):
            result["communication_peer"] = line.split(":")[1].strip()
        elif line.startswith("AE Titles used"):
            result["ae_titles_used"] = line.split(":")[1].strip()
        elif line.startswith("Current Date/Time"):
            result["current_datetime"] = line.split(":", 1)[1].strip()
    return result


def _parse_summary(summary) -> Dict:
    result = {}
    for line in summary:
        if line.startswith("Number of SOP instances"):
            result["sop_instances"] = int(line.split(":")[1])
        elif line.startswith("- sent to the peer"):
            result["sent_to_peer"] = int(line.split(":")[1])
        elif line.startswith("  * with status SUCCESS"):
            result["successfull"] = int(line.split(":")[1])
        elif line.startswith("  * with status SUCCESS"):
            result["error"] = int(line.split(":")[1])
    return result


[docs]def parse(result_file) -> Dict: """Parses the dcmsend result file and returns a python dictionary.""" with result_file.open() as f: content = f.readlines() result = {} for index, element in enumerate(content): if element.startswith("Status Summary"): summary_start = index break else: raise Exception("Failed to parse dcmsend result.") result["summary"] = _parse_summary(content[summary_start:]) # Just take the first 8 lines of the result file, # optimistic guessing length of the header result["header"] = _parse_header(content[:8]) return result
[docs]def create_arg_parser() -> argparse.ArgumentParser: """Creates and returns the ArgumentParser object.""" parser = argparse.ArgumentParser( description="Creates stripped json representation of a dcmsend result text file and prints it out." ) parser.add_argument("resultFile", help="Path to the dcmsend result file.") return parser
if __name__ == "__main__": arg_parser = create_arg_parser() parsed_args = arg_parser.parse_args(sys.argv[1:]) result_file = parsed_args.resultFile result = parse(Path(result_file)) logger.info(json.dumps(result, indent=4, sort_keys=True)) sys.exit(0)