Skip to content

Input Module

This module handles typed options and molecular-system input processing.

Typed Runner Options

config

Strict, source-aware loading for runner options YAML.

InputValidationError

Bases: ValueError

Base class for actionable errors raised before backend execution.

Source code in src/cofolder/modules/input/config.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
class InputValidationError(ValueError):
    """Base class for actionable errors raised before backend execution."""

    error_code = "input_validation_failed"

    def __init__(
        self,
        message: str,
        *,
        source_path: Path | str | None = None,
        field_path: tuple[FieldToken, ...] = (),
        line: int | None = None,
        column: int | None = None,
        entity_id: str | None = None,
        chain_id: str | None = None,
        source_record_id: str | None = None,
        residue_position: int | None = None,
        character: str | None = None,
    ) -> None:
        super().__init__(message)
        self.source_path = Path(source_path) if source_path is not None else None
        self.field_path = tuple(field_path)
        self.line = line
        self.column = column
        self.entity_id = entity_id
        self.chain_id = chain_id
        self.source_record_id = source_record_id
        self.residue_position = residue_position
        self.character = character

    @property
    def details(self) -> dict[str, JSONValue]:
        details: dict[str, JSONValue] = {}
        if self.source_path is not None:
            details["source_path"] = str(self.source_path)
        if self.field_path:
            details["field_path"] = list(self.field_path)
        if self.line is not None:
            details["line"] = self.line
        if self.column is not None:
            details["column"] = self.column
        if self.entity_id is not None:
            details["entity_id"] = self.entity_id
        if self.chain_id is not None:
            details["chain_id"] = self.chain_id
        if self.source_record_id is not None:
            details["source_record_id"] = self.source_record_id
        if self.residue_position is not None:
            details["residue_position"] = self.residue_position
        if self.character is not None:
            details["character"] = self.character
        return details

System

system

System configuration management for Boltz predictions.

This module provides the System class for managing molecular system configurations, including proteins, ligands, and their properties.

SystemChain dataclass

One expanded chain from a system sequence entry.

Source code in src/cofolder/modules/input/system.py 
17
18
19
20
21
22
23
24
@dataclass(frozen=True, slots=True)
class SystemChain:
    """One expanded chain from a system sequence entry."""

    chain_id: str
    entity_type: str
    entity_data: dict[str, Any]
    sequence_index: int

System

Manage molecular system configuration for Boltz predictions.

The System class provides methods to load, update, query, and save molecular system definitions from dictionaries or YAML files. System configurations typically define proteins, ligands, MSAs, and other molecular components.

Parameters:

Name Type Description Default
system dict

Pre-loaded system dictionary containing molecular definitions.

 None
system_path str

Path to YAML file to load the system configuration from.

 None

Raises:

Type Description
ValueError

If both system and system_path are provided, or if neither is provided.

Examples:

>>> # Load from dictionary
>>> system_dict = {"sequences": [{"protein": {"id": "A"}}]}
>>> sys = System(system=system_dict)

>>> # Load from YAML file
>>> sys = System(system_path="system.yaml")
Source code in src/cofolder/modules/input/system.py 
 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
291
292
293
294
295
296
297
298
299
300
class System:
    """Manage molecular system configuration for Boltz predictions.

    The System class provides methods to load, update, query, and save
    molecular system definitions from dictionaries or YAML files. System
    configurations typically define proteins, ligands, MSAs, and other
    molecular components.

    Parameters
    ----------
    system : dict, optional
        Pre-loaded system dictionary containing molecular definitions.
    system_path : str, optional
        Path to YAML file to load the system configuration from.

    Raises
    ------
    ValueError
        If both system and system_path are provided, or if neither is provided.

    Examples
    --------
    >>> # Load from dictionary
    >>> system_dict = {"sequences": [{"protein": {"id": "A"}}]}
    >>> sys = System(system=system_dict)

    >>> # Load from YAML file
    >>> sys = System(system_path="system.yaml")
    """
    def __init__(self, system=None, system_path=None):
        if system is not None and system_path is not None:
            raise ValueError("Provide either 'system' or 'system_path', not both.")

        if system is not None:
            self.system = system
        elif system_path:
            logger.debug(f"Loading system YAML from {system_path}")
            self.system = read.read_yaml(path=system_path)
        else:
            raise ValueError("Either 'system' or 'system_path' must be provided.")

    def update_system(self, value, path=None, parent_key=None, sub_key=None):
        """Update system configuration at a specific path or key.

        Modifies the system dictionary either by navigating a specific path
        through nested structures, or by searching for a parent key and
        optionally a sub-key within it.

        Parameters
        ----------
        value : any
            The value to set at the specified location.
        path : list, optional
            Path to the target location as a list of keys/indices.
            Example: ["sequences", 0, "protein", "fasta"]
        parent_key : str, optional
            Top-level or nested key to search for and update.
        sub_key : str, optional
            Sub-key within the parent_key dictionary to update.

        Raises
        ------
        ValueError
            If the path traversal fails due to type mismatch.
        KeyError
            If the parent_key is not found in the system.

        Notes
        -----
        Either path or parent_key must be provided, but not both.
        The path parameter creates nested structures if they don't exist.
        """
        if path:
            d = self.system
            for key in path[:-1]:
                if isinstance(d, dict):
                    d = d.setdefault(key, {})
                elif isinstance(d, list):
                    idx = int(key)
                    d = d[idx]
                else:
                    raise ValueError(f"Cannot traverse into object at {key} in path {path}")

            last_key = path[-1]
            if isinstance(d, dict):
                d[last_key] = value
            elif isinstance(d, list):
                d[int(last_key)] = value
            else:
                raise ValueError(f"Cannot set value at path {path}")

        elif parent_key:
            def set_key(d):
                if isinstance(d, dict):
                    if parent_key in d:
                        if sub_key:
                            d[parent_key] = d.get(parent_key, {})
                            d[parent_key][sub_key] = value
                        else:
                            d[parent_key] = value
                        return True
                    return any(set_key(v) for v in d.values())
                if isinstance(d, list):
                    return any(set_key(i) for i in d)
                return False

            if not set_key(self.system):
                raise KeyError(f"Parent key '{parent_key}' not found in the system.")

    def delete_system_key(self, path=None, parent_key=None, sub_key=None, keys_to_delete=None):
        """
        Delete specific keys from the system configuration at a given path or key.

        Parameters
        ----------
        path : list, optional
            Path to the target dictionary/list as a list of keys/indices.
            Example: ["sequences", 0, "ligand"]
        parent_key : str, optional
            Top-level or nested key to search for and delete keys in.
        sub_key : str, optional
            Sub-key within the parent_key dictionary to target.
        keys_to_delete : list of str, optional
            List of keys to delete from the target dictionary.

        Raises
        ------
        ValueError
            If path traversal fails due to type mismatch.
        KeyError
            If parent_key is not found in the system.
        """
        if keys_to_delete is None:
            return

        if path:
            d = self.system
            for key in path:
                if isinstance(d, dict):
                    d = d.setdefault(key, {})
                elif isinstance(d, list):
                    d = d[int(key)]
                else:
                    raise ValueError(f"Cannot traverse into object at {key} in path {path}")
            if isinstance(d, dict):
                for k in keys_to_delete:
                    if k in d:
                        del d[k]

        elif parent_key:
            def remove_keys(d):
                if isinstance(d, dict):
                    if parent_key in d:
                        target = d[parent_key]
                        if sub_key:
                            target = target.get(sub_key, {})
                            if isinstance(target, dict):
                                for k in keys_to_delete:
                                    target.pop(k, None)
                        else:
                            if isinstance(target, dict):
                                for k in keys_to_delete:
                                    target.pop(k, None)
                        return True
                    return any(remove_keys(v) for v in d.values())
                if isinstance(d, list):
                    return any(remove_keys(i) for i in d)
                return False

    def find_value(self, key=None, path=None):
        """Retrieve a value by path or by recursively searching for a key.

        Supports two modes of retrieval: direct path navigation or recursive
        key search throughout the nested structure.

        Parameters
        ----------
        key : str, optional
            Key name to search for recursively throughout the system.
        path : list, optional
            Specific path to the value as a list of keys/indices.

        Returns
        -------
        any or None
            The value found at the specified location, or None if not found.

        Raises
        ------
        ValueError
            If path navigation fails or if the key appears multiple times.

        Examples
        --------
        >>> sys = System(system={"sequences": [{"protein": {"id": "A", "fasta": "MKRAAT"}}]})

        >>> # Find by path
        >>> sys.find_value(path=["sequences", 0, "protein", "fasta"])
        'MKRAAT'

        >>> # Find by key
        >>> sys.find_value(key="fasta")
        'MKRAAT'

        Notes
        -----
        When using key search, if the key appears multiple times in the
        system, a ValueError is raised. Use path instead for disambiguation.
        """
        if path:
            cur = self.system
            for p in path:
                if isinstance(cur, dict):
                    cur = cur[p]
                elif isinstance(cur, list):
                    cur = cur[int(p)]
                else:
                    raise ValueError(f"Path {path} is invalid")
            return cur

        if key:
            def search(d):
                results = []
                if isinstance(d, dict):
                    for k, v in d.items():
                        if k == key:
                            results.append(v)
                        results.extend(search(v))
                elif isinstance(d, list):
                    for item in d:
                        results.extend(search(item))
                return results

            found = search(self.system)
            if not found:
                logger.info(f"Key not found in system: {key}")
                return None
            if len(found) > 1:
                raise ValueError(f"Key '{key}' appears multiple times; use path instead.")
            return found[0]

    def save_system_to_yaml(self, path):
        """Persist the current system dictionary to YAML."""
        write.write_yaml(self, path)
update_system 
update_system(value, path=None, parent_key=None, sub_key=None)

Update system configuration at a specific path or key.

Modifies the system dictionary either by navigating a specific path through nested structures, or by searching for a parent key and optionally a sub-key within it.

Parameters:

Name Type Description Default
value any

The value to set at the specified location.

 required
path list

Path to the target location as a list of keys/indices. Example: ["sequences", 0, "protein", "fasta"]

 None
parent_key str

Top-level or nested key to search for and update.

 None
sub_key str

Sub-key within the parent_key dictionary to update.

 None

Raises:

Type Description
ValueError

If the path traversal fails due to type mismatch.
KeyError

If the parent_key is not found in the system.
Notes

Either path or parent_key must be provided, but not both. The path parameter creates nested structures if they don't exist.

Source code in src/cofolder/modules/input/system.py 
 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
def update_system(self, value, path=None, parent_key=None, sub_key=None):
    """Update system configuration at a specific path or key.

    Modifies the system dictionary either by navigating a specific path
    through nested structures, or by searching for a parent key and
    optionally a sub-key within it.

    Parameters
    ----------
    value : any
        The value to set at the specified location.
    path : list, optional
        Path to the target location as a list of keys/indices.
        Example: ["sequences", 0, "protein", "fasta"]
    parent_key : str, optional
        Top-level or nested key to search for and update.
    sub_key : str, optional
        Sub-key within the parent_key dictionary to update.

    Raises
    ------
    ValueError
        If the path traversal fails due to type mismatch.
    KeyError
        If the parent_key is not found in the system.

    Notes
    -----
    Either path or parent_key must be provided, but not both.
    The path parameter creates nested structures if they don't exist.
    """
    if path:
        d = self.system
        for key in path[:-1]:
            if isinstance(d, dict):
                d = d.setdefault(key, {})
            elif isinstance(d, list):
                idx = int(key)
                d = d[idx]
            else:
                raise ValueError(f"Cannot traverse into object at {key} in path {path}")

        last_key = path[-1]
        if isinstance(d, dict):
            d[last_key] = value
        elif isinstance(d, list):
            d[int(last_key)] = value
        else:
            raise ValueError(f"Cannot set value at path {path}")

    elif parent_key:
        def set_key(d):
            if isinstance(d, dict):
                if parent_key in d:
                    if sub_key:
                        d[parent_key] = d.get(parent_key, {})
                        d[parent_key][sub_key] = value
                    else:
                        d[parent_key] = value
                    return True
                return any(set_key(v) for v in d.values())
            if isinstance(d, list):
                return any(set_key(i) for i in d)
            return False

        if not set_key(self.system):
            raise KeyError(f"Parent key '{parent_key}' not found in the system.")
delete_system_key 
delete_system_key(
    path=None, parent_key=None, sub_key=None, keys_to_delete=None
)

Delete specific keys from the system configuration at a given path or key.

Parameters:

Name Type Description Default
path list

Path to the target dictionary/list as a list of keys/indices. Example: ["sequences", 0, "ligand"]

 None
parent_key str

Top-level or nested key to search for and delete keys in.

 None
sub_key str

Sub-key within the parent_key dictionary to target.

 None
keys_to_delete list of str

List of keys to delete from the target dictionary.

 None

Raises:

Type Description
ValueError

If path traversal fails due to type mismatch.
KeyError

If parent_key is not found in the system.
Source code in src/cofolder/modules/input/system.py 
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
def delete_system_key(self, path=None, parent_key=None, sub_key=None, keys_to_delete=None):
    """
    Delete specific keys from the system configuration at a given path or key.

    Parameters
    ----------
    path : list, optional
        Path to the target dictionary/list as a list of keys/indices.
        Example: ["sequences", 0, "ligand"]
    parent_key : str, optional
        Top-level or nested key to search for and delete keys in.
    sub_key : str, optional
        Sub-key within the parent_key dictionary to target.
    keys_to_delete : list of str, optional
        List of keys to delete from the target dictionary.

    Raises
    ------
    ValueError
        If path traversal fails due to type mismatch.
    KeyError
        If parent_key is not found in the system.
    """
    if keys_to_delete is None:
        return

    if path:
        d = self.system
        for key in path:
            if isinstance(d, dict):
                d = d.setdefault(key, {})
            elif isinstance(d, list):
                d = d[int(key)]
            else:
                raise ValueError(f"Cannot traverse into object at {key} in path {path}")
        if isinstance(d, dict):
            for k in keys_to_delete:
                if k in d:
                    del d[k]

    elif parent_key:
        def remove_keys(d):
            if isinstance(d, dict):
                if parent_key in d:
                    target = d[parent_key]
                    if sub_key:
                        target = target.get(sub_key, {})
                        if isinstance(target, dict):
                            for k in keys_to_delete:
                                target.pop(k, None)
                    else:
                        if isinstance(target, dict):
                            for k in keys_to_delete:
                                target.pop(k, None)
                    return True
                return any(remove_keys(v) for v in d.values())
            if isinstance(d, list):
                return any(remove_keys(i) for i in d)
            return False
find_value 
find_value(key=None, path=None)

Retrieve a value by path or by recursively searching for a key.

Supports two modes of retrieval: direct path navigation or recursive key search throughout the nested structure.

Parameters:

Name Type Description Default
key str

Key name to search for recursively throughout the system.

 None
path list

Specific path to the value as a list of keys/indices.

 None

Returns:

Type Description
any or None

The value found at the specified location, or None if not found.

Raises:

Type Description
ValueError

If path navigation fails or if the key appears multiple times.

Examples:

>>> sys = System(system={"sequences": [{"protein": {"id": "A", "fasta": "MKRAAT"}}]})

>>> # Find by path
>>> sys.find_value(path=["sequences", 0, "protein", "fasta"])
'MKRAAT'

>>> # Find by key
>>> sys.find_value(key="fasta")
'MKRAAT'
Notes

When using key search, if the key appears multiple times in the system, a ValueError is raised. Use path instead for disambiguation.

Source code in src/cofolder/modules/input/system.py 
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
def find_value(self, key=None, path=None):
    """Retrieve a value by path or by recursively searching for a key.

    Supports two modes of retrieval: direct path navigation or recursive
    key search throughout the nested structure.

    Parameters
    ----------
    key : str, optional
        Key name to search for recursively throughout the system.
    path : list, optional
        Specific path to the value as a list of keys/indices.

    Returns
    -------
    any or None
        The value found at the specified location, or None if not found.

    Raises
    ------
    ValueError
        If path navigation fails or if the key appears multiple times.

    Examples
    --------
    >>> sys = System(system={"sequences": [{"protein": {"id": "A", "fasta": "MKRAAT"}}]})

    >>> # Find by path
    >>> sys.find_value(path=["sequences", 0, "protein", "fasta"])
    'MKRAAT'

    >>> # Find by key
    >>> sys.find_value(key="fasta")
    'MKRAAT'

    Notes
    -----
    When using key search, if the key appears multiple times in the
    system, a ValueError is raised. Use path instead for disambiguation.
    """
    if path:
        cur = self.system
        for p in path:
            if isinstance(cur, dict):
                cur = cur[p]
            elif isinstance(cur, list):
                cur = cur[int(p)]
            else:
                raise ValueError(f"Path {path} is invalid")
        return cur

    if key:
        def search(d):
            results = []
            if isinstance(d, dict):
                for k, v in d.items():
                    if k == key:
                        results.append(v)
                    results.extend(search(v))
            elif isinstance(d, list):
                for item in d:
                    results.extend(search(item))
            return results

        found = search(self.system)
        if not found:
            logger.info(f"Key not found in system: {key}")
            return None
        if len(found) > 1:
            raise ValueError(f"Key '{key}' appears multiple times; use path instead.")
        return found[0]
save_system_to_yaml 
save_system_to_yaml(path)

Persist the current system dictionary to YAML.

Source code in src/cofolder/modules/input/system.py 
298
299
300
def save_system_to_yaml(self, path):
    """Persist the current system dictionary to YAML."""
    write.write_yaml(self, path)

iter_system_chains 

iter_system_chains(system_obj: Any) -> Iterator[SystemChain]

Yield chains in backend order, expanding list-valued id fields.

Source code in src/cofolder/modules/input/system.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def iter_system_chains(system_obj: Any) -> Iterator[SystemChain]:
    """Yield chains in backend order, expanding list-valued ``id`` fields."""

    if isinstance(system_obj, System):
        sequences = system_obj.system.get("sequences", [])
    elif isinstance(system_obj, dict):
        sequences = system_obj.get("sequences", [])
    else:
        find_value = getattr(system_obj, "find_value", None)
        sequences = find_value(key="sequences") if callable(find_value) else []
    if not isinstance(sequences, list):
        return
    for sequence_index, entry in enumerate(sequences):
        if not isinstance(entry, dict) or len(entry) != 1:
            continue
        entity_type, entity_data = next(iter(entry.items()))
        if not isinstance(entity_data, dict):
            continue
        raw_ids = entity_data.get("id")
        chain_ids = raw_ids if isinstance(raw_ids, list) else [raw_ids]
        for chain_id in chain_ids:
            if chain_id is None:
                continue
            yield SystemChain(
                chain_id=str(chain_id),
                entity_type=str(entity_type).lower(),
                entity_data=entity_data,
                sequence_index=sequence_index,
            )