diff --git a/src/biotite/structure/io/pdbx/cif.py b/src/biotite/structure/io/pdbx/cif.py index 94efc6db3..97638b476 100644 --- a/src/biotite/structure/io/pdbx/cif.py +++ b/src/biotite/structure/io/pdbx/cif.py @@ -4,7 +4,6 @@ __author__ = "Patrick Kunzmann" __all__ = ["CIFFile", "CIFBlock", "CIFCategory", "CIFColumn", "CIFData"] -import itertools from collections.abc import Iterator, Mapping, MutableMapping, Sequence from os import PathLike from typing import IO, Any, Self, TypeAlias, overload @@ -16,6 +15,11 @@ is_open_compatible, is_text, ) +from biotite.rust.structure.io.pdbx import ( + deserialize_category, + split_blocks, + split_categories, +) from biotite.structure.io.pdbx.component import MaskValue, _Component from biotite.typing import NDArray1 @@ -386,24 +390,8 @@ def supercomponent_class() -> type[CIFBlock]: @classmethod def deserialize(cls, text: str) -> Self: - lines = [line.strip() for line in text.splitlines() if not _is_empty(line)] - - if _is_loop_start(lines[0]): - is_looped = True - lines.pop(0) - else: - is_looped = False - - category_name = _parse_category_name(lines[0]) - if category_name is None: - raise DeserializationError("Failed to parse category name") - - lines = _to_single(lines) - if is_looped: - category_dict = CIFCategory._deserialize_looped(lines) - else: - category_dict = CIFCategory._deserialize_single(lines) - return cls(category_dict, category_name) + category_name, columns = deserialize_category(text) + return cls(dict(columns), category_name) def serialize(self) -> str: if self._name is None: @@ -468,79 +456,6 @@ def __eq__(self, other: object) -> bool: return False return True - @staticmethod - def _deserialize_single(lines: list[str]) -> dict[str, CIFColumn]: - """ - Process a category where each field has a single value. - """ - category_dict = {} - line_i = 0 - while line_i < len(lines): - line = lines[line_i] - parts = list(_split_one_line(line)) - if len(parts) == 2: - # Standard case -> name and value in one line - name_part, value_part = parts - line_i += 1 - elif len(parts) == 1: - # Value is a multiline value on the next line - name_part = parts[0] - parts = list(_split_one_line(lines[line_i + 1])) - if len(parts) == 1: - value_part = parts[0] - else: - raise DeserializationError(f"Failed to parse line '{line}'") - line_i += 2 - elif len(parts) == 0: - raise DeserializationError("Empty line within category") - else: - raise DeserializationError(f"Failed to parse line '{line}'") - category_dict[name_part.split(".")[1]] = CIFColumn(value_part) - return category_dict - - @staticmethod - def _deserialize_looped(lines: list[str]) -> dict[str, CIFColumn]: - """ - Process a category where each field has multiple values - (category is a table). - """ - category_dict = {} - column_names = [] - i = 0 - for key_line in lines: - key_line = key_line.strip() - if key_line[0] == "_": - # Key line - key = key_line.split(".")[1] - column_names.append(key) - category_dict[key] = [] - i += 1 - else: - break - - data_lines = lines[i:] - # Rows may be split over multiple lines -> do not rely on - # row-line-alignment at all and simply cycle through columns - column_indices = itertools.cycle(range(len(column_names))) - for data_line in data_lines: - values = _split_one_line(data_line) - for val in values: - column_index = next(column_indices) - column_name = column_names[column_index] - category_dict[column_name].append(val) - - # Check if all columns have the same length - # Otherwise, this would indicate a parsing error or an invalid CIF file - column_index = next(column_indices) - if column_index != 0: - raise DeserializationError( - "Category contains columns with different lengths" - ) - - for key, column in category_dict.items(): - category_dict[key] = CIFColumn(column) - return category_dict - def _serialize_single(self) -> list[str]: if self._name is None: raise SerializationError("Category name is required") @@ -665,29 +580,7 @@ def supercomponent_class() -> type[CIFFile]: @classmethod def deserialize(cls, text: str) -> Self: - lines = text.splitlines() - current_category_name = None - category_starts = [] - category_names = [] - for i, line in enumerate(lines): - if not _is_empty(line): - is_loop_in_line = _is_loop_start(line) - category_name_in_line = _parse_category_name(line) - if is_loop_in_line or ( - category_name_in_line != current_category_name - and category_name_in_line is not None - ): - # Track the new category - if is_loop_in_line: - # In case of lines with "loop_" the category is - # in the next line - category_name_in_line = _parse_category_name(lines[i + 1]) - current_category_name = category_name_in_line - category_starts.append(i) - category_names.append(current_category_name) - return cls( - _create_element_dict(lines, category_names, category_starts) # pyright: ignore[reportArgumentType] - ) + return cls(dict(split_categories(text))) def serialize(self) -> str: if self._name is None: @@ -858,18 +751,7 @@ def supercomponent_class() -> None: @classmethod def deserialize(cls, text: str) -> Self: - lines = text.splitlines() - block_starts = [] - block_names = [] - for i, line in enumerate(lines): - if not _is_empty(line): - data_block_name = _parse_data_block_name(line) - if data_block_name is not None: - block_starts.append(i) - block_names.append(data_block_name) - return cls( - _create_element_dict(lines, block_names, block_starts) # pyright: ignore[reportArgumentType] - ) + return cls(dict(split_blocks(text))) def serialize(self) -> str: text_blocks = [] @@ -977,91 +859,6 @@ def __eq__(self, other: object) -> bool: return True -def _is_empty(line: str) -> bool: - return len(line.strip()) == 0 or line[0] == "#" - - -def _create_element_dict( - lines: list[str], element_names: list[str], element_starts: list[int] -) -> dict[str, str]: - """ - Create a dict mapping the `element_names` to the corresponding - `lines`, which are located between ``element_starts[i]`` and - ``element_starts[i+1]``. - """ - # Add exclusive stop to indices for easier slicing - element_starts.append(len(lines)) - # Lazy deserialization - # -> keep as text for now and deserialize later if needed - return { - element_name: "\n".join(lines[element_starts[i] : element_starts[i + 1]]) + "\n" - for i, element_name in enumerate(element_names) - } - - -def _parse_data_block_name(line: str) -> str | None: - """ - If the line defines a data block, return this name. - Return ``None`` otherwise. - """ - if line.startswith("data_"): - return line[5:] - else: - return None - - -def _parse_category_name(line: str) -> str | None: - """ - If the line defines a category, return this name. - Return ``None`` otherwise. - """ - line = line.strip() - if line[0] != "_": - return None - else: - return line[1 : line.find(".")] - - -def _is_loop_start(line: str) -> bool: - """ - Return whether the line starts a looped category. - """ - return line.startswith("loop_") - - -def _to_single(lines: list[str]) -> list[str]: - r""" - Convert multiline values into singleline values - (in terms of 'lines' list elements). - Linebreaks are preserved as ``'\n'`` characters within a list element. - The initial ``';'`` character is also preserved, while the final ``';'`` character - is removed. - """ - processed_lines = [] - in_multi_line = False - mutli_line_value = [] - for line in lines: - # Multiline value are enclosed by ';' at the start of the beginning and end line - if line[0] == ";": - if not in_multi_line: - # Start of multiline value - in_multi_line = True - mutli_line_value.append(line) - else: - # End of multiline value - in_multi_line = False - # The current line contains only the end character ';' - # Hence this line is not added to the processed lines - processed_lines.append("\n".join(mutli_line_value)) - mutli_line_value = [] - else: - if in_multi_line: - mutli_line_value.append(line) - else: - processed_lines.append(line) - return processed_lines - - def _escape(array: NDArray1[Any, np.str_]) -> NDArray1[Any, np.str_]: """ Escape special characters in values to make them compatible with CIF. @@ -1121,40 +918,6 @@ def _multiline( return "\n;" + value + "\n;\n" -def _split_one_line(line: str) -> Iterator[str]: - """ - Split a line into its fields. - Supporting embedded quotes (' or "), like `'a dog's life'` to `a dog's life` - """ - # Special case of multiline value, where the line starts with ';' - if line[0] == ";": - yield line[1:] - elif "'" in line or '"' in line: - # Quoted values in the line - while line: - # Strip leading whitespace(s) - stripped_line = line.lstrip() - # Split the line on whitespace - word, _, line = stripped_line.partition(" ") - # Handle the case where the word start with a quote - if word.startswith(("'", '"')): - # Set the separator to the quote found - separator = word[0] - # Handle the case of a quoted word without space - if word.endswith(separator) and len(word) > 1: - # Yield the word without the opening and closing quotes - yield word[1:-1] - continue - # split the word on the separator - word, _, line = stripped_line[1:].partition(separator) - - yield word - else: - # No quoted values in the line -> simple whitespace split - for line in line.split(): - yield line - - def _arrayfy(data: Any) -> NDArray1[Any, Any]: if not isinstance(data, (Sequence, np.ndarray)) or isinstance(data, str): data = [data] diff --git a/src/rust/structure/io/pdbx/cif.rs b/src/rust/structure/io/pdbx/cif.rs new file mode 100644 index 000000000..1e66002a7 --- /dev/null +++ b/src/rust/structure/io/pdbx/cif.rs @@ -0,0 +1,413 @@ +use numpy::{IntoPyArray, PyArrayDescr}; +use pyo3::exceptions::PyIndexError; +use pyo3::prelude::*; +use pyo3::types::PyString; +use std::borrow::Cow; + +// Label as a separate module to indicate that this exception comes +// from biotite +mod biotite { + pyo3::import_exception!(biotite.file, DeserializationError); +} + +type Columns<'py> = Vec<(String, Bound<'py, PyAny>)>; + +/// Split the text of a CIF file into its data blocks. +/// +/// Return the name and text of each block, in the order of the file. +#[pyfunction] +pub fn split_blocks(text: &Bound<'_, PyString>) -> PyResult> { + let lines = split_lines(to_str(text)?); + let mut block_names = Vec::new(); + let mut block_starts = Vec::new(); + for (i, line) in lines.iter().enumerate() { + if let Some(block_name) = line.strip_prefix("data_") { + block_starts.push(i); + block_names.push(block_name.to_string()); + } + } + Ok(join_elements(&lines, block_names, &block_starts)) +} + +/// Split the text of a CIF block into its categories. +/// +/// Return the name and text of each category, in the order of the block. +#[pyfunction] +pub fn split_categories(text: &Bound<'_, PyString>) -> PyResult, String)>> { + let lines = split_lines(to_str(text)?); + let mut current_category_name = None; + let mut category_names = Vec::new(); + let mut category_starts = Vec::new(); + for (i, line) in lines.iter().enumerate() { + if is_empty(line) { + continue; + } + let is_loop_in_line = line.starts_with("loop_"); + let mut category_name_in_line = parse_category_name(line); + if is_loop_in_line + || (category_name_in_line != current_category_name && category_name_in_line.is_some()) + { + // Track the new category + if is_loop_in_line { + // In case of lines with "loop_" the category is + // in the next line + let next_line = lines.get(i + 1).ok_or_else(index_error)?; + if next_line.trim_matches(is_space).is_empty() { + return Err(PyIndexError::new_err("string index out of range")); + } + category_name_in_line = parse_category_name(next_line); + } + current_category_name = category_name_in_line; + category_starts.push(i); + category_names.push(current_category_name.map(str::to_string)); + } + } + Ok(join_elements(&lines, category_names, &category_starts)) +} + +/// Deserialize the text of a CIF category into its name and columns. +/// +/// Each column is a NumPy unicode array, as ``np.asarray()`` would create it from +/// the list of value strings. +#[pyfunction] +pub fn deserialize_category<'py>( + py: Python<'py>, + text: &Bound<'py, PyString>, +) -> PyResult<(String, Columns<'py>)> { + let mut lines: Vec<&str> = split_lines(to_str(text)?) + .into_iter() + .filter(|line| !line.starts_with('#')) + .map(|line| line.trim_matches(is_space)) + .filter(|line| !line.is_empty()) + .collect(); + let first_line = *lines.first().ok_or_else(index_error)?; + let is_looped = first_line.starts_with("loop_"); + if is_looped { + lines.remove(0); + } + let category_name = parse_category_name(lines.first().ok_or_else(index_error)?) + .ok_or_else(|| biotite::DeserializationError::new_err("Failed to parse category name"))?; + + let lines = to_single(&lines); + let columns = if is_looped { + deserialize_looped(py, &lines)? + } else { + deserialize_single(py, &lines)? + }; + Ok((category_name.to_string(), columns)) +} + +/// Split a line into its values. +/// +/// Supports embedded quotes (' or "), like ``'a dog's life'`` to ``a dog's life``. +#[pyfunction] +pub fn split_one_line(line: &str) -> Vec<&str> { + let mut values = Vec::new(); + split_values(line, |value| values.push(value)); + values +} + +/// Process a category where each field has a single value. +fn deserialize_single<'py>(py: Python<'py>, lines: &[Cow<'_, str>]) -> PyResult> { + let mut columns = Vec::new(); + let mut line_i = 0; + while line_i < lines.len() { + let line = &lines[line_i]; + let parts = split_one_line(line); + let (name_part, value_part) = match parts[..] { + // Standard case -> name and value in one line + [name_part, value_part] => { + line_i += 1; + (name_part, value_part) + } + // Value is a multiline value on the next line + [name_part] => { + let next_line = lines.get(line_i + 1).ok_or_else(index_error)?; + let [value_part] = split_one_line(next_line)[..] else { + return Err(parse_error(line)); + }; + line_i += 2; + (name_part, value_part) + } + [] => { + return Err(biotite::DeserializationError::new_err( + "Empty line within category", + )) + } + _ => return Err(parse_error(line)), + }; + let width = value_part.chars().count().max(1); + let mut buffer = vec![0; width]; + write_code_points(&mut buffer, value_part); + // A repeated key is resolved by the `dict()` call on the Python side + columns.push(( + column_name(name_part)?.to_string(), + unicode_array(py, buffer, width)?, + )); + } + Ok(columns) +} + +/// Process a category where each field has multiple values +/// (category is a table). +fn deserialize_looped<'py>(py: Python<'py>, lines: &[Cow<'_, str>]) -> PyResult> { + let mut column_names = Vec::new(); + // Maps the position of a key line to its column, + // which differ only if a column name is repeated + let mut column_of_key = Vec::new(); + for key_line in lines.iter().take_while(|line| line.starts_with('_')) { + let key = column_name(key_line)?; + let column_i = match column_names.iter().position(|name| *name == key) { + Some(i) => i, + None => { + column_names.push(key); + column_names.len() - 1 + } + }; + column_of_key.push(column_i); + } + let data_lines = &lines[column_of_key.len()..]; + + // First pass: count the values and find the widest value of each column + let mut lengths = vec![0; column_names.len()]; + let mut widths = vec![1; column_names.len()]; + let key_i = for_each_value(data_lines, &column_of_key, |column_i, value| { + lengths[column_i] += 1; + widths[column_i] = widths[column_i].max(value.chars().count()); + }); + // Check if all columns have the same length + // Otherwise, this would indicate a parsing error or an invalid CIF file + if key_i != 0 { + return Err(biotite::DeserializationError::new_err( + "Category contains columns with different lengths", + )); + } + + // Second pass: write each value directly into its array, + // so the values do not need to be kept until all widths are known + let mut buffers: Vec> = lengths + .iter() + .zip(&widths) + .map(|(length, width)| vec![0; length * width]) + .collect(); + let mut row_counts = vec![0; column_names.len()]; + for_each_value(data_lines, &column_of_key, |column_i, value| { + let start = row_counts[column_i] * widths[column_i]; + write_code_points( + &mut buffers[column_i][start..start + widths[column_i]], + value, + ); + row_counts[column_i] += 1; + }); + + column_names + .into_iter() + .zip(buffers) + .zip(widths) + .map(|((name, buffer), width)| Ok((name.to_string(), unicode_array(py, buffer, width)?))) + .collect() +} + +/// Call `f(column_i, value)` for each value in the data lines of a looped category. +/// +/// Rows may be split over multiple lines -> do not rely on +/// row-line-alignment at all and simply cycle through columns. +/// Return the key index after the last value, which is 0 if all rows are complete. +fn for_each_value<'a>( + data_lines: &'a [Cow<'_, str>], + column_of_key: &[usize], + mut f: impl FnMut(usize, &'a str), +) -> usize { + let mut key_i = 0; + for data_line in data_lines { + split_values(data_line, |value| { + f(column_of_key[key_i], value); + key_i = (key_i + 1) % column_of_key.len(); + }); + } + key_i +} + +/// Call `emit` for each value in the line. +fn split_values<'a>(line: &'a str, mut emit: impl FnMut(&'a str)) { + // Special case of multiline value, where the line starts with ';' + if let Some(value) = line.strip_prefix(';') { + emit(value); + } else if line.contains(['\'', '"']) { + // Quoted values in the line + let mut line = line; + while !line.is_empty() { + let stripped_line = line.trim_start_matches(is_space); + // Split the line on whitespace + let (mut word, mut rest) = partition(stripped_line, ' '); + // Handle the case where the word starts with a quote + if let Some(separator) = word.chars().next().filter(|&c| c == '\'' || c == '"') { + // Handle the case of a quoted word without space + if word.len() > 1 && word.ends_with(separator) { + // Emit the word without the opening and closing quotes + emit(&word[1..word.len() - 1]); + line = rest; + continue; + } + (word, rest) = partition(&stripped_line[1..], separator); + } + emit(word); + line = rest; + } + } else { + // No quoted values in the line -> simple whitespace split + line.split(is_space) + .filter(|value| !value.is_empty()) + .for_each(emit); + } +} + +/// Convert multiline values into singleline values +/// (in terms of 'lines' list elements). +/// Linebreaks are preserved as ``'\n'`` characters within a list element. +/// The initial ``';'`` character is also preserved, while the final ``';'`` character +/// is removed. +fn to_single<'a>(lines: &[&'a str]) -> Vec> { + let mut processed_lines = Vec::new(); + let mut multi_line_value: Option> = None; + for &line in lines { + // Multiline value are enclosed by ';' at the start of the beginning and end line + if line.starts_with(';') { + match multi_line_value.take() { + // Start of multiline value + None => multi_line_value = Some(vec![line]), + // End of multiline value + // The current line contains only the end character ';' + // Hence this line is not added to the processed lines + Some(value_lines) => processed_lines.push(Cow::Owned(value_lines.join("\n"))), + } + } else if let Some(value_lines) = multi_line_value.as_mut() { + value_lines.push(line); + } else { + processed_lines.push(Cow::Borrowed(line)); + } + } + processed_lines +} + +/// Join the lines of each element (block or category) into its text, where element `i` +/// spans from ``starts[i]`` to the start of the next element. +fn join_elements(lines: &[&str], names: Vec, starts: &[usize]) -> Vec<(N, String)> { + names + .into_iter() + .enumerate() + .map(|(i, name)| { + let stop = starts.get(i + 1).copied().unwrap_or(lines.len()); + let mut text = lines[starts[i]..stop].join("\n"); + text.push('\n'); + (name, text) + }) + .collect() +} + +/// Write the code points of `value` into the start of `row`. +/// The rest of `row` keeps its zeros, which NumPy reads as padding. +fn write_code_points(row: &mut [u32], value: &str) { + for (code_point, c) in row.iter_mut().zip(value.chars()) { + *code_point = c as u32; + } +} + +/// View a buffer of UCS4 code points as a NumPy unicode array with the given width, +/// without copying. +fn unicode_array<'py>( + py: Python<'py>, + buffer: Vec, + width: usize, +) -> PyResult> { + let dtype = PyArrayDescr::new(py, format!("U{width}"))?; + buffer.into_pyarray(py).call_method1("view", (dtype,)) +} + +/// Borrow the text of a Python string. +fn to_str<'a>(text: &'a Bound<'_, PyString>) -> PyResult<&'a str> { + // Fails only for lone surrogates, e.g. from a file opened with + // `errors="surrogateescape"` + text.to_str().map_err(|_| { + biotite::DeserializationError::new_err( + "CIF text contains lone surrogate characters, which are not valid Unicode", + ) + }) +} + +/// Equivalent to Python's ``str.isspace()`` for a single character. +fn is_space(c: char) -> bool { + c.is_whitespace() || ('\x1c'..='\x1f').contains(&c) +} + +/// Equivalent to Python's ``str.splitlines()``. +fn split_lines(text: &str) -> Vec<&str> { + let mut lines = Vec::new(); + let mut start = 0; + let mut chars = text.char_indices().peekable(); + while let Some((i, c)) = chars.next() { + if matches!( + c, + '\n' | '\r' + | '\x0b' + | '\x0c' + | '\x1c' + | '\x1d' + | '\x1e' + | '\u{85}' + | '\u{2028}' + | '\u{2029}' + ) { + lines.push(&text[start..i]); + start = i + c.len_utf8(); + if c == '\r' && chars.next_if(|&(_, next)| next == '\n').is_some() { + start += 1; + } + } + } + if start < text.len() { + lines.push(&text[start..]); + } + lines +} + +/// Equivalent to Python's ``str.partition()`` with a single character separator. +fn partition(text: &str, separator: char) -> (&str, &str) { + match text.find(separator) { + Some(i) => (&text[..i], &text[i + separator.len_utf8()..]), + None => (text, ""), + } +} + +fn is_empty(line: &str) -> bool { + line.trim_matches(is_space).is_empty() || line.starts_with('#') +} + +/// If the line defines a category, return this name. +/// Return ``None`` otherwise. +fn parse_category_name(line: &str) -> Option<&str> { + // Leading whitespace is allowed, e.g. in YASARA files + let line = line.trim_matches(is_space); + if !line.starts_with('_') { + return None; + } + // Like ``line[1 : line.find(".")]`` in Python, where a missing '.' gives -1 + let end = line + .find('.') + .unwrap_or_else(|| line.char_indices().last().map_or(0, |(i, _)| i)); + Some(line.get(1..end).unwrap_or("")) +} + +/// The part of ``_category.column`` after the first '.'. +fn column_name(name_part: &str) -> PyResult<&str> { + name_part.split('.').nth(1).ok_or_else(index_error) +} + +fn parse_error(line: &str) -> PyErr { + biotite::DeserializationError::new_err(format!("Failed to parse line '{line}'")) +} + +/// The error Python raises for an index beyond the end of a list. +fn index_error() -> PyErr { + PyIndexError::new_err("list index out of range") +} diff --git a/src/rust/structure/io/pdbx/mod.rs b/src/rust/structure/io/pdbx/mod.rs index 97397b0c5..e89280b91 100644 --- a/src/rust/structure/io/pdbx/mod.rs +++ b/src/rust/structure/io/pdbx/mod.rs @@ -1,11 +1,16 @@ use pyo3::prelude::*; +pub mod cif; pub mod encoding; pub mod matching; pub fn module<'py>(parent_module: &Bound<'py, PyModule>) -> PyResult> { let module = PyModule::new(parent_module.py(), "pdbx")?; module.add_function(wrap_pyfunction!(matching::find_matches, &module)?)?; + module.add_function(wrap_pyfunction!(cif::deserialize_category, &module)?)?; + module.add_function(wrap_pyfunction!(cif::split_blocks, &module)?)?; + module.add_function(wrap_pyfunction!(cif::split_categories, &module)?)?; + module.add_function(wrap_pyfunction!(cif::split_one_line, &module)?)?; module.add_function(wrap_pyfunction!(encoding::run_length_encode, &module)?)?; module.add_function(wrap_pyfunction!(encoding::run_length_decode, &module)?)?; module.add_function(wrap_pyfunction!(encoding::integer_packing_encode, &module)?)?; diff --git a/tests/structure/io/test_pdbx.py b/tests/structure/io/test_pdbx.py index 68c5ba364..c7ad6f893 100644 --- a/tests/structure/io/test_pdbx.py +++ b/tests/structure/io/test_pdbx.py @@ -8,6 +8,7 @@ import biotite.structure as struc import biotite.structure.io.pdb as pdb import biotite.structure.io.pdbx as pdbx +from biotite.rust.structure.io.pdbx import split_one_line from biotite.structure.io.pdbx.bcif import _encode_numpy as encode_numpy from biotite.structure.io.pdbx.compress import _get_decimal_places as get_decimal_places from tests.util import data_dir @@ -102,7 +103,7 @@ def test_split_one_line(cif_line, expected_fields): """ Test whether values that have an embedded quote are properly escaped. """ - assert list(pdbx.cif._split_one_line(cif_line)) == expected_fields + assert list(split_one_line(cif_line)) == expected_fields @pytest.mark.parametrize("model", [None, 1, -1])