<commit_before>class Parsing
  include Sidekiq::Worker
  sidekiq_options :queue => :parse, :retry => 5, :unique => true

  attr_reader :genotype, :temp_table_name, :tempfile, :stats, :start_time

  def perform(genotype_id)
    @stats = {}
    @start_time = Time.current
    logger.info("Started parsing genotype with id #{genotype_id}")
    @genotype = Genotype.find(genotype_id)
    stats[:filetype] = genotype.filetype
    stats[:genotype_id] = genotype.id
    @temp_table_name = "user_snps_temp_#{genotype.id}"
    @tempfile = Tempfile.new("snpr_genotype_#{genotype.id}_")

    create_temp_table
    normalize_csv
    copy_csv_into_temp_table
    insert_into_snps
    insert_into_user_snps
    notify_user

    logger.info("Finished parsing genotype with id #{genotype.id}, cleaning up.")
  rescue => e
    logger.error("Failed with #{e.class}: #{e.message}")
    raise
  ensure
    drop_temp_table
    # TODO: Why doesn't `tempfile.unlink` work here?
    File.delete(tempfile.path)
    stats[:duration] = "#{(Time.current - start_time).round(3)}s"
    logger.info("Stats: #{stats.to_a.map { |s| s.join('=') }.join(', ')}")
  end

  def create_temp_table
    execute("drop table if exists #{temp_table_name}")
    execute(<<-SQL)
      create table #{temp_table_name} (
        snp_name varchar(32),
        chromosome varchar(32),
        position varchar(32),
        local_genotype char(2)
      )
    SQL
  end

  def drop_temp_table
    execute("drop table #{temp_table_name}")
  end

  def normalize_csv
    rows = File.readlines(genotype.genotype.path)
      .reject { |line| line.start_with?('#') } # Skip comments
    stats[:rows_without_comments] = rows.length
    csv = send(:"parse_#{genotype.filetype.sub('-', '_').downcase}", rows)
    known_chromosomes = ['MT', 'X', 'Y', (1..22).map(&:to_s)].flatten
    csv.select! do |row|
      # snp name
      row[0].present? &&
      # chromosome
      known_chromosomes.include?(row[1]) &&
      # position
      row[2].to_i >= 1 && row[2].to_i <= 249_250_621 &&
      # local genotype
      row[3].is_a?(String) && (1..2).include?(row[3].length)
    end
    stats[:rows_after_parsing] = csv.length
    tempfile.write(csv.map { |row| row.join(',') }.join("\n"))
    tempfile.close
    FileUtils.chmod(0644, tempfile.path)
  end

  def copy_csv_into_temp_table
    execute(<<-SQL)
      copy #{temp_table_name} (
        snp_name,
        chromosome,
        position,
        local_genotype
      )
      from '#{tempfile.path}'
      with (FORMAT CSV, HEADER FALSE, DELIMITER ',')
    SQL
  end

  def insert_into_snps
    time = Time.now.utc.iso8601

    snps = execute(<<-SQL)
      select
        #{temp_table_name}.snp_name as name,
        #{temp_table_name}.chromosome,
        #{temp_table_name}.position,
        1 as user_snps_count
      from #{temp_table_name}
      left join snps on #{temp_table_name}.snp_name = snps.name
      where snps.name is null
    SQL
    Snp.create!(snps.to_a)
  end

  def insert_into_user_snps
    known_user_snps = UserSnp.select(:snp_name).where(genotype_id: genotype.id)
    execute(<<-SQL)
      insert into user_snps (snp_name, local_genotype, genotype_id)
      (
        select
          #{temp_table_name}.snp_name,
          #{temp_table_name}.local_genotype,
          #{genotype.id} as genotype_id
        from #{temp_table_name}
        where #{temp_table_name}.snp_name not in (#{known_user_snps.to_sql})
      )
    SQL
  end

  def parse_23andme(rows)
    rows.map do |row|
      fields = row.strip.split("\t")
      [
        fields[0],
        fields[1],
        fields[2],
        fields[3].rstrip
      ]
    end
  end

  def parse_decodeme(rows)
    rows.shift if rows.first.start_with?('Name')
    rows.map do |row|
      fields = row.strip.split(',')
      [
        fields[0],
        fields[2],
        fields[3],
        fields[5]
      ]
    end
  end

  def parse_ancestry(rows)
    rows.shift if rows.first.start_with?('rsid')
    rows.map do |row|
      fields = row.strip.split("\t")
      [
        fields[0],
        fields[1],
        fields[2],
        "#{fields[3]}#{fields[4]}"
      ]
    end
  end

  def parse_ftdna_illumina(rows)<commit_msg>90% of 23andme-exome parsing done<commit_after>class Parsing
  include Sidekiq::Worker
  sidekiq_options :queue => :parse, :retry => 5, :unique => true

  attr_reader :genotype, :temp_table_name, :tempfile, :stats, :start_time

  def perform(genotype_id)
    @stats = {}
    @start_time = Time.current
    logger.info("Started parsing genotype with id #{genotype_id}")
    @genotype = Genotype.find(genotype_id)
    stats[:filetype] = genotype.filetype
    stats[:genotype_id] = genotype.id
    @temp_table_name = "user_snps_temp_#{genotype.id}"
    @tempfile = Tempfile.new("snpr_genotype_#{genotype.id}_")

    create_temp_table
    normalize_csv
    copy_csv_into_temp_table
    insert_into_snps
    insert_into_user_snps
    notify_user

    logger.info("Finished parsing genotype with id #{genotype.id}, cleaning up.")
  rescue => e
    logger.error("Failed with #{e.class}: #{e.message}")
    raise
  ensure
    drop_temp_table
    # TODO: Why doesn't `tempfile.unlink` work here?
    File.delete(tempfile.path)
    stats[:duration] = "#{(Time.current - start_time).round(3)}s"
    logger.info("Stats: #{stats.to_a.map { |s| s.join('=') }.join(', ')}")
  end

  def create_temp_table
    execute("drop table if exists #{temp_table_name}")
    execute(<<-SQL)
      create table #{temp_table_name} (
        snp_name varchar(32),
        chromosome varchar(32),
        position varchar(32),
        local_genotype char(2)
      )
    SQL
  end

  def drop_temp_table
    execute("drop table #{temp_table_name}")
  end

  def normalize_csv
    rows = File.readlines(genotype.genotype.path)
      .reject { |line| line.start_with?('#') } # Skip comments
    stats[:rows_without_comments] = rows.length
    csv = send(:"parse_#{genotype.filetype.gsub('-', '_').downcase}", rows)
    logger.info(csv)
    known_chromosomes = ['MT', 'X', 'Y', (1..22).map(&:to_s)].flatten
    csv.select! do |row|
      # snp name
      row[0].present? &&
      # chromosome
      known_chromosomes.include?(row[1]) &&
      # position
      row[2].to_i >= 1 && row[2].to_i <= 249_250_621 &&
      # local genotype
      row[3].is_a?(String) && (1..2).include?(row[3].length)
    end
    stats[:rows_after_parsing] = csv.length
    tempfile.write(csv.map { |row| row.join(',') }.join("\n"))
    tempfile.close
    FileUtils.chmod(0644, tempfile.path)
  end

  def copy_csv_into_temp_table
    execute(<<-SQL)
      copy #{temp_table_name} (
        snp_name,
        chromosome,
        position,
        local_genotype
      )
      from '#{tempfile.path}'
      with (FORMAT CSV, HEADER FALSE, DELIMITER ',')
    SQL
  end

  def insert_into_snps
    time = Time.now.utc.iso8601

    snps = execute(<<-SQL)
      select
        #{temp_table_name}.snp_name as name,
        #{temp_table_name}.chromosome,
        #{temp_table_name}.position,
        1 as user_snps_count
      from #{temp_table_name}
      left join snps on #{temp_table_name}.snp_name = snps.name
      where snps.name is null
    SQL
    Snp.create!(snps.to_a)
  end

  def insert_into_user_snps
    known_user_snps = UserSnp.select(:snp_name).where(genotype_id: genotype.id)
    execute(<<-SQL)
      insert into user_snps (snp_name, local_genotype, genotype_id)
      (
        select
          #{temp_table_name}.snp_name,
          #{temp_table_name}.local_genotype,
          #{genotype.id} as genotype_id
        from #{temp_table_name}
        where #{temp_table_name}.snp_name not in (#{known_user_snps.to_sql})
      )
    SQL
  end

  def parse_23andme(rows)
    rows.map do |row|
      fields = row.strip.split("\t")
      [
        fields[0],
        fields[1],
        fields[2],
        fields[3].rstrip
      ]
    end
  end

  def parse_23andme_exome_vcf(rows)
    # Rules:
    # Skip lines with IndelType in them
    # Skip lines were SNP name is '.', these are non-standard SNPs
    rows.map do |row|
      next if row.include? 'IndelType'
      fields = row.strip.split("\t")
      next if fields[2] == '.'
      major_allele = fields[3] # C
      minor_allele = fields[4] # A
      trans_dict = {"0" => major_allele, "1" => minor_allele}
      names = fields[-1].split(":")[0].split("/") # ["0", "1"], meaning A/C
      alleles = names.map{ |a| trans_dict[a]}.sort.join # becomes AC
      [
        fields[2],
        fields[1],
        fields[0],
        alleles
      ]
    end.compact # because the above next introduces nil.
    # Slower alternative is to use reject first, but then we'll iterate > 2 times
  end

  def parse_decodeme(rows)
    rows.shift if rows.first.start_with?('Name')
    rows.map do |row|
      fields = row.strip.split(',')
      [
        fields[0],
        fields[2],
        fields[3],
        fields[5]
      ]
    end
  end

  def parse_ancestry(rows)
    rows.shift if rows.first.start_with?('rsid')
    rows.map do |row|
      fields = row.strip.split("\t")
      [
        fields[0],
        fields[1],
        fields[2],
        "#{fields[3]}#{fields[4]}"
      ]
    end
  end

  def parse_ftdna_illumina(rows)
{'fraction_non_alphanumeric': 0.13568712656165127, 'fraction_numerical': 0.013796849538294405, 'mean_word_length': 2.6286953094205754, 'pattern_counts': {'":': 1, '<': 21, '<?xml version=': 0, '>': 16, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4189154', 'n_tokens_mistral': 3340, 'n_tokens_neox': 3235, 'n_words': 633}