#! /usr/bin/env python3 import argparse import logging import os import pandas as pd import pysam class ChemoRun: def __init__(self, database, project, output_dir, name, vcf): self.database = database self.project = project self.output_dir = output_dir self.name = name self.vcf = vcf self.logger = self.setup_logger() def setup_logger(self): logger = logging.getLogger(__name__) logger.setLevel(logging.ERROR) formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s') # 创建控制台处理器 console_handler = logging.StreamHandler() console_handler.setFormatter(formatter) logger.addHandler(console_handler) return logger @staticmethod def get_drug_plan(x, drugsum): tlist = x.split('+') tdeslist = list() for tdes in tlist: if tdes.strip() in drugsum: t1_des = drugsum[tdes.strip()] tdeslist.append(t1_des) if '慎用' in tdeslist or '谨慎' in tdeslist: return '慎用' elif '推荐' in tdeslist: return '推荐' elif '常规' in tdeslist: return '可选' else: return '可选' @staticmethod def build_record_dict(vcf_file): vcfdata = pysam.VariantFile(vcf_file) records = {} for record in vcfdata: chrom = record.chrom pos = record.pos records[(chrom, pos)] = record return records def parsedata(self): """ 处理chemo_drug_rsid_snppos.xlsx, 并生成对应bed文件 """ data = pd.read_excel(self.database, None) drug_rsid = data['drug_rsid'] drug_combine = data['drug_combine'] drug_rsid.fillna('.', inplace=True) drug_combine.fillna('.', inplace=True) drug_rsid_data = drug_rsid[drug_rsid['source'].str.contains(self.project)] drug_combine_data = drug_combine[drug_combine['source'].str.contains(self.project)] if drug_rsid_data.empty: logging.error(f'无此项目！{self.project} ,请查看参数project是否正确！') raise UserWarning(f'无此项目！{self.project} ,请查看参数project是否正确！') # 生成bed文件 beddata = drug_rsid_data[['chr', 'start', 'end', 'rsid']].drop_duplicates() beddata.rename(columns={'chr': '#chr'}, inplace=True) bedpath = os.path.join(self.output_dir, f'{self.project}_chemo.bed') beddata.to_csv(bedpath, sep='\t', index=False) return drug_rsid_data, drug_combine_data, bedpath def parse_vcf(self, vcfpath, drug_rsid_data, drug_combine_data): records = self.build_record_dict(vcfpath) beddata = drug_rsid_data[['chr', 'start', 'end', 'rsid']].drop_duplicates() fliterdata = pd.DataFrame() for _, row in beddata.iterrows(): chrom = row['chr'] end = row['end'] fliter = pd.DataFrame() if (chrom, end) not in records: fliter = pd.concat([fliter, drug_rsid_data[ (drug_rsid_data['chr'] == chrom) & (drug_rsid_data['end'] == end) & (drug_rsid_data['genotype'] == '0/0') ]]) self.logger.info(f'{(chrom, end),}野生型') else: record = records[(chrom, end)] ref = record.ref alt = record.alts[0] gt = '/'.join(list(map(str, sorted(record.samples.get(record.samples.keys()[0]).get('GT'))))) fliter = pd.concat([fliter, drug_rsid_data[ (drug_rsid_data['chr'] == chrom) & (drug_rsid_data['end'] == end) & (drug_rsid_data['ref'] == ref) & (drug_rsid_data['alt'] == alt) & (drug_rsid_data['genotype'] == gt) ]]) self.logger.info(f'{(chrom, end, ref, alt),}, 非野生型') if fliter.empty: self.logger.error( 'chr: %s , end: %s 数据库未能匹配, 野生型0/0也未能匹配' % (chrom, end)) raise UserWarning( 'chr: %s , end: %s 数据库未能匹配, 野生型0/0也未能匹配' % (chrom, end)) fliterdata = pd.concat([fliterdata, fliter]) # 生成过滤之后文件 respath = os.path.join(self.output_dir, f'{self.name}.chemo.res.csv') if not fliterdata.empty: fliterdata['project'] = self.project fliterdata.to_csv(respath, sep='\t', index=False) # 分类汇总 同位点，药物合并 drug.infos.txt drugrsid = fliterdata[['drugname', 'genename', 'rsid', 'result', 'level', 'tips', 'drugsort']] drugrsid = drugrsid.drop_duplicates() resdrugrsid = drugrsid.groupby(['drugname', 'genename', 'rsid', 'result', 'level', 'drugsort'])['tips'].agg( ','.join).reset_index() resdrugrsid.rename(columns= {'drugname': '药物', 'genename': '检测基因', 'rsid': '检测位点', 'result': '基因型', 'level': '证据等级', 'tips': '用药提示'}, inplace=True) resdrugrsid = resdrugrsid.sort_values(by=['drugsort', '药物', '检测基因']) resdrugrsid.to_csv(os.path.join(self.output_dir, f'{self.name}.drug.infos.txt'), index=False, sep='\t') # 药物 药物疗效 推荐程度合并 drug.res.txt drugtypesum = fliterdata[['drugname', 'drugtype', 'rsid', 'weights']] drugtypesum = drugtypesum.drop_duplicates() drugtyperes = list() drugsum = dict() for drug, drugdata in drugtypesum.groupby('drugname'): tipsnum = drugdata.groupby(['drugtype']).agg({'weights': 'sum'}).to_dict('index') sumlist = list() if 'LX' in tipsnum: LX = tipsnum['LX']['weights'] if LX > 0: lxdes = '疗效较好' lxnum = 1 elif LX == 0: lxdes = '疗效一般' lxnum = 0 else: lxdes = '疗效较差' lxnum = -1 sumlist.append(lxdes) else: LX = 0 lxnum = 0 if 'DF' in tipsnum: DF = tipsnum['DF']['weights'] if DF > 0: dfdes = '毒副较低' dfnum = 1 elif DF == 0: dfdes = '毒副一般' dfnum = 0 else: dfdes = '毒副较高' dfnum = -1 sumlist.append(dfdes) else: DF = 0 dfnum = 0 # 评价方式 疗效 1 0 -1, 毒副 1 0 -1 ，可形成9宫格 sumnum = lxnum + dfnum if sumnum > 0: sumdes = '推荐' elif sumnum == 0: sumdes = '常规' else: sumdes = '谨慎' # 特别药物处理 if (drug == "氟尿嘧啶" or drug == "卡培他滨") and DF < 0: sumdes = '谨慎' drugtyperes.append(dict( 药物名称=drug, 疗效=LX, 毒副=DF, 推荐程度=sumdes, 疗效和毒副总结=','.join(sumlist) )) drugsum[drug] = sumdes # 报告中展示药物有顺序 drugsort = fliterdata[['drugname', 'drugsort']].drop_duplicates() drugsort_dict = drugsort.set_index('drugname')['drugsort'].to_dict() drugtyperes_sort = sorted(drugtyperes, key=lambda x: ( drugsort_dict[x['药物名称']] if x['药物名称'] in drugsort_dict else 100, x['药物名称'])) # 生成数据 pd.DataFrame(drugtyperes_sort).to_csv(os.path.join(self.output_dir, f'{self.name}.drug.res.txt'), index=False, sep='\t') # 联合用药 if not drug_combine_data.empty: drug_combine_data['临床提示'] = drug_combine_data['用药方案'].apply(self.get_drug_plan, args=(drugsum,)) drug_combine_data.to_csv(os.path.join(self.output_dir, f'{self.name}.chemo.comb.txt'), index=False, sep='\t') def run(self): try: drug_rsid_data, drug_combine_data, bedpath = self.parsedata() self.parse_vcf(self.vcf, drug_rsid_data, drug_combine_data) except UserWarning as e: self.logger.info("******", e) self.logger.info(f"Chemo Analysis over!") if __name__ == "__main__": parser = argparse.ArgumentParser(description="Chemotherapy Process Script") parser.add_argument('-d', '--database', help="Path to chemo_drug's database", default='/home/zhangchao/project/pipeline/workflow/script/public/chemo_database.xlsx') parser.add_argument('-p', '--project', help="Project name", required=True) parser.add_argument('-n', '--name', help="Name for sample", required=True) parser.add_argument('-v', '--vcf', help="germline vcf", required=True) parser.add_argument('-o', '--output_dir', help="Output directory, default ./", default='') args = parser.parse_args() chemorun = ChemoRun(args.database, args.project, args.output_dir, args.name, args.vcf) chemorun.run()