<SYSTEM_TASK:>
Solve the following problem using Python, implementing the functions described below, one line at a time
<END_TASK>
<USER_TASK:>
Description:
 def _gatk_base_recalibrator(broad_runner, dup_align_bam, ref_file, platform, dbsnp_file, intervals, data):
 """Step 1 of GATK recalibration process, producing table of covariates. For GATK 4 we use local multicore spark runs: https://github.com/broadinstitute/gatk/issues/2345 For GATK3, Large whole genome BAM files take an excessively long time to recalibrate and the extra inputs don't help much beyond a certain point. See the 'Downsampling analysis' plots in the GATK documentation: http://gatkforums.broadinstitute.org/discussion/44/base-quality-score-recalibrator#latest This identifies large files and calculates the fraction to downsample to. spark host and timeout settings help deal with runs on restricted systems where we encounter network and timeout errors """

    target_counts = 1e8  # 100 million reads per read group, 20x the plotted max
    out_file = os.path.join(dd.get_work_dir(data), "align", dd.get_sample_name(data),
                            "%s-recal.grp" % utils.splitext_plus(os.path.basename(dup_align_bam))[0])
    if not utils.file_exists(out_file):
        if has_aligned_reads(dup_align_bam, intervals):
            with file_transaction(data, out_file) as tx_out_file:
                gatk_type = broad_runner.gatk_type()
                assert gatk_type in ["restricted", "gatk4"], \
                    "Require full version of GATK 2.4+ or GATK4 for BQSR"
                params = ["-I", dup_align_bam]
                cores = dd.get_num_cores(data)
                if gatk_type == "gatk4":
                    resources = config_utils.get_resources("gatk-spark", data["config"])
                    spark_opts = [str(x) for x in resources.get("options", [])]
                    params += ["-T", "BaseRecalibratorSpark",
                               "--output", tx_out_file, "--reference", dd.get_ref_file(data)]
                    if spark_opts:
                        params += spark_opts
                    else:
                        params += ["--spark-master", "local[%s]" % cores,
                                   "--conf", "spark.driver.host=localhost", "--conf", "spark.network.timeout=800",
                                   "--conf", "spark.executor.heartbeatInterval=100",
                                   "--conf", "spark.local.dir=%s" % os.path.dirname(tx_out_file)]
                    if dbsnp_file:
                        params += ["--known-sites", dbsnp_file]
                    if intervals:
                        params += ["-L", intervals, "--interval-set-rule", "INTERSECTION"]
                else:
                    params += ["-T", "BaseRecalibrator",
                                "-o", tx_out_file, "-R", ref_file]
                    downsample_pct = bam.get_downsample_pct(dup_align_bam, target_counts, data)
                    if downsample_pct:
                        params += ["--downsample_to_fraction", str(downsample_pct),
                                   "--downsampling_type", "ALL_READS"]
                    if platform.lower() == "solid":
                        params += ["--solid_nocall_strategy", "PURGE_READ",
                                   "--solid_recal_mode", "SET_Q_ZERO_BASE_N"]
                    if dbsnp_file:
                        params += ["--knownSites", dbsnp_file]
                    if intervals:
                        params += ["-L", intervals, "--interval_set_rule", "INTERSECTION"]
                memscale = {"magnitude": 0.9 * cores, "direction": "increase"} if cores > 1 else None
                broad_runner.run_gatk(params, os.path.dirname(tx_out_file), memscale=memscale,
                                      parallel_gc=True)
        else:
            with open(out_file, "w") as out_handle:
                out_handle.write("# No aligned reads")
    return out_file