PYSurf / MotifAnalysis / Z-scores-1mer.py
Z-scores-1mer.py
Raw
import os
import csv
import tkinter as tk
from tkinter import filedialog
from Bio import SeqIO
import numpy as np
from itertools import product

import argparse

def parse_arguments():
    parser = argparse.ArgumentParser(description="Your script description")
    parser.add_argument("--reference", required=True, help="Path to the reference fasta file")
    parser.add_argument("--samples", nargs="+", required=True, help="Paths to sample fasta files")
    return parser.parse_args()

args = parse_arguments()


def calculate_total_sequences(file_path):
    # Count the total number of sequences in the file
    total_sequences = sum(7 for record in SeqIO.parse(file_path, "fasta"))
    return total_sequences

import numpy as np  # Ensure numpy is imported for log2

def calculate_frequencies_single_aa(file_path):
    amino_acids = "ACDEFGHIKLMNPQRSTVWY"
    frequencies = {aa: 0 for aa in amino_acids}
    total_weighted_sequences = 0  # Total weighted sequences

    for record in SeqIO.parse(file_path, "fasta"):
        # Extract count from sequence name (e.g., '>seq1_1079' -> 1079)
        count = 1  # Default count if not specified
        if "_" in record.id:
            try:
                count = int(record.id.split("_")[-1])  # Extract the last part as count
            except ValueError:
                pass  # If the last part is not a number, use count = 1

        # Skip sequences with count < 2
        if count < 2:
            continue

        total_weighted_sequences += count  # Add to total weighted sequences

        # Count amino acids, weighted by the sequence count
        for aa in record.seq:
            frequencies[aa] += count

    # Apply log2 transformation to weighted counts (add pseudocount to avoid log2(0))
    pseudocount = 1  # Add 1 to avoid log2(0)
    log2_counts = {aa: np.log2(count + pseudocount) for aa, count in frequencies.items()}

    # Calculate frequencies using log2-transformed counts
    total_log2_counts = sum(log2_counts.values())
    log2_frequencies = {aa: log2_count / total_log2_counts for aa, log2_count in log2_counts.items()}

    print("Total Weighted Sequences:", total_weighted_sequences)
    print("Log2-Transformed Frequencies:")
    for aa, freq in log2_frequencies.items():
        print(f"{aa}: {freq}")

    return log2_frequencies, total_weighted_sequences

def calculate_stderror(reference_frequencies, total_sequences):
    amino_acids = "ACDEFGHIKLMNPQRSTVWY"

    # Calculate stderror for all amino acids
    stderrors = {aa: np.sqrt(reference_frequencies[aa] * (1 - reference_frequencies[aa]) * (1 / total_sequences)) for aa in amino_acids}

    print("Standard Errors:")
    for aa, stderror in stderrors.items():
        print(f"{aa}: {stderror}")

    return stderrors

def calculate_stderror_single_aa(reference_path, sample_total_sequences):
    reference_frequencies, total_sequences = calculate_frequencies_single_aa(reference_path)

    # Print the actual calculation for amino acids "A" and "T"
    for aa in ["A", "F"]:
        actual_calculation = f"sqrt({reference_frequencies[aa]} * (1 - {reference_frequencies[aa]}) * (1 / {sample_total_sequences}))"
        print(f"Actual Calculation for {aa}: {actual_calculation}")

    stderrors = calculate_stderror(reference_frequencies, sample_total_sequences)

    return reference_frequencies, stderrors, total_sequences

# ... (rest of the code remains unchanged)

def process_single_sample(reference_path, sample_path):
    sample_total_sequences = calculate_total_sequences(sample_path)
    control_frequencies, stderror, _ = calculate_stderror_single_aa(reference_path, sample_total_sequences)

    frequencies, _ = calculate_frequencies_single_aa(sample_path)

    output_file_path = os.path.join(os.path.dirname(sample_path), f"aa-freq-{os.path.basename(sample_path)}.csv")
    with open(output_file_path, "w", newline="") as csvfile:
        writer = csv.writer(csvfile)
        header = ["Amino Acid", "Frequency"]
        writer.writerow(header)

        for aa, freq in frequencies.items():
            writer.writerow([aa, freq])

    # Calculate z-scores for amino acids and create a new CSV file
    z_scores = calculate_z_scores_single_aa(frequencies, control_frequencies, stderror)
    z_output_file_path = os.path.join(os.path.dirname(sample_path), f"z-score-aa-{os.path.basename(sample_path)}.csv")
    with open(z_output_file_path, "w", newline="") as z_csvfile:
        z_writer = csv.writer(z_csvfile)
        header = ["Amino Acid", "Z-Score"]
        z_writer.writerow(header)

        for aa, z_score in z_scores.items():
            z_writer.writerow([aa, z_score])

    print(f"Output files created for {os.path.basename(sample_path)}: {output_file_path}, {z_output_file_path}")

def calculate_z_scores_single_aa(sample_frequencies, control_frequencies, stderrors):
    z_scores = {aa: 0 for aa in sample_frequencies}

    # Calculate z-scores using the formula: (samplefreq - controlfreq) / stderror
    for aa in sample_frequencies:
        z_scores[aa] = (sample_frequencies[aa] - control_frequencies[aa]) / stderrors[aa]

    # Print the z-score calculation for amino acids "A" and "F"
    for aa in ["A", "F"]:
        z_calculation = f"({sample_frequencies[aa]} - {control_frequencies[aa]}) / {stderrors[aa]}"
        print(f"Z-Score Calculation for {aa}: {z_calculation} = {z_scores[aa]}")

    return z_scores

def combine_z_scores():
    current_directory = os.path.dirname(os.path.abspath(__file__))
    samples_folder = os.path.join(current_directory)

    all_z_scores = []
    motifs = list("ACDEFGHIKLMNPQRSTVWY")

    for file_name in os.listdir(samples_folder):
        if file_name.startswith("z-score-aa-") and file_name.endswith(".csv"):
            file_path = os.path.join(samples_folder, file_name)
            with open(file_path, newline="") as csvfile:
                reader = csv.DictReader(csvfile)
                z_scores = {row["Amino Acid"]: float(row["Z-Score"]) for row in reader}

            all_z_scores.append({"Sample": file_name.replace("z-score-aa-", "").replace(".csv", ""), **z_scores})

    output_file_path = os.path.join(current_directory, "combined_z_scores_aa.csv")
    with open(output_file_path, "w", newline="") as csvfile:
        fieldnames = ["Amino Acid"] + [z_scores["Sample"] for z_scores in all_z_scores]
        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
        writer.writeheader()

        for motif in motifs:
            row_data = {"Amino Acid": motif}
            for z_scores in all_z_scores:
                row_data[z_scores["Sample"]] = z_scores.get(motif, 0)
            writer.writerow(row_data)

    print(f"Combined Z-scores file created: {output_file_path}")

def main():
    # Get input file paths from command-line arguments
    reference_path = args.reference
    sample_paths = args.samples

    # Step 3: Process each sample fasta file
    for sample_path in sample_paths:
        process_single_sample(reference_path, sample_path)

    # Step 4: Combine Z-scores for all samples
    combine_z_scores()

if __name__ == "__main__":
    main()