<commit_before>#!/usr/bin/env python
#
# Copyright (C) 2014 <NAME>, <NAME>,
# Regents of the University of California
#
# This implementation utilizes code and methods from Riccardo Vianello
# as well as code structure and inspiration from geoalchemy2

import base64
import contextlib
import functools
import numbers
import operator
import string
import types

import numpy as np

from sqlalchemy import event, Table, bindparam, func
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql import expression, functions, type_coerce, elements, operators
from sqlalchemy.types import (
    UserDefinedType, 
    _Binary, 
    TypeDecorator, 
    BINARY,
    Float,
    Integer,
    String,
    Text,
)
from sqlalchemy.dialects.postgresql.base import (
    DOUBLE_PRECISION,
    ischema_names,
)

from rdkit import Chem, DataStructs
from rdkit.Chem import Descriptors, rdchem
from rdkit.Chem.Draw import MolToImage
from rdkit.Chem.rdDistGeom import EmbedMolecule

_all_bytes = string.maketrans('', '')


class ChemistryError(ValueError):
    pass


class CustomEqualityBinaryExpression_HACK(elements.BinaryExpression):
    custom_opstrings = ()

    def __bool__(self):
        try:
            return super(CustomEqualityBinaryExpression_HACK, self).__bool__()
        except TypeError:
            if self.operator.opstring in self.custom_opstrings:
                return operators.eq(hash(self._orig[0]), hash(self._orig[1]))
            else:
                raise

    __nonzero__ = __bool__

    @classmethod
    def _override_expr(cls, source, opstrings):
        """Create a shallow copy of this ClauseElement.
        This method may be used by a generative API.  Its also used as
        part of the "deep" copy afforded by a traversal that combines
        the _copy_internals() method.
        """
        c = cls.__new__(cls)
        c.__dict__ = source.__dict__.copy()
        elements.ClauseElement._cloned_set._reset(c)
        elements.ColumnElement.comparator._reset(c)

        # this is a marker that helps to "equate" clauses to each other
        # when a Select returns its list of FROM clauses.  the cloning
        # process leaves around a lot of remnants of the previous clause
        # typically in the form of column expressions still attached to the
        # old table.
        c._is_clone_of = source

        c.custom_opstrings = opstrings

        return c


def _remove_control_characters(data):
    if not isinstance(data, basestring):
        raise ValueError("Data must be a string")
    else:
        data = str(data)
    return data.translate(_all_bytes, _all_bytes[:32])


## Datatype Converstions
## TODO: This should be reorganized... do we even want 
##       automatic conversion from ctab to mol?
## Break these out into separate files for each type


## Mol conversions ##################################################

def ensure_mol(mol, sanitize=True):
    if not isinstance(mol, (Chem.Mol, rdchem.Mol)):
        raise ValueError("Not already an instance of rdkit.Chem.Mol")
    if sanitize:
        try:
            Chem.SanitizeMol(mol)
        except ValueError as e:
           raise ChemistryError(str(e))
    return mol

def extract_mol_element(mol, sanitize=True):
    if hasattr(mol, 'as_mol'):
        mol = mol.as_mol
    elif hasattr(mol, 'mol'):
        mol = mol.mol
    else:
        raise ValueError("Not an instance of RawMolElement or compatible")
    if sanitize:
        try:
            Chem.SanitizeMol(mol)
        except ValueError as e:
            raise ChemistryError(str(e))
    return mol

def smiles_to_mol(smiles, sanitize=True):
    smiles = _remove_control_characters(smiles)
    mol = Chem.MolFromSmiles(smiles, sanitize=False)
    if mol is None:
        raise ValueError("Failed to parse SMILES: `{0}`".format(smiles))
    if sanitize:
        try:
            Chem.SanitizeMol(mol)
        except ValueError as e:
            raise ChemistryError(str(e))
    return mol

def smarts_to_mol(smarts, sanitize=True):
    smiles = _remove_control_characters(smarts)
    mol = Chem.MolFromSmarts(smarts, mergeHs=True)
    if mol is None:
        raise ValueError("Failed to parse SMARTS: `{0}`".format(smarts))
    if sanitize:
        Chem.SanitizeMol(mol, catchErrors=True)
    return mol

def binary_to_mol(data, sanitize=True):
    try:
        mol = Chem.Mol(data)
    except Exception:  # This is a proxy for Boost.Python.ArgumentError
        raise ValueError("Invalid binary mol data: `{0}`".format(data))
    if sanitize:
        try:
            Chem.SanitizeMol(mol)
        except ValueError as e:
            raise ChemistryError(str(e))
    return mol

def ctab_to_mol(data, sanitize=True):
    smiles = _remove_control_characters(data)
    mol = Chem.MolFromMolBlock(data, sanitize=sanitize, removeHs=sanitize)
    if mol is None:
        raise ValueError("Failed to parse CTAB")
    return mol

def inchi_to_mol(inchi, sanitize=True):
    smiles = _remove_control_characters(inchi)
    mol = Chem.MolFromInchi(inchi, sanitize=sanitize, removeHs=sanitize)
    if mol is None:
        raise ValueError("Failed to parse InChI: `{0}`".format(inchi))
    return mol

# Want to maintain order
MOL_PARSERS = [
    ('mol', ensure_mol),
    ('element', extract_mol_element),
    ('binary', binary_to_mol),
    ('smiles', smiles_to_mol),
    ('smarts', smarts_to_mol),
    ('ctab', ctab_to_mol),
    ('inchi', inchi_to_mol),
]

def attempt_mol_coersion(data, sanitize=True, exclude=()):<commit_msg>Fixing long, standing bug that did not assign stereochemsitry after parsing<commit_after>#!/usr/bin/env python
#
# Copyright (C) 2014 <NAME>, <NAME>,
# Regents of the University of California
#
# This implementation utilizes code and methods from Riccardo Vianello
# as well as code structure and inspiration from geoalchemy2

import base64
import contextlib
import functools
import numbers
import operator
import string
import types

import numpy as np

from sqlalchemy import event, Table, bindparam, func
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql import expression, functions, type_coerce, elements, operators
from sqlalchemy.types import (
    UserDefinedType, 
    _Binary, 
    TypeDecorator, 
    BINARY,
    Float,
    Integer,
    String,
    Text,
)
from sqlalchemy.dialects.postgresql.base import (
    DOUBLE_PRECISION,
    ischema_names,
)

from rdkit import Chem, DataStructs
from rdkit.Chem import Descriptors, rdchem
from rdkit.Chem.Draw import MolToImage
from rdkit.Chem.rdDistGeom import EmbedMolecule

_all_bytes = string.maketrans('', '')


class ChemistryError(ValueError):
    pass


class CustomEqualityBinaryExpression_HACK(elements.BinaryExpression):
    custom_opstrings = ()

    def __bool__(self):
        try:
            return super(CustomEqualityBinaryExpression_HACK, self).__bool__()
        except TypeError:
            if self.operator.opstring in self.custom_opstrings:
                return operators.eq(hash(self._orig[0]), hash(self._orig[1]))
            else:
                raise

    __nonzero__ = __bool__

    @classmethod
    def _override_expr(cls, source, opstrings):
        """Create a shallow copy of this ClauseElement.
        This method may be used by a generative API.  Its also used as
        part of the "deep" copy afforded by a traversal that combines
        the _copy_internals() method.
        """
        c = cls.__new__(cls)
        c.__dict__ = source.__dict__.copy()
        elements.ClauseElement._cloned_set._reset(c)
        elements.ColumnElement.comparator._reset(c)

        # this is a marker that helps to "equate" clauses to each other
        # when a Select returns its list of FROM clauses.  the cloning
        # process leaves around a lot of remnants of the previous clause
        # typically in the form of column expressions still attached to the
        # old table.
        c._is_clone_of = source

        c.custom_opstrings = opstrings

        return c


def _remove_control_characters(data):
    if not isinstance(data, basestring):
        raise ValueError("Data must be a string")
    else:
        data = str(data)
    return data.translate(_all_bytes, _all_bytes[:32])


## Datatype Converstions
## TODO: This should be reorganized... do we even want 
##       automatic conversion from ctab to mol?
## Break these out into separate files for each type


## Mol conversions ##################################################

def ensure_mol(mol, sanitize=True):
    if not isinstance(mol, (Chem.Mol, rdchem.Mol)):
        raise ValueError("Not already an instance of rdkit.Chem.Mol")
    if sanitize:
        try:
            Chem.SanitizeMol(mol)
        except ValueError as e:
           raise ChemistryError(str(e))
    return mol

def extract_mol_element(mol, sanitize=True):
    if hasattr(mol, 'as_mol'):
        mol = mol.as_mol
    elif hasattr(mol, 'mol'):
        mol = mol.mol
    else:
        raise ValueError("Not an instance of RawMolElement or compatible")
    if sanitize:
        try:
            Chem.SanitizeMol(mol)
        except ValueError as e:
            raise ChemistryError(str(e))
    return mol

def smiles_to_mol(smiles, sanitize=True):
    smiles = _remove_control_characters(smiles)
    mol = Chem.MolFromSmiles(smiles, sanitize=False)
    if mol is None:
        raise ValueError("Failed to parse SMILES: `{0}`".format(smiles))
    if sanitize:
        try:
            Chem.SanitizeMol(mol)
            Chem.AssignStereochemistry(mol)
        except ValueError as e:
            raise ChemistryError(str(e))
    return mol

def smarts_to_mol(smarts, sanitize=True):
    smiles = _remove_control_characters(smarts)
    mol = Chem.MolFromSmarts(smarts, mergeHs=True)
    if mol is None:
        raise ValueError("Failed to parse SMARTS: `{0}`".format(smarts))
    if sanitize:
        Chem.SanitizeMol(mol, catchErrors=True)
        Chem.AssignStereochemistry(mol)
    return mol

def binary_to_mol(data, sanitize=True):
    try:
        mol = Chem.Mol(data)
    except Exception:  # This is a proxy for Boost.Python.ArgumentError
        raise ValueError("Invalid binary mol data: `{0}`".format(data))
    if sanitize:
        try:
            Chem.SanitizeMol(mol)
            Chem.AssignStereochemistry(mol)
        except ValueError as e:
            raise ChemistryError(str(e))
    return mol

def ctab_to_mol(data, sanitize=True):
    smiles = _remove_control_characters(data)
    mol = Chem.MolFromMolBlock(data, sanitize=sanitize, removeHs=sanitize)
    if mol is None:
        raise ValueError("Failed to parse CTAB")
    if sanitize:
        try:
            Chem.SanitizeMol(mol)
            Chem.AssignStereochemistry(mol)
        except ValueError as e:
            raise ChemistryError(str(e))
    return mol

def inchi_to_mol(inchi, sanitize=True):
    smiles = _remove_control_characters(inchi)
    mol = Chem.MolFromInchi(inchi, sanitize=sanitize, removeHs=sanitize)
    if mol is None:
        raise ValueError("Failed to parse InChI: `{0}`".format(inchi))
    if sanitize:
        try:
            Chem.SanitizeMol(mol)
            Chem.AssignStereochemistry(mol)
        except ValueError as e:
            raise ChemistryError(str(e))
    return mol

# Want to maintain order
MOL_PARSERS = [
    ('mol', ensure_mol),
    ('element', extract_mol_element),
    ('binary', binary_to_mol),
    ('smiles', smiles_to_mol),
    ('smarts', smarts_to_mol),
    ('ctab', ctab_to_mol),
    ('inchi', inchi_to_mol),
]

def attempt_mol_coersion(data, sanitize=True, exclude=()):
{'fraction_non_alphanumeric': 0.09376083188908145, 'fraction_numerical': 0.0025996533795493936, 'mean_word_length': 3.0031217481789803, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1513348', 'n_tokens_mistral': 3628, 'n_tokens_neox': 3475, 'n_words': 1002}