<jupyter_start><jupyter_text>Let's code!  Import modules<jupyter_code># OS command
import os

# Sure, we need `pandas`
import pandas as pd
import numpy as np
# Typing
from typing import (
    List,
    Tuple
)
import re
import networkx.algorithms.isomorphism as iso

# Abstract classes we're going to use them  in the section
from abc import ABC, abstractmethod
from itertools import combinations
# Base model
from pydantic import BaseModel
from IPython.display import display
from tqdm import tqdm
from functional_group import get_FG
import networkx as nx<jupyter_output><empty_output><jupyter_text>Abstract Adapter<jupyter_code>class AbstractAdapter(ABC):
    @abstractmethod
    def extract(self):
        """
        Extract input data
        """

    def tranform(self):
        """
        Tranformed data into desired format
        """

    def load(self):
        """
        Load data to next step
        """

    @abstractmethod
    def apply(self):
        """
        Apply data ETL process
        """
        pass<jupyter_output><empty_output><jupyter_text>Abstract class ข้างบนจะบอก adapter process โดยภาพรวมนะครับ โดยผมเขียนให้อยู่ในรูปของ ETL เพื่อให้เข้ากับหลัก Data engineer  CIF Adapter  Adapter desired output<jupyter_code>class CIF2PandasAdapterOutput(BaseModel):
    """
    Set up desired output
    """
    
    metadata: pd.DataFrame
    loops: List[pd.DataFrame]

    class Config:
        """
        Set some config to allow pd.DataFrame to BaseModel
        """
        arbitrary_types_allowed = True<jupyter_output><empty_output><jupyter_text>Set up adapter<jupyter_code>class CIF2PandasAdapter(AbstractAdapter):
    """
    An adapter to convert CIF into Pandas DataFrames
    """

    @staticmethod
    def load_cif(cif_filepath: str) -> List[str]:
        """
        Read CIF file as String and split looping sections
        """
        with open(cif_filepath) as f:
            filename = f.readline().strip()
            dataframes = []
            dataframe = []
            for line in f.readlines():
                columns = [
                    l.strip() for l in line.strip().split(" ") if l and (l != "")
                ]
                if columns:
                    if columns[0] == "loop_":
                        dataframes.append(dataframe)
                        dataframe = []
                    else:
                        if "fapswitch" not in columns and columns != [""]:
                            dataframe.append(columns)
            dataframes.append(dataframe)

        return dataframes

    @staticmethod
    def get_metadata(dataframes: List[List[str]]) -> pd.DataFrame:
        """
        Get metadata of a CIF file
        """
        return pd.DataFrame(dataframes[0])

    @staticmethod
    def get_loops(dataframes: List[List[str]]) -> List[pd.DataFrame]:
        """
        Get loops
        """
        loops = []
        for dataframe in dataframes[1:]:
            loop = pd.DataFrame(dataframe)
            loop_fixed = loop[loop[1].notna()]
            loop_fixed.columns = loop[loop[1].isna()][0]
            loops.append(loop_fixed)

        return loops

    def extract(self, cif_filepath: str) -> List[str]:
        """
        Read filepath and return sectioned string
        """
        return self.load_cif(cif_filepath)

    def transform(self, cif_list: List[str]) -> Tuple[pd.DataFrame, List[pd.DataFrame]]:
        """
        Transform CIF text into datafram
        :params:
        cif_list: List of sectioned CIF text
        :return:
        Tuple of :
        (
            dataframe of CIF metadata,
            Tuple of (
                dataframe of loop_1,
                dataframe of loop_2,
            )
        )
        """
        metadata = self.get_metadata(cif_list)
        extract_loops = self.get_loops(cif_list)
        return metadata, extract_loops

    def load(
        self, metadata: pd.DataFrame, extract_loops: List[pd.DataFrame]
    ) -> CIF2PandasAdapterOutput:
        """
        Load tranformed data into desired output
        """
        output = CIF2PandasAdapterOutput(metadata=metadata, loops=extract_loops)
        return output

    def apply(self, cif_filepath: str) -> List[pd.DataFrame]:
        """
        Apply ETL pipeline
        """
        # Extract
        cif_list = self.extract(cif_filepath)

        # Transform
        metadata = self.get_metadata(cif_list)
        extract_loops = self.get_loops(cif_list)

        # Load
        output = self.load(metadata, extract_loops)

        return output<jupyter_output><empty_output><jupyter_text>get most negative charge<jupyter_code>df_train = pd.read_csv('D:/Data/tmlcc-2021/train_clean_no_outlier_01-10.csv')
df_train['surface_area [m^2/g]'].replace(0,np.nan,inplace = True)
df_train.dropna(inplace = True)
df_train.reset_index(drop = True,inplace = True)
df_test = pd.read_csv('D:/Data/tmlcc-2021/test_set_clean_04-10.csv')
FG = list(df_train['functional_groups 1'].unique())

df_test<jupyter_output><empty_output><jupyter_text>reordering FG 1, 2 to unique key (train&test)<jupyter_code>order_FG = set()
for x,y in zip(df_train['functional_groups 1'].tolist(),df_train['functional_groups 2'].tolist()):
    k = sorted([x,y])
    order_FG.add((k[0],k[1]))
fgs = []
for x,y in zip(df_train['functional_groups 1'].tolist(),df_train['functional_groups 2'].tolist()):
    if (x,y) not in order_FG:
        fgs.append([y,x])
    else:
        fgs.append([x,y])
fgs = pd.DataFrame(fgs,columns = ['order_FG-1','order_FG-2'])
df_train = df_train.join(fgs)
fgs = []
for x,y in zip(df_test['functional_groups 1'].tolist(),df_test['functional_groups 2'].tolist()):
    if (x,y) not in order_FG:
        fgs.append([y,x])
    else:
        fgs.append([x,y])
fgs = pd.DataFrame(fgs,columns = ['order_FG-1','order_FG-2'])
df_test = df_test.join(fgs)
fgs<jupyter_output><empty_output><jupyter_text>mapping function<jupyter_code>map_bond ={
    'S':1,
    'D':2,
    'T':3,
    'A':4,
}
atoms = [
    'Br','C','H','F','N','S','I','Cl','O'
]
atoms2idx = {x:idx for idx,x in enumerate(atoms)}
sequence = {
    'COOH': ['C', 'O', 'O', 'H'],
    'F': ['F'],
    'OMe': ['O', 'C', 'H', 'H', 'H'],
    'H': ['H'],
    'NHMe': ['N', 'H', 'C', 'H', 'H', 'H'],
    'Pr': ['C', 'C', 'C', 'H', 'H', 'H', 'H', 'H', 'H', 'H'],
    'NH2': ['N', 'H', 'H'],
    'Br': ['Br'],
    'HCO': ['C', 'H', 'O'],
    'SO3H': ['S', 'O', 'O', 'O', 'H'],
    'Me': ['C', 'H', 'H', 'H'],
    'OEt': ['O', 'C', 'H', 'H', 'C', 'H', 'H', 'H'],
    'Cl': ['Cl'],
    'CN': ['C', 'N'],
    'Et': ['C', 'C', 'H', 'H', 'H', 'H', 'H'],
    'OH': ['O', 'H'],
    'NO2': ['N', 'O', 'O'],
    'OPr': ['O', 'C', 'H', 'H', 'C', 'H', 'H', 'C', 'H', 'H', 'H'],
    'Ph': ['C', 'C', 'C', 'C', 'C', 'C', 'H', 'H', 'H', 'H', 'H'],
    'I': ['I']
}
FG2idx = {x:idx for idx,x in enumerate(FG)}
idx2FG = {idx:x for idx,x in enumerate(FG)}<jupyter_output><empty_output><jupyter_text>get most negative charge (train)   convert mof to networkx format<jupyter_code>def mof_to_networkx(df_node,df_edge):
    G = nx.Graph()
    node_attrs = []
    for idx, k in enumerate(df_node['_atom_site_type_symbol'].values):
        attrs = {
            'atom':k
        }
        node_attrs.append((idx,attrs))
    list_atom = df_node['_atom_site_label'].tolist()
    edge_attrs = []
    for u,v,bond in zip(df_edge['_geom_bond_atom_site_label_1'].values,df_edge['_geom_bond_atom_site_label_2'].values,\
                   df_edge['_ccdc_geom_bond_type'].values):
        pos_u ,pos_v = list_atom.index(u) , list_atom.index(v)

        attrs = {
            'bond_type':bond
        }
        edge_attrs.append((pos_u,pos_v,attrs))
    G.add_nodes_from(node_attrs)
    G.add_edges_from(edge_attrs)
    return G<jupyter_output><empty_output><jupyter_text>subgraph isomorphism<jupyter_code>def get_mnc(df_node,list_nodes,H= False):
    y = [float(x) for x in df_node['_atom_type_partial_charge'].values[np.array(list_nodes)]]
    if H ==False:    
        return min(y)
    else:
        return max(y)
def subgraph_matching(G,FG):
    nm = iso.categorical_node_match('atom', 'atom')
    em = iso.categorical_edge_match('bond','bond')
    GM = nx.algorithms.isomorphism.GraphMatcher(G,FG,node_match =nm, edge_match = em)
    subgraphs = set()
    for g_sub in GM.subgraph_isomorphisms_iter():
        list_node = sorted([k for k,v in g_sub.items()])
        subgraphs.add(tuple(list_node))
    return subgraphs<jupyter_output><empty_output><jupyter_text>train<jupyter_code>root_path = 'D:/Data/tmlcc-2021/mof_cif_train'
feats = []
for i in tqdm(range(len(df_train))):
    fg1 = df_train.iloc[i]['order_FG-1']
    fg2 = df_train.iloc[i]['order_FG-2']
    name = df_train.iloc[i]['MOFname']+'.cif'
    metal = df_train.iloc[i]['metal_linker']
    ## read cif
    adapter = CIF2PandasAdapter()
    output = adapter.apply(os.path.join(root_path,name))
    ## load FG
    FG1 = get_FG(fg1)
    FG2 = get_FG(fg2)
    ## convert to pandas 
    df_node = output.loops[0].reset_index(drop = True)
    df_edge = output.loops[1].reset_index(drop = True)
    ## convert mof to networkx
    G = mof_to_networkx(df_node,df_edge)
    
    mnc_1 = []
    mnc_2 = []
    
    ## graph matching FG1 
    sub_FG_1 = subgraph_matching(G,FG1)
    for sub in sub_FG_1:
        flag = False
        if fg1 =='H':
            flag = True
        mnc_1.append(get_mnc(df_node,list(sub),flag))
        
    ## graph matching FG2
    sub_FG_2 = subgraph_matching(G,FG2)
    for sub in sub_FG_2:
        flag = False
        if fg2 =='H':
            flag = True
        mnc_2.append(get_mnc(df_node,list(sub),flag))
        
    ## store value
    if not mnc_1 :
        num_mnc_1 = 0
        mean_mnc_1 = 0
    else:
        num_mnc_1 =np.mean(mnc_1)
        mean_mnc_1 = len(mnc_1)
    if not mnc_2 :
        num_mnc_2 = 0
        mean_mnc_2 = 0
    else:
        num_mnc_2 =np.mean(mnc_2)
        mean_mnc_2 = len(mnc_2)
        
    feat = [num_mnc_1,mean_mnc_1,num_mnc_2,mean_mnc_2]
    feats.append(feat)<jupyter_output>100%|████████████████████████████████████████████████████████████████████████████| 56737/56737 [24:35<00:00, 38.45it/s]<jupyter_text>count H<jupyter_code>most_negative= pd.DataFrame(feats,columns = ['num_fg-1','mnc_fg-1','num_fg-2','mnc_fg-2'])
most_negative.head(10)<jupyter_output><empty_output><jupyter_text>count H with C<jupyter_code>most_negative= pd.DataFrame(feats,columns = ['num_fg-1','mnc_fg-1','num_fg-2','mnc_fg-2'])
most_negative.head(10)
#most_negative.to_csv('train_mnc_fg_1.csv',index = False)<jupyter_output><empty_output><jupyter_text>test<jupyter_code>root_path = 'D:/Data/tmlcc-2021/mof_cif_test'
feats = []
for i in tqdm(range(len(df_test))):
    fg1 = df_test.iloc[i]['order_FG-1']
    fg2 = df_test.iloc[i]['order_FG-2']
    name = df_test.iloc[i]['MOFname']+'.cif'
    metal = df_test.iloc[i]['metal_linker']
    ## read cif
    adapter = CIF2PandasAdapter()
    output = adapter.apply(os.path.join(root_path,name))
    ## load FG
    FG1 = get_FG(fg1)
    FG2 = get_FG(fg2)
    ## convert to pandas 
    df_node = output.loops[0].reset_index(drop = True)
    df_edge = output.loops[1].reset_index(drop = True)
    ## convert mof to networkx
    G = mof_to_networkx(df_node,df_edge)
    
    mnc_1 = []
    mnc_2 = []
    
    ## graph matching FG1 
    sub_FG_1 = subgraph_matching(G,FG1)
    for sub in sub_FG_1:
        flag = False
        if fg1 =='H':
            flag = True
        mnc_1.append(get_mnc(df_node,list(sub),flag))
        
    ## graph matching FG2
    sub_FG_2 = subgraph_matching(G,FG2)
    for sub in sub_FG_2:
        flag = False
        if fg2 =='H':
            flag = True
        mnc_2.append(get_mnc(df_node,list(sub),flag))
        
    if not mnc_1 :
        num_mnc_1 = 0
        mean_mnc_1 = 0
    else:
        num_mnc_1 =np.mean(mnc_1)
        mean_mnc_1 = len(mnc_1)
        
    if not mnc_2 :
        num_mnc_2 = 0
        mean_mnc_2 = 0
    else:
        num_mnc_2 =np.mean(mnc_2)
        mean_mnc_2 = len(mnc_2)
        
    feat = [num_mnc_1,mean_mnc_1,num_mnc_2,mean_mnc_2]
    feats.append(feat)

most_negative= pd.DataFrame(feats,columns = ['num_fg-1','mnc_fg-1','num_fg-2','mnc_fg-2'])
most_negative.head(10)
most_negative= pd.DataFrame(feats,columns = ['num_fg-1','mnc_fg-1','num_fg-2','mnc_fg-2'])
most_negative.head(10)
#most_negative.to_csv('test_mnc_fg_1.csv',index = False)




nm = iso.categorical_node_match('atom', 'atom')
em = iso.categorical_edge_match('bond','bond')
GM = nx.algorithms.isomorphism.GraphMatcher(G,FG,node_match =nm, edge_match = em)
for subgraph in GM.subgraph_isomorphisms_iter():
    print(subgraph)
print GM = iso.GraphMatcher(B,A,node_match=iso.categorical_node_match(['material', 'size'],['metal',1]))
G.nodes(data = True)
GM.mapping

subgraphs[153].nodes(data = True)
subgraphs[153].edges(data = True)
nx.draw_networkx(subgraphs[153])
FG.nodes(data = True)
FG.edges(data = True)

nx.is_isomorphic(FG,subgraphs[120],node_match = True , edge_match = True)

test = nx.Graph()
test.add_node(1,atom = 'O')
test.add_node(2,atom = 'H')
test.add_node(3,atom = 'C')
test.add_node(4,atom = 'H')
test.add_node(5,atom = 'H')
#c
test.add_edge(3,1,bond_type=map_bond['S'])
test.add_edge(3,4,bond_type=map_bond['S'])
test.add_edge(3,5,bond_type=map_bond['S'])
test.add_edge(3,2,bond_type=map_bond['S'])
nx.draw_networkx(test)
import networkx.algorithms.isomorphism as iso
nm = iso.categorical_node_match('atom', 'atom')
em = iso.categorical_edge_match('bond','bond')
nx.is_isomorphic(FG,test,node_match = nm , edge_match = em)





import networkx as nx
# ref:https://github.com/maxhodak/keras-molecules/pull/32/files

def mol_to_nx(mol):
    G = nx.Graph()
    for atom in mol.GetAtoms():
        G.add_node(atom.GetIdx())
    for bond in mol.GetBonds():
        G.add_edge(bond.GetBeginAtomIdx(),
                   bond.GetEndAtomIdx(),
                   bond_type=bond.GetBondType())
    return G<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.15542332783712093, 'fraction_numerical': 0.017062154993189475, 'mean_word_length': 2.961942629934678, 'pattern_counts': {'":': 1, '<': 53, '<?xml version=': 0, '>': 59, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 3, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '13', 'dir': 'jupyter-structured-clean-dedup', 'id': '51012', 'license': 'MIT', 'n_tokens_mistral': 5571, 'n_tokens_neox': 5293, 'n_words': 952, 'path': 'nicky/mnc_functional_group.ipynb', 'repo_name': 'WitMaster98/capturing_CO2_working_cap_MOFs', 'size': '13949'}