Merge pull request #63 from tahiri-lab/typer-integration

Typer integrationImplement Typer for Improved CLI Functionality

aphylogeo/genetic_trees.py

@@ -63,7 +63,7 @@ def load_trees_from_file(cls, file_name: str, format: str = "newick") -> dict:
         with open(file_name, "r") as f:
             str_trees = f.read()
         trees = cls.load_trees_from_json(str_trees)
-        return cls(trees)
+        return trees
 
     @classmethod
     def load_trees_from_json(cls, trees: str, format: str = "newick") -> dict:

aphylogeo/main.py

@@ -4,6 +4,8 @@
 from aphylogeo.params import Params
 from aphylogeo import utils
 from aphylogeo.genetic_trees import GeneticTrees
+import typer
+import os
 
 # from aphylogeo.utils import climaticPipeline, geneticPipeline, filterResults, loadSequenceFile
 
@@ -19,18 +21,42 @@
                               \/__/
 """  # https://patorjk.com/software/taag/#p=display&f=Larry%203D&t=aphylogeo%20
 
-if __name__ == "__main__":
-    print(titleCard + "\n")
+app = typer.Typer(invoke_without_command=True)
 
-    # geneticTrees = GeneticTrees.load_trees_from_file("./results/geneticTreesTest.json")
-    # loaded_seq_alignment = Alignment.load_from_json("./results/aligned_sequences.json")
+@app.command()
+def climate_pipeline(
+        file_name: str = typer.Option(Params.PARAMETER_KEYS["file_name"], help="The name of the file containing the climatic data."),
+        output: str = typer.Option("./datasets/example/climaticTrees.nwk", help="The name of the file to save the climatic trees."),
+    ):
+    """
+    This function is used to run the climatic pipeline that creates the climatic trees.
+    Args:
+        file_name (str): The name of the file containing the climatic data.
+    """
+    Params.load_from_file()
+
+    climatic_data = pd.read_csv(file_name)
+    climaticTrees = utils.climaticPipeline(climatic_data)
+
+    try:
+        utils.save_climatic_trees(climaticTrees, output)
+    except Exception as e:
+        print(f"Error saving the file: {e}")
 
-    # load params 
+@app.command()
+def genetic_pipeline(
+        reference_gene_filepath: str = typer.Option(os.path.join(Params.PARAMETER_KEYS["reference_gene_dir"], Params.PARAMETER_KEYS["reference_gene_file"]) , help="The path to the reference gene file."),
+        output: str = typer.Option("./datasets/example/geneticTrees.json", help="The name of the file to save the genetic trees."),
+    ):
+    """
+    This function is used to run the genetic pipeline that creates the genetic trees.
+    Args:
+        reference_gene_filepath (str): The path to the reference gene file.
+    """
     Params.load_from_file()
-    sequenceFile = utils.loadSequenceFile(Params.reference_gene_filepath)
-    align_sequence = AlignSequences(sequenceFile)
 
-    climatic_data = pd.read_csv(Params.file_name)
+    sequenceFile = utils.loadSequenceFile(reference_gene_filepath)
+    align_sequence = AlignSequences(sequenceFile)
 
     print("\nStarting alignement")
     start_time = time.time()
@@ -41,9 +67,62 @@
     elapsed_time = round(end_time - start_time, 3)
     print(f"Elapsed time: {elapsed_time} seconds")
 
-    climaticTrees = utils.climaticPipeline(climatic_data)
+    try:
+        trees.save_trees_to_json(output)
+    except Exception as e:
+        print(f"Error saving the file: {e}")
+
+@app.callback()
+def main(
+    climatic_tree: str = typer.Option(None, help="The name of the file containing the climatic trees."),
+    genetic_tree: str = typer.Option(None, help="The name of the file containing the genetic trees."),
+):
+
+    # geneticTrees = GeneticTrees.load_trees_from_file("./results/geneticTreesTest.json")
+    # loaded_seq_alignment = Alignment.load_from_json("./results/aligned_sequences.json")
+
+    # load params
+    Params.load_from_file()
+
+    # genetic pipeline
+    alignements = None
+
+    if genetic_tree is not None and os.path.exists(genetic_tree):
+        geneticTrees = GeneticTrees.load_trees_from_file(genetic_tree)
+        geneticTrees = geneticTrees.trees
+        trees = GeneticTrees(trees_dict=geneticTrees, format="newick")
+    else:
+
+        typer.echo(typer.style(titleCard, fg=typer.colors.GREEN))
+
+        sequenceFile = utils.loadSequenceFile(Params.reference_gene_filepath)
+        align_sequence = AlignSequences(sequenceFile)
+
+        print("\nStarting alignement")
+        start_time = time.time()
+        alignements = align_sequence.align()
+        geneticTrees = utils.geneticPipeline(alignements.msa)
+        trees = GeneticTrees(trees_dict=geneticTrees, format="newick")
+        end_time = time.time()
+        elapsed_time = round(end_time - start_time, 3)
+        print(f"Elapsed time: {elapsed_time} seconds")
+
+    # climatic sequence
+    if climatic_tree is not None and os.path.exists(climatic_tree):
+        climaticTrees = utils.load_climatic_trees(climatic_tree)
+        climatic_data = utils.reverse_climatic_pipeline(climaticTrees)
+    else:
+        climatic_data = pd.read_csv(Params.file_name)
+        climaticTrees = utils.climaticPipeline(climatic_data)
+
     utils.filterResults(climaticTrees, geneticTrees, climatic_data)
 
     # save results
-    alignements.save_to_json(f"./results/aligned_{Params.reference_gene_file}.json")
+    if alignements is not None:
+        alignements.save_to_json(f"./results/aligned_{Params.reference_gene_file}.json")
+
     trees.save_trees_to_json("./results/geneticTrees.json")
+
+
+if __name__ == "__main__":
+    app()

aphylogeo/utils.py

@@ -200,6 +200,20 @@ def climaticPipeline(df):
         trees[names[i]] = createTree(dm)
     return trees
 
+def reverse_climatic_pipeline(trees):
+    """
+    Converts a dictionary of climatic trees back into a DataFrame.
+
+    :param trees: The climatic trees dictionary.
+    :return: The DataFrame containing the climatic data.
+    :rtype: pd.DataFrame
+    """
+    data = []
+    for key, tree in trees.items():
+        for leaf in tree.get_terminals():
+            data.append([leaf.name, key, leaf.branch_length])
+    df = pd.DataFrame(data, columns=[Params.names[0], 'Variable', 'Value'])
+    return df
 
 def createBoostrap(msaSet: dict, bootstrapAmount):
     """
@@ -665,3 +679,35 @@ def loadSequenceFile(file):
         for sequence in SeqIO.parse(sequencesFile, "fasta"):
             sequences[sequence.id] = sequence.seq
     return sequences
+
+def save_climatic_trees(climatic_trees, file_path):
+    """
+    Save the climatic trees to a file.
+
+    :param dict climatic_trees: The climatic trees to save.
+    :param str file_path: The path to the file where the climatic trees will be saved.
+    """
+
+    with open(file_path, 'w') as file:
+        for key, tree in climatic_trees.items():
+            file.write(f">{key}\n")
+            Phylo.write(tree, file, "newick")
+
+
+def load_climatic_trees(file_path):
+    """
+    Load climatic trees from a file.
+
+    :param str file_path: Path to the target file where the climatic trees will be loaded
+    :returns: The loaded climatic trees
+    :rtype: dict
+    """
+    climatic_trees = {}
+
+    with open(file_path, 'r') as file:
+        lines = file.readlines()
+        for i in range(0, len(lines), 2):
+            key = lines[i].strip()[1:]
+            tree = Phylo.read(StringIO(lines[i + 1]), "newick")
+            climatic_trees[key] = tree
+    return climatic_trees

pyproject.toml

@@ -22,6 +22,7 @@ robinson-foulds = "^1.2"
 ete3 = "^3.1.3"
 dendropy = "^4.6.1"
 textdistance = "^4.6.0"
+typer = "^0.13.1"
 
 
 [tool.poetry.group.dev.dependencies]

setup.py

@@ -20,6 +20,7 @@
         "ete3==3.1.3",
         "dendropy==4.6.1",
         "textdistance==4.6.0",
+        "typer==0.13.1"
     ],
     python_requires=">=3.9.0",
     packages=find_packages(),