synplan.utils package

Submodules

synplan.utils.config module

Module containing configuration classes.

class synplan.utils.config.BaseConfigModel

Bases: BaseModel

Base class for all SynPlanner configuration models.

Replaces ConfigABC. Provides backward-compatible from_dict(), from_yaml(), to_dict(), to_yaml() methods as thin wrappers over Pydantic’s native API.

classmethod from_dict(config_dict: dict[str, Any])

Create instance from dictionary. Backward-compatible wrapper.

classmethod from_yaml(file_path: str)

Create instance from YAML file. Backward-compatible wrapper.

model_config = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

to_dict() → dict[str, Any]

Convert to dictionary. Backward-compatible wrapper.

to_yaml(file_path: str) → None

Serialize to YAML file. Backward-compatible wrapper.

class synplan.utils.config.CombinedPolicyConfig(*, filtering_weights_path: str | Path, ranking_weights_path: str | Path, top_rules: Annotated[int, Gt(gt=0)] = 50, rule_prob_threshold: float = 0.0, ranking_weight: Annotated[float, Gt(gt=0)] = 1.0, temperature: Annotated[float, Gt(gt=0)] = 1.0)

Bases: BaseConfigModel

Configuration for combined filtering + ranking policy.

Combines filtering and ranking policies by weighted addition of logits:

combined_logits = filtering_logits + ranking_weight * ranking_logits combined_probs = softmax(combined_logits / temperature)

The filtering policy provides applicability scores (trained on multi-label applicability). The ranking policy provides feasibility scores (trained on actual reactions).

Parameters:

• filtering_weights_path – Path to the filtering policy network weights.

• ranking_weights_path – Path to the ranking policy network weights.

• top_rules – Number of top rules to return.

• rule_prob_threshold – Minimum probability threshold for returning a rule.

• ranking_weight – Weight for ranking logits (default 1.0). Values > 1.0 give more weight to ranking (feasibility).

• temperature – Temperature for softmax (default 1.0). Values > 1.0 produce softer distributions (more exploration).

filtering_weights_path: str | Path

model_config = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

ranking_weight: float

ranking_weights_path: str | Path

rule_prob_threshold: float

temperature: float

top_rules: int

class synplan.utils.config.PolicyEvaluationConfig(*, normalize: bool = False)

Bases: BaseConfigModel

Configuration for policy-based evaluation strategy.

Uses policy network probabilities as evaluation scores.

Parameters:

normalize – Whether to normalize scores to [0, 1].

model_config = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

normalize: bool

class synplan.utils.config.PolicyNetworkConfig(*, policy_type: Literal['filtering', 'ranking'] = 'ranking', embedder_type: Literal['gcn', 'gcn_concat', 'gps'] = 'gcn', vector_dim: Annotated[int, Gt(gt=0)] = 256, batch_size: Annotated[int, Gt(gt=0)] = 500, dropout: Annotated[float, Ge(ge=0.0), Le(le=1.0)] = 0.4, learning_rate: Annotated[float, Gt(gt=0)] = 0.008, num_conv_layers: Annotated[int, Gt(gt=0)] = 5, num_epoch: Annotated[int, Gt(gt=0)] = 100, weights_path: str | Path | None = None, heads: Annotated[int, Gt(gt=0)] = 4, attn_type: Literal['performer', 'multihead'] = 'performer', attn_dropout: Annotated[float, Ge(ge=0.0), Le(le=1.0)] = 0.5, logger: dict | None = None, trainer: dict | None = None, log_grad_norm: bool = False, priority_rules_fraction: Annotated[float, Ge(ge=0)] = 0.5, rule_prob_threshold: Annotated[float, Ge(ge=0)] = 0.0, top_rules: Annotated[int, Gt(gt=0)] = 50)

Bases: BaseConfigModel

Configuration class for the policy network.

Parameters:

• vector_dim – Dimension of the input vectors.

• batch_size – Number of samples per batch.

• dropout – Dropout rate for regularization.

• learning_rate – Learning rate for the optimizer.

• num_conv_layers – Number of convolutional layers in the network.

• num_epoch – Number of training epochs.

• policy_type – Mode of operation, either ‘filtering’ or ‘ranking’.

• logger – Training logger configuration. None disables logging. A dict with "type" key ("csv", "tensorboard", "mlflow", or "wandb") and optional logger-specific parameters passed to the PyTorch Lightning logger constructor.

attn_dropout: float

attn_type: Literal['performer', 'multihead']

batch_size: int

dropout: float

embedder_type: Literal['gcn', 'gcn_concat', 'gps']

heads: int

learning_rate: float

log_grad_norm: bool

logger: dict | None

model_config = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

num_conv_layers: int

num_epoch: int

policy_type: Literal['filtering', 'ranking']

priority_rules_fraction: float

rule_prob_threshold: float

top_rules: int

trainer: dict | None

vector_dim: int

weights_path: str | Path | None

class synplan.utils.config.RDKitEvaluationConfig(*, score_function: Literal['sascore', 'weight', 'heavyAtomCount', 'weightXsascore', 'WxWxSAS'] = 'sascore', normalize: bool = False)

Bases: BaseConfigModel

Configuration for RDKit-based evaluation strategy.

Uses molecular descriptors like SA score, molecular weight, etc.

Parameters:

• score_function – Name of the scoring function to use. Options: “sascore”, “weight”, “heavyAtomCount”, “weightXsascore”, “WxWxSAS”.

• normalize – Whether to normalize scores to [0, 1].

model_config = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

normalize: bool

score_function: Literal['sascore', 'weight', 'heavyAtomCount', 'weightXsascore', 'WxWxSAS']

class synplan.utils.config.RandomEvaluationConfig(*, normalize: bool = False)

Bases: BaseConfigModel

Configuration for random evaluation strategy.

Assigns random scores - useful for testing and baseline comparisons.

Parameters:

normalize – Whether to normalize scores to [0, 1].

model_config = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

normalize: bool

class synplan.utils.config.ReactorConfig(*, automorphism_filter: bool = True, delete_atoms: bool = False, one_shot: bool = True, fix_aromatic_rings: bool = True, fix_tautomers: bool = True)

Bases: BaseConfigModel

Configuration for chython Reactor instances.

Controls how Reactor objects are created when loading reaction rules. Intended for programmatic use, not config files.

Parameters:

• automorphism_filter – If True, skip substructure matches to the same set of atoms. Should be False for asymmetric product templates (e.g. Suzuki) where different match orientations produce different products.

• delete_atoms – If True, atoms in reactants but not in products are removed.

• one_shot – If True, do only single reaction center per application.

• fix_aromatic_rings – Proceed kekule and thiele on products.

• fix_tautomers – Fix tautomers in products.

automorphism_filter: bool

delete_atoms: bool

fix_aromatic_rings: bool

fix_tautomers: bool

model_config = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

one_shot: bool

to_reactor_kwargs() → dict

Convert to kwargs dict for Reactor constructor.

class synplan.utils.config.RolloutEvaluationConfig(*, policy_network: Any, reaction_rules: Any, building_blocks: Any, min_mol_size: Annotated[int, Ge(ge=0)] = 0, max_depth: Annotated[int, Gt(gt=0)] = 6, normalize: bool = False, stochastic: bool = False)

Bases: BaseConfigModel

Configuration for rollout-based evaluation strategy.

Contains all dependencies needed for rollout simulation.

Parameters:

• policy_network – Policy network function for rollout simulation.

• reaction_rules – List of reaction rules for applying transformations.

• building_blocks – Set of building block molecules.

• min_mol_size – Minimum molecule size to consider for expansion.

• max_depth – Maximum depth for rollout simulation.

• normalize – Whether to normalize scores to [0, 1].

• stochastic – If True, sample from valid rules using policy probabilities. If False (default), use greedy selection (first successful rule).

building_blocks: Any

max_depth: int

min_mol_size: int

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

normalize: bool

policy_network: Any

reaction_rules: Any

stochastic: bool

class synplan.utils.config.RuleExtractionConfig(*, keep_metadata: bool = False, reactor_validation: bool = True, reverse_rule: bool = True, as_query_container: bool = True, single_product_only: bool = True, environment_atom_count: int = 1, min_popularity: int = 3, include_rings: bool = True, multicenter_rules: bool = True, include_func_groups: bool = False, keep_leaving_groups: bool = True, keep_incoming_groups: bool = True, keep_reagents: bool = False, func_groups_list: list[str] = <factory>, atom_info_retention: dict[str, dict[str, bool]]=<factory>)

Bases: BaseConfigModel

Configuration class for extracting reaction rules.

Parameters:

• multicenter_rules – If True, extracts a single rule encompassing all centers. If False, extracts separate reaction rules for each reaction center in a multicenter reaction.

• as_query_container – If True, the extracted rules are generated as QueryContainer objects, analogous to SMARTS objects for pattern matching in chemical structures.

• reverse_rule – If True, reverses the direction of the reaction for rule extraction.

• reactor_validation – If True, validates each generated rule in a chemical reactor to ensure correct generation of products from reactants.

• include_func_groups – If True, includes specific functional groups in the reaction rule in addition to the reaction center and its environment.

• func_groups_list – A list of functional groups to be considered when include_func_groups is True.

• include_rings – If True, includes ring structures in the reaction rules.

• keep_leaving_groups – If True, retains leaving groups in the extracted reaction rule.

• keep_incoming_groups – If True, retains incoming groups in the extracted reaction rule.

• keep_reagents – If True, includes reagents in the extracted reaction rule.

• environment_atom_count – Defines the size of the environment around the reaction center to be included in the rule (0 for only the reaction center, 1 for the first environment, etc.).

• min_popularity – Minimum number of times a rule must be applied to be considered for further analysis.

• keep_metadata – If True, retains metadata associated with the reaction in the extracted rule.

• single_product_only – If True, skips reactions that have more than one product (after reagent removal).

• atom_info_retention – Controls the amount of information about each atom to retain (‘none’, ‘reaction_center’, or ‘all’).

as_query_container: bool

atom_info_retention: dict[str, dict[str, bool]]

environment_atom_count: int

func_groups_list: list[str]

include_func_groups: bool

include_rings: bool

keep_incoming_groups: bool

keep_leaving_groups: bool

keep_metadata: bool

keep_reagents: bool

min_popularity: int

model_config = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

multicenter_rules: bool

reactor_validation: bool

reverse_rule: bool

single_product_only: bool

class synplan.utils.config.TreeConfig(*, max_iterations: Annotated[int, Gt(gt=0)] = 100, max_tree_size: Annotated[int, Gt(gt=0)] = 1000000, max_time: Annotated[float, Gt(gt=0)] = 600, max_depth: Annotated[int, Gt(gt=0)] = 6, exclude_small: bool = True, min_mol_size: Annotated[int, Ge(ge=0)] = 6, silent: bool = False, algorithm: str = 'uct', normalize_scores: bool = False, max_rules_applied: int = 10, stop_at_first: bool = False, enable_pruning: bool = False, search_strategy: Literal['expansion_first', 'evaluation_first'] = 'expansion_first', ucb_type: Literal['uct', 'puct', 'value'] = 'uct', c_ucb: Annotated[float, Ge(ge=0)] = 0.1, backprop_type: Literal['muzero', 'cumulative'] = 'muzero', evaluation_agg: Literal['max', 'average'] = 'max', epsilon: Annotated[float, Ge(ge=0.0), Le(le=1.0)] = 0.0, init_node_value: Annotated[float, Ge(ge=0.0), Le(le=1.0)] = 0.5, beam_width: Annotated[int, Gt(gt=0)] = 10, nmcs_level: Annotated[int, Gt(gt=0)] = 2, nmcs_playout_mode: Literal['greedy', 'random', 'policy'] = 'greedy', lnmcs_ratio: Annotated[float, Ge(ge=0.0), Le(le=1.0)] = 0.2)

Bases: BaseConfigModel

Configuration class for the tree search algorithm.

Parameters:

• max_iterations – The number of iterations to run the algorithm for.

• max_tree_size – The maximum number of nodes in the tree.

• max_time – The time limit (in seconds) for the algorithm to run.

• max_depth – The maximum depth of the tree.

• ucb_type – Type of UCB used in the search algorithm. Options are “puct”, “uct”, “value”, defaults to “uct”.

• c_ucb – The exploration-exploitation balance coefficient used in Upper Confidence Bound (UCB).

• backprop_type – Type of backpropagation algorithm. Options are “muzero”, “cumulative”, defaults to “muzero”.

• search_strategy – The strategy used for tree search. Options are “expansion_first”, “evaluation_first”.

• exclude_small – Whether to exclude small molecules during the search.

• evaluation_agg – Method for aggregating evaluation scores. Options are “max”, “average”, defaults to “max”.

• normalize_scores – Whether to normalize evaluation scores to [0, 1].

• init_node_value – Initial value for a new node.

• epsilon – A parameter in the epsilon-greedy search strategy representing the chance of random selection of reaction rules during the selection stage in Monte Carlo Tree Search, specifically during Upper Confidence Bound estimation. It balances between exploration and exploitation.

• min_mol_size – Defines the minimum size of a molecule that is have to be synthesized. Molecules with 6 or fewer heavy atoms are assumed to be building blocks by definition, thus setting the threshold for considering larger molecules in the search, defaults to 6.

• silent – Whether to suppress progress output.

• nmcs_level – Nesting level for NMCS and LazyNMCS algorithms. Higher levels provide more thorough search but are more computationally expensive. Defaults to 2.

• nmcs_playout_mode – Playout mode for NMCS base-level rollouts. Options are “greedy” (best value), “random”, or “policy” (best policy probability). Defaults to “greedy”.

• lnmcs_ratio – Pruning percentile for LazyNMCS algorithm. Only candidates scoring above this percentile threshold are explored. Value in range [0.0, 1.0]. Defaults to 0.2.

algorithm: str

backprop_type: Literal['muzero', 'cumulative']

beam_width: int

c_ucb: float

enable_pruning: bool

epsilon: float

evaluation_agg: Literal['max', 'average']

exclude_small: bool

init_node_value: float

lnmcs_ratio: float

max_depth: int

max_iterations: int

max_rules_applied: int

max_time: float

max_tree_size: int

min_mol_size: int

model_config = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

nmcs_level: int

nmcs_playout_mode: Literal['greedy', 'random', 'policy']

normalize_scores: bool

search_strategy: Literal['expansion_first', 'evaluation_first']

silent: bool

stop_at_first: bool

ucb_type: Literal['uct', 'puct', 'value']

class synplan.utils.config.TuningConfig(*, batch_size: Annotated[int, Gt(gt=0)] = 100, num_simulations: int = 1)

Bases: BaseConfigModel

Configuration class for the network training.

Parameters:

• batch_size – The number of targets per batch in the planning simulation step.

• num_simulations – The number of planning simulations.

batch_size: int

model_config = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

num_simulations: int

class synplan.utils.config.ValueNetworkConfig(*, weights_path: str | Path | None = None, vector_dim: Annotated[int, Gt(gt=0)] = 256, batch_size: Annotated[int, Gt(gt=0)] = 500, dropout: Annotated[float, Ge(ge=0.0), Le(le=1.0)] = 0.4, learning_rate: Annotated[float, Gt(gt=0)] = 0.008, num_conv_layers: Annotated[int, Gt(gt=0)] = 5, num_epoch: Annotated[int, Gt(gt=0)] = 100)

Bases: BaseConfigModel

Configuration class for the value network.

Parameters:

• vector_dim – Dimension of the input vectors.

• batch_size – Number of samples per batch.

• dropout – Dropout rate for regularization.

• learning_rate – Learning rate for the optimizer.

• num_conv_layers – Number of convolutional layers in the network.

• num_epoch – Number of training epochs.

batch_size: int

dropout: float

learning_rate: float

model_config = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

num_conv_layers: int

num_epoch: int

vector_dim: int

weights_path: str | Path | None

class synplan.utils.config.ValueNetworkEvaluationConfig(*, weights_path: str | Path, normalize: bool = False)

Bases: BaseConfigModel

Configuration for value network-based evaluation strategy.

Parameters:

• weights_path – Path to the value network weights file.

• normalize – Whether to normalize scores to [0, 1].

model_config = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

normalize: bool

weights_path: str | Path

synplan.utils.files module

Module containing classes and functions needed for reactions/molecules data reading/writing.

class synplan.utils.files.FileHandler(filename: str | Path, **kwargs)

Bases: object

General class to handle chemical files.

close()

class synplan.utils.files.MoleculeReader(filename: str | Path, **kwargs)

Bases: Reader

class synplan.utils.files.MoleculeWriter(filename: str | Path, mapping: bool = True, **kwargs)

Bases: Writer

write(molecule: MoleculeContainer)

Function to write a specific molecule to the file.

Parameters:

molecule – The path and name of the file.

Returns:

None.

class synplan.utils.files.RawReactionReader(filename: str | Path, batch_size: int = 1)

Bases: object

Yields raw unparsed items: str lines for SMILES, str blocks for RDF.

count() → int

Count the number of records without parsing them.

iter_chunks(chunk_size: int) → Iterator[tuple[int, list]]

Yield (offset, items) chunks of up to chunk_size records.

class synplan.utils.files.ReactionReader(filename: str | Path, **kwargs)

Bases: Reader

class synplan.utils.files.ReactionWriter(filename: str | Path, mapping: bool = True, **kwargs)

Bases: Writer

write(reaction: ReactionContainer)

Function to write a specific reaction to the file.

Parameters:

reaction – The path and name of the file.

Returns:

None.

class synplan.utils.files.Reader(filename: str | Path, **kwargs)

Bases: FileHandler

class synplan.utils.files.SMILESRead(filename: str | Path, **kwargs)

Bases: object

close()

read()

Parse the whole SMILES file.

Returns:

List of parsed molecules or reactions.

class synplan.utils.files.Writer(filename: str | Path, mapping: bool = True, **kwargs)

Bases: FileHandler

synplan.utils.files.count_rdf_records(path: str | Path) → int

Count number of RDF records (by $RFMT/$MFMT markers).

synplan.utils.files.count_sdf_records(path: str | Path) → int

Count number of SDF records (by ‘$$$$’ separators).

synplan.utils.files.count_smiles_records(path: str | Path) → int

Count number of non-empty SMILES records (lines).

synplan.utils.files.init_parse_worker(fmt: str)

Initializer: store the input format so parse_one knows how to parse.

synplan.utils.files.iter_csv_smiles(path: str | Path, *, header: bool = True, delimiter: str = ',', smiles_column: str = 'SMILES') → Iterator[str]

Yield SMILES strings from a CSV/CSV.GZ file.

Parameters:

• path – Path to the CSV file. If it ends with “.gz”, it’s treated as gzipped CSV.

• header – If True, treat the first row as a header and read from smiles_column. If False, treat the first column as SMILES.

• delimiter – CSV delimiter (default: “,”).

• smiles_column – Column name containing SMILES (used when header=True).

synplan.utils.files.iter_csv_smiles_blocks(path: str | Path, records_per_block: int, *, header: bool = True, delimiter: str = ',', smiles_column: str = 'SMILES') → Iterator[list[str]]

Yield SMILES lists of up to records_per_block items from a CSV/CSV.GZ file.

synplan.utils.files.iter_rdf_text_blocks(path: str | Path, records_per_block: int) → Iterator[str]

Yield RDF text blocks of up to records_per_block records.

Each block is a string containing one or more $RFMT/$MFMT records, parseable via RDFRead(StringIO(block)).

synplan.utils.files.iter_sdf_text_blocks(path: str | Path, records_per_block: int) → Iterator[str]

Yield SDF text blocks containing up to records_per_block molecules.

Records are delimited by lines equal to ‘$$$$’. The block is a concatenated string of lines that can be fed to StringIO and parsed by SDFRead.

synplan.utils.files.iter_smiles(path: str | Path) → Iterator[str]

Yield first whitespace-delimited token (SMILES) per non-empty line.

synplan.utils.files.iter_smiles_blocks(path: str | Path, records_per_block: int) → Iterator[list[str]]

Yield SMILES lists of up to records_per_block items from file.

synplan.utils.files.load_rule_index_mapping_tsv(tsv_path: str | Path) → dict

Load reaction-to-rule index mapping from a rules TSV file.

The TSV is already sorted by descending popularity (same order as the pickle), so the rule index is simply the row number (0-based).

Parameters:

tsv_path – Path to the TSV file with columns rule_smarts, popularity, reaction_indices.

Returns:

Dict mapping reaction_index → rule_index.

synplan.utils.files.open_text(path: str | Path) → TextIO

Open a text file that may be gzip-compressed.

If the path ends with “.gz”, the file is opened via gzip in text mode.

synplan.utils.files.parse_one(item: str)

Parse a single raw item → (ReactionContainer | None, error | None).

item is a SMILES string (for .smi) or an RDF text block (for .rdf). The format is set once per worker via init_parse_worker().

synplan.utils.files.parse_reaction(item: str | ReactionContainer, fmt: str = 'smi') → ReactionContainer

Parse a raw string into a ReactionContainer.

If item is already a ReactionContainer it is returned as-is.

Parameters:

• item – A SMILES string, an RDF text block, or a ReactionContainer.

• fmt – "smi" for SMILES strings, "rdf" for RDF text blocks.

Returns:

The parsed ReactionContainer.

synplan.utils.files.to_reaction_smiles_record(reaction: ReactionContainer) → str

Converts the reaction to the SMILES record. Needed for reaction/molecule writers.

Parameters:

reaction – The reaction to be written.

Returns:

The SMILES record to be written.

synplan.utils.loading module

synplan.utils.logging module

Generic logging helpers for scripts and notebooks.

class synplan.utils.logging.DisableLogger

Bases: object

Context‑manager that suppresses all logging inside its scope.

class synplan.utils.logging.HiddenPrints

Bases: object

Context‑manager that suppresses print output inside its scope.

class synplan.utils.logging.TqdmHandler(stream=None)

Bases: StreamHandler

Write via tqdm.write so log lines don’t break progress bars.

emit(record)

Emit a record.

If a formatter is specified, it is used to format the record. The record is then written to the stream with a trailing newline. If exception information is present, it is formatted using traceback.print_exception and appended to the stream. If the stream has an ‘encoding’ attribute, it is used to determine how to do the output to the stream.

synplan.utils.logging.init_logger(*, name: str = 'app', console_level: str | int = 'ERROR', file_level: str | int = 'INFO', log_dir: str | PathLike = '.', redirect_tqdm: bool = True) → tuple[Logger, str]

Initialise (or fetch) a namespaced logger that works in scripts & notebooks. Idempotent ‑ safe to call multiple times.

Returns:

Configured logger instance.

Return type:

logging.Logger

synplan.utils.logging.silence_logger(logger_name: str, level: int | str = 40)

Raise the threshold of a chatty library logger. Useful inside worker process initializers.

synplan.utils.visualisation module

Module contents