in pathology/transformation_pipeline/ingestion_lib/dicom_util/dicom_iod_util.py [0:0]
def decode_dataset(self, line_stack: List[UnDecodedTableLine]):
"""Decode list of table lines to create the dicom dataset.
Dicom tables define contents using a combination of
inline and linked definitions.
Example cut example from (table_name": "8.8-1",):
{"type": "LinkedObject", "prefix": "Include", ...
{"type": "InlineObject", "name": "Equivalent, ...
{"type": "LinkedObject", "prefix": ">Include" ...
The above are 3 lines from the table.
Order of the rows matters!
{"type": "LinkedObject", "prefix": "Include", ...
First row indicates that a contents of another table
is to be dumped directly in to the table being described.
{"type": "InlineObject", "name": "Equivalent, ...
Directly describe the definition of dicom tag describing a
dicom sequence tag (VR=SQ). The structure of the leafs nodes
described in the sequence is then described by the following line.
{"type": "LinkedObject", "prefix": ">Include" ...
This line again references another table. It indicates that the
tables content describes the sequence because the include is prefixed
by a >,
Tables can have links or sequences which are self recursive.
This method decodes a target node of the IOD tree.
the method works by maintaining a stack of nodes to decode.
Nodes are poped from the stack. Nodes applying to the decoded node
are decoded and added to the node. Nodes pertaining to child SQ or
deeper levels are pushed to the child node and decoded only if the child
is subsquently decoded.
Args:
line_stack: List of table lines to decode.
"""
line_stack.reverse()
last_tag = None
while line_stack:
line_item = line_stack.pop() # get table line
module_usage = line_item.requirement
table_line = line_item.line
if isinstance(table_line, dcm_util.InlineTableLine):
# Inline Table definition.
tag_table_level = table_line.name.count('>')
if tag_table_level == 0:
# line describes dicom tag in current dataset
standard_tag_def = self._dicom_standard.get_tag(table_line.address)
if standard_tag_def is None:
raise dcm_util.DICOMSpecMetadataError(
'Table line specifies tag not defined'
' in the parsed spec '
f'{table_line.address}'
)
last_tag = DICOMTag(
standard_tag_def, table_line.required, module_usage
)
self.add_tag(last_tag)
else:
if last_tag is None or not last_tag.is_sq():
raise dcm_util.DICOMSpecMetadataError(
'Cannot tag to Null or Non-SQ Tag'
)
# line describes dicom tag in child dataset
# push tag definition to child. decode later if child is decoded.
name = DICOMDataset._remove_child_level_indicator(
table_line.name, table_line
)
new_line = dcm_util.InlineTableLine(
name, table_line.address, table_line.required, table_line.comment
)
inner_ds = self._get_child_tag_sq_dataset(last_tag)
inner_ds.level_line_queue_append(
UnDecodedTableLine(module_usage, new_line)
)
elif isinstance(table_line, dcm_util.LinkedTableLine):
# Line references another table.
tag_table_level = table_line.prefix.count('>')
if tag_table_level == 0:
# table is inline reference. push inline tags into parse stack.
if table_line.table_name == 'IODFunctionalGroupMacros':
iod_module_list = (
self._dicom_standard.get_iod_functional_group_modules(
self._iod_name
)
)
for module_table_lines in self.get_module_tables(iod_module_list):
module_table_lines = copy.copy(module_table_lines)
module_table_lines.reverse()
for line in module_table_lines:
line_stack.append(line)
else:
lnk_tbl = self._dicom_standard.get_table(table_line.table_name)
linked_tablelines = copy.copy(lnk_tbl.table_lines)
linked_tablelines.reverse()
# push tags onto decoding stack.
for linked_line in linked_tablelines:
line_stack.append(UnDecodedTableLine(module_usage, linked_line))
else:
if last_tag is None or not last_tag.is_sq():
raise dcm_util.DICOMSpecMetadataError(
'Cannot tag to Null or Non-SQ Tag'
)
# Line references another table which is included at a child level.
# decode later.
prefix = DICOMDataset._remove_child_level_indicator(
table_line.prefix, table_line
)
new_line = dcm_util.LinkedTableLine(
prefix, table_line.linked_table, table_line.usage
)
inner_ds = self._get_child_tag_sq_dataset(last_tag)
inner_ds.level_line_queue_append(
UnDecodedTableLine(module_usage, new_line)
)