En este tutorial, creamos un marco compacto y eficiente que demuestra cómo convertir la documentación de herramientas en interfaces estandarizadas e invocables, registrar esas herramientas en un sistema central y ejecutarlas como parte de un proceso automatizado. A medida que avanzamos en cada etapa, creamos un convertidor simple, diseñamos herramientas bioinformáticas simuladas, las organizamos en un registro y comparamos ejecuciones de procesos individuales y de varios pasos. A través de este proceso, exploramos cómo las interfaces de herramientas estructuradas y la automatización pueden optimizar y modularizar los flujos de trabajo de datos. Consulta los CÓDIGOS COMPLETOS aquí.
salidas: dict[str, str]
def parse_doc_to_spec(nombre: str, doc: str) -> ToolSpec: desc = doc.strip().splitlines()[0].strip() si doc.strip() sino nombre arg_block = “n”.join([l for l in doc.splitlines() if “–” in l or “:” in l]) entradas = {} para la línea en arg_block.splitlines(): m = re.findall(r”(–?w[w-]*|bw+b)s*[:=]?s*(w+)?”, línea) para la clave, escriba m: k = key.lstrip(“-“) si k y k no están en las entradas y k no están en [“Returns”,”Output”,”Outputs”]: entradas[k] = (typ o “str”) si no es entradas: entradas = {“in”: “str”} return ToolSpec(nombre=nombre, descripción=desc, entradas=entradas, salidas={“out”:”json”})
Comenzamos definiendo la estructura de nuestras herramientas y escribiendo un analizador simple que convierte documentación simple en una especificación de herramienta estandarizada. Esto nos ayuda a extraer automáticamente parámetros y resultados de descripciones textuales. Consulta los CÓDIGOS COMPLETOS aquí.
q30 = suma(l>=min_len para l en lente)/max(1,len(lente)) gc = suma(c en “GCgc” para s en seqs para c en s)/max(1,sum(lente)) return {“n_seqs”:len(lens),”len_mean”:(sum(lens)/max(1,len(lens))),”pct_q30″:q30,”gc”:gc} def tool_bowtie2_like(ref:str, reads:str, mode:str=”end-to-end”) -> Dict[str,Any]: def revcomp(s): t=str.maketrans(“ACGTacgt”,”TGCAtgca”); devolver s.translate
lista_lecturas=[r for r in re.split(r”>[^n]*n”, se lee)[1:]]ref_seq=””.join(ref.splitlines()[1:]) aciertos=[]
para i,r en enumerate(reads_list): rseq=””.join(r.split()) alineado = (rseq en ref_seq) o (revcomp(rseq) en ref_seq) hits.append({“read_id”:i,”aligned”:bool(aligned),”pos”:ref_seq.find(rseq)}) return {“n”:len(golpes),”alineado”:suma(h[“aligned”] para h en visitas),”mode”:mode,”hits”:hits} def tool_bcftools_like(ref:str, alt:str, win:int=15) -> Dict[str,Any]: ref_seq=””.join(ref.splitlines()[1:]); alt_seq=””.join(alt.splitlines()[1:]) n=min(len(ref_seq),len(alt_seq)); vars=[]
para i en rango (n): si ref_seq[i]!=alt_seq[i]: vars.append({“pos”:i,”ref”:ref_seq[i],”alt”:alt_seq[i]}) devuelve {“n_sites”:n,”n_var”:len(vars),”variantes”:vars[:win]} FASTQC_DOC = “””Control de calidad tipo FastQC para FASTA –seq_fasta: str –min_len: int Salidas: json””” BOWTIE_DOC = “””Alineador tipo Bowtie2 –ref: str –reads: str –mode: str Salidas: json””” BCF_DOC = “””Variante de llamada tipo bcftools –ref: str –alt: cadena –win: int Salidas: json”””
Creamos implementaciones simuladas de herramientas bioinformáticas como FastQC, Bowtie2 y Bcftools. Definimos sus entradas y salidas esperadas para que puedan ejecutarse de manera consistente a través de una interfaz unificada. Consulta los CÓDIGOS COMPLETOS aquí.
def call_tool(self, nombre:cadena, argumentos:Dict[str,Any]) -> Dictar[str,Any]: si el nombre no está en self.tools: aumente KeyError(f”herramienta {nombre} no encontrada”) spec = self.tools[name].spec kwargs={k:args.get(k) para k en spec.inputs.keys()} devolver self.tools[name].fn(**kwargs) server=MCPServer() server.register(“fastqc”, FASTQC_DOC, tool_fastqc) server.register(“bowtie2”, BOWTIE_DOC, tool_bowtie2_like) server.register(“bcftools”, BCF_DOC, tool_bcftools_like) Tarea = Tupla[str, Dict[str,Any]]TUBERÍAS = { “rnaseq_qc_align_call”:[
(“fastqc”, {“seq_fasta”:”{reads}”, “min_len”:30}),
(“bowtie2”, {“ref”:”{ref}”, “reads”:”{reads}”, “mode”:”end-to-end”}),
(“bcftools”, {“ref”:”{ref}”, “alt”:”{alt}”, “win”:15}),
]
} def compile_pipeline(nl_request:str) -> Lista[Task]: clave = “rnaseq_qc_align_call” if re.search(r”rna|qc|align|variant|call”, nl_request, re.I) else “rnaseq_qc_align_call” return TUBERÍAS[key]
Construimos un servidor liviano que registra herramientas, enumera sus especificaciones y nos permite llamarlas mediante programación. También definimos una estructura básica de canalización que describe la secuencia en la que se deben ejecutar las herramientas. Consulta los CÓDIGOS COMPLETOS aquí.
Preparamos pequeños datos FASTA sintéticos para probar e implementamos una función que ejecuta todo el proceso. Aquí, pasamos dinámicamente los parámetros de la herramienta y ejecutamos cada paso de la secuencia. Consulta los CÓDIGOS COMPLETOS aquí.
(“fastqc”, {“seq_fasta”:READS,”min_len”:25}),
(“bowtie2”, {“ref”:REF,”reads”:READS,”mode”:”end-to-end”}),
(“bcftools”, {“ref”:REF,”alt”:ALT,”win”:10}),
]
filas =[]
para nombre, argumentos en casos: t0=time.time(); vale=Verdadero; err=Ninguno; out=Ninguno intente: out=server.call_tool(name,args) excepto Excepción como e: ok=False; err=str(e) rows.append({“tool”:name,”ok”:ok,”ms”:int((time.time()-t0)*1000),”out_keys”:list(out.keys()) si está bien [],”err”:err}) devuelve filas def bench_pipeline() -> Dict[str,Any]: t0=time.time() res=run_pipeline(“Ejecutar control de calidad, alineación y llamada de variante de RNA-seq.”, {“ref”:REF,”reads”:READS,”alt”:ALT}) ok = all(paso[“output”] para paso en res[“results”]) devuelve {“pipeline”:”rnaseq_qc_align_call”,”ok”:ok,”ms”:int((time.time()-t0)*1000),”n_steps”:len(res[“results”])} print(“== HERRAMIENTAS ==”); print(json.dumps(server.list_tools(), sangría=2)) print(“n== BANCO INDIVIDUAL ==”); print(json.dumps(bench_individual(), indent=2)) print(“n== BANCO DE TUBERÍA ==”); print(json.dumps(bench_pipeline(), sangría=2)) print(“n== PIPELINE RUN ==”); print(json.dumps(run_pipeline(“Ejecutar control de calidad, alineación y llamada de variante de RNA-seq.”, {“ref”:REF,”reads”:READS,”alt”:ALT}), indent=2))
Comparamos tanto las herramientas individuales como el proceso completo, capturando sus resultados y métricas de rendimiento. Finalmente, imprimimos los resultados para verificar que cada etapa del flujo de trabajo se ejecute correctamente y se integre sin problemas.
En conclusión, desarrollamos una comprensión clara de cómo la conversión, el registro y la orquestación de herramientas livianas pueden funcionar juntas en un solo entorno. Observamos cómo una interfaz unificada nos permite conectar múltiples herramientas sin problemas, ejecutarlas en secuencia y medir su rendimiento. Este ejercicio práctico nos ayuda a apreciar cómo los principios de diseño simples, la estandarización, la automatización y la modularidad pueden mejorar la reproducibilidad y la eficiencia de los flujos de trabajo computacionales en cualquier dominio.
Consulta los CÓDIGOS COMPLETOS aquí. No dude en consultar nuestra página de GitHub para tutoriales, códigos y cuadernos. Además, no dude en seguirnos en Twitter y no olvide unirse a nuestro SubReddit de más de 100.000 ML y suscribirse a nuestro boletín. ¡Esperar! estas en telegrama? Ahora también puedes unirte a nosotros en Telegram.
Asif Razzaq es el director ejecutivo de Marktechpost Media Inc.. Como empresario e ingeniero visionario, Asif está comprometido a aprovechar el potencial de la inteligencia artificial para el bien social. Su esfuerzo más reciente es el lanzamiento de una plataforma de medios de inteligencia artificial, Marktechpost, que se destaca por su cobertura en profundidad del aprendizaje automático y las noticias sobre aprendizaje profundo que es técnicamente sólida y fácilmente comprensible para una amplia audiencia. La plataforma cuenta con más de 2 millones de visitas mensuales, lo que ilustra su popularidad entre el público.
🙌 Siga MARKTECHPOST: agréguenos como fuente preferida en Google.