1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86
|
from __future__ import print_function
import argparse
import logging
import odil
from .print_ import find_max_name_length, print_data_set
def add_subparser(subparsers):
parser = subparsers.add_parser(
"find", help="DICOM query (C-FIND)",
formatter_class=argparse.ArgumentDefaultsHelpFormatter)
parser.add_argument("host", help="Remote host address")
parser.add_argument("port", type=int, help="Remote host port")
parser.add_argument(
"calling_ae_title", help="AE title of the calling application")
parser.add_argument(
"called_ae_title", help="AE title of the called application")
parser.add_argument(
"level", choices=["patient", "study"], help="Root object of the query")
parser.add_argument("keys", nargs="+", help="Query keys")
parser.add_argument(
"--decode-uids", "-u", action="store_true",
help="Print human-friendly name of known UIDs")
parser.set_defaults(function=find)
return parser
def find(host, port, calling_ae_title, called_ae_title, level, keys, decode_uids):
query = odil.DataSet()
for key in keys:
if "=" in key:
key, value = key.split("=", 1)
value = value.split("\\")
else:
value = None
tag = getattr(odil.registry, key)
if value is not None:
vr = odil.registry.public_dictionary[tag].vr
if vr in ["DS", "FL", "FD"]:
value = [float(x) for x in value]
elif vr in ["IS", "SL", "SS", "UL", "US"]:
value = [int(x) for x in value]
query.add(tag, value)
else:
query.add(tag)
sop_class = getattr(
odil.registry,
"{}RootQueryRetrieveInformationModelFind".format(level.capitalize()))
find_pc = odil.AssociationParameters.PresentationContext(
1, sop_class,
[
odil.registry.ImplicitVRLittleEndian,
odil.registry.ExplicitVRLittleEndian
], odil.AssociationParameters.PresentationContext.Role.SCU
)
association = odil.Association()
association.set_peer_host(host)
association.set_peer_port(port)
association.update_parameters()\
.set_calling_ae_title(calling_ae_title)\
.set_called_ae_title(called_ae_title) \
.set_presentation_contexts([find_pc])
association.associate()
logging.info("Association established")
find = odil.FindSCU(association)
find.set_affected_sop_class(sop_class)
data_sets = find.find(query)
print("{} answer{}".format(len(data_sets), "s" if len(data_sets)>1 else ""))
max_length = 0
for data_set in data_sets:
max_length = max(max_length, find_max_name_length(data_set))
for data_set in data_sets:
print_data_set(data_set, decode_uids, "", max_length, odil.Value.Strings())
print()
association.release()
logging.info("Association released")
|