Skip to content

IODSpecBuilder

dcmspec.iod_spec_builder.IODSpecBuilder

Orchestrates the construction of a expanded DICOM IOD specification model.

The IODSpecBuilder uses a factory to build the IOD Modules model and, for each referenced module, uses a (possibly different) factory to build and cache the Module models. It then assembles a new model with the IOD nodes and their referenced module nodes as children, and caches the expanded model.

Source code in src/dcmspec/iod_spec_builder.py 
 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
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
class IODSpecBuilder:
    """Orchestrates the construction of a expanded DICOM IOD specification model.

    The IODSpecBuilder uses a factory to build the IOD Modules model and, for each referenced module,
    uses a (possibly different) factory to build and cache the Module models. It then assembles a new
    model with the IOD nodes and their referenced module nodes as children, and caches the expanded model.
    """

    def __init__(
        self,
        iod_factory: SpecFactory = None,
        module_factory: SpecFactory = None,
        logger: logging.Logger = None,
    ):
        """Initialize the IODSpecBuilder.

        If no factory is provided, a default SpecFactory is used for both IOD and module models.

        Args:
            iod_factory (Optional[SpecFactory]): Factory for building the IOD model. If None, uses SpecFactory().
            module_factory (Optional[SpecFactory]): Factory for building module models. If None, uses iod_factory.
            logger (Optional[logging.Logger]): Logger instance to use. If None, a default logger is created.

        The builder is initialized with factories for the IOD and module models. By default, the same
        factory is used for both, but a different factory can be provided for modules if needed.

        """
        self.logger = logger or logging.getLogger(self.__class__.__name__)

        self.iod_factory = iod_factory or SpecFactory(logger=self.logger)
        self.module_factory = module_factory or self.iod_factory
        self.dom_utils = DOMUtils(logger=self.logger)

    def build_from_url(
        self,
        url: str,
        cache_file_name: str,
        table_id: str,
        force_download: bool = False,
        progress_observer: 'Optional[ProgressObserver]' = None,
        # BEGIN LEGACY SUPPORT: Remove for int progress callback deprecation
        progress_callback: 'Optional[Callable[[int], None]]' = None,
        # END LEGACY SUPPORT
        json_file_name: str = None,
        **kwargs: object,
    ) -> SpecModel:
        """Build and cache a DICOM IOD specification model from a URL.

        This method orchestrates the full workflow:
        - Loads or downloads the IOD table and builds/caches the IOD model using the iod_factory.
        - Finds all nodes in the IOD model with a "ref" attribute, indicating a referenced module.
        - For each referenced module, loads or parses and caches the module model using the module_factory.
        - Assembles a new expanded model, where each IOD node has its referenced module's content node as a child.
        - Uses the first module's metadata header and version for the expanded model's metadata.
        - Caches the expanded model if a json_file_name is provided.

        Args:
            url (str): The URL to download the input file from.
            cache_file_name (str): Filename of the cached input file.
            table_id (str): The ID of the IOD table to parse.
            force_download (bool): If True, always download the input file and generate the model even if cached.
            progress_observer (Optional[ProgressObserver]): Optional observer to report download progress.
                See the Note below for details on the progress events and their properties.
            progress_callback (Optional[Callable[[int], None]]): [LEGACY, Deprecated] Optional callback to
                report progress as an integer percent (0-100, or -1 if indeterminate). Use progress_observer
                instead. Will be removed in a future release.            
            json_file_name (str, optional): Filename to save the cached expanded JSON model.
            **kwargs: Additional arguments for model construction.

        Returns:
            SpecModel: The expanded model with IOD and module content.

        Note:
            If a progress observer accepting a Progress object is provided, progress events are as follows:

            - **Step 1 (DOWNLOADING_IOD):** Events include `status=DOWNLOADING_IOD`, `step=1`,
            `total_steps=4`, and a meaningful `percent` value.
            - **Step 2 (PARSING_IOD_MODULE_LIST):** Events include `status=PARSING_IOD_MODULE_LIST`, `step=2`,
            `total_steps=4`, and `percent == -1` (indeterminate).
            - **Step 3 (PARSING_IOD_MODULES):** Events include `status=PARSING_IOD_MODULES`, `step=3`,
                `total_steps=4`, and a meaningful `percent` value.
            - **Step 4 (SAVING_IOD_MODEL):** Events include `status=SAVING_IOD_MODEL`, `step=4`,
                `total_steps=4`, and `percent == -1` (indeterminate).

            For example usage in a client application,
            see [`ProgressStatus`](progress.md#dcmspec.progress.ProgressStatus).

        """
        # BEGIN LEGACY SUPPORT: Remove for int progress callback deprecation
        progress_observer = handle_legacy_callback(progress_observer, progress_callback)
        # END LEGACY SUPPORT
        # Load from cache if the expanded IOD model is already present
        cached_model = self._load_expanded_model_from_cache(json_file_name, force_download)
        if cached_model is not None:
            cached_model.logger = self.logger
            return cached_model

        total_steps = 4  # 1=download, 2=parse IOD, 3=build modules, 4=save

        # --- Step 1: Load the DOM from cache file or download and cache DOM in memory ---
        if progress_observer:
            @add_progress_step(step=1, total_steps=total_steps, status=ProgressStatus.DOWNLOADING_IOD)
            def step1_progress_observer(progress):
                progress_observer(progress)
        else:
            step1_progress_observer = None
        dom = self.iod_factory.load_document(
            url=url,
            cache_file_name=cache_file_name,
            force_download=force_download,
            progress_observer=step1_progress_observer,
            # BEGIN LEGACY SUPPORT: Remove for int progress callback deprecation
            progress_callback=progress_observer,
            # END LEGACY SUPPORT
        )

        # --- Step 2: Build the IOD Module List model from the DOM ---
        if progress_observer:
            progress_observer(
                Progress(-1, status=ProgressStatus.PARSING_IOD_MODULE_LIST, step=2, total_steps=total_steps)
                )
        iodmodules_model = self.iod_factory.build_model(
            doc_object=dom,
            table_id=table_id,
            url=url,
            json_file_name=json_file_name,
        )

        # --- Step 3: Build or load model for each module in the IOD ---
        if progress_observer:
            progress_observer(
                Progress(-1, status=ProgressStatus.PARSING_IOD_MODULES, step=3, total_steps=total_steps)
            )

        # Find all nodes with a "ref" attribute in the IOD Modules model
        nodes_with_ref = [node for node in iodmodules_model.content.children if hasattr(node, "ref")]

        # Build or load module models for each referenced section
        module_models = self._build_module_models(
            nodes_with_ref, dom, url, step=3, total_steps=total_steps, progress_observer=progress_observer
        )
        # Fail if no module models were found.
        if not module_models:
            raise RuntimeError("No module models were found for the referenced modules in the IOD table.")

        # --- Step 4: Create and store the expanded model with IOD and module content ---
        if progress_observer:
            progress_observer(Progress(-1, status=ProgressStatus.SAVING_IOD_MODEL, step=4, total_steps=total_steps))

        # Create the expanded model from the IOD modules and module models
        iod_model = self._create_expanded_model(iodmodules_model, module_models)

        # Cache the expanded model if a json_file_name was provided
        if json_file_name:
            iod_json_file_path = os.path.join(
                self.iod_factory.config.get_param("cache_dir"), "model", json_file_name
            )
            try:
                self.iod_factory.model_store.save(iod_model, iod_json_file_path)
            except Exception as e:
                self.logger.warning(f"Failed to cache expanded model to {iod_json_file_path}: {e}")
        else:
            self.logger.info("No json_file_name specified; IOD model not cached.")

        return iod_model

    def _load_expanded_model_from_cache(self, json_file_name: str, force_download: bool) -> SpecModel | None:
        """Return the cached expanded IOD model if available and not force_download, else None."""
        iod_json_file_path = None
        if json_file_name:
            iod_json_file_path = os.path.join(
                self.iod_factory.config.get_param("cache_dir"), "model", json_file_name
            )
        if iod_json_file_path and os.path.exists(iod_json_file_path) and not force_download:
            try:
                return self.iod_factory.model_store.load(iod_json_file_path)
            except Exception as e:
                self.logger.warning(
                    f"Failed to load expanded IOD model from cache {iod_json_file_path}: {e}"
                )
        return None

    def _build_module_models(
        self,
        nodes_with_ref: List[Any],
        dom: Any,
        url: str,
        step: int,
        total_steps: int,
        progress_observer: Optional['ProgressObserver'] = None
    ) -> Dict[str, Any]:        
        """Build or load module models for each referenced section, reporting progress."""
        module_models: Dict[str, Any] = {}
        total_modules = len(nodes_with_ref)
        if progress_observer and total_modules > 0:
            progress_observer(
                Progress(0, status=ProgressStatus.PARSING_IOD_MODULES, step=step, total_steps=total_steps)
                )
        for idx, node in enumerate(nodes_with_ref):
            ref_value = getattr(node, "ref", None)
            if not ref_value:
                continue
            section_id = f"sect_{ref_value}"
            module_table_id = self.dom_utils.get_table_id_from_section(dom, section_id)
            if not module_table_id:
                self.logger.warning(f"No table found for section id {section_id}")
                continue

            module_json_file_name = f"{module_table_id}.json"
            module_json_file_path = os.path.join(
                self.module_factory.config.get_param("cache_dir"), "model", module_json_file_name
            )
            if os.path.exists(module_json_file_path):
                try:
                    module_model = self.module_factory.model_store.load(module_json_file_path)
                except Exception as e:
                    self.logger.warning(f"Failed to load module model from cache {module_json_file_path}: {e}")
                    module_model = self.module_factory.build_model(
                        doc_object=dom,
                        table_id=module_table_id,
                        url=url,
                        json_file_name=module_json_file_name,
                        progress_observer=progress_observer,
                    )
            else:
                module_model = self.module_factory.build_model(
                    doc_object=dom,
                    table_id=module_table_id,
                    url=url,
                    json_file_name=module_json_file_name,
                    progress_observer=progress_observer,
                )
            module_models[ref_value] = module_model
            if progress_observer and total_modules > 0:
                percent = calculate_percent(idx + 1, total_modules)
                progress_observer(Progress(
                    percent,
                    status=ProgressStatus.PARSING_IOD_MODULES,
                    step=step,
                    total_steps=total_steps
                ))
        return module_models

    def _create_expanded_model(self, iodmodules_model: SpecModel, module_models: dict) -> SpecModel:
        """Create the expanded model by attaching Module nodes content to IOD nodes."""
        # Use the first module's metadata node for the expanded model
        first_module = next(iter(module_models.values()))
        iod_metadata = first_module.metadata
        iod_metadata.table_id = iodmodules_model.metadata.table_id

        # The content node will have as children the IOD model's nodes,
        # and for each referenced module, its content's children will be attached directly under the iod node
        iod_content = Node("content")
        for iod_node in iodmodules_model.content.children:
            ref_value = getattr(iod_node, "ref", None)
            if ref_value and ref_value in module_models:
                module_content = module_models[ref_value].content
                for child in list(module_content.children):
                    child.parent = iod_node
            iod_node.parent = iod_content

        # Create and return the expanded model
        return SpecModel(metadata=iod_metadata, content=iod_content)

__init__(iod_factory=None, module_factory=None, logger=None)

Initialize the IODSpecBuilder.

If no factory is provided, a default SpecFactory is used for both IOD and module models.

PARAMETER DESCRIPTION
iod_factory

Factory for building the IOD model. If None, uses SpecFactory().

TYPE: Optional[SpecFactory] DEFAULT: None

module_factory

Factory for building module models. If None, uses iod_factory.

TYPE: Optional[SpecFactory] DEFAULT: None

logger

Logger instance to use. If None, a default logger is created.

TYPE: Optional[Logger] DEFAULT: None

The builder is initialized with factories for the IOD and module models. By default, the same factory is used for both, but a different factory can be provided for modules if needed.

Source code in src/dcmspec/iod_spec_builder.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def __init__(
    self,
    iod_factory: SpecFactory = None,
    module_factory: SpecFactory = None,
    logger: logging.Logger = None,
):
    """Initialize the IODSpecBuilder.

    If no factory is provided, a default SpecFactory is used for both IOD and module models.

    Args:
        iod_factory (Optional[SpecFactory]): Factory for building the IOD model. If None, uses SpecFactory().
        module_factory (Optional[SpecFactory]): Factory for building module models. If None, uses iod_factory.
        logger (Optional[logging.Logger]): Logger instance to use. If None, a default logger is created.

    The builder is initialized with factories for the IOD and module models. By default, the same
    factory is used for both, but a different factory can be provided for modules if needed.

    """
    self.logger = logger or logging.getLogger(self.__class__.__name__)

    self.iod_factory = iod_factory or SpecFactory(logger=self.logger)
    self.module_factory = module_factory or self.iod_factory
    self.dom_utils = DOMUtils(logger=self.logger)

build_from_url(url, cache_file_name, table_id, force_download=False, progress_observer=None, progress_callback=None, json_file_name=None, **kwargs)

Build and cache a DICOM IOD specification model from a URL.

This method orchestrates the full workflow: - Loads or downloads the IOD table and builds/caches the IOD model using the iod_factory. - Finds all nodes in the IOD model with a "ref" attribute, indicating a referenced module. - For each referenced module, loads or parses and caches the module model using the module_factory. - Assembles a new expanded model, where each IOD node has its referenced module's content node as a child. - Uses the first module's metadata header and version for the expanded model's metadata. - Caches the expanded model if a json_file_name is provided.

PARAMETER DESCRIPTION
url

The URL to download the input file from.

TYPE: str

cache_file_name

Filename of the cached input file.

TYPE: str

table_id

The ID of the IOD table to parse.

TYPE: str

force_download

If True, always download the input file and generate the model even if cached.

TYPE: bool DEFAULT: False

progress_observer

Optional observer to report download progress. See the Note below for details on the progress events and their properties.

TYPE: Optional[ProgressObserver] DEFAULT: None

progress_callback

[LEGACY, Deprecated] Optional callback to report progress as an integer percent (0-100, or -1 if indeterminate). Use progress_observer instead. Will be removed in a future release.

TYPE: Optional[Callable[[int], None]] DEFAULT: None

json_file_name

Filename to save the cached expanded JSON model.

TYPE: str DEFAULT: None

**kwargs

Additional arguments for model construction.

TYPE: object DEFAULT: {}

RETURNS DESCRIPTION
SpecModel

The expanded model with IOD and module content.

TYPE: SpecModel

Note

If a progress observer accepting a Progress object is provided, progress events are as follows:

  • Step 1 (DOWNLOADING_IOD): Events include status=DOWNLOADING_IOD, step=1, total_steps=4, and a meaningful percent value.
  • Step 2 (PARSING_IOD_MODULE_LIST): Events include status=PARSING_IOD_MODULE_LIST, step=2, total_steps=4, and percent == -1 (indeterminate).
  • Step 3 (PARSING_IOD_MODULES): Events include status=PARSING_IOD_MODULES, step=3, total_steps=4, and a meaningful percent value.
  • Step 4 (SAVING_IOD_MODEL): Events include status=SAVING_IOD_MODEL, step=4, total_steps=4, and percent == -1 (indeterminate).

For example usage in a client application, see ProgressStatus.

Source code in src/dcmspec/iod_spec_builder.py 
 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
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
def build_from_url(
    self,
    url: str,
    cache_file_name: str,
    table_id: str,
    force_download: bool = False,
    progress_observer: 'Optional[ProgressObserver]' = None,
    # BEGIN LEGACY SUPPORT: Remove for int progress callback deprecation
    progress_callback: 'Optional[Callable[[int], None]]' = None,
    # END LEGACY SUPPORT
    json_file_name: str = None,
    **kwargs: object,
) -> SpecModel:
    """Build and cache a DICOM IOD specification model from a URL.

    This method orchestrates the full workflow:
    - Loads or downloads the IOD table and builds/caches the IOD model using the iod_factory.
    - Finds all nodes in the IOD model with a "ref" attribute, indicating a referenced module.
    - For each referenced module, loads or parses and caches the module model using the module_factory.
    - Assembles a new expanded model, where each IOD node has its referenced module's content node as a child.
    - Uses the first module's metadata header and version for the expanded model's metadata.
    - Caches the expanded model if a json_file_name is provided.

    Args:
        url (str): The URL to download the input file from.
        cache_file_name (str): Filename of the cached input file.
        table_id (str): The ID of the IOD table to parse.
        force_download (bool): If True, always download the input file and generate the model even if cached.
        progress_observer (Optional[ProgressObserver]): Optional observer to report download progress.
            See the Note below for details on the progress events and their properties.
        progress_callback (Optional[Callable[[int], None]]): [LEGACY, Deprecated] Optional callback to
            report progress as an integer percent (0-100, or -1 if indeterminate). Use progress_observer
            instead. Will be removed in a future release.            
        json_file_name (str, optional): Filename to save the cached expanded JSON model.
        **kwargs: Additional arguments for model construction.

    Returns:
        SpecModel: The expanded model with IOD and module content.

    Note:
        If a progress observer accepting a Progress object is provided, progress events are as follows:

        - **Step 1 (DOWNLOADING_IOD):** Events include `status=DOWNLOADING_IOD`, `step=1`,
        `total_steps=4`, and a meaningful `percent` value.
        - **Step 2 (PARSING_IOD_MODULE_LIST):** Events include `status=PARSING_IOD_MODULE_LIST`, `step=2`,
        `total_steps=4`, and `percent == -1` (indeterminate).
        - **Step 3 (PARSING_IOD_MODULES):** Events include `status=PARSING_IOD_MODULES`, `step=3`,
            `total_steps=4`, and a meaningful `percent` value.
        - **Step 4 (SAVING_IOD_MODEL):** Events include `status=SAVING_IOD_MODEL`, `step=4`,
            `total_steps=4`, and `percent == -1` (indeterminate).

        For example usage in a client application,
        see [`ProgressStatus`](progress.md#dcmspec.progress.ProgressStatus).

    """
    # BEGIN LEGACY SUPPORT: Remove for int progress callback deprecation
    progress_observer = handle_legacy_callback(progress_observer, progress_callback)
    # END LEGACY SUPPORT
    # Load from cache if the expanded IOD model is already present
    cached_model = self._load_expanded_model_from_cache(json_file_name, force_download)
    if cached_model is not None:
        cached_model.logger = self.logger
        return cached_model

    total_steps = 4  # 1=download, 2=parse IOD, 3=build modules, 4=save

    # --- Step 1: Load the DOM from cache file or download and cache DOM in memory ---
    if progress_observer:
        @add_progress_step(step=1, total_steps=total_steps, status=ProgressStatus.DOWNLOADING_IOD)
        def step1_progress_observer(progress):
            progress_observer(progress)
    else:
        step1_progress_observer = None
    dom = self.iod_factory.load_document(
        url=url,
        cache_file_name=cache_file_name,
        force_download=force_download,
        progress_observer=step1_progress_observer,
        # BEGIN LEGACY SUPPORT: Remove for int progress callback deprecation
        progress_callback=progress_observer,
        # END LEGACY SUPPORT
    )

    # --- Step 2: Build the IOD Module List model from the DOM ---
    if progress_observer:
        progress_observer(
            Progress(-1, status=ProgressStatus.PARSING_IOD_MODULE_LIST, step=2, total_steps=total_steps)
            )
    iodmodules_model = self.iod_factory.build_model(
        doc_object=dom,
        table_id=table_id,
        url=url,
        json_file_name=json_file_name,
    )

    # --- Step 3: Build or load model for each module in the IOD ---
    if progress_observer:
        progress_observer(
            Progress(-1, status=ProgressStatus.PARSING_IOD_MODULES, step=3, total_steps=total_steps)
        )

    # Find all nodes with a "ref" attribute in the IOD Modules model
    nodes_with_ref = [node for node in iodmodules_model.content.children if hasattr(node, "ref")]

    # Build or load module models for each referenced section
    module_models = self._build_module_models(
        nodes_with_ref, dom, url, step=3, total_steps=total_steps, progress_observer=progress_observer
    )
    # Fail if no module models were found.
    if not module_models:
        raise RuntimeError("No module models were found for the referenced modules in the IOD table.")

    # --- Step 4: Create and store the expanded model with IOD and module content ---
    if progress_observer:
        progress_observer(Progress(-1, status=ProgressStatus.SAVING_IOD_MODEL, step=4, total_steps=total_steps))

    # Create the expanded model from the IOD modules and module models
    iod_model = self._create_expanded_model(iodmodules_model, module_models)

    # Cache the expanded model if a json_file_name was provided
    if json_file_name:
        iod_json_file_path = os.path.join(
            self.iod_factory.config.get_param("cache_dir"), "model", json_file_name
        )
        try:
            self.iod_factory.model_store.save(iod_model, iod_json_file_path)
        except Exception as e:
            self.logger.warning(f"Failed to cache expanded model to {iod_json_file_path}: {e}")
    else:
        self.logger.info("No json_file_name specified; IOD model not cached.")

    return iod_model