DrugDesign Data Analysis
Module of the DrugDesign project responsible for loading and pre-processing data from ChEMBL and PubChem
All Classes Namespaces Files Functions Variables Pages
files_funcs Namespace Reference

Functions

 CombineCSVInFolder (str folder_name, str combined_file_name)
 
 DeleteFilesInFolder (str folder_name, list[str]|None except_items=None, bool delete_folders=False)
 
bool IsFileInFolder (str file_name, str folder_name)
 
bool IsFolderEmpty (str folder_name)
 
 MoveFileToFolder (str file_name, str curr_folder_name, str folder_name)
 
 SaveMolfilesToSDF (pd.DataFrame data, str file_name, str molecule_id_column_name, pd.DataFrame extra_data=pd.DataFrame(), bool indexing_lists=False)
 

Detailed Description

Utils/files_funcs.py Этот модуль содержит набор функций для работы с файлами и директориями, включая удаление, перемещение, объединение файлов, а также сохранение молекулярных структур в формате SDF.

Function Documentation

◆ CombineCSVInFolder()

files_funcs.CombineCSVInFolder ( str folder_name,
str combined_file_name )
Склеивает все .csv файлы в папке в один. Args: folder_name (str): имя папки с .csv файлами. combined_file_name (str): имя склеенного .csv файла.
121def CombineCSVInFolder(folder_name: str, combined_file_name: str):
122 """
123 Склеивает все .csv файлы в папке в один.
124
125 Args:
126 folder_name (str): имя папки с .csv файлами.
127 combined_file_name (str): имя склеенного .csv файла.
128 """
129
130 # получаем конфигурацию для объединения CSV-файлов.
131 combine_config: Config = config["Utils"]["CombineCSVInFolder"]
132
133 # получаем индекс формата логгера.
134 restore_index: int = (
135 v_logger.UpdateFormat(combine_config["logger_label"], combine_config["logger_color"])
136 - 1
137 )
138
139 v_logger.info("Start combining downloads...")
140 v_logger.info("-", LogMode.VERBOSELY)
141
142 # если файл уже существует и нужно пропускать скачанные, выходим.
143 if (
144 IsFileInFolder(folder_name, f"{combined_file_name}.csv") and config["skip_downloaded"]
145 ):
146 v_logger.info(
147 f"File '{combined_file_name}.csv' is in folder, no need to combine.",
148 LogMode.VERBOSELY,
149 )
150
151 # восстанавливаем формат логгера.
152 v_logger.RestoreFormat(restore_index)
153 return
154
155 # создаем пустой DataFrame для объединения.
156 combined_df = pd.DataFrame()
157
158 # если папка пуста, выходим.
159 if len(os.listdir(folder_name)) == 0:
160 v_logger.info(f"{folder_name} is empty, no need to combine.")
161
162 # восстанавливаем формат логгера.
163 v_logger.RestoreFormat(restore_index)
164 return
165
166 # итерируемся по файлам в папке.
167 for file_name in os.listdir(folder_name):
168 # проверяем, является ли файл CSV-файлом и не является ли он
169 # результирующим.
170 if file_name.endswith(".csv") and file_name != f"{combined_file_name}.csv":
171 # формируем полный путь к файлу.
172 full_file_name: str = os.path.join(folder_name, file_name)
173
174 v_logger.info(f"Collecting '{file_name}' to pandas.DataFrame...", LogMode.VERBOSELY)
175
176 # читаем CSV-файл в DataFrame.
177 df = pd.read_csv(full_file_name, sep=config["csv_separator"], low_memory=False)
178
179 v_logger.success(
180 f"Collecting '{file_name}' to pandas.DataFrame!", LogMode.VERBOSELY
181 )
182 v_logger.info(
183 f"Concatenating '{file_name}' to combined_data_frame...", LogMode.VERBOSELY
184 )
185
186 # объединяем DataFrame с общим DataFrame.
187 combined_df = pd.concat([combined_df, df], ignore_index=True)
188
189 v_logger.success(
190 f"Concatenating '{file_name}' to combined_data_frame!", LogMode.VERBOSELY
191 )
192 v_logger.info("-", LogMode.VERBOSELY)
193
194 v_logger.info(
195 f"Collecting to combined .csv file in '{folder_name}'...", LogMode.VERBOSELY
196 )
197
198 # сохраняем объединенный DataFrame в CSV-файл.
199 combined_df.to_csv(
200 f"{folder_name}/{combined_file_name}.csv", sep=config["csv_separator"], index=False
201 )
202
203 v_logger.success(
204 f"Collecting to combined .csv file in '{folder_name}'!", LogMode.VERBOSELY
205 )
206 v_logger.info("-", LogMode.VERBOSELY)
207 v_logger.success("End combining downloads!")
208
209 # восстанавливаем формат логгера.
210 v_logger.RestoreFormat(restore_index)
211
212

◆ DeleteFilesInFolder()

files_funcs.DeleteFilesInFolder ( str folder_name,
list[str] | None except_items = None,
bool delete_folders = False )
Удаляет все файлы в указанной папке, кроме файлов в списке исключений. Args: folder_name (str): путь к папке. except_items (list[str], optional): список исключений (файлов или папок), которые не нужно удалять. Defaults to []. delete_folders (bool, optional): удалять ли вложенные папки. Defaults to False.
21):
22 """
23 Удаляет все файлы в указанной папке, кроме файлов в списке исключений.
24
25 Args:
26 folder_name (str): путь к папке.
27 except_items (list[str], optional): список исключений (файлов или папок),
28 которые не нужно удалять.
29 Defaults to [].
30 delete_folders (bool, optional): удалять ли вложенные папки.
31 Defaults to False.
32 """
33
34 # итерируемся по элементам в папке.
35 if except_items is None:
36 except_items = []
37 for item in os.listdir(folder_name):
38 # формируем путь к текущему элементу.
39 item_path: str = os.path.join(folder_name, item)
40
41 # если элемент не в списке исключений.
42 if item not in except_items:
43 # если это файл, удаляем его.
44 if os.path.isfile(item_path):
45 os.remove(item_path)
46
47 # если нужно удалять папки и это папка, удаляем её.
48 elif delete_folders and os.path.isdir(item_path):
49 shutil.rmtree(item_path)
50
51

◆ IsFileInFolder()

bool files_funcs.IsFileInFolder ( str file_name,
str folder_name )
Проверяет, существует ли файл в указанной папке. Args: file_name (str): имя файла для проверки. folder_name (str): путь к папке. Returns: bool: True, если файл существует, иначе False.
52def IsFileInFolder(file_name: str, folder_name: str) -> bool:
53 """
54 Проверяет, существует ли файл в указанной папке.
55
56 Args:
57 file_name (str): имя файла для проверки.
58 folder_name (str): путь к папке.
59
60 Returns:
61 bool: True, если файл существует, иначе False.
62 """
63
64 # формируем полный путь к файлу.
65 full_file_name = os.path.join(folder_name, file_name)
66 # возвращаем True, если файл существует, иначе False.
67 return os.path.exists(full_file_name)
68
69

◆ IsFolderEmpty()

bool files_funcs.IsFolderEmpty ( str folder_name)
Проверяет, является ли папка пустой. Args: folder_name (str): путь к папке. Returns: bool: True, если папка пуста, иначе False.
70def IsFolderEmpty(folder_name: str) -> bool:
71 """
72 Проверяет, является ли папка пустой.
73
74 Args:
75 folder_name (str): путь к папке.
76
77 Returns:
78 bool: True, если папка пуста, иначе False.
79 """
80
81 try:
82 # если количество элементов в папке равно нулю, возвращаем True.
83 return len(os.listdir(folder_name)) == 0
84
85 # если папка не найдена, возвращаем True.
86 except FileNotFoundError:
87 return True
88
89

◆ MoveFileToFolder()

files_funcs.MoveFileToFolder ( str file_name,
str curr_folder_name,
str folder_name )
Перемещает файл в указанную папку. Args: file_name (str): имя файла. curr_folder_name (str): путь к текущей папке файла. folder_name (str): путь к целевой папке.
90def MoveFileToFolder(file_name: str, curr_folder_name: str, folder_name: str):
91 """
92 Перемещает файл в указанную папку.
93
94 Args:
95 file_name (str): имя файла.
96 curr_folder_name (str): путь к текущей папке файла.
97 folder_name (str): путь к целевой папке.
98 """
99
100 # создаем целевую папку, если она не существует.
101 os.makedirs(folder_name, exist_ok=True)
102
103 # формируем полный путь к файлу.
104 full_file_name = os.path.join(curr_folder_name, file_name)
105 # формируем полный путь к файлу в целевой папке.
106 next_file_name = os.path.join(folder_name, file_name)
107
108 # если файл с таким именем уже существует в целевой папке, удаляем его.
109 if os.path.exists(next_file_name):
110 os.remove(next_file_name)
111
112 # если файл существует, перемещаем его.
113 if os.path.exists(full_file_name):
114 shutil.move(full_file_name, folder_name)
115
116 # если файл не существует, логируем предупреждение.
117 else:
118 v_logger.warning(f"{full_file_name} does not exist!", LogMode.VERBOSELY)
119
120

◆ SaveMolfilesToSDF()

files_funcs.SaveMolfilesToSDF ( pd.DataFrame data,
str file_name,
str molecule_id_column_name,
pd.DataFrame extra_data = pd.DataFrame(),
bool indexing_lists = False )
Сохраняет molfiles из pd.DataFrame в .sdf файл. Args: data (pd.DataFrame): DataFrame с колонками molfile и id. file_name (str): имя файла (без ".sdf"). molecule_id_column_name (str): имя колонки с id соединения. extra_data (pd.DataFrame, optional): дополнительная информация. Defaults to pd.DataFrame(). indexing_lists (bool, optional): нужно ли индексировать списки. Defaults to False.
219):
220 """
221 Сохраняет molfiles из pd.DataFrame в .sdf файл.
222
223 Args:
224 data (pd.DataFrame): DataFrame с колонками molfile и id.
225 file_name (str): имя файла (без ".sdf").
226 molecule_id_column_name (str): имя колонки с id соединения.
227 extra_data (pd.DataFrame, optional): дополнительная информация.
228 Defaults to pd.DataFrame().
229 indexing_lists (bool, optional): нужно ли индексировать списки.
230 Defaults to False.
231 """
232
233 def WriteColumnAndValueToSDF(file: TextIOWrapper, value: Any, column: str = ""):
234 """
235 Записывает столбец и значение в .sdf файл.
236
237 Args:
238 file (TextIOWrapper): открытый файл для записи.
239 value (Any): значение, которое нужно записать.
240 column (str, optional): имя столбца. Defaults to "".
241 """
242
243 # если столбец не задан, выходим.
244 if not column:
245 return
246
247 # если значение - список или pd.Series.
248 if isinstance(value, list) or isinstance(value, pd.Series):
249 file.write(f"> <{column}>\n")
250
251 i: int = 0
252 # итерируемся по элементам списка.
253 for elem in value:
254 # если элемент - словарь.
255 if isinstance(elem, dict):
256 # рекурсивно вызываем функцию для записи словаря.
257 WriteColumnAndValueToSDF(file, elem)
258
259 # если элемент - не словарь.
260 else:
261 # преобразуем элемент в строку.
262 elem = str(elem)
263
264 # если элемент не пустой, записываем его.
265 if elem not in {"nan", "None", ""}:
266 # если нужно индексировать списки.
267 if indexing_lists:
268 file.write(f"{i}: {elem}\n")
269 # если не нужно индексировать списки.
270 else:
271 file.write(f"{elem}\n")
272 i += 1
273
274 # если значение - словарь.
275 elif isinstance(value, dict):
276 file.write(f"> <{column}>\n")
277
278 # итерируемся по элементам словаря.
279 for key, elem in value.items():
280 # преобразуем элемент в строку.
281 elem = str(elem)
282
283 # если элемент не пустой, записываем его.
284 if elem not in {"nan", "None", ""}:
285 file.write(f"{key}: {elem}\n")
286
287 # если значение - не список и не словарь.
288 else:
289 # преобразуем значение в строку.
290 value = str(value)
291
292 # если значение не пустое, записываем его.
293 if value not in {"nan", "None", ""}:
294 file.write(f"> <{column}>\n")
295
296 file.write(f"{value}\n")
297
298 # записываем пустую строку для разделения значений.
299 file.write("\n")
300
301 # открываем файл для записи.
302 with open(f"{file_name}.sdf", "w", encoding="utf-8") as f:
303 # итерируемся по строкам DataFrame.
304 for value in data.to_numpy():
305 # получаем id молекулы и molfile.
306 molecule_id, molfile = value
307
308 # записываем id молекулы и molfile в файл.
309 f.write(f"{molecule_id}{molfile}\n\n")
310
311 # если есть дополнительная информация.
312 if not extra_data.empty:
313 # устанавливаем id молекулы в качестве индекса.
314 df = extra_data.set_index(f"{molecule_id_column_name}")
315
316 # итерируемся по столбцам DataFrame.
317 for column in df.columns:
318 # записываем столбец и значение в файл.
319 WriteColumnAndValueToSDF(f, df.loc[molecule_id, column], column)
320
321 # записываем разделитель между молекулами.
322 f.write("$$$$\n")
323
324 v_logger.info(f"Writing {molecule_id} data to .sdf file...", LogMode.VERBOSELY)
325
326 # переоткрываем файл, чтобы исправить избыточные переносы строк
327 # (да, заново открыть его - это самый простой способ)
328 with open(f"{file_name}.sdf", encoding="utf-8") as f:
329 sdf_content = f.read()
330
331 # максимальное кол-во переносов строк
332 max_n_amounts = 0
333 while "\n" * max_n_amounts in sdf_content:
334 max_n_amounts += 1
335
336 # заменяем все идущие подряд переносы
337 # (вплоть до "\n\n" невключительно)
338 for amount in range(max_n_amounts, 2, -1):
339 sdf_content = sdf_content.replace("\n" * amount, "\n\n")
340
341 # после окончания блока должен быть лишь 1 перенос
342 sdf_content = sdf_content.replace("$$$$\n\n", "$$$$\n")
343
344 # перезаписываем файл
345 with open(f"{file_name}.sdf", "w", encoding="utf-8") as f:
346 f.write(sdf_content)