SAFE

SAFE Encoder-Decoder

SAFEConverter(slicer: Optional[Union[str, List[str], Callable]] = 'brics', require_hs: Optional[bool] = None, use_original_opener_for_attach: bool = True, ignore_stereo: bool = False)

Molecule line notation conversion from SMILES to SAFE

A SAFE representation is a string based representation of a molecule decomposition into fragment components, separated by a dot ('.'). Note that each component (fragment) might not be a valid molecule by themselves, unless explicitely correct to add missing hydrogens.

Slicing algorithms

By default SAFE strings are generated using BRICS, however, the following alternative are supported:

• Hussain-Rea (hr)
• RECAP (recap)
• RDKit's MMPA (mmpa)
• Any possible attachment points (attach)

Furthermore, you can also provide your own slicing algorithm, which should return a pair of atoms corresponding to the bonds to break.

Constructor for the SAFE converter

Parameters:

• slicer (Optional[Union[str, List[str], Callable]], default: 'brics' ) –

  slicer algorithm to use for encoding. Can either be one of the supported slicing algorithm (SUPPORTED_SLICERS) or a custom callable that returns the bond ids that can be sliced.

• require_hs (Optional[bool], default: None ) –

  whether the slicing algorithm require the molecule to have hydrogen explictly added. attach slicer requires adding hydrogens.

• use_original_opener_for_attach (bool, default: True ) –

  whether to use the original branch opener digit when adding back mapping number to attachment points, or use simple enumeration.

• ignore_stereo (bool, default: False ) –

  whether to discard input stereochemistry explicitly. When false, stereochemistry-changing cuts are skipped and the encoded graph is verified.

Source code in safe/converter.py

def __init__(
    self,
    slicer: Optional[Union[str, List[str], Callable]] = "brics",
    require_hs: Optional[bool] = None,
    use_original_opener_for_attach: bool = True,
    ignore_stereo: bool = False,
):
    """Constructor for the SAFE converter

    Args:
        slicer: slicer algorithm to use for encoding.
            Can either be one of the supported slicing algorithm (SUPPORTED_SLICERS)
            or a custom callable that returns the bond ids that can be sliced.
        require_hs: whether the slicing algorithm require the molecule to have hydrogen explictly added.
            `attach` slicer requires adding hydrogens.
        use_original_opener_for_attach: whether to use the original branch opener digit when adding back
            mapping number to attachment points, or use simple enumeration.
        ignore_stereo: whether to discard input stereochemistry explicitly. When false,
            stereochemistry-changing cuts are skipped and the encoded graph is verified.

    """
    self.slicer = slicer
    if isinstance(slicer, str) and slicer.lower() in self.SUPPORTED_SLICERS:
        self.slicer = self.__SLICE_SMARTS.get(slicer.lower(), slicer)
    if self.slicer != "brics" and isinstance(self.slicer, str):
        self.slicer = [self.slicer]
    if isinstance(self.slicer, (list, tuple)):
        self.slicer = [dm.from_smarts(x) for x in self.slicer]
        if any(x is None for x in self.slicer):
            raise ValueError(f"Slicer: {slicer} cannot be valid")
    self.require_hs = require_hs or (slicer == "attach")
    self.use_original_opener_for_attach = use_original_opener_for_attach
    self.ignore_stereo = ignore_stereo

SUPPORTED_SLICERS = ['hr', 'rotatable', 'recap', 'mmpa', 'attach', 'brics'] class-attribute instance-attribute

__SLICE_SMARTS = {'hr': ['[*]!@-[*]'], 'recap': ['[$([C;!$(C([#7])[#7])](=!@[O]))]!@[$([#7;+0;!D1])]', '[$(C=!@O)]!@[$([O;+0])]', '[$([N;!D1;+0;!$(N-C=[#7,#8,#15,#16])](-!@[*]))]-!@[$([*])]', '[$(C(=!@O)([#7;+0;D2,D3])!@[#7;+0;D2,D3])]!@[$([#7;+0;D2,D3])]', '[$([O;+0](-!@[#6!$(C=O)])-!@[#6!$(C=O)])]-!@[$([#6!$(C=O)])]', 'C=!@C', '[N;+1;D4]!@[#6]', '[$([n;+0])]-!@C', '[$([O]=[C]-@[N;+0])]-!@[$([C])]', 'c-!@c', '[$([#7;+0;D2,D3])]-!@[$([S](=[O])=[O])]'], 'mmpa': ['[#6+0;!$(*=,#[!#6])]!@!=!#[*]'], 'attach': ['[*]!@[*]'], 'rotatable': ['[!$(*#*)&!D1]-&!@[!$(*#*)&!D1]']} class-attribute instance-attribute

ignore_stereo = ignore_stereo instance-attribute

require_hs = require_hs or slicer == 'attach' instance-attribute

slicer = slicer instance-attribute

use_original_opener_for_attach = use_original_opener_for_attach instance-attribute

decoder(inp: str, as_mol: bool = False, canonical: bool = False, fix: bool = True, remove_dummies: bool = True, remove_added_hs: bool = True)

Convert input SAFE representation to smiles

Parameters:

• inp (str) –

  input SAFE representation to decode as a valid molecule or smiles

• as_mol (bool, default: False ) –

  whether to return a molecule object or a smiles string

• canonical (bool, default: False ) –

  whether to return a canonical

• fix (bool, default: True ) –

  whether to fix the SAFE representation to take into account non-connected attachment points

• remove_dummies (bool, default: True ) –

  whether to remove dummy atoms from the SAFE representation. Set this to False when decoding an open SAFE fragment or scaffold if attachment points must be preserved.

• remove_added_hs (bool, default: True ) –

  whether to remove all the added hydrogen atoms after applying dummy removal for recovery

Source code in safe/converter.py

def decoder(
    self,
    inp: str,
    as_mol: bool = False,
    canonical: bool = False,
    fix: bool = True,
    remove_dummies: bool = True,
    remove_added_hs: bool = True,
):
    """Convert input SAFE representation to smiles

    Args:
        inp: input SAFE representation to decode as a valid molecule or smiles
        as_mol: whether to return a molecule object or a smiles string
        canonical: whether to return a canonical
        fix: whether to fix the SAFE representation to take into account non-connected attachment points
        remove_dummies: whether to remove dummy atoms from the SAFE representation. Set this to
            ``False`` when decoding an open SAFE fragment or scaffold if attachment points
            must be preserved.
        remove_added_hs: whether to remove all the added hydrogen atoms after applying dummy removal for recovery
    """

    if fix:
        inp = self._ensure_valid(inp)
    mol = dm.to_mol(inp)
    if mol is None:
        raise ValueError("SAFE string could not be parsed into a molecule")
    if remove_dummies:
        dummy_query = dm.from_smarts("[$([#0]!-!:*);$([#0;D1])]")
        if any(atom.GetAtomicNum() == 0 for atom in mol.GetAtoms()):
            replacements = Chem.ReplaceSubstructs(
                mol,
                dummy_query,
                dm.to_mol("C"),
                True,
            )
            mol = dm.remove_dummies(replacements[0])
    if as_mol:
        if remove_added_hs:
            mol = dm.remove_hs(mol, update_explicit_count=True)
        return mol
    out = dm.to_smiles(mol, canonical=canonical, explicit_hs=(not remove_added_hs))
    has_stereo = any(
        atom.GetChiralTag() != Chem.ChiralType.CHI_UNSPECIFIED for atom in mol.GetAtoms()
    ) or any(bond.GetStereo() != Chem.BondStereo.STEREONONE for bond in mol.GetBonds())
    mol_graph = self._canonical_isomeric_graph(mol)
    if has_stereo and self._canonical_isomeric_graph(out) != mol_graph:
        # RDKit's non-canonical writer can choose an inconsistent parity
        # for rare symmetry-dependent stereocentres. Canonical writing is
        # deterministic and preserves the graph in those cases.
        canonical_out = dm.to_smiles(
            mol,
            canonical=True,
            explicit_hs=(not remove_added_hs),
        )
        out = (
            canonical_out if self._canonical_isomeric_graph(canonical_out) == mol_graph else inp
        )
    return out

encoder(inp: Union[str, dm.Mol], canonical: bool = True, randomize: Optional[bool] = False, seed: Optional[int] = None, constraints: Optional[List[dm.Mol]] = None, allow_empty: bool = False, rdkit_safe: bool = True)

Convert input smiles to SAFE representation

Parameters:

• inp (Union[str, Mol]) –

  input smiles

• canonical (bool, default: True ) –

  whether to return canonical smiles string. Defaults to True

• randomize (Optional[bool], default: False ) –

  whether to randomize the safe string encoding. Will be ignored if canonical is provided

• seed (Optional[int], default: None ) –

  optional seed to use when allowing randomization of the SAFE encoding. Randomization happens at two steps: 1. at the original smiles representation by randomization the atoms. 2. at the SAFE conversion by randomizing fragment orders

• constraints (Optional[List[Mol]], default: None ) –

  List of molecules or pattern to preserve during the SAFE construction. Any bond slicing would happen outside of a substructure matching one of the patterns.

• allow_empty (bool, default: False ) –

  whether to allow the slicing algorithm to return empty bonds

• rdkit_safe (bool, default: True ) –

  whether to apply rdkit-safe digit standardization to the output SAFE string.

Source code in safe/converter.py

def encoder(
    self,
    inp: Union[str, dm.Mol],
    canonical: bool = True,
    randomize: Optional[bool] = False,
    seed: Optional[int] = None,
    constraints: Optional[List[dm.Mol]] = None,
    allow_empty: bool = False,
    rdkit_safe: bool = True,
):
    """Convert input smiles to SAFE representation

    Args:
        inp: input smiles
        canonical: whether to return canonical smiles string. Defaults to True
        randomize: whether to randomize the safe string encoding. Will be ignored if canonical is provided
        seed: optional seed to use when allowing randomization of the SAFE encoding.
            Randomization happens at two steps:
            1. at the original smiles representation by randomization the atoms.
            2. at the SAFE conversion by randomizing fragment orders
        constraints: List of molecules or pattern to preserve during the SAFE construction. Any bond slicing would
            happen outside of a substructure matching one of the patterns.
        allow_empty: whether to allow the slicing algorithm to return empty bonds
        rdkit_safe: whether to apply rdkit-safe digit standardization to the output SAFE string.
    """
    source_text = inp if isinstance(inp, str) else None
    source_mol = dm.to_mol(inp, remove_hs=False)
    if source_mol is None:
        raise ValueError("Input could not be parsed into a molecule")
    if not self.ignore_stereo and source_mol.GetStereoGroups():
        raise SAFEEncodeError(
            "Enhanced CXSMILES stereo groups are not representable in SAFE 1.0; "
            "resolve them to a single stereoisomer or set ignore_stereo=True explicitly"
        )

    rng = None
    should_randomize = bool(randomize and not canonical)
    if should_randomize:
        rng = np.random.default_rng(seed)
        inp = self.randomize(source_mol, rng)

    if isinstance(inp, dm.Mol):
        inp = dm.to_smiles(inp, canonical=canonical, randomize=False, ordered=False)
    elif canonical:
        # Canonical SAFE must not depend on which equivalent SMILES spelling
        # the caller supplied. Molecule inputs already followed this path;
        # normalize string inputs before choosing rooted fragment atoms.
        inp = dm.to_smiles(source_mol, canonical=True, randomize=False, ordered=False)

    # EN: we first normalize the attachment if the molecule is a query:
    # inp = dm.reactions.convert_attach_to_isotope(inp, as_smiles=True)

    # RDKit's extended ring-closure form ('%(nnn)', up to 5 digits) is used for
    # labels >= 100; see `_format_ring_closure`.
    # https://www.rdkit.org/docs/RDKit_Book.html#ring-closures
    branch_numbers = self._find_branch_number(inp)

    mol = dm.to_mol(inp, remove_hs=False)
    if mol is None:
        raise ValueError("Input could not be parsed into a molecule")
    # Inspect explicit tags on the original graph. FindPotentialStereo can
    # omit symmetry-dependent tags in constrained peroxide systems, and
    # atom renumbering must never disable the final identity guard.
    has_specified_stereo = any(
        atom.GetChiralTag() != Chem.ChiralType.CHI_UNSPECIFIED for atom in source_mol.GetAtoms()
    ) or any(bond.GetStereo() != Chem.BondStereo.STEREONONE for bond in source_mol.GetBonds())
    if self.ignore_stereo:
        mol = dm.remove_stereochemistry(mol)

    bond_map_id = 1
    open_attachment_ids = set()
    for atom in mol.GetAtoms():
        if atom.GetAtomicNum() == 0:
            # Preserve the distinction between an explicitly labelled
            # attachment point (for example ``[1*]`` or ``[*:1]``), or a
            # terminal ``[*]``, and a literal wildcard atom embedded in a
            # structure (for example ``C1*CCC1``). All are normalised below
            # so fragment labels remain unique, but only attachment points
            # must survive as unmatched SAFE ring closures for constrained
            # generation.
            if atom.GetDegree() == 1:
                open_attachment_ids.add(bond_map_id)
            atom.SetAtomMapNum(0)
            atom.SetIsotope(bond_map_id)
            bond_map_id += 1

    if self.require_hs:
        mol = dm.add_hs(mol)
    matching_bonds = self._fragment(mol, allow_empty=allow_empty)
    substructed_ignored = []
    if constraints is not None:
        substructed_ignored = list(
            itertools.chain(
                *[
                    mol.GetSubstructMatches(constraint, uniquify=True)
                    for constraint in constraints
                ]
            )
        )

    bonds = []
    for i_a, i_b in matching_bonds:
        # if both atoms of the bond are found in a disallowed substructure, we cannot consider them
        # on the other end, a bond between two substructure to preserved independently is perfectly fine
        if any((i_a in ignore_x and i_b in ignore_x) for ignore_x in substructed_ignored):
            continue
        obond = mol.GetBondBetweenAtoms(i_a, i_b)
        bonds.append(obond.GetIdx())

    if len(bonds) > 0:
        mol = Chem.FragmentOnBonds(
            mol,
            bonds,
            dummyLabels=[(i + bond_map_id, i + bond_map_id) for i in range(len(bonds))],
        )
    # here we need to be clever and disable rooted atom as the atom with mapping

    frags = list(Chem.GetMolFrags(mol, asMols=True))
    if should_randomize:
        frags = rng.permutation(frags).tolist()
    elif canonical:
        frags = sorted(
            frags,
            key=lambda x: x.GetNumAtoms(),
            reverse=True,
        )

    frags_str = []
    for frag in frags:
        non_map_atom_idxs = [
            atom.GetIdx() for atom in frag.GetAtoms() if atom.GetAtomicNum() != 0
        ]
        frags_str.append(
            Chem.MolToSmiles(
                frag,
                isomericSmiles=True,
                canonical=True,  # needs to always be true
                rootedAtAtom=non_map_atom_idxs[0] if non_map_atom_idxs else -1,
            )
        )

    scaffold_str = ".".join(frags_str)
    # EN: fix for https://github.com/datamol-io/safe/issues/37
    # we were using the wrong branch number count which did not take into account
    # possible change in digit utilization after bond slicing
    scf_branch_num = self._find_branch_number(scaffold_str) + branch_numbers

    # don't capture atom mapping in the scaffold
    attach_pos = set(re.findall(r"(\[\d+\*\]|!\[[^:]*:\d+\])", scaffold_str))
    # Set iteration made non-canonical encodings, and therefore seeded
    # model prompts, depend on PYTHONHASHSEED. Retain the historical seed-0
    # ordering explicitly while canonical encodings keep ascending order.
    attach_pos = sorted(attach_pos, reverse=not canonical)
    starting_num = 1 if len(scf_branch_num) == 0 else max(scf_branch_num) + 1
    for attach in attach_pos:
        val = self._format_ring_closure(starting_num)
        # we cannot have anything of the form "\([@=-#-$/\]*\d+\)"
        attach_regexp = re.compile(r"(" + re.escape(attach) + r")")
        # check if we have at least 2 matches, if not, we have a dummy
        n_matches = len(attach_regexp.findall(scaffold_str))
        attachment_match = re.fullmatch(r"\[(\d+)\*\]", attach)
        is_explicit_attachment = (
            attachment_match is not None
            and int(attachment_match.group(1)) in open_attachment_ids
        )
        scaffold_str = (
            attach_regexp.sub(val, scaffold_str)
            if n_matches > 1 or is_explicit_attachment
            else scaffold_str.replace(attach, "*")
        )
        starting_num += 1

    # now we need to remove all the parenthesis around digit only number
    wrong_attach = re.compile(r"(?<!%)\((%\(\d+\)|[\%\d]*)\)")
    scaffold_str = wrong_attach.sub(r"\g<1>", scaffold_str)
    # furthermore, we autoapply rdkit-compatible digit standardization.
    if rdkit_safe:
        pattern = r"\(([=-@#\/\\]{0,2})(%\(\d+\)|%?\d{1,2})\)"
        replacement = r"\g<1>\g<2>"
        scaffold_str = re.sub(pattern, replacement, scaffold_str)
    if not self.ignore_stereo and has_specified_stereo:
        source_graph = self._canonical_isomeric_graph(source_mol)
        encoded_graph = self._canonical_isomeric_graph(
            self.decoder(
                scaffold_str,
                canonical=True,
                remove_dummies=False,
            )
        )
        if source_graph is None or source_graph != encoded_graph:
            # Some constrained stereochemical systems can change their
            # RDKit assignment after fragmentation even when no directly
            # stereogenic bond was cut. Preserve the valid input intact.
            if source_text is not None:
                return source_text
            return Chem.MolToSmiles(
                source_mol,
                canonical=True,
                isomericSmiles=True,
            )
    return scaffold_str

randomize(mol: dm.Mol, rng: Optional[int] = None) staticmethod

Randomize the position of the atoms in a mol.

Parameters:

• mol (Mol) –

  molecules to randomize

• rng (Optional[int], default: None ) –

  optional seed to use

Source code in safe/converter.py

@staticmethod
def randomize(mol: dm.Mol, rng: Optional[int] = None):
    """Randomize the position of the atoms in a mol.

    Args:
        mol: molecules to randomize
        rng: optional seed to use
    """
    if isinstance(rng, int):
        rng = np.random.default_rng(rng)
    elif rng is None:
        rng = np.random.default_rng()
    if mol.GetNumAtoms() == 0:
        return mol
    atom_indices = list(range(mol.GetNumAtoms()))
    atom_indices = rng.permutation(atom_indices).tolist()
    return Chem.RenumberAtoms(mol, atom_indices)

encode(inp: Union[str, dm.Mol], canonical: bool = True, randomize: Optional[bool] = False, seed: Optional[int] = None, slicer: Optional[Union[List[str], str, Callable]] = None, require_hs: Optional[bool] = None, constraints: Optional[List[dm.Mol]] = None, ignore_stereo: Optional[bool] = False, allow_empty: bool = False)

Convert input smiles to SAFE representation

Parameters:

• inp (Union[str, Mol]) –

  input smiles

• canonical (bool, default: True ) –

  whether to return canonical SAFE string. Defaults to True

• randomize (Optional[bool], default: False ) –

  whether to randomize the safe string encoding. Will be ignored if canonical is provided

• seed (Optional[int], default: None ) –

  optional seed to use when allowing randomization of the SAFE encoding.

• slicer (Optional[Union[List[str], str, Callable]], default: None ) –

  slicer algorithm to use for encoding. Defaults to "brics".

• require_hs (Optional[bool], default: None ) –

  whether the slicing algorithm require the molecule to have hydrogen explictly added.

• constraints (Optional[List[Mol]], default: None ) –

  List of molecules or pattern to preserve during the SAFE construction.

• ignore_stereo (Optional[bool], default: False ) –

  whether to discard input stereochemistry explicitly. When false, stereochemistry-changing cuts are skipped and the encoded graph is verified.

• allow_empty (bool, default: False ) –

  whether to tolerate molecules the slicer cannot cut. When True, an input with no breakable bonds (for example a rigid ring, a single atom, or the components of a salt) is returned as a single unfragmented SAFE block instead of raising SAFEFragmentationError.

Source code in safe/converter.py

def encode(
    inp: Union[str, dm.Mol],
    canonical: bool = True,
    randomize: Optional[bool] = False,
    seed: Optional[int] = None,
    slicer: Optional[Union[List[str], str, Callable]] = None,
    require_hs: Optional[bool] = None,
    constraints: Optional[List[dm.Mol]] = None,
    ignore_stereo: Optional[bool] = False,
    allow_empty: bool = False,
):
    """
    Convert input smiles to SAFE representation

    Args:
        inp: input smiles
        canonical: whether to return canonical SAFE string. Defaults to True
        randomize: whether to randomize the safe string encoding. Will be ignored if canonical is provided
        seed: optional seed to use when allowing randomization of the SAFE encoding.
        slicer: slicer algorithm to use for encoding. Defaults to "brics".
        require_hs: whether the slicing algorithm require the molecule to have hydrogen explictly added.
        constraints: List of molecules or pattern to preserve during the SAFE construction.
        ignore_stereo: whether to discard input stereochemistry explicitly. When false,
            stereochemistry-changing cuts are skipped and the encoded graph is verified.
        allow_empty: whether to tolerate molecules the slicer cannot cut. When True,
            an input with no breakable bonds (for example a rigid ring, a single atom,
            or the components of a salt) is returned as a single unfragmented SAFE
            block instead of raising ``SAFEFragmentationError``.
    """
    if slicer is None:
        slicer = "brics"
    with dm.without_rdkit_log():
        safe_obj = SAFEConverter(slicer=slicer, require_hs=require_hs, ignore_stereo=ignore_stereo)
        try:
            encoded = safe_obj.encoder(
                inp,
                canonical=canonical,
                randomize=randomize,
                constraints=constraints,
                seed=seed,
                allow_empty=allow_empty,
            )
        except (SAFEEncodeError, SAFEFragmentationError) as e:
            raise e
        except Exception as e:
            raise SAFEEncodeError(f"Failed to encode {inp} with {slicer}") from e
        return encoded

decode(safe_str: str, as_mol: bool = False, canonical: bool = False, fix: bool = True, remove_added_hs: bool = True, remove_dummies: bool = True, ignore_errors: bool = False)

Convert input SAFE representation to smiles Args: safe_str: input SAFE representation to decode as a valid molecule or smiles as_mol: whether to return a molecule object or a smiles string canonical: whether to return a canonical smiles or a randomized smiles fix: whether to fix the SAFE representation to take into account non-connected attachment points remove_added_hs: whether to remove the hydrogen atoms that have been added to fix the string. remove_dummies: whether to remove dummy atoms from the SAFE representation ignore_errors: whether to ignore error and return None on decoding failure or raise an error

Source code in safe/converter.py

def decode(
    safe_str: str,
    as_mol: bool = False,
    canonical: bool = False,
    fix: bool = True,
    remove_added_hs: bool = True,
    remove_dummies: bool = True,
    ignore_errors: bool = False,
):
    """Convert input SAFE representation to smiles
    Args:
        safe_str: input SAFE representation to decode as a valid molecule or smiles
        as_mol: whether to return a molecule object or a smiles string
        canonical: whether to return a canonical smiles or a randomized smiles
        fix: whether to fix the SAFE representation to take into account non-connected attachment points
        remove_added_hs: whether to remove the hydrogen atoms that have been added to fix the string.
        remove_dummies: whether to remove dummy atoms from the SAFE representation
        ignore_errors: whether to ignore error and return None on decoding failure or raise an error

    """
    with dm.without_rdkit_log():
        safe_obj = SAFEConverter()
        try:
            decoded = safe_obj.decoder(
                safe_str,
                as_mol=as_mol,
                canonical=canonical,
                fix=fix,
                remove_dummies=remove_dummies,
                remove_added_hs=remove_added_hs,
            )

        except Exception as e:
            if ignore_errors:
                return None
            raise SAFEDecodeError(f"Failed to decode {safe_str}") from e
        return decoded

---

SAFE Design

SAFEDesign(model: Union[SAFEDoubleHeadsModel, str], tokenizer: Union[str, SAFETokenizer], generation_config: Optional[Union[str, GenerationConfig]] = None, safe_encoder: Optional[sf.SAFEConverter] = None, verbose: bool = True)

Design molecules with a pretrained SAFE language model.

SAFEDesign constructor

Info

Design methods in SAFE are not deterministic when it comes to the token sampling step. If a method accepts a random_seed, it's for the SAFE-related algorithms and not the sampling from the autoregressive model. To ensure you get a deterministic sampling, please set the seed at the transformers package level.

import safe as sf
import transformers
my_seed = 100
designer = sf.SAFEDesign(...)

transformers.set_seed(100) # use this before calling a design function
designer.linker_generation(...)

Parameters:

• model (Union[SAFEDoubleHeadsModel, str]) –

  input SAFEDoubleHeadsModel to use for generation

• tokenizer (Union[str, SAFETokenizer]) –

  input SAFETokenizer to use for generation

• generation_config (Optional[Union[str, GenerationConfig]], default: None ) –

  input GenerationConfig to use for generation

• safe_encoder (Optional[SAFEConverter], default: None ) –

  custom safe encoder to use

• verbose (bool, default: True ) –

  whether to print out logging information during generation

Source code in safe/sample.py

def __init__(
    self,
    model: Union[SAFEDoubleHeadsModel, str],
    tokenizer: Union[str, SAFETokenizer],
    generation_config: Optional[Union[str, GenerationConfig]] = None,
    safe_encoder: Optional[sf.SAFEConverter] = None,
    verbose: bool = True,
):
    """SAFEDesign constructor

    !!! info
        Design methods in SAFE are not deterministic when it comes to the token sampling step.
        If a method accepts a `random_seed`, it's for the SAFE-related algorithms and not the
        sampling from the autoregressive model. To ensure you get a deterministic sampling,
        please set the seed at the `transformers` package level.

        ```python
        import safe as sf
        import transformers
        my_seed = 100
        designer = sf.SAFEDesign(...)

        transformers.set_seed(100) # use this before calling a design function
        designer.linker_generation(...)
        ```


    Args:
        model: input SAFEDoubleHeadsModel to use for generation
        tokenizer: input SAFETokenizer to use for generation
        generation_config: input GenerationConfig to use for generation
        safe_encoder: custom safe encoder to use
        verbose: whether to print out logging information during generation
    """

    if isinstance(model, (str, os.PathLike)):
        model = SAFEDoubleHeadsModel.from_pretrained(model)

    if isinstance(tokenizer, (str, os.PathLike)):
        tokenizer = SAFETokenizer.load(tokenizer)

    model.eval()
    self.model = model
    self.tokenizer = tokenizer
    if isinstance(generation_config, (str, os.PathLike)):
        generation_config = GenerationConfig.from_pretrained(generation_config)
    if generation_config is None:
        generation_config = GenerationConfig.from_model_config(model.config)
    self.generation_config = generation_config
    for special_token_id in ["bos_token_id", "eos_token_id", "pad_token_id"]:
        if getattr(self.generation_config, special_token_id) is None:
            setattr(
                self.generation_config, special_token_id, getattr(tokenizer, special_token_id)
            )

    self.verbose = verbose
    self.safe_encoder = safe_encoder or sf.SAFEConverter()
    self._constrained_generator = None

generation_config = generation_config instance-attribute

model = model instance-attribute

safe_encoder = safe_encoder or sf.SAFEConverter() instance-attribute

tokenizer = tokenizer instance-attribute

verbose = verbose instance-attribute

__mix_sequences(prefix_sequences: List[str], suffix_sequences: List[str], prefix: str, suffix: str, n_samples: int, mol_linker_slicer)

Use generated prefix and suffix sequences to form new molecules that will be the merging of both. This is the two step scaffold morphing and linker generation scheme Args: prefix_sequences: list of prefix sequences suffix_sequences: list of suffix sequences prefix: decoded smiles of the prefix suffix: decoded smiles of the suffix n_samples: number of samples to generate

Source code in safe/sample.py

def __mix_sequences(
    self,
    prefix_sequences: List[str],
    suffix_sequences: List[str],
    prefix: str,
    suffix: str,
    n_samples: int,
    mol_linker_slicer,
):
    """Use generated prefix and suffix sequences to form new molecules
    that will be the merging of both. This is the two step scaffold morphing and linker generation scheme
    Args:
        prefix_sequences: list of prefix sequences
        suffix_sequences: list of suffix sequences
        prefix: decoded smiles of the prefix
        suffix: decoded smiles of the suffix
        n_samples: number of samples to generate
    """
    prefix_linkers = []
    suffix_linkers = []
    prefix_query = dm.from_smarts(prefix)
    suffix_query = dm.from_smarts(suffix)

    for x in prefix_sequences:
        molecule = dm.to_mol(x)
        if molecule is not None:
            prefix_linkers.append(mol_linker_slicer(molecule, prefix_query)[1])
    for x in suffix_sequences:
        molecule = dm.to_mol(x)
        if molecule is not None:
            suffix_linkers.append(mol_linker_slicer(molecule, suffix_query)[1])
    linked = []
    linkers = dict.fromkeys(
        linker for linker in prefix_linkers + suffix_linkers if linker is not None
    )
    for linker in linkers:
        linked.extend(mol_linker_slicer.link_fragments(linker, prefix, suffix))
        linked = list(dict.fromkeys(x for x in linked if x))
        if len(linked) >= n_samples:
            break
    return linked[:n_samples]

de_novo_generation(n_samples_per_trial: int = 10, sanitize: bool = False, n_trials: Optional[int] = None, refine: bool = False, **kwargs: Optional[Dict[Any, Any]])

Perform de novo generation using the pretrained SAFE model.

De novo generation is equivalent to not having any prefix.

Parameters:

• n_samples_per_trial (int, default: 10 ) –

  number of new molecules to generate

• sanitize (bool, default: False ) –

  whether to perform sanitization, aka, perform control to ensure what is asked is what is returned

• n_trials (Optional[int], default: None ) –

  number of randomization to perform

• refine (bool, default: False ) –

  quality mode. Oversample, then return only valid, deduplicated molecules up to the requested count; for completion tasks the result is also constrained to a single connected molecule.

• kwargs (Optional[Dict[Any, Any]], default: {} ) –

  any argument to provide to the underlying generation function

Source code in safe/sample.py

def de_novo_generation(
    self,
    n_samples_per_trial: int = 10,
    sanitize: bool = False,
    n_trials: Optional[int] = None,
    refine: bool = False,
    **kwargs: Optional[Dict[Any, Any]],
):
    """Perform de novo generation using the pretrained SAFE model.

    De novo generation is equivalent to not having any prefix.

    Args:
        n_samples_per_trial: number of new molecules to generate
        sanitize: whether to perform sanitization, aka, perform control to ensure what is asked is what is returned
        n_trials: number of randomization to perform
        refine: quality mode. Oversample, then
            return only valid, deduplicated molecules up to the requested count; for
            completion tasks the result is also constrained to a single connected molecule.
        kwargs: any argument to provide to the underlying generation function
    """
    kwargs.setdefault("how", "random")
    if kwargs["how"] != "random" and not kwargs.get("do_sample"):
        logger.warning(
            "Deterministic decoding can return repeated de novo samples; use "
            "do_sample=True or how='random' when diversity is required"
        )

    total_sequences = []
    n_trials = n_trials or 1
    candidates_per_trial = self._candidate_count(n_samples_per_trial, refine)
    for _ in tqdm(range(n_trials), disable=(not self.verbose), leave=False):
        sequences = self._generate(n_samples=candidates_per_trial, **kwargs)
        total_sequences.extend(sequences)
    total_sequences = self._decode_safe(
        total_sequences, canonical=True, remove_invalid=sanitize or refine
    )

    if sanitize and self.verbose:
        logger.info(
            f"After sanitization, {len(total_sequences)} / {n_samples_per_trial*n_trials} ({len(total_sequences)*100/(n_samples_per_trial*n_trials):.2f} %) generated molecules are valid !"
        )
    return self._finalize_samples(
        total_sequences,
        n_samples_per_trial * n_trials,
        refine,
    )

linker_generation(*groups: Union[str, dm.Mol], n_samples_per_trial: int = 10, n_trials: Optional[int] = 1, sanitize: bool = False, do_not_fragment_further: Optional[bool] = True, random_seed: Optional[int] = None, model_only: Optional[bool] = False, refine: bool = False, **kwargs: Optional[Dict[Any, Any]])

Perform linker generation using the pretrained SAFE model. Linker generation is really just scaffold morphing underlying.

Parameters:

• groups (Union[str, Mol], default: () ) –

  list of fragments to link together, they are joined in the order provided

• n_samples_per_trial (int, default: 10 ) –

  number of new molecules to generate for each randomization

• n_trials (Optional[int], default: 1 ) –

  number of randomization to perform

• do_not_fragment_further (Optional[bool], default: True ) –

  whether to fragment the scaffold further or not

• sanitize (bool, default: False ) –

  whether to sanitize the generated molecules

• random_seed (Optional[int], default: None ) –

  random seed to use

• model_only (Optional[bool], default: False ) –

  whether to use the model only ability and nothing more.

• refine (bool, default: False ) –

  quality mode. Oversample, then return only valid, deduplicated molecules up to the requested count; for completion tasks the result is also constrained to a single connected molecule.

• kwargs (Optional[Dict[Any, Any]], default: {} ) –

  any argument to provide to the underlying generation function

Source code in safe/sample.py

def linker_generation(
    self,
    *groups: Union[str, dm.Mol],
    n_samples_per_trial: int = 10,
    n_trials: Optional[int] = 1,
    sanitize: bool = False,
    do_not_fragment_further: Optional[bool] = True,
    random_seed: Optional[int] = None,
    model_only: Optional[bool] = False,
    refine: bool = False,
    **kwargs: Optional[Dict[Any, Any]],
):
    """Perform linker generation using the pretrained SAFE model.
    Linker generation is really just scaffold morphing underlying.

    Args:
        groups: list of fragments to link together, they are joined in the order provided
        n_samples_per_trial: number of new molecules to generate for each randomization
        n_trials: number of randomization to perform
        do_not_fragment_further: whether to fragment the scaffold further or not
        sanitize: whether to sanitize the generated molecules
        random_seed: random seed to use
        model_only: whether to use the model only ability and nothing more.
        refine: quality mode. Oversample, then
            return only valid, deduplicated molecules up to the requested count; for
            completion tasks the result is also constrained to a single connected molecule.
        kwargs: any argument to provide to the underlying generation function
    """
    side_chains = list(groups)

    if len(side_chains) != 2:
        raise ValueError(
            "Linker generation only works when providing two groups as side chains"
        )

    return self._fragment_linking(
        side_chains=side_chains,
        n_samples_per_trial=n_samples_per_trial,
        n_trials=n_trials,
        sanitize=sanitize,
        do_not_fragment_further=do_not_fragment_further,
        random_seed=random_seed,
        is_linking=True,
        model_only=model_only,
        refine=refine,
        **kwargs,
    )

load_default(model_dir: Optional[str] = None, model_revision: Optional[str] = None, device: str = None, verbose: bool = False, **kwargs: Any) -> SAFEDesign classmethod

Load default SAFEGenerator model

Parameters:

• verbose (bool, default: False ) –

  whether to print out logging information during generation

• model_dir (Optional[str], default: None ) –

  Optional path to model folder to use instead of the default one. If provided the tokenizer should be in the model_dir named as tokenizer.json

• model_revision (Optional[str], default: None ) –

  Hugging Face revision to load. The reviewed default model revision is pinned when model_dir is omitted.

• device (str, default: None ) –

  optional device where to move the model

• kwargs (Any, default: {} ) –

  any additional argument to pass to the init function

Source code in safe/sample.py

@classmethod
def load_default(
    cls,
    model_dir: Optional[str] = None,
    model_revision: Optional[str] = None,
    device: str = None,
    verbose: bool = False,
    **kwargs: Any,
) -> "SAFEDesign":
    """Load default SAFEGenerator model

    Args:
        verbose: whether to print out logging information during generation
        model_dir: Optional path to model folder to use instead of the default one.
            If provided the tokenizer should be in the model_dir named as `tokenizer.json`
        model_revision: Hugging Face revision to load. The reviewed default
            model revision is pinned when `model_dir` is omitted.
        device: optional device where to move the model
        kwargs: any additional argument to pass to the init function
    """
    use_default_model = model_dir is None or not model_dir
    if use_default_model:
        model_dir = cls._DEFAULT_MODEL_PATH
        model_revision = model_revision or cls._DEFAULT_MODEL_REVISION
    load_kwargs = {"revision": model_revision} if model_revision is not None else {}
    model = SAFEDoubleHeadsModel.from_pretrained(model_dir, **load_kwargs)
    tokenizer = SAFETokenizer.from_pretrained(model_dir, **load_kwargs)
    gen_config = GenerationConfig.from_pretrained(model_dir, **load_kwargs)
    if device is not None:
        model = model.to(device)
    return cls(
        model=model,
        tokenizer=tokenizer,
        generation_config=gen_config,
        verbose=verbose,
        **kwargs,
    )

load_from_wandb(artifact_path: str, device: Optional[str] = None, verbose: bool = True, **kwargs: Any) -> SAFEDesign classmethod

Load a SAFE model and tokenizer from a Weights & Biases artifact.

When SAFE_MODEL_ROOT is set, the artifact is downloaded into that directory.

Parameters:

• artifact_path (str) –

  The path to the wandb artifact in the format entity/project/artifact:version.

• device (Optional[str], default: None ) –

  The device where the model should be loaded ('cpu' or 'cuda'). If None, it defaults to the available device.

• verbose (bool, default: True ) –

  Whether to print out logging information during generation.

Returns:

• SAFEDesign ( SAFEDesign ) –

  An instance of SAFEDesign class with the model, tokenizer, and generation config loaded from wandb.

Source code in safe/sample.py

@classmethod
def load_from_wandb(
    cls, artifact_path: str, device: Optional[str] = None, verbose: bool = True, **kwargs: Any
) -> "SAFEDesign":
    """Load a SAFE model and tokenizer from a Weights & Biases artifact.

    When ``SAFE_MODEL_ROOT`` is set, the artifact is downloaded into that
    directory.

    Args:
        artifact_path: The path to the wandb artifact in the format `entity/project/artifact:version`.
        device: The device where the model should be loaded ('cpu' or 'cuda'). If None, it defaults to the available device.
        verbose: Whether to print out logging information during generation.

    Returns:
        SAFEDesign: An instance of SAFEDesign class with the model, tokenizer, and generation config loaded from wandb.
    """
    import wandb

    artifact_path = artifact_path.replace("wandb://", "")

    # Parse the artifact path to extract project and artifact name
    parts = artifact_path.split("/", 1)
    if len(parts) > 1:
        project_name, artifact_name = parts
    else:
        project_name = os.getenv("SAFE_WANDB_PROJECT", "safe-models")
        artifact_name = artifact_path

    if ":" not in artifact_name:
        artifact_name += ":latest"

    artifact_path = f"{project_name}/{artifact_name}"

    # Check if SAFE_MODEL_ROOT environment variable is defined
    cache_path = os.getenv("SAFE_MODEL_ROOT", None)
    if cache_path is not None:
        # Ensure the cache path exists
        cache_path = Path(cache_path)
        cache_path.mkdir(parents=True, exist_ok=True)
        artifact_subfolder = artifact_path.replace("/", "_").replace(":", "_")
        cache_dir = cache_path / artifact_subfolder
        cache_path = cache_dir.as_posix()

    api = wandb.Api()
    # Download the artifact from wandb to the cache directory
    artifact = api.artifact(artifact_path, type="model")
    artifact_dir = artifact.download(root=cache_path)

    # Load the model, tokenizer, and generation config from the artifact directory
    model = SAFEDoubleHeadsModel.from_pretrained(artifact_dir)
    tokenizer = SAFETokenizer.from_pretrained(artifact_dir)
    gen_config = GenerationConfig.from_pretrained(artifact_dir)

    # Move model to the specified device if provided
    if device is not None:
        model = model.to(device)

    return cls(
        model=model,
        tokenizer=tokenizer,
        generation_config=gen_config,
        verbose=verbose,
        **kwargs,
    )

motif_extension(motif: Union[str, dm.Mol], n_samples_per_trial: int = 10, n_trials: Optional[int] = 1, sanitize: bool = False, do_not_fragment_further: Optional[bool] = True, random_seed: Optional[int] = None, refine: bool = False, **kwargs: Optional[Dict[Any, Any]])

Deprecated alias for :meth:scaffold_decoration.

Source code in safe/sample.py

def motif_extension(
    self,
    motif: Union[str, dm.Mol],
    n_samples_per_trial: int = 10,
    n_trials: Optional[int] = 1,
    sanitize: bool = False,
    do_not_fragment_further: Optional[bool] = True,
    random_seed: Optional[int] = None,
    refine: bool = False,
    **kwargs: Optional[Dict[Any, Any]],
):
    """Deprecated alias for :meth:`scaffold_decoration`."""
    warnings.warn(
        "motif_extension() is an alias of scaffold_decoration() and will be removed in "
        "SAFE 2.0; call scaffold_decoration() directly.",
        FutureWarning,
        stacklevel=2,
    )
    return self.scaffold_decoration(
        motif,
        n_samples_per_trial=n_samples_per_trial,
        n_trials=n_trials,
        sanitize=sanitize,
        do_not_fragment_further=do_not_fragment_further,
        random_seed=random_seed,
        add_dot=True,
        refine=refine,
        **kwargs,
    )

pattern_decoration(scaffold: Union[str, dm.Mol], n_samples_per_trial: int = 10, n_trials: int = 1, do_not_fragment_further: bool = True, sanitize: bool = False, random_seed: Optional[int] = None, add_dot: bool = True, n_scaff_random: Optional[int] = 3, n_scaff_samples: Optional[int] = 10, scaff_temperature: float = 1.0, refine: bool = False, **kwargs: Optional[Dict[Any, Any]]) -> List[str]

Perform pattern decoration using the pretrained SAFE model. The pattern decoration algorithm works by first examplifying the patterns as a set of scaffold then performing scaffold decoration on each scaffold.

Warning

Designing molecules from a given molecule pattern is more challenging than fragment-constrained design. SAFE does not currently support complex SMARTS pattern schemes (e.g., valence or connectivity constraints, some ring constraints). This function works best when sampling given a list of atoms. However, sampling depends on the model's conditional probabilities, meaning that if the model assigns zero probability to a token, you are unlikely to see it.

Parameters:

• scaffold (Union[str, Mol]) –

  Scaffold (with attachment points) to decorate.

• n_samples_per_trial (int, default: 10 ) –

  Number of new molecules to generate for each randomization.

• n_trials (int, default: 1 ) –

  Number of randomizations to perform.

• do_not_fragment_further (bool, default: True ) –

  Whether to prevent further fragmentation of the scaffold.

• sanitize (bool, default: False ) –

  Whether to sanitize the generated molecules and ensure the scaffold is present.

• random_seed (Optional[int], default: None ) –

  Seed for randomization.

• n_scaff_random (Optional[int], default: 3 ) –

  Number of scaffold randomizations to try (to reposition constraints in the string and increase rollout likelihood). Increasing this will improve sampling, but will require more time.

• n_scaff_samples (Optional[int], default: 10 ) –

  Maximum number of samples to sample for a given scaffold from the pattern. Increasing this will make sure you have more diversity in the scaffold coming from the pattern

• scaff_temperature (float, default: 1.0 ) –

  Temperature to use when sampling valid scaffolds from the pattern. Higher temperature means more diverse scaffold

• refine (bool, default: False ) –

  quality mode. Oversample, then return only valid, deduplicated molecules up to the requested count; for completion tasks the result is also constrained to a single connected molecule.

• kwargs (Optional[Dict[Any, Any]], default: {} ) –

  Additional arguments for the underlying generation function.

Returns:

• List[str] –

  List of decorated molecule sequences.

Source code in safe/sample.py

def pattern_decoration(
    self,
    scaffold: Union[str, dm.Mol],
    n_samples_per_trial: int = 10,
    n_trials: int = 1,
    do_not_fragment_further: bool = True,
    sanitize: bool = False,
    random_seed: Optional[int] = None,
    add_dot: bool = True,
    n_scaff_random: Optional[int] = 3,
    n_scaff_samples: Optional[int] = 10,
    scaff_temperature: float = 1.0,
    refine: bool = False,
    **kwargs: Optional[Dict[Any, Any]],
) -> List[str]:
    """
    Perform pattern decoration using the pretrained SAFE model. The pattern decoration algorithm works by first examplifying the patterns
    as a set of scaffold then performing scaffold decoration on each scaffold.

    !!! warning
        Designing molecules from a given molecule pattern is more challenging than fragment-constrained design.
        SAFE does not currently support complex SMARTS pattern schemes (e.g., valence or connectivity constraints, some ring constraints).
        This function works best when sampling given a list of atoms. However, sampling depends on the model's conditional probabilities,
        meaning that if the model assigns zero probability to a token, you are unlikely to see it.

    Args:
        scaffold: Scaffold (with attachment points) to decorate.
        n_samples_per_trial: Number of new molecules to generate for each randomization.
        n_trials: Number of randomizations to perform.
        do_not_fragment_further: Whether to prevent further fragmentation of the scaffold.
        sanitize: Whether to sanitize the generated molecules and ensure the scaffold is present.
        random_seed: Seed for randomization.
        n_scaff_random: Number of scaffold randomizations to try (to reposition constraints in the string and increase rollout likelihood).
            Increasing this will improve sampling, but will require more time.
        n_scaff_samples: Maximum number of samples to sample for a given scaffold from the pattern.
            Increasing this will make sure you have more diversity in the scaffold coming from the pattern
        scaff_temperature: Temperature to use when sampling valid scaffolds from the pattern. Higher temperature means more diverse scaffold
        refine: quality mode. Oversample, then
            return only valid, deduplicated molecules up to the requested count; for
            completion tasks the result is also constrained to a single connected molecule.
        kwargs: Additional arguments for the underlying generation function.

    Returns:
        List of decorated molecule sequences.
    """

    rng = random.Random(random_seed)
    n_trials = n_trials or 1
    smarts_scaffolds = [scaffold]
    if n_scaff_random and n_scaff_random > 0:
        smarts_scaffolds = PatternConstraint.randomize(
            scaffold,
            n_scaff_random,
            seed=random_seed,
        )

    all_scaffolds = {}
    scaffold_sample_count = (
        n_samples_per_trial
        if n_scaff_samples is None
        else min(n_samples_per_trial, n_scaff_samples)
    )
    for sm in smarts_scaffolds:
        cur_dec_pattern = PatternConstraint(sm, self.tokenizer, temperature=scaff_temperature)
        decorator = PatternSampler(self.model, cur_dec_pattern)
        cur_scaffolds = decorator.sample_scaffolds(
            n_samples=scaffold_sample_count,
            n_trials=1,
            random_seed=rng.randint(1, 2**32 - 1),
        )
        all_scaffolds.update(dict.fromkeys(cur_scaffolds))

    # Pattern sampling resolves atom queries to concrete tokens. Parse the
    # result as a molecule rather than preserving it as a query molecule:
    # completion requires a chemically valid molecular graph.
    parsed_scaffolds = []
    for sampled_scaffold in all_scaffolds:
        scaffold_mol = dm.to_mol(sampled_scaffold, remove_hs=False)
        if scaffold_mol is not None:
            parsed_scaffolds.append(scaffold_mol)

    total_sequences = []
    for scaffold_mol in parsed_scaffolds:
        with dm.without_rdkit_log():
            cur_sequences = self._completion(
                fragment=scaffold_mol,
                n_samples_per_trial=int(n_samples_per_trial / max(len(parsed_scaffolds), 1))
                + 1,
                n_trials=n_trials,
                do_not_fragment_further=do_not_fragment_further,
                sanitize=sanitize or refine,
                random_seed=rng.randint(1, 2**32 - 1),
                add_dot=add_dot,
                refine=refine,
                **kwargs,
            )
            total_sequences.extend(cur_sequences)

    rng.shuffle(total_sequences)
    if sanitize or refine:
        total_sequences = sf.utils.filter_by_substructure_constraints(total_sequences, scaffold)
        if self.verbose:
            logger.info(
                f"After sanitization, {len(total_sequences)} / {n_samples_per_trial * n_trials} "
                f"({len(total_sequences) * 100 / (n_samples_per_trial * n_trials):.2f}%) generated molecules are valid!"
            )

    return self._finalize_samples(
        total_sequences,
        n_samples_per_trial * n_trials,
        refine,
    )

scaffold_decoration(scaffold: Union[str, dm.Mol], n_samples_per_trial: int = 10, n_trials: Optional[int] = 1, do_not_fragment_further: Optional[bool] = True, sanitize: bool = False, random_seed: Optional[int] = None, add_dot: Optional[bool] = True, refine: bool = False, **kwargs: Optional[Dict[Any, Any]])

Perform scaffold decoration using the pretrained SAFE model

For scaffold decoration, we basically starts with a prefix with the attachment point. We first convert the prefix into valid safe string.

Parameters:

• scaffold (Union[str, Mol]) –

  scaffold (with attachment points) to decorate

• n_samples_per_trial (int, default: 10 ) –

  number of new molecules to generate for each randomization

• n_trials (Optional[int], default: 1 ) –

  number of randomization to perform

• do_not_fragment_further (Optional[bool], default: True ) –

  whether to fragment the scaffold further or not

• sanitize (bool, default: False ) –

  whether to sanitize the generated molecules and check if the scaffold is still present

• random_seed (Optional[int], default: None ) –

  random seed to use

• refine (bool, default: False ) –

  quality mode. Oversample, then return only valid, deduplicated molecules up to the requested count; for completion tasks the result is also constrained to a single connected molecule.

• kwargs (Optional[Dict[Any, Any]], default: {} ) –

  any argument to provide to the underlying generation function

Source code in safe/sample.py

def scaffold_decoration(
    self,
    scaffold: Union[str, dm.Mol],
    n_samples_per_trial: int = 10,
    n_trials: Optional[int] = 1,
    do_not_fragment_further: Optional[bool] = True,
    sanitize: bool = False,
    random_seed: Optional[int] = None,
    add_dot: Optional[bool] = True,
    refine: bool = False,
    **kwargs: Optional[Dict[Any, Any]],
):
    """Perform scaffold decoration using the pretrained SAFE model

    For scaffold decoration, we basically starts with a prefix with the attachment point.
    We first convert the prefix into valid safe string.

    Args:
        scaffold: scaffold (with attachment points) to decorate
        n_samples_per_trial: number of new molecules to generate for each randomization
        n_trials: number of randomization to perform
        do_not_fragment_further: whether to fragment the scaffold further or not
        sanitize: whether to sanitize the generated molecules and check if the scaffold is still present
        random_seed: random seed to use
        refine: quality mode. Oversample, then
            return only valid, deduplicated molecules up to the requested count; for
            completion tasks the result is also constrained to a single connected molecule.
        kwargs: any argument to provide to the underlying generation function
    """

    n_trials = n_trials or 1
    total_sequences = self._completion(
        fragment=scaffold,
        n_samples_per_trial=n_samples_per_trial,
        n_trials=n_trials,
        do_not_fragment_further=do_not_fragment_further,
        sanitize=sanitize or refine,
        random_seed=random_seed,
        add_dot=add_dot,
        refine=refine,
        **kwargs,
    )
    # if we require sanitization
    # then we should filter out molecules that do not match the requested
    if sanitize or refine:
        total_sequences = sf.utils.filter_by_substructure_constraints(total_sequences, scaffold)
        if self.verbose:
            logger.info(
                f"After sanitization, {len(total_sequences)} / {n_samples_per_trial*n_trials} ({len(total_sequences)*100/(n_samples_per_trial*n_trials):.2f} %)  generated molecules are valid !"
            )
    return self._finalize_samples(
        total_sequences,
        n_samples_per_trial * n_trials,
        refine,
    )

scaffold_morphing(side_chains: Optional[Union[dm.Mol, str, List[Union[str, dm.Mol]]]] = None, mol: Optional[Union[dm.Mol, str]] = None, core: Optional[Union[dm.Mol, str]] = None, n_samples_per_trial: int = 10, n_trials: Optional[int] = 1, sanitize: bool = False, do_not_fragment_further: Optional[bool] = True, random_seed: Optional[int] = None, refine: bool = False, **kwargs: Optional[Dict[Any, Any]])

Perform scaffold morphing decoration using the pretrained SAFE model

For scaffold morphing, we try to replace the core by a new one. If the side_chains are provided, we use them. If a combination of molecule and core is provided, then, we use them to extract the side chains and performing the scaffold morphing then.

Finding the side chains

The algorithm to find the side chains from core assumes that the core we get as input has attachment points. Those attachment points are never considered as part of the query, rather they are used to define the attachment points. See ~sf.utils.compute_side_chains for more information.

Parameters:

• side_chains (Optional[Union[Mol, str, List[Union[str, Mol]]]], default: None ) –

  side chains to use to perform scaffold morphing (joining as best as possible the set of fragments)

• mol (Optional[Union[Mol, str]], default: None ) –

  input molecules when side_chains are not provided

• core (Optional[Union[Mol, str]], default: None ) –

  core to morph into another scaffold

• n_samples_per_trial (int, default: 10 ) –

  number of new molecules to generate for each randomization

• n_trials (Optional[int], default: 1 ) –

  number of randomization to perform

• do_not_fragment_further (Optional[bool], default: True ) –

  whether to fragment the scaffold further or not

• sanitize (bool, default: False ) –

  whether to sanitize the generated molecules

• random_seed (Optional[int], default: None ) –

  random seed to use

• refine (bool, default: False ) –

  quality mode. Oversample, then return only valid, deduplicated molecules up to the requested count; for completion tasks the result is also constrained to a single connected molecule.

• kwargs (Optional[Dict[Any, Any]], default: {} ) –

  any argument to provide to the underlying generation function

Source code in safe/sample.py

def scaffold_morphing(
    self,
    side_chains: Optional[Union[dm.Mol, str, List[Union[str, dm.Mol]]]] = None,
    mol: Optional[Union[dm.Mol, str]] = None,
    core: Optional[Union[dm.Mol, str]] = None,
    n_samples_per_trial: int = 10,
    n_trials: Optional[int] = 1,
    sanitize: bool = False,
    do_not_fragment_further: Optional[bool] = True,
    random_seed: Optional[int] = None,
    refine: bool = False,
    **kwargs: Optional[Dict[Any, Any]],
):
    """Perform scaffold morphing decoration using the pretrained SAFE model

    For scaffold morphing, we try to replace the core by a new one. If the side_chains are provided, we use them.
    If a combination of molecule and core is provided, then, we use them to extract the side chains and performing the
    scaffold morphing then.

    !!! note "Finding the side chains"
        The algorithm to find the side chains from core assumes that the core we get as input has attachment points.
        Those attachment points are never considered as part of the query, rather they are used to define the attachment points.
        See ~sf.utils.compute_side_chains for more information.

    Args:
        side_chains: side chains to use to perform scaffold morphing (joining as best as possible the set of fragments)
        mol: input molecules when side_chains are not provided
        core: core to morph into another scaffold
        n_samples_per_trial: number of new molecules to generate for each randomization
        n_trials: number of randomization to perform
        do_not_fragment_further: whether to fragment the scaffold further or not
        sanitize: whether to sanitize the generated molecules
        random_seed: random seed to use
        refine: quality mode. Oversample, then
            return only valid, deduplicated molecules up to the requested count; for
            completion tasks the result is also constrained to a single connected molecule.
        kwargs: any argument to provide to the underlying generation function
    """

    return self._fragment_linking(
        side_chains=side_chains,
        mol=mol,
        core=core,
        n_samples_per_trial=n_samples_per_trial,
        n_trials=n_trials,
        sanitize=sanitize,
        do_not_fragment_further=do_not_fragment_further,
        random_seed=random_seed,
        is_linking=False,
        refine=refine,
        **kwargs,
    )

super_structure(core: Union[str, dm.Mol], n_samples_per_trial: int = 10, n_trials: Optional[int] = 1, sanitize: bool = False, do_not_fragment_further: Optional[bool] = True, random_seed: Optional[int] = None, attachment_point_depth: Optional[int] = None, refine: bool = False, **kwargs: Optional[Dict[Any, Any]])

Perform super structure generation using the pretrained SAFE model.

To generate super-structure, we basically just create various attachment points to the input core, then perform scaffold decoration.

Parameters:

• core (Union[str, Mol]) –

  input substructure to use. We aim to generate super structures of this molecule

• n_samples_per_trial (int, default: 10 ) –

  number of new molecules to generate for each randomization

• n_trials (Optional[int], default: 1 ) –

  number of different attachment points to consider

• do_not_fragment_further (Optional[bool], default: True ) –

  whether to fragment the scaffold further or not

• sanitize (bool, default: False ) –

  whether to sanitize the generated molecules

• random_seed (Optional[int], default: None ) –

  random seed to use

• attachment_point_depth (Optional[int], default: None ) –

  depth of opening the attachment points. Increasing this, means you increase the number of substitution point to consider.

• refine (bool, default: False ) –

  quality mode. Oversample, then return only valid, deduplicated molecules up to the requested count; for completion tasks the result is also constrained to a single connected molecule.

• kwargs (Optional[Dict[Any, Any]], default: {} ) –

  any argument to provide to the underlying generation function

Source code in safe/sample.py

def super_structure(
    self,
    core: Union[str, dm.Mol],
    n_samples_per_trial: int = 10,
    n_trials: Optional[int] = 1,
    sanitize: bool = False,
    do_not_fragment_further: Optional[bool] = True,
    random_seed: Optional[int] = None,
    attachment_point_depth: Optional[int] = None,
    refine: bool = False,
    **kwargs: Optional[Dict[Any, Any]],
):
    """Perform super structure generation using the pretrained SAFE model.

    To generate super-structure, we basically just create various attachment points to the input core,
    then perform scaffold decoration.

    Args:
        core: input substructure to use. We aim to generate super structures of this molecule
        n_samples_per_trial: number of new molecules to generate for each randomization
        n_trials: number of different attachment points to consider
        do_not_fragment_further: whether to fragment the scaffold further or not
        sanitize: whether to sanitize the generated molecules
        random_seed: random seed to use
        attachment_point_depth: depth of opening the attachment points.
            Increasing this, means you increase the number of substitution point to consider.
        refine: quality mode. Oversample, then
            return only valid, deduplicated molecules up to the requested count; for
            completion tasks the result is also constrained to a single connected molecule.
        kwargs: any argument to provide to the underlying generation function
    """

    core = dm.to_mol(core)
    # Keep the original core: ``core`` is reassigned inside the trial loop
    # below, but the requested substructure to enforce is this input.
    requested_core = core
    cores = sf.utils.list_individual_attach_points(core, depth=attachment_point_depth)
    # get the fully open mol, everytime too.
    cores.append(dm.to_smiles(dm.reactions.open_attach_points(core)))
    cores = list(dict.fromkeys(cores))
    rng = random.Random(random_seed)
    rng.shuffle(cores)
    # now also get the single openining of an attachment point
    total_sequences = []
    n_trials = n_trials or 1
    for _ in tqdm(range(n_trials), disable=(not self.verbose), leave=False):
        core = cores[_ % len(cores)]
        try:
            with sf.utils.attr_as(self, "verbose", False):
                out = self._completion(
                    fragment=core,
                    n_samples_per_trial=n_samples_per_trial,
                    n_trials=1,
                    do_not_fragment_further=do_not_fragment_further,
                    sanitize=sanitize or refine,
                    random_seed=rng.randint(1, 2**32 - 1),
                    refine=refine,
                    **kwargs,
                )
                total_sequences.extend(out)
        except (sf.SAFEEncodeError, ValueError) as e:
            if self.verbose:
                logger.error(e)

    # Match the other constrained methods: verify the requested core is
    # actually present in the generated superstructures.
    if sanitize or refine:
        total_sequences = sf.utils.filter_by_substructure_constraints(
            total_sequences, requested_core
        )
    if sanitize and self.verbose:
        logger.info(
            f"After sanitization, {len(total_sequences)} / {n_samples_per_trial*n_trials} ({len(total_sequences)*100/(n_samples_per_trial*n_trials):.2f} %)  generated molecules are valid !"
        )
    return self._finalize_samples(
        total_sequences,
        n_samples_per_trial * n_trials,
        refine,
    )

---

SAFE Tokenizer

SAFESplitter(pattern=None)

Split a SAFE string into notation tokens.

Source code in safe/_tokenizer_utils.py

def __init__(self, pattern=None):
    if pattern is None:
        pattern = self.REGEX_PATTERN
    self.regex = re.compile(pattern)

REGEX_PATTERN = '(\\[[^\\]]+]|Br?|Cl?|N|O|S|P|F|I|b|c|n|o|s|p|\\(|\\)|\\.|=|#|-|\\+|\\\\|\\/|:|~|@|\\?|>>?|\\*|\\$|\\%\\([0-9]{1,5}\\)|\\%[0-9]{2}|[0-9])' class-attribute instance-attribute

name = 'safe' class-attribute instance-attribute

regex = re.compile(pattern) instance-attribute

detokenize(chars)

Reconstruct a SAFE string from tokens.

Source code in safe/_tokenizer_utils.py

def detokenize(self, chars):
    """Reconstruct a SAFE string from tokens."""
    if isinstance(chars, str):
        chars = chars.split(" ")
    return "".join(value.strip() for value in chars)

pre_tokenize(pretok)

Apply this splitter to a Hugging Face pretokenizer.

Source code in safe/_tokenizer_utils.py

def pre_tokenize(self, pretok):
    """Apply this splitter to a Hugging Face pretokenizer."""
    pretok.split(self.split)

split(_index, normalized)

Pretokenize a value for Hugging Face Tokenizers.

Source code in safe/_tokenizer_utils.py

def split(self, _index, normalized):
    """Pretokenize a value for Hugging Face Tokenizers."""
    return self.tokenize(normalized)

tokenize(line)

Tokenize a SAFE string.

Source code in safe/_tokenizer_utils.py

def tokenize(self, line):
    """Tokenize a SAFE string."""
    if isinstance(line, str):
        tokens = list(self.regex.findall(line))
        reconstruction = "".join(tokens)
        if line != reconstruction:
            logger.error(
                f"Tokens different from sample:\ntokens {reconstruction}\nsample {line}."
            )
            raise ValueError(line)
    else:
        idxs = re.finditer(self.regex, str(line))
        tokens = [line[match.start(0) : match.end(0)] for match in idxs]
    return tokens

SAFETokenizer(tokenizer_type: str = 'bpe', splitter: Optional[str] = 'safe', trainer_args=None, decoder_args=None, token_model_args=None)

Bases: PushToHubMixin

Class to initialize and train a tokenizer for SAFE string Once trained, you can use the converted version of the tokenizer to an HuggingFace PreTrainedTokenizerFast

Source code in safe/tokenizer.py

def __init__(
    self,
    tokenizer_type: str = "bpe",
    splitter: Optional[str] = "safe",
    trainer_args=None,
    decoder_args=None,
    token_model_args=None,
):
    super().__init__()
    self.tokenizer_type = tokenizer_type
    self.trainer_args = trainer_args or {}
    self.decoder_args = decoder_args or {}
    self.token_model_args = token_model_args or {}
    if tokenizer_type is not None and tokenizer_type.startswith("bpe"):
        self.model = BPE(unk_token=UNK_TOKEN, **self.token_model_args)
        self.trainer = BpeTrainer(special_tokens=SPECIAL_TOKENS, **self.trainer_args)

    else:
        self.model = WordLevel(unk_token=UNK_TOKEN, **self.token_model_args)
        self.trainer = WordLevelTrainer(special_tokens=SPECIAL_TOKENS, **self.trainer_args)

    self.tokenizer = Tokenizer(self.model)
    self.splitter = None
    if splitter == "safe":
        self.splitter = SAFESplitter()
        self.tokenizer.pre_tokenizer = PreTokenizer.custom(self.splitter)
    self.tokenizer.post_processor = TemplateProcessing(
        single=TEMPLATE_SINGLE,
        pair=TEMPLATE_PAIR,
        special_tokens=TEMPLATE_SPECIAL_TOKENS,
    )
    self.tokenizer.decoder = decoders.BPEDecoder(**self.decoder_args)
    self.tokenizer = self.set_special_tokens(self.tokenizer)

bos_token_id property

Get the bos token id

cls_token_id property

Get the cls token id

decoder_args = decoder_args or {} instance-attribute

eos_token_id property

Get the eos token id

mask_token_id property

Get the mask token id

model = BPE(unk_token=UNK_TOKEN, **self.token_model_args) instance-attribute

pad_token_id property

Get the pad token id

sep_token_id property

Get the sep token id

splitter = None instance-attribute

token_model_args = token_model_args or {} instance-attribute

tokenizer = self.set_special_tokens(self.tokenizer) instance-attribute

tokenizer_type = tokenizer_type instance-attribute

trainer = BpeTrainer(special_tokens=SPECIAL_TOKENS, **self.trainer_args) instance-attribute

trainer_args = trainer_args or {} instance-attribute

unk_token_id property

Get the unk token id

vocab_files_names: str = 'tokenizer.json' class-attribute instance-attribute

__getstate__()

Getting state to allow pickling

Source code in safe/tokenizer.py

def __getstate__(self):
    """Getting state to allow pickling"""
    with attr_as(self.tokenizer, "pre_tokenizer", Whitespace()):
        d = copy.deepcopy(self.__dict__)
    d["custom_pre_tokenizer"] = self.splitter is not None
    # copy back tokenizer level attribute
    d["tokenizer_attrs"] = self.tokenizer.__dict__.copy()
    d["tokenizer"].pre_tokenizer = Whitespace()
    return d

__len__()

Gets the count of tokens in vocab along with special tokens.

Source code in safe/tokenizer.py

def __len__(self):
    r"""
    Gets the count of tokens in vocab along with special tokens.
    """
    return len(self.tokenizer.get_vocab().keys())

__setstate__(d)

Setting state during reloading pickling

Source code in safe/tokenizer.py

def __setstate__(self, d):
    """Setting state during reloading pickling"""
    use_pretokenizer = d.get("custom_pre_tokenizer")
    if use_pretokenizer:
        d["tokenizer"].pre_tokenizer = PreTokenizer.custom(SAFESplitter())
    d["tokenizer"].__dict__.update(d.get("tokenizer_attrs", {}))
    self.__dict__.update(d)

decode(ids: list, skip_special_tokens: bool = True, ignore_stops: bool = False, stop_token_ids: Optional[List[int]] = None) -> str

Decodes a list of ids to molecular representation in the format in which this tokenizer was created.

Parameters:

• ids (list) –

  list of IDs

• skip_special_tokens (bool, default: True ) –

  whether to skip all special tokens when encountering them

• ignore_stops (bool, default: False ) –

  whether to ignore the stop tokens, thus decoding till the end

• stop_token_ids (Optional[List[int]], default: None ) –

  optional list of stop token ids to use

Returns:

• sequence ( str ) –

  str representation of molecule

Source code in safe/tokenizer.py

def decode(
    self,
    ids: list,
    skip_special_tokens: bool = True,
    ignore_stops: bool = False,
    stop_token_ids: Optional[List[int]] = None,
) -> str:
    r"""
    Decodes a list of ids to molecular representation in the format in which this tokenizer was created.

    Args:
        ids: list of IDs
        skip_special_tokens: whether to skip all special tokens when encountering them
        ignore_stops: whether to ignore the stop tokens, thus decoding till the end
        stop_token_ids: optional list of stop token ids to use

    Returns:
        sequence: str representation of molecule
    """
    if len(ids) == 0:
        return ""
    old_id_list = ids
    if not isinstance(ids[0], (list, np.ndarray)) and not torch.is_tensor(ids[0]):
        old_id_list = [ids]
    if not stop_token_ids:
        stop_token_ids = [self.tokenizer.token_to_id(self.tokenizer.eos_token)]

    new_ids_list = []
    for ids in old_id_list:
        new_ids = ids
        if not ignore_stops:
            new_ids = []
            # if first tokens are stop, we just remove it
            # this is because of bart essentially
            pos = 0
            if len(ids) > 1:
                while pos < len(ids) and ids[pos] in stop_token_ids:
                    pos += 1
            # we only ignore when there is a list of tokens
            ids = ids[pos:]
            for pos, id in enumerate(ids):
                if int(id) in stop_token_ids:
                    break
                new_ids.append(id)
        new_ids_list.append(new_ids)
    if len(new_ids_list) == 1:
        return self.tokenizer.decode(
            list(new_ids_list[0]), skip_special_tokens=skip_special_tokens
        )
    return self.tokenizer.decode_batch(
        list(new_ids_list), skip_special_tokens=skip_special_tokens
    )

encode(sample_str: str, ids_only: bool = True, **kwargs) -> list

Encodes a given molecule string once training is done

Parameters:

• sample_str (str) –

  Sample string to encode molecule

• ids_only (bool, default: True ) –

  whether to return only the ids or the encoding objet

Returns:

• object ( list ) –

  Returns encoded list of IDs

Source code in safe/tokenizer.py

def encode(self, sample_str: str, ids_only: bool = True, **kwargs) -> list:
    r"""
    Encodes a given molecule string once training is done

    Args:
        sample_str: Sample string to encode molecule
        ids_only: whether to return only the ids or the encoding objet

    Returns:
        object: Returns encoded list of IDs
    """
    if isinstance(sample_str, str):
        enc = self.tokenizer.encode(sample_str, **kwargs)
        if ids_only:
            return enc.ids
        return enc

    encs = self.tokenizer.encode_batch(sample_str, **kwargs)
    if ids_only:
        return [enc.ids for enc in encs]
    return encs

from_dict(data: dict) classmethod

Load tokenizer from dict

Parameters:

• data (dict) –

  dictionary containing the tokenizer info

Source code in safe/tokenizer.py

@classmethod
def from_dict(cls, data: dict):
    """Load tokenizer from dict

    Args:
        data: dictionary containing the tokenizer info
    """
    tokenizer_type = data.pop("tokenizer_type", "safe")
    tokenizer_attrs = data.pop("tokenizer_attrs", None)
    custom_pre_tokenizer = data.pop("custom_pre_tokenizer", False)
    tokenizer = Tokenizer.from_str(json.dumps(data))
    if custom_pre_tokenizer:
        tokenizer.pre_tokenizer = PreTokenizer.custom(SAFESplitter())
    mol_tokenizer = cls(tokenizer_type)
    mol_tokenizer.tokenizer = mol_tokenizer.set_special_tokens(tokenizer)
    if tokenizer_attrs and isinstance(tokenizer_attrs, dict):
        mol_tokenizer.tokenizer.__dict__.update(tokenizer_attrs)
    return mol_tokenizer

from_pretrained(pretrained_model_name_or_path: Union[str, os.PathLike], cache_dir: Optional[Union[str, os.PathLike]] = None, force_download: bool = False, local_files_only: bool = False, token: Optional[Union[str, bool]] = None, return_fast_tokenizer: Optional[bool] = False, proxies: Optional[Dict[str, str]] = None, **kwargs) classmethod

Instantiate a [~tokenization_utils_base.PreTrainedTokenizerBase] (or a derived class) from a predefined tokenizer.

Parameters:

• pretrained_model_name_or_path (Union[str, PathLike]) –

  Can be either:

  • A string, the model id of a predefined tokenizer hosted inside a model repo on huggingface.co. Valid model ids can be located at the root-level, like bert-base-uncased, or namespaced under a user or organization name, like dbmdz/bert-base-german-cased.
  • A path to a directory containing vocabulary files required by the tokenizer, for instance saved using the [~tokenization_utils_base.PreTrainedTokenizerBase.save_pretrained] method, e.g., ./my_model_directory/.
  • (Deprecated, not applicable to all derived classes) A path or url to a single saved vocabulary file (if and only if the tokenizer only requires a single vocabulary file like Bert or XLNet), e.g., ./my_model_directory/vocab.txt.
• cache_dir (Optional[Union[str, PathLike]], default: None ) –

  Path to a directory in which a downloaded predefined tokenizer vocabulary files should be cached if the standard cache should not be used.

• force_download (bool, default: False ) –

  Whether or not to force the (re-)download the vocabulary files and override the cached versions if they exist.

• proxies (Optional[Dict[str, str]], default: None ) –

  A dictionary of proxy servers to use by protocol or endpoint, e.g., {'http': 'foo.bar:3128', 'http://hostname': 'foo.bar:4012'}. The proxies are used on each request.

• token (Optional[Union[str, bool]], default: None ) –

  The token to use as HTTP bearer authorization for remote files. If True, will use the token generated when running huggingface-cli login (stored in ~/.huggingface).

• local_files_only (bool, default: False ) –

  Whether or not to only rely on local files and not to attempt to download any files.

• return_fast_tokenizer (Optional[bool], default: False ) –

  Whether to return fast tokenizer or not.

Examples:

    # We can't instantiate directly the base class *PreTrainedTokenizerBase* so let's show our examples on a derived class: BertTokenizer
    # Download vocabulary from huggingface.co and cache.
    tokenizer = SAFETokenizer.from_pretrained("datamol-io/safe-gpt")

    # If vocabulary files are in a directory (e.g. tokenizer was saved using *save_pretrained('./test/saved_model/')*)
    tokenizer = SAFETokenizer.from_pretrained("./test/saved_model/")

    # If the tokenizer uses a single vocabulary file, you can point directly to this file
    tokenizer = BertTokenizer.from_pretrained("./test/saved_model/tokenizer.json")

Source code in safe/tokenizer.py

@classmethod
def from_pretrained(
    cls,
    pretrained_model_name_or_path: Union[str, os.PathLike],
    cache_dir: Optional[Union[str, os.PathLike]] = None,
    force_download: bool = False,
    local_files_only: bool = False,
    token: Optional[Union[str, bool]] = None,
    return_fast_tokenizer: Optional[bool] = False,
    proxies: Optional[Dict[str, str]] = None,
    **kwargs,
):
    r"""
    Instantiate a [`~tokenization_utils_base.PreTrainedTokenizerBase`] (or a derived class) from a predefined
    tokenizer.

    Args:
        pretrained_model_name_or_path:
            Can be either:

            - A string, the *model id* of a predefined tokenizer hosted inside a model repo on huggingface.co.
              Valid model ids can be located at the root-level, like `bert-base-uncased`, or namespaced under a
              user or organization name, like `dbmdz/bert-base-german-cased`.
            - A path to a *directory* containing vocabulary files required by the tokenizer, for instance saved
              using the [`~tokenization_utils_base.PreTrainedTokenizerBase.save_pretrained`] method, e.g.,
              `./my_model_directory/`.
            - (**Deprecated**, not applicable to all derived classes) A path or url to a single saved vocabulary
              file (if and only if the tokenizer only requires a single vocabulary file like Bert or XLNet), e.g.,
              `./my_model_directory/vocab.txt`.
        cache_dir: Path to a directory in which a downloaded predefined tokenizer vocabulary files should be cached if the
            standard cache should not be used.
        force_download: Whether or not to force the (re-)download the vocabulary files and override the cached versions if they exist.
        proxies: A dictionary of proxy servers to use by protocol or endpoint, e.g.,
            `{'http': 'foo.bar:3128', 'http://hostname': 'foo.bar:4012'}`. The proxies are used on each request.
        token: The token to use as HTTP bearer authorization for remote files.
            If `True`, will use the token generated when running `huggingface-cli login` (stored in `~/.huggingface`).
        local_files_only: Whether or not to only rely on local files and not to attempt to download any files.
        return_fast_tokenizer: Whether to return fast tokenizer or not.

    Examples:
    ``` py
        # We can't instantiate directly the base class *PreTrainedTokenizerBase* so let's show our examples on a derived class: BertTokenizer
        # Download vocabulary from huggingface.co and cache.
        tokenizer = SAFETokenizer.from_pretrained("datamol-io/safe-gpt")

        # If vocabulary files are in a directory (e.g. tokenizer was saved using *save_pretrained('./test/saved_model/')*)
        tokenizer = SAFETokenizer.from_pretrained("./test/saved_model/")

        # If the tokenizer uses a single vocabulary file, you can point directly to this file
        tokenizer = BertTokenizer.from_pretrained("./test/saved_model/tokenizer.json")
    ```
    """
    kwargs.pop("resume_download", None)
    use_auth_token = kwargs.pop("use_auth_token", None)
    subfolder = kwargs.pop("subfolder", None)
    from_pipeline = kwargs.pop("_from_pipeline", None)
    from_auto_class = kwargs.pop("_from_auto", False)
    kwargs.pop("_commit_hash", None)
    revision = kwargs.pop("revision", None)

    if use_auth_token is not None:
        warnings.warn(
            "The `use_auth_token` argument is deprecated and will be removed in v5 of Transformers.",
            FutureWarning,
        )
        if token is not None:
            raise ValueError(
                "`token` and `use_auth_token` are both specified. Please set only the argument `token`."
            )
        token = use_auth_token

    user_agent = {
        "file_type": "tokenizer",
        "from_auto_class": from_auto_class,
        "is_fast": "Fast" in cls.__name__,
    }
    if from_pipeline is not None:
        user_agent["using_pipeline"] = from_pipeline

    pretrained_model_name_or_path = str(pretrained_model_name_or_path)

    file_path: Optional[Union[str, os.PathLike]] = None
    if os.path.isfile(pretrained_model_name_or_path):
        file_path = pretrained_model_name_or_path
    elif os.path.isdir(pretrained_model_name_or_path):
        file_path = Path(pretrained_model_name_or_path)
        if subfolder:
            file_path /= subfolder
        file_path /= cls.vocab_files_names
    elif urlparse(pretrained_model_name_or_path).scheme in {"http", "https"}:
        if proxies:
            warnings.warn(
                "Per-call proxies are not supported by Hugging Face Hub 1.x; "
                "set HTTP_PROXY or HTTPS_PROXY instead.",
                UserWarning,
                stacklevel=2,
            )
        with urlopen(pretrained_model_name_or_path) as response:
            tokenizer = cls.from_dict(json.loads(response.read().decode("utf-8")))
        return tokenizer.get_pretrained() if return_fast_tokenizer else tokenizer
    else:
        if proxies:
            warnings.warn(
                "Per-call proxies are not supported by Hugging Face Hub 1.x; "
                "set HTTP_PROXY or HTTPS_PROXY instead.",
                UserWarning,
                stacklevel=2,
            )
        filename = cls.vocab_files_names
        if subfolder:
            filename = f"{subfolder.strip('/')}/{filename}"
        file_path = hf_hub_download(
            repo_id=pretrained_model_name_or_path,
            filename=filename,
            cache_dir=cache_dir,
            force_download=force_download,
            local_files_only=local_files_only,
            user_agent=user_agent,
            revision=revision,
            token=token,
        )

    if file_path is None or not os.path.isfile(file_path):
        raise OSError(
            f"Could not resolve {cls.vocab_files_names!r} from "
            f"{pretrained_model_name_or_path!r}."
        )

    tokenizer = cls.load(file_path)
    if return_fast_tokenizer:
        return tokenizer.get_pretrained()
    return tokenizer

get_pretrained(**kwargs) -> PreTrainedTokenizerFast

Get a pretrained tokenizer from this tokenizer

Returns:

• PreTrainedTokenizerFast –

  Returns pre-trained fast tokenizer for hugging face models.

Source code in safe/tokenizer.py

def get_pretrained(self, **kwargs) -> PreTrainedTokenizerFast:
    r"""
    Get a pretrained tokenizer from this tokenizer

    Returns:
        Returns pre-trained fast tokenizer for hugging face models.
    """
    with attr_as(self.tokenizer, "pre_tokenizer", Whitespace()):
        tk = PreTrainedTokenizerFast(tokenizer_object=self.tokenizer)
    tk._tokenizer.pre_tokenizer = self.tokenizer.pre_tokenizer
    # now we need to add special_tokens
    tk.add_special_tokens(
        {
            "cls_token": self.tokenizer.cls_token,
            "bos_token": self.tokenizer.bos_token,
            "eos_token": self.tokenizer.eos_token,
            "mask_token": self.tokenizer.mask_token,
            "pad_token": self.tokenizer.pad_token,
            "unk_token": self.tokenizer.unk_token,
            "sep_token": self.tokenizer.sep_token,
        }
    )
    if (
        tk.model_max_length is None
        or tk.model_max_length > 1e8
        and hasattr(self.tokenizer, "model_max_length")
    ):
        tk.model_max_length = self.tokenizer.model_max_length
        setattr(
            tk,
            "model_max_length",
            getattr(self.tokenizer, "model_max_length"),
        )
    return tk

load(file_name) classmethod

Load the current tokenizer from file

Source code in safe/tokenizer.py

@classmethod
def load(cls, file_name):
    """Load the current tokenizer from file"""
    with fsspec.open(file_name, "r") as OUT:
        data_str = OUT.read()
    data = json.loads(data_str)
    # EN: the rust json parser of tokenizers has a predefined structure
    # the next two lines are important
    return cls.from_dict(data)

push_to_hub(repo_id: str, use_temp_dir: Optional[bool] = None, commit_message: Optional[str] = None, private: Optional[bool] = None, token: Optional[Union[bool, str]] = None, max_shard_size: Optional[Union[int, str]] = '10GB', create_pr: bool = False, safe_serialization: bool = False, **deprecated_kwargs) -> str

Upload the tokenizer to the 🤗 Model Hub.

Parameters:

• repo_id (str) –

  The name of the repository you want to push your {object} to. It should contain your organization name when pushing to a given organization.

• use_temp_dir (Optional[bool], default: None ) –

  Whether or not to use a temporary directory to store the files saved before they are pushed to the Hub. Will default to True if there is no directory named like repo_id, False otherwise.

• commit_message (Optional[str], default: None ) –

  Message to commit while pushing. Will default to "Upload {object}".

• private (Optional[bool], default: None ) –

  Whether or not the repository created should be private.

• token (Optional[Union[bool, str]], default: None ) –

  The token to use as HTTP bearer authorization for remote files. If True, will use the token generated when running huggingface-cli login (stored in ~/.huggingface). Will default to True if repo_url is not specified.

• max_shard_size (Optional[Union[int, str]], default: '10GB' ) –

  Only applicable for models. The maximum size for a checkpoint before being sharded. Checkpoints shard will then be each of size lower than this size. If expressed as a string, needs to be digits followed by a unit (like "5MB").

• create_pr (bool, default: False ) –

  Whether or not to create a PR with the uploaded files or directly commit.

• safe_serialization (bool, default: False ) –

  Whether or not to convert the model weights in safetensors format for safer serialization.

Source code in safe/tokenizer.py

def push_to_hub(
    self,
    repo_id: str,
    use_temp_dir: Optional[bool] = None,
    commit_message: Optional[str] = None,
    private: Optional[bool] = None,
    token: Optional[Union[bool, str]] = None,
    max_shard_size: Optional[Union[int, str]] = "10GB",
    create_pr: bool = False,
    safe_serialization: bool = False,
    **deprecated_kwargs,
) -> str:
    """
    Upload the tokenizer to the 🤗 Model Hub.

    Args:
        repo_id: The name of the repository you want to push your {object} to. It should contain your organization name
            when pushing to a given organization.
        use_temp_dir: Whether or not to use a temporary directory to store the files saved before they are pushed to the Hub.
            Will default to `True` if there is no directory named like `repo_id`, `False` otherwise.
        commit_message: Message to commit while pushing. Will default to `"Upload {object}"`.
        private: Whether or not the repository created should be private.
        token: The token to use as HTTP bearer authorization for remote files. If `True`, will use the token generated
            when running `huggingface-cli login` (stored in `~/.huggingface`). Will default to `True` if `repo_url`
            is not specified.
        max_shard_size: Only applicable for models. The maximum size for a checkpoint before being sharded. Checkpoints shard
            will then be each of size lower than this size. If expressed as a string, needs to be digits followed
            by a unit (like `"5MB"`).
        create_pr: Whether or not to create a PR with the uploaded files or directly commit.
        safe_serialization: Whether or not to convert the model weights in safetensors format for safer serialization.
    """
    use_auth_token = deprecated_kwargs.pop("use_auth_token", None)
    if use_auth_token is not None:
        warnings.warn(
            "The `use_auth_token` argument is deprecated and will be removed in v5 of Transformers.",
            FutureWarning,
        )
        if token is not None:
            raise ValueError(
                "`token` and `use_auth_token` are both specified. Please set only the argument `token`."
            )
        token = use_auth_token

    repo_path_or_name = deprecated_kwargs.pop("repo_path_or_name", None)
    if repo_path_or_name is not None:
        warnings.warn(
            "`repo_path_or_name` is no longer supported; pass `repo_id` instead.",
            FutureWarning,
            stacklevel=2,
        )
        raise ValueError("Pass `repo_id` directly.")

    for removed_name in ("repo_url", "organization"):
        if deprecated_kwargs.pop(removed_name, None) is not None:
            warnings.warn(
                f"`{removed_name}` is no longer supported; include the namespace in `repo_id`.",
                FutureWarning,
                stacklevel=2,
            )
    if deprecated_kwargs:
        unknown = ", ".join(sorted(deprecated_kwargs))
        raise TypeError(f"Unexpected keyword argument(s): {unknown}")

    # Transformers 5 removed the temporary-directory upload helpers that
    # this class historically called. The public Hub API is stable and
    # avoids coupling SAFE to those private Transformers internals.
    del use_temp_dir, max_shard_size, safe_serialization
    api = HfApi(token=token)
    api.create_repo(repo_id=repo_id, private=private, exist_ok=True)
    with tempfile.TemporaryDirectory(prefix="safe-tokenizer-") as work_dir:
        tokenizer_path = self.save_pretrained(work_dir)[0]
        return api.upload_file(
            path_or_fileobj=tokenizer_path,
            path_in_repo=self.vocab_files_names,
            repo_id=repo_id,
            commit_message=commit_message,
            token=token,
            create_pr=create_pr,
        )

save(file_name=None)

Saves the :class:~tokenizers.Tokenizer to the file at the given path.

Parameters:

• file_name (str, default: None ) –

  File where to save tokenizer

Source code in safe/tokenizer.py

def save(self, file_name=None):
    r"""
    Saves the :class:`~tokenizers.Tokenizer` to the file at the given path.

    Args:
        file_name (str, optional): File where to save tokenizer
    """
    # EN: whole logic here assumes noone is going to mess with the special token
    tk_data = self.to_dict()
    with fsspec.open(file_name, "w", encoding="utf-8") as OUT:
        out_str = json.dumps(tk_data, ensure_ascii=False)
        OUT.write(out_str)

save_pretrained(save_directory, **kwargs)

Save the tokenizer in a Hugging Face-compatible directory.

Source code in safe/tokenizer.py

def save_pretrained(self, save_directory, **kwargs):
    """Save the tokenizer in a Hugging Face-compatible directory."""
    del kwargs
    os.makedirs(save_directory, exist_ok=True)
    tokenizer_path = os.path.join(save_directory, self.vocab_files_names)
    self.save(tokenizer_path)
    return (tokenizer_path,)

set_special_tokens(tokenizer: Tokenizer, bos_token: str = CLS_TOKEN, eos_token: str = SEP_TOKEN) classmethod

Set special tokens for a tokenizer

Parameters:

• tokenizer (Tokenizer) –

  tokenizer for which special tokens will be set

• bos_token (str, default: CLS_TOKEN ) –

  Optional bos token to use

• eos_token (str, default: SEP_TOKEN ) –

  Optional eos token to use

Source code in safe/tokenizer.py

@classmethod
def set_special_tokens(
    cls,
    tokenizer: Tokenizer,
    bos_token: str = CLS_TOKEN,
    eos_token: str = SEP_TOKEN,
):
    """Set special tokens for a tokenizer

    Args:
        tokenizer: tokenizer for which special tokens will be set
        bos_token: Optional bos token to use
        eos_token: Optional eos token to use
    """
    tokenizer.pad_token = PADDING_TOKEN
    tokenizer.cls_token = CLS_TOKEN
    tokenizer.sep_token = SEP_TOKEN
    tokenizer.mask_token = MASK_TOKEN
    tokenizer.unk_token = UNK_TOKEN
    tokenizer.eos_token = eos_token
    tokenizer.bos_token = bos_token

    if isinstance(tokenizer, Tokenizer):
        tokenizer.add_special_tokens(
            [
                PADDING_TOKEN,
                CLS_TOKEN,
                SEP_TOKEN,
                MASK_TOKEN,
                UNK_TOKEN,
                eos_token,
                bos_token,
            ]
        )
    return tokenizer

to_dict(**kwargs)

Convert tokenizer to dict

Source code in safe/tokenizer.py

def to_dict(self, **kwargs):
    """Convert tokenizer to dict"""
    # we need to do this because HuggingFace tokenizers doesnt save with custom pre-tokenizers
    if self.splitter is None:
        tk_data = json.loads(self.tokenizer.to_str())
    else:
        with attr_as(self.tokenizer, "pre_tokenizer", Whitespace()):
            # temporary replace pre tokenizer with whitespace
            tk_data = json.loads(self.tokenizer.to_str())
            tk_data["custom_pre_tokenizer"] = True
    tk_data["tokenizer_type"] = self.tokenizer_type
    tk_data["tokenizer_attrs"] = self.tokenizer.__dict__
    return tk_data

train(files: Optional[List[str]], **kwargs)

This is to train a new tokenizer from either a list of file or some input data

Args files (str): file in which your molecules are separated by new line kwargs (dict): optional args for the tokenizer train

Source code in safe/tokenizer.py

def train(self, files: Optional[List[str]], **kwargs):
    r"""
    This is to train a new tokenizer from either a list of file or some input data

    Args
        files (str): file in which your molecules are separated by new line
        kwargs (dict): optional args for the tokenizer `train`
    """
    if isinstance(files, str):
        files = [files]
    self.tokenizer.train(files=files, trainer=self.trainer)

train_from_iterator(data: Iterator, **kwargs: Any)

Train the Tokenizer using the provided iterator.

You can provide anything that is a Python Iterator * A list of sequences :obj:List[str] * A generator that yields :obj:str or :obj:List[str] * A Numpy array of strings

Parameters:

• data (Iterator) –

  data iterator

• **kwargs (Any, default: {} ) –

  additional keyword argument for the tokenizer train_from_iterator

Source code in safe/tokenizer.py

def train_from_iterator(self, data: Iterator, **kwargs: Any):
    """Train the Tokenizer using the provided iterator.

    You can provide anything that is a Python Iterator
        * A list of sequences :obj:`List[str]`
        * A generator that yields :obj:`str` or :obj:`List[str]`
        * A Numpy array of strings

    Args:
        data: data iterator
        **kwargs: additional keyword argument for the tokenizer `train_from_iterator`
    """
    self.tokenizer.train_from_iterator(data, trainer=self.trainer, **kwargs)

---

Utils

__implicit_carbon_query = dm.from_smarts('[#6;h]') module-attribute

__mmpa_query = dm.from_smarts('[*;!$(*=,#[!#6])]!@!=!#[*]') module-attribute

MolSlicer(shortest_linker: bool = False, min_linker_size: int = 0, require_ring_system: bool = True, verbose: bool = False)

Slice a molecule into head-linker-tail

Constructor of bond slicer.

Parameters:

• shortest_linker (bool, default: False ) –

  whether to consider longuest or shortest linker. Does not have any effect when expected_head group is provided during splitting

• min_linker_size (int, default: 0 ) –

  minimum linker size

• require_ring_system (bool, default: True ) –

  whether all fragment needs to have a ring system

• verbose (bool, default: False ) –

  whether to allow verbosity in logging

Source code in safe/utils.py

def __init__(
    self,
    shortest_linker: bool = False,
    min_linker_size: int = 0,
    require_ring_system: bool = True,
    verbose: bool = False,
):
    """
    Constructor of bond slicer.

    Args:
        shortest_linker: whether to consider longuest or shortest linker.
            Does not have any effect when expected_head group is provided during splitting
        min_linker_size: minimum linker size
        require_ring_system: whether all fragment needs to have a ring system
        verbose: whether to allow verbosity in logging
    """

    self.bond_splitters = [dm.from_smarts(x) for x in self.BOND_SPLITTERS]
    self.shortest_linker = shortest_linker
    self.min_linker_size = min_linker_size
    self.require_ring_system = require_ring_system
    self.verbose = verbose

BOND_SPLITTERS = ['[R:1]-&!@[!R;!D1:2]', '[R:1]-&!@[R:2]'] class-attribute instance-attribute

MAX_CUTS = 2 class-attribute instance-attribute

bond_splitters = [dm.from_smarts(x) for x in self.BOND_SPLITTERS] instance-attribute

min_linker_size = min_linker_size instance-attribute

require_ring_system = require_ring_system instance-attribute

shortest_linker = shortest_linker instance-attribute

verbose = verbose instance-attribute

__call__(mol: Union[dm.Mol, str], expected_head: Union[dm.Mol, str] = None)

Perform slicing of the input molecule

Parameters:

• mol (Union[Mol, str]) –

  input molecule

• expected_head (Union[Mol, str], default: None ) –

  substructure that should be part of the head. The small fragment containing this substructure would be kept as head

Source code in safe/utils.py

def __call__(self, mol: Union[dm.Mol, str], expected_head: Union[dm.Mol, str] = None):
    """Perform slicing of the input molecule

    Args:
        mol: input molecule
        expected_head: substructure that should be part of the head.
            The small fragment containing this substructure would be kept as head
    """

    mol = dm.to_mol(mol)
    # remove salt and solution
    mol = dm.keep_largest_fragment(mol)
    Chem.rdDepictor.Compute2DCoords(mol)
    dist_mat = Chem.rdmolops.GetDistanceMatrix(mol)

    if expected_head is not None:
        if isinstance(expected_head, str):
            expected_head = dm.to_mol(expected_head)
        if not mol.HasSubstructMatch(expected_head):
            if self.verbose:
                logger.info(
                    "Expected head was provided, but does not match molecules. It will be ignored"
                )
            expected_head = None

    candidate_bonds = self._get_bonds_to_cut(mol)

    # we have all the candidate bonds we can cut
    # now we need to pick the most plausible bonds
    selected_bonds = [mol.GetBondBetweenAtoms(a1, a2) for (a1, a2) in candidate_bonds]

    # CASE 1: no bond to cut ==> only head
    if len(selected_bonds) == 0:
        return (mol, None, None)

    # CASE 2: only one bond ==> linker is empty
    if len(selected_bonds) == 1:
        # there is not linker
        tmp = Chem.rdmolops.FragmentOnBonds(mol, [b.GetIdx() for b in selected_bonds])
        head, tail = Chem.GetMolFrags(tmp, asMols=True)
        return (head, None, tail)

    # CASE 3a: we select the most plausible bond to cut on ourselves
    if expected_head is None:
        choice = self._bond_selection_from_max_cuts(candidate_bonds, dist_mat)
        if choice is None:
            return (mol, None, None)
        selected_bonds = [selected_bonds[c] for c in choice]
        return self._fragment_mol(mol, selected_bonds)

    # CASE 3b: slightly more complex case where we want the head to be the smallest graph containing the
    # provided substructure
    bond_combination = list(itertools.combinations(selected_bonds, self.MAX_CUTS))
    bond_score = float("inf")
    linker_score = float("inf")
    head, linker, tail = (None, None, None)
    for split_bonds in bond_combination:
        cur_head, cur_linker, cur_tail = self._fragment_mol(mol, split_bonds)
        # head can also be tail
        head_match = cur_head.GetSubstructMatch(expected_head)
        tail_match = cur_tail.GetSubstructMatch(expected_head)
        if not head_match and not tail_match:
            continue
        if not head_match and tail_match:
            cur_head, cur_tail = cur_tail, cur_head
        cur_bond_score = cur_head.GetNumHeavyAtoms()
        # compute linker score
        cur_linker_score = self._compute_linker_score(cur_linker)
        if (cur_bond_score < bond_score) or (
            cur_bond_score < self._BOND_BUFFER + bond_score and cur_linker_score < linker_score
        ):
            head, linker, tail = cur_head, cur_linker, cur_tail
            bond_score = cur_bond_score
            linker_score = cur_linker_score

    return (head, linker, tail)

get_ring_system(mol: dm.Mol)

Get the list of ring system from a molecule

Parameters:

• mol (Mol) –

  input molecule for which we are computing the ring system

Source code in safe/utils.py

def get_ring_system(self, mol: dm.Mol):
    """Get the list of ring system from a molecule

    Args:
        mol: input molecule for which we are computing the ring system
    """
    mol.UpdatePropertyCache()
    ri = mol.GetRingInfo()
    systems = []
    for ring in ri.AtomRings():
        ring_atoms = set(ring)
        cur_system = []  # keep a track of ring system
        for system in systems:
            if len(ring_atoms.intersection(system)) > 0:
                ring_atoms = ring_atoms.union(system)  # merge ring system that overlap
            else:
                cur_system.append(system)
        cur_system.append(ring_atoms)
        systems = cur_system
    return systems

link_fragments(linker: Union[dm.Mol, str], head: Union[dm.Mol, str], tail: Union[dm.Mol, str]) classmethod

Link fragments together using the provided linker

Parameters:

• linker (Union[Mol, str]) –

  linker to use

• head (Union[Mol, str]) –

  head fragment

• tail (Union[Mol, str]) –

  tail fragment

Source code in safe/utils.py

@classmethod
def link_fragments(
    cls, linker: Union[dm.Mol, str], head: Union[dm.Mol, str], tail: Union[dm.Mol, str]
):
    """Link fragments together using the provided linker

    Args:
        linker: linker to use
        head: head fragment
        tail: tail fragment
    """
    if isinstance(linker, dm.Mol):
        linker = dm.to_smiles(linker)
    linker = standardize_attach(linker)
    reactants = [dm.to_mol(head), dm.to_mol(tail), dm.to_mol(linker)]
    return dm.reactions.apply_reaction(
        cls._MERGING_RXN, reactants, as_smiles=True, sanitize=True, product_index=0
    )

attr_as(obj: Any, field: str, value: Any)

Temporary replace the value of an object

Parameters:

• obj (Any) –

  object to temporary patch

• field (str) –

  name of the key to change

• value (Any) –

  value of key to be temporary changed

Source code in safe/utils.py

@contextmanager
def attr_as(obj: Any, field: str, value: Any):
    """Temporary replace the value of an object

    Args:
        obj: object to temporary patch
        field: name of the key to change
        value: value of key to be temporary changed
    """
    old_value = getattr(obj, field, None)
    setattr(obj, field, value)
    try:
        yield
    finally:
        setattr(obj, field, old_value)

compute_side_chains(mol: dm.Mol, core: dm.Mol, label_by_index: bool = False)

Compute the side chain of a molecule given a core

Finding the side chains

The algorithm to find the side chains from core assumes that the core we get as input has attachment points. Those attachment points are never considered as part of the query, rather they are used to define the attachment points on the side chains. Removing the attachment points from the core is exactly the same as keeping them.

mol = "CC1=C(C(=NO1)C2=CC=CC=C2Cl)C(=O)NC3C4N(C3=O)C(C(S4)(C)C)C(=O)O"
core0 = "CC1(C)CN2C(CC2=O)S1"
core1 = "CC1(C)SC2C(-*)C(=O)N2C1-*"
core2 = "CC1N2C(SC1(C)C)C(N)C2=O"
side_chain = compute_side_chain(core=core0, mol=mol)
dm.to_image([side_chain, core0, mol])

Therefore on the above, core0 and core1 are equivalent for the molecule mol, but core2 is not.

Parameters:

• mol (Mol) –

  molecule to split

• core (Mol) –

  core to use for deriving the side chains

Source code in safe/utils.py

def compute_side_chains(mol: dm.Mol, core: dm.Mol, label_by_index: bool = False):
    """Compute the side chain of a molecule given a core

    !!! note "Finding the side chains"
        The algorithm to find the side chains from core assumes that the core we get as input has attachment points.
        Those attachment points are never considered as part of the query, rather they are used to define the attachment points
        on the side chains. Removing the attachment points from the core is exactly the same as keeping them.

        ```python
        mol = "CC1=C(C(=NO1)C2=CC=CC=C2Cl)C(=O)NC3C4N(C3=O)C(C(S4)(C)C)C(=O)O"
        core0 = "CC1(C)CN2C(CC2=O)S1"
        core1 = "CC1(C)SC2C(-*)C(=O)N2C1-*"
        core2 = "CC1N2C(SC1(C)C)C(N)C2=O"
        side_chain = compute_side_chain(core=core0, mol=mol)
        dm.to_image([side_chain, core0, mol])
        ```
        Therefore on the above, core0 and core1 are equivalent for the molecule `mol`, but core2 is not.

    Args:
        mol: molecule to split
        core: core to use for deriving the side chains
    """

    if isinstance(mol, str):
        mol = dm.to_mol(mol)
    if isinstance(core, str):
        core = dm.to_mol(core)
    core_query_param = AdjustQueryParameters()
    core_query_param.makeDummiesQueries = True
    core_query_param.adjustDegree = False
    core_query_param.aromatizeIfPossible = True
    core_query_param.makeBondsGeneric = False
    core_query = AdjustQueryProperties(core, core_query_param)
    return ReplaceCore(
        mol, core_query, labelByIndex=label_by_index, replaceDummies=False, requireDummyMatch=False
    )

convert_to_safe(mol: dm.Mol, canonical: bool = False, randomize: bool = False, seed: Optional[int] = 1, slicer: str = 'brics', split_fragment: bool = True, fraction_hs: Optional[float] = None, resolution: Optional[float] = 0.5)

Convert a molecule to a safe representation

Parameters:

• mol (Mol) –

  molecule to convert

• canonical (bool, default: False ) –

  whether to use canonical encoding

• randomize (bool, default: False ) –

  whether to randomize the encoding

• seed (Optional[int], default: 1 ) –

  random seed

• slicer (str, default: 'brics' ) –

  the slicer to use for fragmentation

• split_fragment (bool, default: True ) –

  whether to split fragments

• fraction_hs (Optional[float], default: None ) –

  proportion of random atom to which we will add explicit hydrogens

• resolution (Optional[float], default: 0.5 ) –

  resolution for the partitioning algorithm

• seed (Optional[int], default: 1 ) –

  random seed

Source code in safe/utils.py

def convert_to_safe(
    mol: dm.Mol,
    canonical: bool = False,
    randomize: bool = False,
    seed: Optional[int] = 1,
    slicer: str = "brics",
    split_fragment: bool = True,
    fraction_hs: Optional[float] = None,
    resolution: Optional[float] = 0.5,
):
    """Convert a molecule to a safe representation

    Args:
        mol: molecule to convert
        canonical: whether to use canonical encoding
        randomize: whether to randomize the encoding
        seed: random seed
        slicer: the slicer to use for fragmentation
        split_fragment: whether to split fragments
        fraction_hs: proportion of random atom to which we will add explicit hydrogens
        resolution: resolution for the partitioning algorithm
        seed: random seed
    """
    x = None
    try:
        x = sf.encode(mol, canonical=canonical, randomize=randomize, slicer=slicer, seed=seed)
    except sf.SAFEFragmentationError:
        if split_fragment:
            if isinstance(mol, str) and "." in mol:
                return None
            try:
                x = sf.encode(
                    mol,
                    canonical=False,
                    randomize=randomize,
                    seed=seed,
                    slicer=partial(
                        fragment_aware_spliting,
                        fraction_hs=fraction_hs,
                        resolution=resolution,
                        seed=seed,
                    ),
                )
            except (sf.SAFEEncodeError, sf.SAFEFragmentationError):
                # logger.exception(e)
                return x
        # we need to resplit using attachment point but here we are only adding
    except sf.SAFEEncodeError:
        return x
    return x

filter_by_substructure_constraints(sequences: List[Union[str, dm.Mol]], substruct: Union[str, dm.Mol], n_jobs: int = -1)

Check whether the input substructures are present in each of the molecule in the sequences

Parameters:

• sequences (List[Union[str, Mol]]) –

  list of molecules to validate

• substruct (Union[str, Mol]) –

  substructure to use as query

• n_jobs (int, default: -1 ) –

  number of jobs to use for parallelization

Source code in safe/utils.py

def filter_by_substructure_constraints(
    sequences: List[Union[str, dm.Mol]], substruct: Union[str, dm.Mol], n_jobs: int = -1
):
    """Check whether the input substructures are present in each of the molecule in the sequences

    Args:
        sequences: list of molecules to validate
        substruct: substructure to use as query
        n_jobs: number of jobs to use for parallelization

    """

    # Normalize both string and molecule queries the same way: attachment
    # points must behave as wildcards, otherwise a ``dm.Mol`` query carrying
    # dummy atoms would only match other dummies and silently reject every
    # decorated molecule.
    if isinstance(substruct, dm.Mol):
        substruct = dm.to_smiles(substruct)
    if isinstance(substruct, str):
        substruct = standardize_attach(substruct)
        substruct = dm.from_smarts(substruct)
    if substruct is None:
        raise ValueError("Substructure constraint could not be parsed")

    def _check_match(mol):
        mol = dm.to_mol(mol)
        return mol is not None and mol.HasSubstructMatch(substruct)

    matches = dm.parallelized(_check_match, sequences, n_jobs=n_jobs)
    return list(compress(sequences, matches))

find_partition_edges(G: nx.Graph, partition: List[List]) -> List[Tuple]

Find the edges connecting the subgraphs in a given partition of a graph.

Parameters:

• G (Graph) –

  The original graph.

• partition (list of list of nodes) –

  The partition of the graph where each element is a list of nodes representing a subgraph.

Returns:

• list ( List[Tuple] ) –

  A list of edges connecting the subgraphs in the partition.

Source code in safe/utils.py

def find_partition_edges(G: nx.Graph, partition: List[List]) -> List[Tuple]:
    """
    Find the edges connecting the subgraphs in a given partition of a graph.

    Args:
        G (networkx.Graph): The original graph.
        partition (list of list of nodes): The partition of the graph where each element is a list of nodes representing a subgraph.

    Returns:
        list: A list of edges connecting the subgraphs in the partition.
    """
    partition_edges = []
    for subgraph1, subgraph2 in combinations(partition, 2):
        edges = nx.edge_boundary(G, subgraph1, subgraph2)
        partition_edges.extend(edges)
    return partition_edges

fragment_aware_spliting(mol: dm.Mol, fraction_hs: Optional[float] = None, **kwargs: Any)

Custom splitting algorithm for dataset building.

This slicing strategy will cut any bond including bonding with hydrogens However, only one cut per atom is allowed

Parameters:

• mol (Mol) –

  molecule to split

• fraction_hs (Optional[float], default: None ) –

  proportion of random atom to which we will add explicit hydrogens

• kwargs (Any, default: {} ) –

  additional arguments to pass to the partitioning algorithm

Source code in safe/utils.py

def fragment_aware_spliting(mol: dm.Mol, fraction_hs: Optional[float] = None, **kwargs: Any):
    """Custom splitting algorithm for dataset building.

    This slicing strategy will cut any bond including bonding with hydrogens
    However, only one cut per atom is allowed

    Args:
        mol: molecule to split
        fraction_hs: proportion of random atom to which we will add explicit hydrogens
        kwargs: additional arguments to pass to the partitioning algorithm
    """
    seed = kwargs.get("seed", 1)
    rng = random.Random(seed)
    mol = dm.to_mol(mol, remove_hs=False)
    mol = _selective_add_hs(mol, fraction_hs=fraction_hs, rng=rng)
    graph = dm.graph.to_graph(mol)
    d = mol_partition(mol, **kwargs)
    q = deque(d)
    partition = q.pop()
    return find_partition_edges(graph, partition)

list_individual_attach_points(mol: dm.Mol, depth: Optional[int] = None)

List all individual attachement points.

We do not allow multiple attachment points per substitution position.

Parameters:

• mol (Mol) –

  molecule for which we need to open the attachment points

Source code in safe/utils.py

def list_individual_attach_points(mol: dm.Mol, depth: Optional[int] = None):
    """List all individual attachement points.

    We do not allow multiple attachment points per substitution position.

    Args:
        mol: molecule for which we need to open the attachment points

    """
    ATTACHING_RXN = ReactionFromSmarts("[*;h;!$([*][#0]):1]>>[*:1][*]")
    mols = [mol]
    curated_prods = set()
    num_attachs = len(mol.GetSubstructMatches(dm.from_smarts("[*;h:1]"), uniquify=True))
    depth = depth or 1
    depth = min(max(depth, 1), num_attachs)
    while depth > 0:
        prods = set()
        for mol in mols:
            mol = dm.to_mol(mol)
            for p in ATTACHING_RXN.RunReactants((mol,)):
                try:
                    m = dm.sanitize_mol(p[0])
                    sm = dm.to_smiles(m, canonical=True)
                    sm = dm.reactions.add_brackets_to_attachment_points(sm)
                    prods.add(dm.reactions.convert_attach_to_isotope(sm, as_smiles=True))
                except Exception as e:
                    logger.error(e)
        curated_prods.update(prods)
        mols = prods
        depth -= 1
    return list(curated_prods)

mol_partition(mol: dm.Mol, query: Optional[dm.Mol] = None, seed: Optional[int] = None, **kwargs: Any)

Partition a molecule into fragments using a bond query

Parameters:

• mol (Mol) –

  molecule to split

• query (Optional[Mol], default: None ) –

  bond query to use for splitting

• seed (Optional[int], default: None ) –

  random seed

• kwargs (Any, default: {} ) –

  additional arguments to pass to the partitioning algorithm

Source code in safe/utils.py

@py_random_state("seed")
def mol_partition(
    mol: dm.Mol, query: Optional[dm.Mol] = None, seed: Optional[int] = None, **kwargs: Any
):
    """Partition a molecule into fragments using a bond query

    Args:
        mol: molecule to split
        query: bond query to use for splitting
        seed: random seed
        kwargs: additional arguments to pass to the partitioning algorithm

    """
    resolution = kwargs.get("resolution", 1.0)
    threshold = kwargs.get("threshold", 1e-7)
    weight = kwargs.get("weight", "weight")

    if query is None:
        query = __mmpa_query

    G = dm.graph.to_graph(mol)
    bond_partition = [
        tuple(sorted(match)) for match in mol.GetSubstructMatches(query, uniquify=True)
    ]

    def get_relevant_edges(e1, e2):
        return tuple(sorted([e1, e2])) not in bond_partition

    subgraphs = nx.subgraph_view(G, filter_edge=get_relevant_edges)

    partition = [{u} for u in G.nodes()]
    inner_partition = sorted(nx.connected_components(subgraphs), key=lambda x: min(x))
    mod = nx.algorithms.community.modularity(
        G, inner_partition, resolution=resolution, weight=weight
    )
    is_directed = G.is_directed()
    graph = G.__class__()
    graph.add_nodes_from(G)
    graph.add_weighted_edges_from(G.edges(data=weight, default=1))
    graph = nx.algorithms.community.louvain._gen_graph(graph, inner_partition)
    m = graph.size(weight="weight")
    partition, inner_partition, improvement = nx.algorithms.community.louvain._one_level(
        graph, m, inner_partition, resolution, is_directed, seed
    )
    improvement = True
    while improvement:
        # gh-5901 protect the sets in the yielded list from further manipulation here
        yield [s.copy() for s in partition]
        new_mod = nx.algorithms.community.modularity(
            graph, inner_partition, resolution=resolution, weight="weight"
        )
        if new_mod - mod <= threshold:
            return
        mod = new_mod
        graph = nx.algorithms.community.louvain._gen_graph(graph, inner_partition)
        partition, inner_partition, improvement = nx.algorithms.community.louvain._one_level(
            graph, m, partition, resolution, is_directed, seed
        )

standardize_attach(inputs: str, standard_attach: str = '[*]')

Standardize the attachment points of a molecule

Parameters:

• inputs (str) –

  input molecule

• standard_attach (str, default: '[*]' ) –

  standard attachment point to use

Source code in safe/utils.py

def standardize_attach(inputs: str, standard_attach: str = "[*]"):
    """Standardize the attachment points of a molecule

    Args:
        inputs: input molecule
        standard_attach: standard attachment point to use
    """

    for attach_regex in _SMILES_ATTACHMENT_POINTS:
        inputs = re.sub(attach_regex, standard_attach, inputs)
    return inputs