Skip to content

CIF Model

Core types (Rust extension)

cifflow_core

Builder

cifflow.cifmodel.builder

CifBuilder — constructs a CifFile from the CifParserEvents stream.

CifBuilder implements CifParserEvents and is wired directly to the Rust parser:

builder = CifBuilder(on_error=handler.on_error)
version = cifflow_core.parse(source, builder)
cif = builder.result

Semantic errors (empty loop, row-count mismatch) are reported via the on_error callable using error_type='semantic'. In strict mode the builder stops accumulating after the first semantic error; in pad mode it continues and pads incomplete loop rows with '?' placeholders.

Multiline text field values (ValueType.MULTILINE_STRING) are passed through the transformation pipeline (prefix removal + line unfolding) before storage. All other value types are stored as raw strings.

CifBuilder

Implements :class:~cifflow.types.CifParserEvents; accumulates events into a CifFile.

Parameters:

Name Type Description Default
on_error Callable[[ParseError], None]

Called with a :class:~cifflow.types.ParseError for each semantic error detected by the IR layer. Pass handler.on_error to unify parser and IR errors into a single stream.

required
mode Literal['strict', 'pad']

'strict' — stop accumulating on the first semantic error. 'pad' — continue and pad incomplete loop rows with '?' placeholders.

'pad'
Source code in src/cifflow/cifmodel/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
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
283
284
285
286
287
288
289
290
class CifBuilder:
    """
    Implements :class:`~cifflow.types.CifParserEvents`; accumulates events into a CifFile.

    Parameters
    ----------
    on_error
        Called with a :class:`~cifflow.types.ParseError` for each semantic error
        detected by the IR layer.  Pass ``handler.on_error`` to unify parser and
        IR errors into a single stream.
    mode
        ``'strict'`` — stop accumulating on the first semantic error.
        ``'pad'`` — continue and pad incomplete loop rows with ``'?'`` placeholders.
    """

    def __init__(
        self,
        on_error: Callable[[ParseError], None],
        mode: Literal['strict', 'pad'] = 'pad',
    ) -> None:
        self._on_error = on_error
        self._mode = mode
        self._file = CifFile()

        # Block / save-frame state
        self._block: CifBlock | None = None
        self._save_frame: CifSaveFrame | None = None

        # Active tag (awaiting a value)
        self._active_tag: str | None = None

        # Loop state
        self._in_loop = False
        self._loop_tags: list[str] = []
        self._loop_value_index = 0
        self._loop_buffers: dict[str, list[CifValue]] = {}

        # Container nesting stack
        self._container_stack: list[_Container] = []

        # Set to True in strict mode after a semantic error
        self._stopped = False

    # ── Result ────────────────────────────────────────────────────────────────

    @property
    def result(self) -> CifFile:
        """The CifFile accumulated so far."""
        return self._file

    # ── Helpers ───────────────────────────────────────────────────────────────

    @property
    def _current_ns(self) -> CifBlock | CifSaveFrame | None:
        """Return the active namespace: the current save frame, or the current block."""
        return self._save_frame if self._save_frame is not None else self._block

    def _semantic_error(self, message: str, recovery: str) -> None:
        """Emit a semantic error event and set stopped=True in strict mode."""
        self._on_error(ParseError(
            error_type='semantic',
            message=message,
            line=0, column=0,
            context='CifBuilder',
            recovery_action=recovery,
        ))
        if self._mode == 'strict':
            self._stopped = True

    def _dispatch_value(self, value: CifValue) -> None:
        """Route a complete value (scalar or closed container) to its destination."""
        if self._container_stack:
            top = self._container_stack[-1]
            if isinstance(top, list):
                top.append(value)
            else:
                if top.current_key is not None:
                    top.data[top.current_key] = value
                    top.current_key = None
            return

        if self._in_loop:
            n = len(self._loop_tags)
            tag = self._loop_tags[self._loop_value_index % n]
            self._loop_buffers[tag].append(value)
            self._loop_value_index += 1
            return

        ns = self._current_ns
        if ns is not None and self._active_tag is not None:
            ns._append_value(self._active_tag, value)
            self._active_tag = None

    # ── CifParserEvents ───────────────────────────────────────────────────────

    def on_data_block(self, name: str) -> None:
        """Casefold name, emit error on duplicate, and open a new data block."""
        if self._stopped:
            return
        name = _casefold(name)
        if name in self._file:
            self._semantic_error(
                message=f'duplicate data block name: {name!r}',
                recovery='duplicate block stored with distinct internal id',
            )
        self._block = CifBlock(name)
        self._file._add_block(self._block)
        self._save_frame = None
        self._active_tag = None
        self._in_loop = False
        self._loop_tags = []
        self._loop_value_index = 0
        self._loop_buffers = {}
        self._container_stack = []

    def on_save_frame_start(self, name: str) -> None:
        """Casefold name, emit error on duplicate, and open a new save frame."""
        if self._stopped or self._block is None:
            return
        name = _casefold(name)
        if name in self._block:
            self._semantic_error(
                message=f'duplicate save frame name: {name!r}',
                recovery='duplicate save frame stored with distinct internal id',
            )
        self._save_frame = CifSaveFrame(name)

    def on_save_frame_end(self) -> None:
        """Close the current save frame and register it with the block."""
        if self._stopped or self._block is None:
            return
        if self._save_frame is not None:
            self._block._add_save_frame(self._save_frame)
        self._save_frame = None

    def add_tag(self, tag_name: str) -> None:
        """Casefold and register the pending tag name."""
        if self._stopped:
            return
        self._active_tag = _casefold(tag_name)

    def add_value(self, value: str, value_type: ValueType) -> None:
        """Transform multiline fields; quote bare dots/questions; dispatch the value."""
        if self._stopped:
            return
        if value_type == ValueType.MULTILINE_STRING:
            value = transform_multiline(value)
        elif value_type != ValueType.PLACEHOLDER and value in ('.', '?'):
            value = f'"{value}"'
        self._dispatch_value(value)

    def on_list_start(self) -> None:
        """Push a new list container onto the nesting stack."""
        if self._stopped:
            return
        self._container_stack.append([])

    def on_list_end(self) -> None:
        """Pop the completed list and dispatch it as a value."""
        if self._stopped or not self._container_stack:
            return
        completed = self._container_stack.pop()
        self._dispatch_value(completed)

    def on_table_start(self) -> None:
        """Push a new table container onto the nesting stack."""
        if self._stopped:
            return
        self._container_stack.append(_TableInProgress())

    def on_table_key(self, key: str, value_type: ValueType) -> None:
        """Set the current key on the pending table container."""
        if self._stopped or not self._container_stack:
            return
        top = self._container_stack[-1]
        if isinstance(top, _TableInProgress):
            top.current_key = key

    def on_table_end(self) -> None:
        """Pop the completed table and dispatch it as a value."""
        if self._stopped or not self._container_stack:
            return
        top = self._container_stack.pop()
        if isinstance(top, _TableInProgress):
            self._dispatch_value(top.data)

    def on_loop_start(self, tags: list[str]) -> None:
        """Initialise loop tracking with casefolded tag names and empty buffers."""
        if self._stopped:
            return
        self._in_loop = True
        self._loop_tags = [_casefold(t) for t in tags]
        self._loop_value_index = 0
        self._loop_buffers = {_casefold(t): [] for t in tags}

    def on_loop_end(self) -> None:
        """Validate row count; pad in pad mode; commit loop to the namespace."""
        if self._stopped:
            return
        n = len(self._loop_tags)
        total = self._loop_value_index
        ns = self._current_ns

        if n == 0:
            self._in_loop = False
            return

        if total % n != 0:
            missing = n - (total % n)
            tag_list = ', '.join(self._loop_tags)
            self._semantic_error(
                message=(
                    f'loop value count {total} is not divisible by tag count {n} '
                    f'({missing} value(s) missing from final row); '
                    f'tags: {tag_list}'
                ),
                recovery='stopped' if self._mode == 'strict' else f'padded {missing} placeholder(s)',
            )
            if self._stopped:
                self._in_loop = False
                return
            # Pad mode: fill incomplete final row with '?'
            for _ in range(missing):
                tag = self._loop_tags[self._loop_value_index % n]
                self._loop_buffers[tag].append('?')
                self._loop_value_index += 1

        if ns is not None:
            ns._add_loop(self._loop_tags, self._loop_buffers)

        self._in_loop = False
        self._loop_tags = []
        self._loop_value_index = 0
        self._loop_buffers = {}

    def on_error(self, error: ParseError) -> None:
        """Forward the parse error to the registered on_error callback."""
        self._on_error(error)

result property

The CifFile accumulated so far.

on_data_block(name)

Casefold name, emit error on duplicate, and open a new data block.

Source code in src/cifflow/cifmodel/builder.py
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def on_data_block(self, name: str) -> None:
    """Casefold name, emit error on duplicate, and open a new data block."""
    if self._stopped:
        return
    name = _casefold(name)
    if name in self._file:
        self._semantic_error(
            message=f'duplicate data block name: {name!r}',
            recovery='duplicate block stored with distinct internal id',
        )
    self._block = CifBlock(name)
    self._file._add_block(self._block)
    self._save_frame = None
    self._active_tag = None
    self._in_loop = False
    self._loop_tags = []
    self._loop_value_index = 0
    self._loop_buffers = {}
    self._container_stack = []

on_save_frame_start(name)

Casefold name, emit error on duplicate, and open a new save frame.

Source code in src/cifflow/cifmodel/builder.py
168
169
170
171
172
173
174
175
176
177
178
def on_save_frame_start(self, name: str) -> None:
    """Casefold name, emit error on duplicate, and open a new save frame."""
    if self._stopped or self._block is None:
        return
    name = _casefold(name)
    if name in self._block:
        self._semantic_error(
            message=f'duplicate save frame name: {name!r}',
            recovery='duplicate save frame stored with distinct internal id',
        )
    self._save_frame = CifSaveFrame(name)

on_save_frame_end()

Close the current save frame and register it with the block.

Source code in src/cifflow/cifmodel/builder.py
180
181
182
183
184
185
186
def on_save_frame_end(self) -> None:
    """Close the current save frame and register it with the block."""
    if self._stopped or self._block is None:
        return
    if self._save_frame is not None:
        self._block._add_save_frame(self._save_frame)
    self._save_frame = None

add_tag(tag_name)

Casefold and register the pending tag name.

Source code in src/cifflow/cifmodel/builder.py
188
189
190
191
192
def add_tag(self, tag_name: str) -> None:
    """Casefold and register the pending tag name."""
    if self._stopped:
        return
    self._active_tag = _casefold(tag_name)

add_value(value, value_type)

Transform multiline fields; quote bare dots/questions; dispatch the value.

Source code in src/cifflow/cifmodel/builder.py
194
195
196
197
198
199
200
201
202
def add_value(self, value: str, value_type: ValueType) -> None:
    """Transform multiline fields; quote bare dots/questions; dispatch the value."""
    if self._stopped:
        return
    if value_type == ValueType.MULTILINE_STRING:
        value = transform_multiline(value)
    elif value_type != ValueType.PLACEHOLDER and value in ('.', '?'):
        value = f'"{value}"'
    self._dispatch_value(value)

on_list_start()

Push a new list container onto the nesting stack.

Source code in src/cifflow/cifmodel/builder.py
204
205
206
207
208
def on_list_start(self) -> None:
    """Push a new list container onto the nesting stack."""
    if self._stopped:
        return
    self._container_stack.append([])

on_list_end()

Pop the completed list and dispatch it as a value.

Source code in src/cifflow/cifmodel/builder.py
210
211
212
213
214
215
def on_list_end(self) -> None:
    """Pop the completed list and dispatch it as a value."""
    if self._stopped or not self._container_stack:
        return
    completed = self._container_stack.pop()
    self._dispatch_value(completed)

on_table_start()

Push a new table container onto the nesting stack.

Source code in src/cifflow/cifmodel/builder.py
217
218
219
220
221
def on_table_start(self) -> None:
    """Push a new table container onto the nesting stack."""
    if self._stopped:
        return
    self._container_stack.append(_TableInProgress())

on_table_key(key, value_type)

Set the current key on the pending table container.

Source code in src/cifflow/cifmodel/builder.py
223
224
225
226
227
228
229
def on_table_key(self, key: str, value_type: ValueType) -> None:
    """Set the current key on the pending table container."""
    if self._stopped or not self._container_stack:
        return
    top = self._container_stack[-1]
    if isinstance(top, _TableInProgress):
        top.current_key = key

on_table_end()

Pop the completed table and dispatch it as a value.

Source code in src/cifflow/cifmodel/builder.py
231
232
233
234
235
236
237
def on_table_end(self) -> None:
    """Pop the completed table and dispatch it as a value."""
    if self._stopped or not self._container_stack:
        return
    top = self._container_stack.pop()
    if isinstance(top, _TableInProgress):
        self._dispatch_value(top.data)

on_loop_start(tags)

Initialise loop tracking with casefolded tag names and empty buffers.

Source code in src/cifflow/cifmodel/builder.py
239
240
241
242
243
244
245
246
def on_loop_start(self, tags: list[str]) -> None:
    """Initialise loop tracking with casefolded tag names and empty buffers."""
    if self._stopped:
        return
    self._in_loop = True
    self._loop_tags = [_casefold(t) for t in tags]
    self._loop_value_index = 0
    self._loop_buffers = {_casefold(t): [] for t in tags}

on_loop_end()

Validate row count; pad in pad mode; commit loop to the namespace.

Source code in src/cifflow/cifmodel/builder.py
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
283
284
285
286
def on_loop_end(self) -> None:
    """Validate row count; pad in pad mode; commit loop to the namespace."""
    if self._stopped:
        return
    n = len(self._loop_tags)
    total = self._loop_value_index
    ns = self._current_ns

    if n == 0:
        self._in_loop = False
        return

    if total % n != 0:
        missing = n - (total % n)
        tag_list = ', '.join(self._loop_tags)
        self._semantic_error(
            message=(
                f'loop value count {total} is not divisible by tag count {n} '
                f'({missing} value(s) missing from final row); '
                f'tags: {tag_list}'
            ),
            recovery='stopped' if self._mode == 'strict' else f'padded {missing} placeholder(s)',
        )
        if self._stopped:
            self._in_loop = False
            return
        # Pad mode: fill incomplete final row with '?'
        for _ in range(missing):
            tag = self._loop_tags[self._loop_value_index % n]
            self._loop_buffers[tag].append('?')
            self._loop_value_index += 1

    if ns is not None:
        ns._add_loop(self._loop_tags, self._loop_buffers)

    self._in_loop = False
    self._loop_tags = []
    self._loop_value_index = 0
    self._loop_buffers = {}

on_error(error)

Forward the parse error to the registered on_error callback.

Source code in src/cifflow/cifmodel/builder.py
288
289
290
def on_error(self, error: ParseError) -> None:
    """Forward the parse error to the registered on_error callback."""
    self._on_error(error)

build(source, *, mode='pad')

Parse CIF source text and return a CifFile.

Uses the Rust IR builder directly — no per-token Python callbacks and no intermediate Python dict. Both parser-level and IR-level errors are returned in emission order.

Parameters:

Name Type Description Default
source str

Full CIF source text.

required
mode Literal['strict', 'pad']

Error-handling mode: 'pad' (default) or 'strict'.

'pad'

Returns:

Type Description
tuple[CifFile, list[ParseError]]

A (cif, errors) pair. errors is empty for well-formed input.

Source code in src/cifflow/cifmodel/builder.py
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
def build(
    source: str,
    *,
    mode: Literal['strict', 'pad'] = 'pad',
) -> tuple[CifFile, list[ParseError]]:
    """
    Parse CIF source text and return a CifFile.

    Uses the Rust IR builder directly — no per-token Python callbacks and no
    intermediate Python dict.  Both parser-level and IR-level errors are
    returned in emission order.

    Parameters
    ----------
    source
        Full CIF source text.
    mode
        Error-handling mode: ``'pad'`` (default) or ``'strict'``.

    Returns
    -------
    tuple[CifFile, list[ParseError]]
        A ``(cif, errors)`` pair.  ``errors`` is empty for well-formed input.
    """
    from cifflow import cifflow_core  # noqa: PLC0415
    cif, error_dicts = cifflow_core.parse_cif(source, mode)
    errors = [ParseError(**e) for e in error_dicts]
    return cif, errors

build_arrow(source, *, mode='pad')

Parse CIF source text and return Arrow RecordBatches.

Each batch covers one logical namespace section: the scalar tags of a block/save frame, or one loop within it. The per-batch schema contains five metadata columns plus one column per tag present in that section.

Parameters:

Name Type Description Default
source str

Full CIF source text.

required
mode Literal['strict', 'pad']

Error-handling mode: 'pad' (default) or 'strict'.

'pad'

Returns:

Type Description
tuple[list, list[ParseError]]

A (batches, errors) pair.

Source code in src/cifflow/cifmodel/builder.py
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
def build_arrow(
    source: str,
    *,
    mode: Literal['strict', 'pad'] = 'pad',
) -> tuple[list, list[ParseError]]:
    """
    Parse CIF source text and return Arrow RecordBatches.

    Each batch covers one logical namespace section: the scalar tags of a
    block/save frame, or one loop within it.  The per-batch schema contains
    five metadata columns plus one column per tag present in that section.

    Parameters
    ----------
    source
        Full CIF source text.
    mode
        Error-handling mode: ``'pad'`` (default) or ``'strict'``.

    Returns
    -------
    tuple[list, list[ParseError]]
        A ``(batches, errors)`` pair.
    """
    from cifflow import cifflow_core  # noqa: PLC0415
    batches, error_dicts = cifflow_core.parse_arrow(source, mode)
    errors = [ParseError(**e) for e in error_dicts]
    return batches, errors

build_arrow_file(path, *, mode='pad')

Parse a CIF file from disk and return Arrow RecordBatches.

File I/O is performed entirely in Rust — no Python file objects are created.

Parameters:

Name Type Description Default
path str

Filesystem path to the CIF file.

required
mode Literal['strict', 'pad']

Error-handling mode: 'pad' (default) or 'strict'.

'pad'

Returns:

Type Description
tuple[list, list[ParseError]]

A (batches, errors) pair.

Source code in src/cifflow/cifmodel/builder.py
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
def build_arrow_file(
    path: str,
    *,
    mode: Literal['strict', 'pad'] = 'pad',
) -> tuple[list, list[ParseError]]:
    """
    Parse a CIF file from disk and return Arrow RecordBatches.

    File I/O is performed entirely in Rust — no Python file objects are created.

    Parameters
    ----------
    path
        Filesystem path to the CIF file.
    mode
        Error-handling mode: ``'pad'`` (default) or ``'strict'``.

    Returns
    -------
    tuple[list, list[ParseError]]
        A ``(batches, errors)`` pair.
    """
    from cifflow import cifflow_core  # noqa: PLC0415
    batches, error_dicts = cifflow_core.parse_arrow_file(path, mode)
    errors = [ParseError(**e) for e in error_dicts]
    return batches, errors

Writer

cifflow.cifmodel.writer

Programmatic CIF construction API.

CifWriter — file-level container; manages blocks and holds the CifFile. BlockWriter — write handle for one CifBlock. SaveFrameWriter — write handle for one CifSaveFrame (base class for BlockWriter).

Usage::

writer = CifWriter(CifVersion.CIF_2_0)
block = writer.add_block("my_block")
block.set_tag("_cell_length_a", "5.0")
cif = writer.build()

SaveFrameWriter

Write handle for one CifSaveFrame; obtained via BlockWriter.add_save_frame or get_save_frame.

Source code in src/cifflow/cifmodel/writer.py
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
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
class SaveFrameWriter:
    """Write handle for one CifSaveFrame; obtained via BlockWriter.add_save_frame or get_save_frame."""

    def __init__(self, ns: CifSaveFrame, version: CifVersion) -> None:
        self._ns = ns
        self._version = version

    def _tag_in_any_loop(self, tag: str) -> bool:
        """Return True if tag appears in any loop in this namespace."""
        return any(tag in loop for loop in self._ns._loops)

    # ── Inspection ────────────────────────────────────────────────────────────

    @property
    def tags(self) -> list[str]:
        """Ordered list of tag names present in this namespace."""
        return self._ns.tags

    @property
    def loops(self) -> list[list[str]]:
        """Loop definitions as a list of tag-name groups."""
        return self._ns.loops

    def __getitem__(self, tag: str) -> list:
        return self._ns[_casefold(tag)]

    def get(self, tag: str, default: list | None = None) -> list | None:
        """Return the value list for tag, or default if the tag is absent."""
        try:
            return self._ns[_casefold(tag)]
        except KeyError:
            return default

    # ── Tag / scalar values ───────────────────────────────────────────────────

    def set_tag(self, tag: str, value: CifInput) -> 'SaveFrameWriter':
        """
        Add a new scalar tag with the given value.

        Parameters
        ----------
        tag
            Fully qualified CIF tag name, e.g. ``_cell.length_a``.
        value
            Scalar value or CIF 2.0 container (list or dict).

        Returns
        -------
        SaveFrameWriter
            Returns self for method chaining.

        Raises
        ------
        ValueError
            If the tag name is not a legal CIF identifier for the current
            version, or if the tag already exists in this namespace.
        """
        _check_tag(tag, self._version)
        tag = _casefold(tag)
        if tag in self._ns._tags:
            raise ValueError(
                f"Tag {tag!r} already exists in namespace {self._ns.name!r}"
            )
        self._ns._append_value(tag, _infer(value))
        return self

    # ── Loop values ───────────────────────────────────────────────────────────

    def add_loop(self, columns: dict[str, list[CifInput]]) -> 'SaveFrameWriter':
        """
        Add a new loop with the given columns.

        Parameters
        ----------
        columns
            Mapping of tag name to value list.  All lists must have the same
            length and no tag may already exist in this namespace.

        Returns
        -------
        SaveFrameWriter
            Returns self for method chaining.

        Raises
        ------
        ValueError
            If ``columns`` is empty, any tag name is not a legal CIF identifier,
            any tag already exists, or column lengths differ.
        """
        if not columns:
            raise ValueError("columns must not be empty")
        for tag in columns:
            _check_tag(tag, self._version)
        columns = {_casefold(k): v for k, v in columns.items()}
        tags = list(columns.keys())
        for tag in tags:
            if tag in self._ns._tags:
                raise ValueError(
                    f"Tag {tag!r} already exists in namespace {self._ns.name!r}"
                )
        lengths = [len(v) for v in columns.values()]
        if len(set(lengths)) > 1:
            raise ValueError(
                f"All loop columns must have the same length; got {lengths}"
            )
        buffers: dict[str, list[CifValue]] = {
            tag: _infer_column(vals) for tag, vals in columns.items()
        }
        self._ns._add_loop(tags, buffers)
        return self

    def add_loop_column(
        self,
        loop_tag: str,
        new_tag: str,
        values: list[CifInput],
    ) -> 'SaveFrameWriter':
        """
        Append a new column to an existing loop.

        Parameters
        ----------
        loop_tag
            Any tag already in the target loop (used to identify it).
            Raises ``KeyError`` if not found in any loop.
        new_tag
            Tag name for the new column.
        values
            Value list; length must equal the loop's current row count.

        Returns
        -------
        SaveFrameWriter
            Returns self for method chaining.

        Raises
        ------
        ValueError
            If ``new_tag`` is not a legal CIF identifier, ``new_tag`` already
            exists, or ``values`` length does not match the loop row count.
        """
        loop_idx = _find_loop_index(self._ns, _casefold(loop_tag))
        _check_tag(new_tag, self._version)
        new_tag = _casefold(new_tag)
        if new_tag in self._ns._tags:
            raise ValueError(
                f"Tag {new_tag!r} already exists in namespace {self._ns.name!r}"
            )
        row_count = _loop_row_count(self._ns, loop_idx)
        if len(values) != row_count:
            raise ValueError(
                f"Column length {len(values)} does not match loop row count {row_count}"
            )
        converted = _infer_column(values)
        # Insert into _tag_order immediately after the loop's last tag
        last_tag = self._ns._loops[loop_idx][-1]
        insert_pos = next(
            i for i in range(len(self._ns._tag_order) - 1, -1, -1)
            if self._ns._tag_order[i] == last_tag
        )
        self._ns._tag_order.insert(insert_pos + 1, new_tag)
        self._ns._tags[new_tag] = converted
        self._ns._loops[loop_idx].append(new_tag)
        return self

    def reorder_loop_tags(
        self,
        loop_tag: str,
        new_order: list[str],
    ) -> 'SaveFrameWriter':
        """
        Reorder the columns of an existing loop.

        Parameters
        ----------
        loop_tag
            Any tag already in the target loop (used to identify it).
            Raises ``KeyError`` if not found in any loop.
        new_order
            Complete list of tag names in the desired order; must be a
            permutation of the loop's current tag list.

        Returns
        -------
        SaveFrameWriter
            Returns self for method chaining.

        Raises
        ------
        ValueError
            If ``new_order`` is not a permutation of the current loop tags.
        """
        loop_idx = _find_loop_index(self._ns, _casefold(loop_tag))
        new_order = [_casefold(t) for t in new_order]
        current = self._ns._loops[loop_idx]
        if sorted(new_order) != sorted(current):
            raise ValueError(
                f"new_order {new_order!r} is not a permutation of {current!r}"
            )
        # Find start of contiguous block in _tag_order and overwrite
        start = next(
            i for i, t in enumerate(self._ns._tag_order) if t in current
        )
        for i, t in enumerate(new_order):
            self._ns._tag_order[start + i] = t
        self._ns._loops[loop_idx] = list(new_order)
        return self

    def get_loop_tags(self, loop_tag: str) -> list[str]:
        """Return a copy of the ordered tag list for the loop containing loop_tag.

        Parameters
        ----------
        loop_tag
            Any tag already in the target loop (used to identify it).

        Returns
        -------
        list[str]
            Ordered list of tag names in the loop.  Raises ``KeyError`` if
            ``loop_tag`` is not found in any loop.
        """
        loop_idx = _find_loop_index(self._ns, _casefold(loop_tag))
        return list(self._ns._loops[loop_idx])

    def add_loop_row(
        self,
        loop_tag: str,
        row: list[CifInput],
    ) -> 'SaveFrameWriter':
        """
        Append one row to an existing loop.

        Parameters
        ----------
        loop_tag
            Any tag already in the target loop (used to identify it).
            Raises ``KeyError`` if not found in any loop.
        row
            Ordered list of values matching the loop's tag order.

        Returns
        -------
        SaveFrameWriter
            Returns self for method chaining.

        Raises
        ------
        ValueError
            If ``row`` length does not match the loop's tag count.
        """
        loop_idx = _find_loop_index(self._ns, _casefold(loop_tag))
        loop_tags = self._ns._loops[loop_idx]
        if len(row) != len(loop_tags):
            raise ValueError(
                f"Row length {len(row)} does not match loop tag count {len(loop_tags)}"
            )
        for tag, val in zip(loop_tags, _infer_column(row)):
            self._ns._tags[tag].append(val)
        return self

    # ── Mutation — tags ───────────────────────────────────────────────────────

    def reassign_tag(
        self,
        tag: str,
        value: 'CifInput | list[CifInput]',
    ) -> 'SaveFrameWriter':
        """
        Replace the value(s) of an existing tag.

        Parameters
        ----------
        tag
            Tag name to update.
        value
            New value.  For loop columns, pass a list of values with the same
            length as the loop row count.

        Returns
        -------
        SaveFrameWriter
            Returns self for method chaining.

        Raises
        ------
        KeyError
            If the tag does not exist in this namespace.
        ValueError
            If the tag is a loop column and ``value`` is not a list.
        """
        tag = _casefold(tag)
        if tag not in self._ns._tags:
            raise KeyError(tag)
        if self._tag_in_any_loop(tag):
            if not isinstance(value, list):
                raise ValueError(
                    f"Tag {tag!r} is a loop column; value must be a list"
                )
            self._ns._tags[tag] = _infer_column(value)
        else:
            if isinstance(value, list) and len(value) == 1:
                value = value[0]
            self._ns._tags[tag] = [_infer(value)]
        return self

    def delete_tag(self, tag: str) -> 'SaveFrameWriter':
        """
        Delete a tag; removes it from its loop if it is a loop column.

        Parameters
        ----------
        tag
            Tag name to delete.

        Returns
        -------
        SaveFrameWriter
            Returns self for method chaining.

        Raises
        ------
        KeyError
            If the tag does not exist in this namespace.
        """
        tag = _casefold(tag)
        if tag not in self._ns._tags:
            raise KeyError(tag)
        if self._tag_in_any_loop(tag):
            loop_idx = _find_loop_index(self._ns, tag)
            self._remove_loop_tag_impl(loop_idx, tag)
        else:
            del self._ns._tags[tag]
            self._ns._tag_order.remove(tag)
        return self

    # ── Mutation — loops ──────────────────────────────────────────────────────

    def remove_loop_tag(
        self,
        loop_tag: str,
        tag_to_remove: str,
    ) -> 'SaveFrameWriter':
        """
        Remove one tag from a loop, deleting the loop entirely if it becomes empty.

        Parameters
        ----------
        loop_tag
            Any tag already in the target loop (used to identify it).
        tag_to_remove
            Tag name to remove from the loop.

        Returns
        -------
        SaveFrameWriter
            Returns self for method chaining.

        Raises
        ------
        KeyError
            If ``loop_tag`` is not found in any loop, or if ``tag_to_remove``
            is not in the identified loop.
        """
        loop_idx = _find_loop_index(self._ns, _casefold(loop_tag))
        tag_to_remove = _casefold(tag_to_remove)
        if tag_to_remove not in self._ns._loops[loop_idx]:
            raise KeyError(tag_to_remove)
        self._remove_loop_tag_impl(loop_idx, tag_to_remove)
        return self

    def _remove_loop_tag_impl(self, loop_idx: int, tag: str) -> None:
        """Remove tag from loop and tags dicts; delete the loop list if now empty."""
        self._ns._loops[loop_idx].remove(tag)
        del self._ns._tags[tag]
        self._ns._tag_order.remove(tag)
        if not self._ns._loops[loop_idx]:
            del self._ns._loops[loop_idx]

tags property

Ordered list of tag names present in this namespace.

loops property

Loop definitions as a list of tag-name groups.

get(tag, default=None)

Return the value list for tag, or default if the tag is absent.

Source code in src/cifflow/cifmodel/writer.py
160
161
162
163
164
165
def get(self, tag: str, default: list | None = None) -> list | None:
    """Return the value list for tag, or default if the tag is absent."""
    try:
        return self._ns[_casefold(tag)]
    except KeyError:
        return default

set_tag(tag, value)

Add a new scalar tag with the given value.

Parameters:

Name Type Description Default
tag str

Fully qualified CIF tag name, e.g. _cell.length_a.

required
value CifInput

Scalar value or CIF 2.0 container (list or dict).

required

Returns:

Type Description
SaveFrameWriter

Returns self for method chaining.

Raises:

Type Description
ValueError

If the tag name is not a legal CIF identifier for the current version, or if the tag already exists in this namespace.

Source code in src/cifflow/cifmodel/writer.py
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
def set_tag(self, tag: str, value: CifInput) -> 'SaveFrameWriter':
    """
    Add a new scalar tag with the given value.

    Parameters
    ----------
    tag
        Fully qualified CIF tag name, e.g. ``_cell.length_a``.
    value
        Scalar value or CIF 2.0 container (list or dict).

    Returns
    -------
    SaveFrameWriter
        Returns self for method chaining.

    Raises
    ------
    ValueError
        If the tag name is not a legal CIF identifier for the current
        version, or if the tag already exists in this namespace.
    """
    _check_tag(tag, self._version)
    tag = _casefold(tag)
    if tag in self._ns._tags:
        raise ValueError(
            f"Tag {tag!r} already exists in namespace {self._ns.name!r}"
        )
    self._ns._append_value(tag, _infer(value))
    return self

add_loop(columns)

Add a new loop with the given columns.

Parameters:

Name Type Description Default
columns dict[str, list[CifInput]]

Mapping of tag name to value list. All lists must have the same length and no tag may already exist in this namespace.

required

Returns:

Type Description
SaveFrameWriter

Returns self for method chaining.

Raises:

Type Description
ValueError

If columns is empty, any tag name is not a legal CIF identifier, any tag already exists, or column lengths differ.

Source code in src/cifflow/cifmodel/writer.py
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
def add_loop(self, columns: dict[str, list[CifInput]]) -> 'SaveFrameWriter':
    """
    Add a new loop with the given columns.

    Parameters
    ----------
    columns
        Mapping of tag name to value list.  All lists must have the same
        length and no tag may already exist in this namespace.

    Returns
    -------
    SaveFrameWriter
        Returns self for method chaining.

    Raises
    ------
    ValueError
        If ``columns`` is empty, any tag name is not a legal CIF identifier,
        any tag already exists, or column lengths differ.
    """
    if not columns:
        raise ValueError("columns must not be empty")
    for tag in columns:
        _check_tag(tag, self._version)
    columns = {_casefold(k): v for k, v in columns.items()}
    tags = list(columns.keys())
    for tag in tags:
        if tag in self._ns._tags:
            raise ValueError(
                f"Tag {tag!r} already exists in namespace {self._ns.name!r}"
            )
    lengths = [len(v) for v in columns.values()]
    if len(set(lengths)) > 1:
        raise ValueError(
            f"All loop columns must have the same length; got {lengths}"
        )
    buffers: dict[str, list[CifValue]] = {
        tag: _infer_column(vals) for tag, vals in columns.items()
    }
    self._ns._add_loop(tags, buffers)
    return self

add_loop_column(loop_tag, new_tag, values)

Append a new column to an existing loop.

Parameters:

Name Type Description Default
loop_tag str

Any tag already in the target loop (used to identify it). Raises KeyError if not found in any loop.

required
new_tag str

Tag name for the new column.

required
values list[CifInput]

Value list; length must equal the loop's current row count.

required

Returns:

Type Description
SaveFrameWriter

Returns self for method chaining.

Raises:

Type Description
ValueError

If new_tag is not a legal CIF identifier, new_tag already exists, or values length does not match the loop row count.

Source code in src/cifflow/cifmodel/writer.py
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
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
def add_loop_column(
    self,
    loop_tag: str,
    new_tag: str,
    values: list[CifInput],
) -> 'SaveFrameWriter':
    """
    Append a new column to an existing loop.

    Parameters
    ----------
    loop_tag
        Any tag already in the target loop (used to identify it).
        Raises ``KeyError`` if not found in any loop.
    new_tag
        Tag name for the new column.
    values
        Value list; length must equal the loop's current row count.

    Returns
    -------
    SaveFrameWriter
        Returns self for method chaining.

    Raises
    ------
    ValueError
        If ``new_tag`` is not a legal CIF identifier, ``new_tag`` already
        exists, or ``values`` length does not match the loop row count.
    """
    loop_idx = _find_loop_index(self._ns, _casefold(loop_tag))
    _check_tag(new_tag, self._version)
    new_tag = _casefold(new_tag)
    if new_tag in self._ns._tags:
        raise ValueError(
            f"Tag {new_tag!r} already exists in namespace {self._ns.name!r}"
        )
    row_count = _loop_row_count(self._ns, loop_idx)
    if len(values) != row_count:
        raise ValueError(
            f"Column length {len(values)} does not match loop row count {row_count}"
        )
    converted = _infer_column(values)
    # Insert into _tag_order immediately after the loop's last tag
    last_tag = self._ns._loops[loop_idx][-1]
    insert_pos = next(
        i for i in range(len(self._ns._tag_order) - 1, -1, -1)
        if self._ns._tag_order[i] == last_tag
    )
    self._ns._tag_order.insert(insert_pos + 1, new_tag)
    self._ns._tags[new_tag] = converted
    self._ns._loops[loop_idx].append(new_tag)
    return self

reorder_loop_tags(loop_tag, new_order)

Reorder the columns of an existing loop.

Parameters:

Name Type Description Default
loop_tag str

Any tag already in the target loop (used to identify it). Raises KeyError if not found in any loop.

required
new_order list[str]

Complete list of tag names in the desired order; must be a permutation of the loop's current tag list.

required

Returns:

Type Description
SaveFrameWriter

Returns self for method chaining.

Raises:

Type Description
ValueError

If new_order is not a permutation of the current loop tags.

Source code in src/cifflow/cifmodel/writer.py
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
def reorder_loop_tags(
    self,
    loop_tag: str,
    new_order: list[str],
) -> 'SaveFrameWriter':
    """
    Reorder the columns of an existing loop.

    Parameters
    ----------
    loop_tag
        Any tag already in the target loop (used to identify it).
        Raises ``KeyError`` if not found in any loop.
    new_order
        Complete list of tag names in the desired order; must be a
        permutation of the loop's current tag list.

    Returns
    -------
    SaveFrameWriter
        Returns self for method chaining.

    Raises
    ------
    ValueError
        If ``new_order`` is not a permutation of the current loop tags.
    """
    loop_idx = _find_loop_index(self._ns, _casefold(loop_tag))
    new_order = [_casefold(t) for t in new_order]
    current = self._ns._loops[loop_idx]
    if sorted(new_order) != sorted(current):
        raise ValueError(
            f"new_order {new_order!r} is not a permutation of {current!r}"
        )
    # Find start of contiguous block in _tag_order and overwrite
    start = next(
        i for i, t in enumerate(self._ns._tag_order) if t in current
    )
    for i, t in enumerate(new_order):
        self._ns._tag_order[start + i] = t
    self._ns._loops[loop_idx] = list(new_order)
    return self

get_loop_tags(loop_tag)

Return a copy of the ordered tag list for the loop containing loop_tag.

Parameters:

Name Type Description Default
loop_tag str

Any tag already in the target loop (used to identify it).

required

Returns:

Type Description
list[str]

Ordered list of tag names in the loop. Raises KeyError if loop_tag is not found in any loop.

Source code in src/cifflow/cifmodel/writer.py
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
def get_loop_tags(self, loop_tag: str) -> list[str]:
    """Return a copy of the ordered tag list for the loop containing loop_tag.

    Parameters
    ----------
    loop_tag
        Any tag already in the target loop (used to identify it).

    Returns
    -------
    list[str]
        Ordered list of tag names in the loop.  Raises ``KeyError`` if
        ``loop_tag`` is not found in any loop.
    """
    loop_idx = _find_loop_index(self._ns, _casefold(loop_tag))
    return list(self._ns._loops[loop_idx])

add_loop_row(loop_tag, row)

Append one row to an existing loop.

Parameters:

Name Type Description Default
loop_tag str

Any tag already in the target loop (used to identify it). Raises KeyError if not found in any loop.

required
row list[CifInput]

Ordered list of values matching the loop's tag order.

required

Returns:

Type Description
SaveFrameWriter

Returns self for method chaining.

Raises:

Type Description
ValueError

If row length does not match the loop's tag count.

Source code in src/cifflow/cifmodel/writer.py
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
def add_loop_row(
    self,
    loop_tag: str,
    row: list[CifInput],
) -> 'SaveFrameWriter':
    """
    Append one row to an existing loop.

    Parameters
    ----------
    loop_tag
        Any tag already in the target loop (used to identify it).
        Raises ``KeyError`` if not found in any loop.
    row
        Ordered list of values matching the loop's tag order.

    Returns
    -------
    SaveFrameWriter
        Returns self for method chaining.

    Raises
    ------
    ValueError
        If ``row`` length does not match the loop's tag count.
    """
    loop_idx = _find_loop_index(self._ns, _casefold(loop_tag))
    loop_tags = self._ns._loops[loop_idx]
    if len(row) != len(loop_tags):
        raise ValueError(
            f"Row length {len(row)} does not match loop tag count {len(loop_tags)}"
        )
    for tag, val in zip(loop_tags, _infer_column(row)):
        self._ns._tags[tag].append(val)
    return self

reassign_tag(tag, value)

Replace the value(s) of an existing tag.

Parameters:

Name Type Description Default
tag str

Tag name to update.

required
value 'CifInput | list[CifInput]'

New value. For loop columns, pass a list of values with the same length as the loop row count.

required

Returns:

Type Description
SaveFrameWriter

Returns self for method chaining.

Raises:

Type Description
KeyError

If the tag does not exist in this namespace.

ValueError

If the tag is a loop column and value is not a list.

Source code in src/cifflow/cifmodel/writer.py
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
def reassign_tag(
    self,
    tag: str,
    value: 'CifInput | list[CifInput]',
) -> 'SaveFrameWriter':
    """
    Replace the value(s) of an existing tag.

    Parameters
    ----------
    tag
        Tag name to update.
    value
        New value.  For loop columns, pass a list of values with the same
        length as the loop row count.

    Returns
    -------
    SaveFrameWriter
        Returns self for method chaining.

    Raises
    ------
    KeyError
        If the tag does not exist in this namespace.
    ValueError
        If the tag is a loop column and ``value`` is not a list.
    """
    tag = _casefold(tag)
    if tag not in self._ns._tags:
        raise KeyError(tag)
    if self._tag_in_any_loop(tag):
        if not isinstance(value, list):
            raise ValueError(
                f"Tag {tag!r} is a loop column; value must be a list"
            )
        self._ns._tags[tag] = _infer_column(value)
    else:
        if isinstance(value, list) and len(value) == 1:
            value = value[0]
        self._ns._tags[tag] = [_infer(value)]
    return self

delete_tag(tag)

Delete a tag; removes it from its loop if it is a loop column.

Parameters:

Name Type Description Default
tag str

Tag name to delete.

required

Returns:

Type Description
SaveFrameWriter

Returns self for method chaining.

Raises:

Type Description
KeyError

If the tag does not exist in this namespace.

Source code in src/cifflow/cifmodel/writer.py
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
def delete_tag(self, tag: str) -> 'SaveFrameWriter':
    """
    Delete a tag; removes it from its loop if it is a loop column.

    Parameters
    ----------
    tag
        Tag name to delete.

    Returns
    -------
    SaveFrameWriter
        Returns self for method chaining.

    Raises
    ------
    KeyError
        If the tag does not exist in this namespace.
    """
    tag = _casefold(tag)
    if tag not in self._ns._tags:
        raise KeyError(tag)
    if self._tag_in_any_loop(tag):
        loop_idx = _find_loop_index(self._ns, tag)
        self._remove_loop_tag_impl(loop_idx, tag)
    else:
        del self._ns._tags[tag]
        self._ns._tag_order.remove(tag)
    return self

remove_loop_tag(loop_tag, tag_to_remove)

Remove one tag from a loop, deleting the loop entirely if it becomes empty.

Parameters:

Name Type Description Default
loop_tag str

Any tag already in the target loop (used to identify it).

required
tag_to_remove str

Tag name to remove from the loop.

required

Returns:

Type Description
SaveFrameWriter

Returns self for method chaining.

Raises:

Type Description
KeyError

If loop_tag is not found in any loop, or if tag_to_remove is not in the identified loop.

Source code in src/cifflow/cifmodel/writer.py
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
def remove_loop_tag(
    self,
    loop_tag: str,
    tag_to_remove: str,
) -> 'SaveFrameWriter':
    """
    Remove one tag from a loop, deleting the loop entirely if it becomes empty.

    Parameters
    ----------
    loop_tag
        Any tag already in the target loop (used to identify it).
    tag_to_remove
        Tag name to remove from the loop.

    Returns
    -------
    SaveFrameWriter
        Returns self for method chaining.

    Raises
    ------
    KeyError
        If ``loop_tag`` is not found in any loop, or if ``tag_to_remove``
        is not in the identified loop.
    """
    loop_idx = _find_loop_index(self._ns, _casefold(loop_tag))
    tag_to_remove = _casefold(tag_to_remove)
    if tag_to_remove not in self._ns._loops[loop_idx]:
        raise KeyError(tag_to_remove)
    self._remove_loop_tag_impl(loop_idx, tag_to_remove)
    return self

BlockWriter

Bases: SaveFrameWriter

Write handle for one CifBlock; obtained via CifWriter.add_block or get_block.

Source code in src/cifflow/cifmodel/writer.py
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
class BlockWriter(SaveFrameWriter):
    """Write handle for one CifBlock; obtained via CifWriter.add_block or get_block."""

    def __init__(self, block: CifBlock, version: CifVersion) -> None:
        super().__init__(block, version)
        self._block = block

    # ── Covariant return overrides ────────────────────────────────────────────

    def set_tag(self, tag: str, value: CifInput) -> 'BlockWriter':
        """Add a new scalar tag; returns BlockWriter for method chaining."""
        super().set_tag(tag, value)
        return self

    def add_loop(self, columns: dict[str, list[CifInput]]) -> 'BlockWriter':
        """Add a new loop; returns BlockWriter for method chaining."""
        super().add_loop(columns)
        return self

    def add_loop_column(self, loop_tag: str, new_tag: str, values: list[CifInput]) -> 'BlockWriter':
        """Append a column to an existing loop; returns BlockWriter for method chaining."""
        super().add_loop_column(loop_tag, new_tag, values)
        return self

    def reorder_loop_tags(self, loop_tag: str, new_order: list[str]) -> 'BlockWriter':
        """Reorder loop columns; returns BlockWriter for method chaining."""
        super().reorder_loop_tags(loop_tag, new_order)
        return self

    def add_loop_row(self, loop_tag: str, row: list[CifInput]) -> 'BlockWriter':
        """Append a loop row; returns BlockWriter for method chaining."""
        super().add_loop_row(loop_tag, row)
        return self

    def reassign_tag(self, tag: str, value: 'CifInput | list[CifInput]') -> 'BlockWriter':
        """Replace a tag value; returns BlockWriter for method chaining."""
        super().reassign_tag(tag, value)
        return self

    def delete_tag(self, tag: str) -> 'BlockWriter':
        """Delete a tag; returns BlockWriter for method chaining."""
        super().delete_tag(tag)
        return self

    def remove_loop_tag(self, loop_tag: str, tag_to_remove: str) -> 'BlockWriter':
        """Remove a loop column; returns BlockWriter for method chaining."""
        super().remove_loop_tag(loop_tag, tag_to_remove)
        return self

    # ── Inspection ────────────────────────────────────────────────────────────

    @property
    def save_frames(self) -> list[str]:
        """Ordered list of save frame names in this block."""
        return self._block.save_frames

    # ── Save frame management ─────────────────────────────────────────────────

    def add_save_frame(self, name: str) -> SaveFrameWriter:
        """
        Add a new save frame to this block and return its writer.

        Parameters
        ----------
        name
            Save frame name (without the ``save_`` prefix).

        Returns
        -------
        SaveFrameWriter
            Writer handle for the new save frame.

        Raises
        ------
        ValueError
            If ``name`` is not a legal CIF identifier for the current version,
            or if a save frame with that name already exists in the block.
        """
        _check_name(name, self._version, "save-frame")
        name = _casefold(name)
        if name in self._block._save_frames:
            raise ValueError(
                f"Save frame {name!r} already exists in block {self._block.name!r}"
            )
        frame = CifSaveFrame(name)
        self._block._add_save_frame(frame)
        return SaveFrameWriter(frame, self._version)

    def get_save_frame(self, name: str, index: int = 0) -> SaveFrameWriter:
        """
        Return a writer handle for an existing save frame.

        Parameters
        ----------
        name
            Save frame name to look up (case-insensitive).
        index
            Which occurrence to return when there are duplicate names; 0-based.

        Returns
        -------
        SaveFrameWriter
            Writer handle for the requested save frame.

        Raises
        ------
        KeyError
            If no save frame with that name exists in the block.
        """
        name = _casefold(name)
        matches = [sf for sf in self._block._save_frame_list if sf.name == name]
        if not matches:
            raise KeyError(name)
        return SaveFrameWriter(matches[index], self._version)

    def remove_save_frame(self, name: str, *, from_end: bool = False) -> 'BlockWriter':
        """
        Remove one save frame from this block.

        Parameters
        ----------
        name
            Save frame name to remove (case-insensitive).
        from_end
            If ``True``, remove the last occurrence when names are duplicated;
            otherwise remove the first occurrence.

        Returns
        -------
        BlockWriter
            Returns self for method chaining.

        Raises
        ------
        KeyError
            If no save frame with that name exists in the block.
        """
        name = _casefold(name)
        if name not in self._block._save_frames:
            raise KeyError(name)
        lst = self._block._save_frame_list
        if from_end:
            idx = max(i for i, sf in enumerate(lst) if sf.name == name)
        else:
            idx = next(i for i, sf in enumerate(lst) if sf.name == name)
        lst.pop(idx)
        survivors = [sf for sf in lst if sf.name == name]
        if survivors:
            self._block._save_frames[name] = survivors[0]
        else:
            del self._block._save_frames[name]
        return self

    def rename_save_frame(self, old_name: str, new_name: str) -> 'BlockWriter':
        """
        Rename a save frame.

        Parameters
        ----------
        old_name
            Current save frame name (case-insensitive).
        new_name
            New save frame name.

        Returns
        -------
        BlockWriter
            Returns self for method chaining.

        Raises
        ------
        KeyError
            If ``old_name`` does not exist in the block.
        ValueError
            If ``new_name`` is not a legal CIF identifier for the current
            version, or if ``new_name`` already exists in the block.
        """
        old_name = _casefold(old_name)
        if old_name not in self._block._save_frames:
            raise KeyError(old_name)
        _check_name(new_name, self._version, "save-frame")
        new_name = _casefold(new_name)
        if new_name in self._block._save_frames:
            raise ValueError(
                f"Save frame {new_name!r} already exists in block {self._block.name!r}"
            )
        frame = self._block._save_frames.pop(old_name)
        frame.name = new_name
        self._block._save_frames[new_name] = frame
        return self

save_frames property

Ordered list of save frame names in this block.

set_tag(tag, value)

Add a new scalar tag; returns BlockWriter for method chaining.

Source code in src/cifflow/cifmodel/writer.py
527
528
529
530
def set_tag(self, tag: str, value: CifInput) -> 'BlockWriter':
    """Add a new scalar tag; returns BlockWriter for method chaining."""
    super().set_tag(tag, value)
    return self

add_loop(columns)

Add a new loop; returns BlockWriter for method chaining.

Source code in src/cifflow/cifmodel/writer.py
532
533
534
535
def add_loop(self, columns: dict[str, list[CifInput]]) -> 'BlockWriter':
    """Add a new loop; returns BlockWriter for method chaining."""
    super().add_loop(columns)
    return self

add_loop_column(loop_tag, new_tag, values)

Append a column to an existing loop; returns BlockWriter for method chaining.

Source code in src/cifflow/cifmodel/writer.py
537
538
539
540
def add_loop_column(self, loop_tag: str, new_tag: str, values: list[CifInput]) -> 'BlockWriter':
    """Append a column to an existing loop; returns BlockWriter for method chaining."""
    super().add_loop_column(loop_tag, new_tag, values)
    return self

reorder_loop_tags(loop_tag, new_order)

Reorder loop columns; returns BlockWriter for method chaining.

Source code in src/cifflow/cifmodel/writer.py
542
543
544
545
def reorder_loop_tags(self, loop_tag: str, new_order: list[str]) -> 'BlockWriter':
    """Reorder loop columns; returns BlockWriter for method chaining."""
    super().reorder_loop_tags(loop_tag, new_order)
    return self

add_loop_row(loop_tag, row)

Append a loop row; returns BlockWriter for method chaining.

Source code in src/cifflow/cifmodel/writer.py
547
548
549
550
def add_loop_row(self, loop_tag: str, row: list[CifInput]) -> 'BlockWriter':
    """Append a loop row; returns BlockWriter for method chaining."""
    super().add_loop_row(loop_tag, row)
    return self

reassign_tag(tag, value)

Replace a tag value; returns BlockWriter for method chaining.

Source code in src/cifflow/cifmodel/writer.py
552
553
554
555
def reassign_tag(self, tag: str, value: 'CifInput | list[CifInput]') -> 'BlockWriter':
    """Replace a tag value; returns BlockWriter for method chaining."""
    super().reassign_tag(tag, value)
    return self

delete_tag(tag)

Delete a tag; returns BlockWriter for method chaining.

Source code in src/cifflow/cifmodel/writer.py
557
558
559
560
def delete_tag(self, tag: str) -> 'BlockWriter':
    """Delete a tag; returns BlockWriter for method chaining."""
    super().delete_tag(tag)
    return self

remove_loop_tag(loop_tag, tag_to_remove)

Remove a loop column; returns BlockWriter for method chaining.

Source code in src/cifflow/cifmodel/writer.py
562
563
564
565
def remove_loop_tag(self, loop_tag: str, tag_to_remove: str) -> 'BlockWriter':
    """Remove a loop column; returns BlockWriter for method chaining."""
    super().remove_loop_tag(loop_tag, tag_to_remove)
    return self

add_save_frame(name)

Add a new save frame to this block and return its writer.

Parameters:

Name Type Description Default
name str

Save frame name (without the save_ prefix).

required

Returns:

Type Description
SaveFrameWriter

Writer handle for the new save frame.

Raises:

Type Description
ValueError

If name is not a legal CIF identifier for the current version, or if a save frame with that name already exists in the block.

Source code in src/cifflow/cifmodel/writer.py
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
def add_save_frame(self, name: str) -> SaveFrameWriter:
    """
    Add a new save frame to this block and return its writer.

    Parameters
    ----------
    name
        Save frame name (without the ``save_`` prefix).

    Returns
    -------
    SaveFrameWriter
        Writer handle for the new save frame.

    Raises
    ------
    ValueError
        If ``name`` is not a legal CIF identifier for the current version,
        or if a save frame with that name already exists in the block.
    """
    _check_name(name, self._version, "save-frame")
    name = _casefold(name)
    if name in self._block._save_frames:
        raise ValueError(
            f"Save frame {name!r} already exists in block {self._block.name!r}"
        )
    frame = CifSaveFrame(name)
    self._block._add_save_frame(frame)
    return SaveFrameWriter(frame, self._version)

get_save_frame(name, index=0)

Return a writer handle for an existing save frame.

Parameters:

Name Type Description Default
name str

Save frame name to look up (case-insensitive).

required
index int

Which occurrence to return when there are duplicate names; 0-based.

0

Returns:

Type Description
SaveFrameWriter

Writer handle for the requested save frame.

Raises:

Type Description
KeyError

If no save frame with that name exists in the block.

Source code in src/cifflow/cifmodel/writer.py
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
def get_save_frame(self, name: str, index: int = 0) -> SaveFrameWriter:
    """
    Return a writer handle for an existing save frame.

    Parameters
    ----------
    name
        Save frame name to look up (case-insensitive).
    index
        Which occurrence to return when there are duplicate names; 0-based.

    Returns
    -------
    SaveFrameWriter
        Writer handle for the requested save frame.

    Raises
    ------
    KeyError
        If no save frame with that name exists in the block.
    """
    name = _casefold(name)
    matches = [sf for sf in self._block._save_frame_list if sf.name == name]
    if not matches:
        raise KeyError(name)
    return SaveFrameWriter(matches[index], self._version)

remove_save_frame(name, *, from_end=False)

Remove one save frame from this block.

Parameters:

Name Type Description Default
name str

Save frame name to remove (case-insensitive).

required
from_end bool

If True, remove the last occurrence when names are duplicated; otherwise remove the first occurrence.

False

Returns:

Type Description
BlockWriter

Returns self for method chaining.

Raises:

Type Description
KeyError

If no save frame with that name exists in the block.

Source code in src/cifflow/cifmodel/writer.py
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
def remove_save_frame(self, name: str, *, from_end: bool = False) -> 'BlockWriter':
    """
    Remove one save frame from this block.

    Parameters
    ----------
    name
        Save frame name to remove (case-insensitive).
    from_end
        If ``True``, remove the last occurrence when names are duplicated;
        otherwise remove the first occurrence.

    Returns
    -------
    BlockWriter
        Returns self for method chaining.

    Raises
    ------
    KeyError
        If no save frame with that name exists in the block.
    """
    name = _casefold(name)
    if name not in self._block._save_frames:
        raise KeyError(name)
    lst = self._block._save_frame_list
    if from_end:
        idx = max(i for i, sf in enumerate(lst) if sf.name == name)
    else:
        idx = next(i for i, sf in enumerate(lst) if sf.name == name)
    lst.pop(idx)
    survivors = [sf for sf in lst if sf.name == name]
    if survivors:
        self._block._save_frames[name] = survivors[0]
    else:
        del self._block._save_frames[name]
    return self

rename_save_frame(old_name, new_name)

Rename a save frame.

Parameters:

Name Type Description Default
old_name str

Current save frame name (case-insensitive).

required
new_name str

New save frame name.

required

Returns:

Type Description
BlockWriter

Returns self for method chaining.

Raises:

Type Description
KeyError

If old_name does not exist in the block.

ValueError

If new_name is not a legal CIF identifier for the current version, or if new_name already exists in the block.

Source code in src/cifflow/cifmodel/writer.py
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
def rename_save_frame(self, old_name: str, new_name: str) -> 'BlockWriter':
    """
    Rename a save frame.

    Parameters
    ----------
    old_name
        Current save frame name (case-insensitive).
    new_name
        New save frame name.

    Returns
    -------
    BlockWriter
        Returns self for method chaining.

    Raises
    ------
    KeyError
        If ``old_name`` does not exist in the block.
    ValueError
        If ``new_name`` is not a legal CIF identifier for the current
        version, or if ``new_name`` already exists in the block.
    """
    old_name = _casefold(old_name)
    if old_name not in self._block._save_frames:
        raise KeyError(old_name)
    _check_name(new_name, self._version, "save-frame")
    new_name = _casefold(new_name)
    if new_name in self._block._save_frames:
        raise ValueError(
            f"Save frame {new_name!r} already exists in block {self._block.name!r}"
        )
    frame = self._block._save_frames.pop(old_name)
    frame.name = new_name
    self._block._save_frames[new_name] = frame
    return self

CifWriter

File-level container for programmatic CIF construction.

Parameters:

Name Type Description Default
version CifVersion

CIF specification version to validate new names and values against.

required
cif CifFile | None

Existing :class:~cifflow.cifflow_core.CifFile to wrap for editing. If None, a new empty :class:~cifflow.cifflow_core.CifFile is created.

None
Source code in src/cifflow/cifmodel/writer.py
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
class CifWriter:
    """
    File-level container for programmatic CIF construction.

    Parameters
    ----------
    version
        CIF specification version to validate new names and values against.
    cif
        Existing :class:`~cifflow.cifflow_core.CifFile` to wrap for editing.
        If ``None``, a new empty :class:`~cifflow.cifflow_core.CifFile` is created.
    """

    def __init__(
        self,
        version: CifVersion,
        cif: CifFile | None = None,
    ) -> None:
        self._version = version
        if cif is not None:
            if cif.version == CifVersion.CIF_2_0 and version == CifVersion.CIF_1_1:
                warnings.warn(
                    "Wrapping a CIF 2.0 file with a CIF 1.1 writer — existing "
                    "CIF 2.0 constructs will not be removed but new ones will be rejected.",
                    UserWarning,
                    stacklevel=2,
                )
            self._file = cif
        else:
            self._file = CifFile(version=version)

    @property
    def version(self) -> CifVersion:
        """CIF specification version this writer validates new names against."""
        return self._version

    # ── Inspection ────────────────────────────────────────────────────────────

    @property
    def blocks(self) -> list[str]:
        """Ordered list of block names in this CifWriter."""
        return self._file.blocks

    # ── Read access ───────────────────────────────────────────────────────────

    def __getitem__(self, name: str) -> CifBlock:
        return self._file[name]

    def __contains__(self, name: str) -> bool:
        return name in self._file

    def get(self, name: str, default: CifBlock | None = None) -> CifBlock | None:
        """Return the CifBlock for name, or default if the block is absent."""
        if name in self._file:
            return self._file[name]
        return default

    # ── Block management ──────────────────────────────────────────────────────

    def add_block(self, name: str) -> BlockWriter:
        """
        Add a new block to this CifFile and return its writer.

        Parameters
        ----------
        name
            Block name (without the ``data_`` prefix).

        Returns
        -------
        BlockWriter
            Writer handle for the new block.

        Raises
        ------
        ValueError
            If ``name`` is not a legal CIF identifier for the current version,
            or if a block with that name already exists.
        """
        _check_name(name, self._version, "block")
        name = _casefold(name)
        if name in self._file:
            raise ValueError(f"Block {name!r} already exists in this CifWriter")
        block = CifBlock(name)
        self._file._add_block(block)
        return BlockWriter(block, self._version)

    def get_block(self, name: str, index: int = 0) -> BlockWriter:
        """
        Return a writer handle for an existing block.

        Parameters
        ----------
        name
            Block name to look up (case-insensitive).
        index
            Which occurrence to return when names are duplicated; 0-based.

        Returns
        -------
        BlockWriter
            Writer handle for the requested block.

        Raises
        ------
        KeyError
            If no block with that name exists.
        """
        name = _casefold(name)
        matches = [b for b in self._file._block_list if b.name == name]
        if not matches:
            raise KeyError(name)
        return BlockWriter(matches[index], self._version)

    def remove_block(self, name: str, *, from_end: bool = False) -> 'CifWriter':
        """
        Remove one block from this CifFile.

        Parameters
        ----------
        name
            Block name to remove (case-insensitive).
        from_end
            If ``True``, remove the last occurrence when names are duplicated;
            otherwise remove the first occurrence.

        Returns
        -------
        CifWriter
            Returns self for method chaining.

        Raises
        ------
        KeyError
            If no block with that name exists.
        """
        name = _casefold(name)
        if name not in self._file:
            raise KeyError(name)
        lst = self._file._block_list
        if from_end:
            idx = max(i for i, b in enumerate(lst) if b.name == name)
        else:
            idx = next(i for i, b in enumerate(lst) if b.name == name)
        lst.pop(idx)
        survivors = [b for b in lst if b.name == name]
        if survivors:
            self._file._blocks[name] = survivors[0]
        else:
            del self._file._blocks[name]
        return self

    def rename_block(self, old_name: str, new_name: str) -> 'CifWriter':
        """
        Rename a block.

        Parameters
        ----------
        old_name
            Current block name (case-insensitive).
        new_name
            New block name.

        Returns
        -------
        CifWriter
            Returns self for method chaining.

        Raises
        ------
        KeyError
            If ``old_name`` does not exist.
        ValueError
            If ``new_name`` is not a legal CIF identifier for the current
            version, or if ``new_name`` already exists.
        """
        old_name = _casefold(old_name)
        if old_name not in self._file:
            raise KeyError(old_name)
        _check_name(new_name, self._version, "block")
        new_name = _casefold(new_name)
        if new_name in self._file:
            raise ValueError(f"Block {new_name!r} already exists in this CifWriter")
        block = self._file._blocks.pop(old_name)
        block.name = new_name
        self._file._blocks[new_name] = block
        return self

    # ── Result ────────────────────────────────────────────────────────────────

    def build(self) -> CifFile:
        """
        Validate and return the completed CifFile.

        Returns
        -------
        CifFile
            The constructed CifFile, ready for ingestion or emission.

        Raises
        ------
        ValueError
            If the file is empty, any loop has unequal column lengths or zero
            rows, any scalar tag has a value count other than 1, or any tag
            holds a container value in CIF 1.1 mode.
        """
        if not self._file._block_list:
            raise ValueError("CifFile must contain at least one block")
        errors: list[str] = []
        for block in self._file._block_list:
            _validate_namespace(block, self._version, errors)
            for sf in block._save_frame_list:
                _validate_namespace(sf, self._version, errors)
        if errors:
            raise ValueError(
                "CifWriter.build() validation failed:\n" + "\n".join(errors)
            )
        return self._file

version property

CIF specification version this writer validates new names against.

blocks property

Ordered list of block names in this CifWriter.

get(name, default=None)

Return the CifBlock for name, or default if the block is absent.

Source code in src/cifflow/cifmodel/writer.py
765
766
767
768
769
def get(self, name: str, default: CifBlock | None = None) -> CifBlock | None:
    """Return the CifBlock for name, or default if the block is absent."""
    if name in self._file:
        return self._file[name]
    return default

add_block(name)

Add a new block to this CifFile and return its writer.

Parameters:

Name Type Description Default
name str

Block name (without the data_ prefix).

required

Returns:

Type Description
BlockWriter

Writer handle for the new block.

Raises:

Type Description
ValueError

If name is not a legal CIF identifier for the current version, or if a block with that name already exists.

Source code in src/cifflow/cifmodel/writer.py
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
def add_block(self, name: str) -> BlockWriter:
    """
    Add a new block to this CifFile and return its writer.

    Parameters
    ----------
    name
        Block name (without the ``data_`` prefix).

    Returns
    -------
    BlockWriter
        Writer handle for the new block.

    Raises
    ------
    ValueError
        If ``name`` is not a legal CIF identifier for the current version,
        or if a block with that name already exists.
    """
    _check_name(name, self._version, "block")
    name = _casefold(name)
    if name in self._file:
        raise ValueError(f"Block {name!r} already exists in this CifWriter")
    block = CifBlock(name)
    self._file._add_block(block)
    return BlockWriter(block, self._version)

get_block(name, index=0)

Return a writer handle for an existing block.

Parameters:

Name Type Description Default
name str

Block name to look up (case-insensitive).

required
index int

Which occurrence to return when names are duplicated; 0-based.

0

Returns:

Type Description
BlockWriter

Writer handle for the requested block.

Raises:

Type Description
KeyError

If no block with that name exists.

Source code in src/cifflow/cifmodel/writer.py
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
def get_block(self, name: str, index: int = 0) -> BlockWriter:
    """
    Return a writer handle for an existing block.

    Parameters
    ----------
    name
        Block name to look up (case-insensitive).
    index
        Which occurrence to return when names are duplicated; 0-based.

    Returns
    -------
    BlockWriter
        Writer handle for the requested block.

    Raises
    ------
    KeyError
        If no block with that name exists.
    """
    name = _casefold(name)
    matches = [b for b in self._file._block_list if b.name == name]
    if not matches:
        raise KeyError(name)
    return BlockWriter(matches[index], self._version)

remove_block(name, *, from_end=False)

Remove one block from this CifFile.

Parameters:

Name Type Description Default
name str

Block name to remove (case-insensitive).

required
from_end bool

If True, remove the last occurrence when names are duplicated; otherwise remove the first occurrence.

False

Returns:

Type Description
CifWriter

Returns self for method chaining.

Raises:

Type Description
KeyError

If no block with that name exists.

Source code in src/cifflow/cifmodel/writer.py
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
def remove_block(self, name: str, *, from_end: bool = False) -> 'CifWriter':
    """
    Remove one block from this CifFile.

    Parameters
    ----------
    name
        Block name to remove (case-insensitive).
    from_end
        If ``True``, remove the last occurrence when names are duplicated;
        otherwise remove the first occurrence.

    Returns
    -------
    CifWriter
        Returns self for method chaining.

    Raises
    ------
    KeyError
        If no block with that name exists.
    """
    name = _casefold(name)
    if name not in self._file:
        raise KeyError(name)
    lst = self._file._block_list
    if from_end:
        idx = max(i for i, b in enumerate(lst) if b.name == name)
    else:
        idx = next(i for i, b in enumerate(lst) if b.name == name)
    lst.pop(idx)
    survivors = [b for b in lst if b.name == name]
    if survivors:
        self._file._blocks[name] = survivors[0]
    else:
        del self._file._blocks[name]
    return self

rename_block(old_name, new_name)

Rename a block.

Parameters:

Name Type Description Default
old_name str

Current block name (case-insensitive).

required
new_name str

New block name.

required

Returns:

Type Description
CifWriter

Returns self for method chaining.

Raises:

Type Description
KeyError

If old_name does not exist.

ValueError

If new_name is not a legal CIF identifier for the current version, or if new_name already exists.

Source code in src/cifflow/cifmodel/writer.py
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
def rename_block(self, old_name: str, new_name: str) -> 'CifWriter':
    """
    Rename a block.

    Parameters
    ----------
    old_name
        Current block name (case-insensitive).
    new_name
        New block name.

    Returns
    -------
    CifWriter
        Returns self for method chaining.

    Raises
    ------
    KeyError
        If ``old_name`` does not exist.
    ValueError
        If ``new_name`` is not a legal CIF identifier for the current
        version, or if ``new_name`` already exists.
    """
    old_name = _casefold(old_name)
    if old_name not in self._file:
        raise KeyError(old_name)
    _check_name(new_name, self._version, "block")
    new_name = _casefold(new_name)
    if new_name in self._file:
        raise ValueError(f"Block {new_name!r} already exists in this CifWriter")
    block = self._file._blocks.pop(old_name)
    block.name = new_name
    self._file._blocks[new_name] = block
    return self

build()

Validate and return the completed CifFile.

Returns:

Type Description
CifFile

The constructed CifFile, ready for ingestion or emission.

Raises:

Type Description
ValueError

If the file is empty, any loop has unequal column lengths or zero rows, any scalar tag has a value count other than 1, or any tag holds a container value in CIF 1.1 mode.

Source code in src/cifflow/cifmodel/writer.py
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
def build(self) -> CifFile:
    """
    Validate and return the completed CifFile.

    Returns
    -------
    CifFile
        The constructed CifFile, ready for ingestion or emission.

    Raises
    ------
    ValueError
        If the file is empty, any loop has unequal column lengths or zero
        rows, any scalar tag has a value count other than 1, or any tag
        holds a container value in CIF 1.1 mode.
    """
    if not self._file._block_list:
        raise ValueError("CifFile must contain at least one block")
    errors: list[str] = []
    for block in self._file._block_list:
        _validate_namespace(block, self._version, errors)
        for sf in block._save_frame_list:
            _validate_namespace(sf, self._version, errors)
    if errors:
        raise ValueError(
            "CifWriter.build() validation failed:\n" + "\n".join(errors)
        )
    return self._file

Clean

cifflow.cifmodel.clean

Cleaning API for parser-produced CifFile objects.

clean() removes well-known parse-time artefacts: - orphan error values (_cifflow_error_value synthetic tag) - duplicate blocks, save frames, and scalar tags - loop padding added by CifBuilder in pad mode

Returns a new CifFile (copy=True, default) or mutates in place (copy=False). Every removal produces a CleanWarning — nothing is silently discarded.

CleanWarning dataclass

Record of one removal action performed by :func:clean.

Attributes:

Name Type Description
category str

Cleaning step name, e.g. 'deduplicate_blocks'.

block str | None

Name of the affected block, or None if not block-scoped.

save_frame str | None

Name of the affected save frame, or None if not save-frame-scoped.

message str

Human-readable description of what was removed.

Source code in src/cifflow/cifmodel/clean.py
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
@dataclass
class CleanWarning:
    """Record of one removal action performed by :func:`clean`.

    Attributes
    ----------
    category
        Cleaning step name, e.g. ``'deduplicate_blocks'``.
    block
        Name of the affected block, or ``None`` if not block-scoped.
    save_frame
        Name of the affected save frame, or ``None`` if not save-frame-scoped.
    message
        Human-readable description of what was removed.
    """

    category: str
    block: str | None
    save_frame: str | None
    message: str

clean(cif, *, copy=True, remove_error_values=True, deduplicate_blocks='first', deduplicate_save_frames='first', deduplicate_tags='first', strip_loop_padding=True)

Remove common parse-time artefacts from a CifFile.

Parameters:

Name Type Description Default
cif CifFile

The source CifFile to clean.

required
copy bool

If True (default), operate on a deep copy and leave the original unchanged. If False, mutate the input in place and return it.

True
remove_error_values bool

Remove orphan _cifflow_error_value synthetic tags left by the parser's error-recovery path.

True
deduplicate_blocks Keep | Literal[False]

'first' or 'last' — which duplicate block to keep — or False to skip deduplication.

'first'
deduplicate_save_frames Keep | Literal[False]

'first' or 'last' — which duplicate save frame to keep — or False to skip deduplication.

'first'
deduplicate_tags Keep | Literal[False]

'first' or 'last' — which duplicate scalar tag value to keep — or False to skip deduplication.

'first'
strip_loop_padding bool

Remove trailing '?' placeholder rows added by :class:CifBuilder in pad mode to complete incomplete loop rows.

True

Returns:

Type Description
tuple[CifFile, list[CleanWarning]]

A (cleaned_cif, warnings) pair. Each :class:CleanWarning describes one removal action.

Source code in src/cifflow/cifmodel/clean.py
 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
def clean(
    cif: CifFile,
    *,
    copy: bool = True,
    remove_error_values: bool = True,
    deduplicate_blocks: Keep | Literal[False] = 'first',
    deduplicate_save_frames: Keep | Literal[False] = 'first',
    deduplicate_tags: Keep | Literal[False] = 'first',
    strip_loop_padding: bool = True,
) -> tuple[CifFile, list[CleanWarning]]:
    """
    Remove common parse-time artefacts from a CifFile.

    Parameters
    ----------
    cif
        The source CifFile to clean.
    copy
        If ``True`` (default), operate on a deep copy and leave the original
        unchanged.  If ``False``, mutate the input in place and return it.
    remove_error_values
        Remove orphan ``_cifflow_error_value`` synthetic tags left by the
        parser's error-recovery path.
    deduplicate_blocks
        ``'first'`` or ``'last'`` — which duplicate block to keep — or
        ``False`` to skip deduplication.
    deduplicate_save_frames
        ``'first'`` or ``'last'`` — which duplicate save frame to keep — or
        ``False`` to skip deduplication.
    deduplicate_tags
        ``'first'`` or ``'last'`` — which duplicate scalar tag value to keep
        — or ``False`` to skip deduplication.
    strip_loop_padding
        Remove trailing ``'?'`` placeholder rows added by :class:`CifBuilder`
        in pad mode to complete incomplete loop rows.

    Returns
    -------
    tuple[CifFile, list[CleanWarning]]
        A ``(cleaned_cif, warnings)`` pair.  Each :class:`CleanWarning`
        describes one removal action.
    """
    target = cif.deepcopy() if copy else cif
    out: list[CleanWarning] = []
    writer = CifWriter(version=target.version, cif=target)

    if remove_error_values:
        _step_remove_error_values(target, writer, out)

    if deduplicate_blocks is not False:
        _step_deduplicate_blocks(target, writer, deduplicate_blocks, out)

    if deduplicate_save_frames is not False:
        _step_deduplicate_save_frames(target, writer, deduplicate_save_frames, out)

    if deduplicate_tags is not False:
        _step_deduplicate_tags(target, writer, deduplicate_tags, out)

    if strip_loop_padding:
        _step_strip_loop_padding(target, writer, out)

    return target, out