in python/website/research_pacs/website/main.py [0:0]
def search_page():
"""Page Search DICOM Instances"""
def get_unique_series_ids(instances):
"""Retrieve the list of series ID related to the instances"""
result = []
for instance in instances:
series_id = instance[1]
if not series_id in result:
result.append(series_id)
return result
# Retrieve the parameters from the query string
query = request.args.get('query', default='')
display_action = 'display' in request.args
export_action = 'export' in request.args
offset = int(request.args.get('offset', default=0))
# Display the page with no results if the form was not submitted
if display_action is False and export_action is False:
return flask.render_template('search.html')
# Translate the query and the user's instance access permissions into a JSON Path query
try:
jsonpath_query = client.permissions.get_jsonpath_query(query)
logger.debug(f'Search - JSON Path Query: {jsonpath_query}')
except ValueError:
return flask.render_template('search.html', error_message='Your query is invalid.')
g.db = DB(env.pg_host, env.pg_port, env.pg_user, env.pg_pwd, env.pg_db)
db_dicom_json = DBDicomJson(g.db)
# If the "Display" button was pressed
if display_action is True:
def generate_header(header_keywords, dicom_json):
"""Generate the accodion headers"""
lines = []
for keyword in header_keywords.split(','):
value = rpacs_dicom_json.get_top_level_elem_value_from_dict(dicom_json, keyword)
lines.append(f'{keyword}: <strong>{value}</strong>')
return '<br>'.join(lines)
def rewrite_full_path_new_offset(new_offset):
args = request.args.copy()
args['offset'] = new_offset
return f'{request.path}?{url_encode(args)}'
# Retrieve the number of instances and series that match the query, and associated details
# for up to `env.results_per_page` from the offset `offset`
try:
total_instances, total_series = db_dicom_json.count_instances(jsonpath_query)
instances_in_page = db_dicom_json.search_instances_with_series(jsonpath_query, limit=env.results_per_page, offset=offset)
series_ids_in_page = get_unique_series_ids(instances_in_page)
except:
logger.warning(f'Page {request.path} - Query: {query}')
return flask.render_template('search.html', error_message='Failed to query the database. Please check your query and retry.')
# Prepare a dict `results` that is used by the Jinja template to display the instances and
# series for the current page
results = []
for series_id in series_ids_in_page:
series = {
'SeriesId': series_id,
'Instances': []
}
for instance in instances_in_page:
instance_id = instance[0]
instance_series_id = instance[1]
index_in_series = instance[2]
instance_json = instance[3]
if instance_series_id == series_id:
instance_json_keywords = rpacs_dicom_json.add_keywords_to_dicom_json(instance_json)
series['Instances'].append({
'InstanceId': instance_id,
'IndexInSeries': index_in_series,
'InstanceHeader': generate_header(env.instance_header_keywords, instance_json),
'InstanceJSON': json.dumps(instance_json_keywords, indent=4, sort_keys=True)
})
if not 'SeriesHeader' in series:
series['SeriesHeader'] = generate_header(env.series_header_keywords, instance_json)
series['Instances'] = sorted(series['Instances'], key=lambda k: k['IndexInSeries'])
results.append(series)
# Calculate the pagination information
pagination = {
'TotalInstances': total_instances,
'TotalSeries': total_series,
}
if offset > 0:
pagination['PreviousEnabled'] = True
left_new_offset = max(0, offset - env.results_per_page)
pagination['PreviousLink'] = rewrite_full_path_new_offset(left_new_offset)
pagination['PreviousRange'] = f'{left_new_offset+1} - {left_new_offset+env.results_per_page}'
if offset + env.results_per_page < total_instances:
pagination['NextEnabled'] = True
right_new_offset = offset + env.results_per_page
pagination['NextLink'] = rewrite_full_path_new_offset(right_new_offset)
pagination['NextRange'] = f'{right_new_offset+1} - {min(right_new_offset+env.results_per_page, total_instances)}'
orthanc_access = client.permissions.has_access_to_orthanc()
response = flask.render_template('search.html', pagination=pagination, results=results, orthanc_access=orthanc_access)
client.access_logger.log_search("Display", query, jsonpath_query, total_instances, total_series)
# If the "Export" button was pressed, return a formatted JSON document for each of the DICOM
# instances matching the query
if export_action is True:
try:
instances = db_dicom_json.search_instances_with_series(jsonpath_query)
series_ids = get_unique_series_ids(instances)
except:
return flask.render_template('search.html', error_message='Failed to query the database. Please check your query and retry.')
file_content = {'Series': []}
for series_id in series_ids:
series = {
'SeriesId': series_id,
'Instances': []
}
for instance in instances:
instance_id = instance[0]
instance_series_id = instance[1]
index_in_series = instance[2]
instance_json = instance[3]
if instance_series_id == series_id:
series['Instances'].append({
'InstanceId': instance_id,
'IndexInSeries': index_in_series,
'InstanceJSON': rpacs_dicom_json.add_keywords_to_dicom_json(instance_json)
})
series['Instances'] = sorted(series['Instances'], key=lambda k: k['IndexInSeries'])
file_content['Series'].append(series)
file_json = json.dumps(file_content, indent=2, sort_keys=True)
file_bytes = BytesIO(file_json.encode())
response = flask.send_file(file_bytes, mimetype='application/json', as_attachment=True, attachment_filename='results.json')
client.access_logger.log_search("Export", query, jsonpath_query, len(instances), len(series_ids))
return response