<jupyter_start><jupyter_text>Setup spark<jupyter_code>def spark_start(local_dir):
    from pyspark.sql import SparkSession
    
    spark = (
            SparkSession.builder
            .appName("LSD2")
            .config("spark.sql.warehouse.dir", local_dir)
            #.config('spark.master', "local[6]")
            .config("spark.master", "local[32]") # yolo mode
            .config('spark.driver.memory', '8G') # 128
            .config('spark.local.dir', local_dir)
            .config('spark.memory.offHeap.enabled', 'true')
            .config('spark.memory.offHeap.size', '4G') # 256
            .config("spark.sql.execution.arrow.enabled", "true")
            .config("spark.driver.maxResultSize", "6G")
            .config("spark.driver.extraJavaOptions", f"-Dderby.system.home={local_dir}")
            .enableHiveSupport()
            .getOrCreate()
                    )   

    return spark

spark_session = spark_start("/epyc/users/kyboone/spark-tmp/")

catalog = axs.AxsCatalog(spark_session)<jupyter_output><empty_output><jupyter_text>Load ZTF data<jupyter_code>ztf = catalog.load('ztf_oct19')<jupyter_output><empty_output><jupyter_text>Load the saved axs table<jupyter_code>wtf = catalog.load('wtf_full_oct19_4')
wtf.count()
# Recalculate the scores
rescored_wtf = (
    wtf.select(
        '*',
        dipper.detect_dippers_udf(
            wtf['mjd_g'],
            wtf['mag_g'],
            wtf['magerr_g'],
            wtf['xpos_g'],
            wtf['ypos_g'],
            wtf['catflags_g'],
            wtf['mjd_r'],
            wtf['mag_r'],
            wtf['magerr_r'],
            wtf['xpos_r'],
            wtf['ypos_r'],
            wtf['catflags_r']
        ).alias('rescored_dipper'),
    )
)
wtf.where(sparkfunc.col("dipper.significance") > 20.).count()
ztf.count()
%%time

# Get the best entries
res = rescored_wtf.sort(rescored_wtf['rescored_dipper.intmag'].desc()).head(1000)
%matplotlib inline

from ipywidgets import interact, IntSlider

def interact_lightcurve(idx, zoom=False):
    show_lightcurve(res[idx], zoom=zoom)

interact(interact_lightcurve, idx=IntSlider(0, 0, len(res) - 1))
for idx in range(10):
    show_lightcurve(res[idx], verbose=False)<jupyter_output><empty_output><jupyter_text>Label how the dip detection works<jupyter_code>detect_dippers_row(res[0])
band_colors = {
    'g': 'tab:green',
    'r': 'tab:red',
    'i': 'tab:purple'
}

def label_dip(row):
    plt.figure(figsize=(8, 6), dpi=100)

    for band in ['g', 'r', 'i']:
        mjd, mag, magerr = parse_observations(
            row[f'mjd_{band}'],
            row[f'mag_{band}'],
            row[f'magerr_{band}'],
            row[f'xpos_{band}'],
            row[f'ypos_{band}'],
            row[f'catflags_{band}'],
        )
        
        plt.errorbar(mjd, mag, magerr, fmt='o', c=band_colors[band], label=f'ZTF-{band}')
        
    

    plt.xlabel('MJD')
    if parsed:
        plt.ylabel('Magnitude + offset')
    else:
        plt.ylabel('Magnitude')
    plt.legend()
    plt.title('objid %d' % row['objid'])
    plt.gca().invert_yaxis()<jupyter_output><empty_output><jupyter_text>Joining<jupyter_code>gaia = catalog.load('gaia_dr2_1am_dup')
wtf_df = rescored_wtf.crossmatch(gaia.select('ra', 'dec', 'zone', 'dup', 'parallax', 'parallax_over_error', 'phot_g_mean_mag',
                                'bp_rp')).toPandas()
plt.figure()
abs_mag = wtf_df['phot_g_mean_mag'] - 5 * np.log10(1000 / wtf_df['parallax']) + 5

cut = (wtf_df['parallax_over_error'] > 5)
plt.scatter(wtf_df['bp_rp'][cut], abs_mag[cut], s=1, alpha=0.1)

cut = (wtf_df['parallax_over_error'] > 5) & (wtf_df['new_score'] > 2.5)
plt.scatter(wtf_df['bp_rp'][cut], abs_mag[cut], s=10)


plt.gca().invert_yaxis()
def print_links(row):
    print("http://simbad.u-strasbg.fr/simbad/sim-coo?Coord=%.6f%+.6f&CooFrame=FK5&CooEpoch=2000&CooEqui=2000&CooDefinedFrames=none&Radius=20&Radius.unit=arcsec&submit=submit+query&CoordList=" % (row['ra'], row['dec']))
    print("RA+Dec: %.6f%+.6f" % (row['ra'], row['dec']))
    print("RA:     %.6f" % row['ra'])
    print("Dec:    %.6f" % row['dec'])

def show_lightcurve(idx):
    row = wtf_df.iloc[idx]  
    #print_links(row)    
    plot_lightcurve(row)
    plt.title(idx)

    print("Score:  %.3f" % detect_dippers_row(row))
a = np.where(cut & (df['bp_rp'] > 0.5) & (df['bp_rp'] < 2) & (abs_mag < 1000) & (abs_mag > 5.))
for i in a[0]:
    show_lightcurve(i)<jupyter_output>Score:  2.566
Score:  2.795
Score:  2.650
Score:  2.909
Score:  2.530
Score:  2.687
Score:  2.845
Score:  2.584
Score:  2.648
Score:  3.264
Score:  3.082<jupyter_text>Run the spark query  Run and save the query<jupyter_code>%%time

# Run on spark
res = (
    ztf
    .exclude_duplicates()
    #.region(ra1=295, ra2=296, dec1=20, dec2=21)
    .where(
        (sparkfunc.col("nobs_g") >= 10)
        | (sparkfunc.col("nobs_r") >= 10)
        | (sparkfunc.col("nobs_i") >= 10)
    )
    .select(
        '*',
        detect_dippers_udf(
            ztf['mjd_g'],
            ztf['mag_g'],
            ztf['magerr_g'],
            ztf['xpos_g'],
            ztf['ypos_g'],
            ztf['catflags_g'],
            ztf['mjd_r'],
            ztf['mag_r'],
            ztf['magerr_r'],
            ztf['xpos_r'],
            ztf['ypos_r'],
            ztf['catflags_r']
        ).alias('dipper'),
    )
    .where(
        (sparkfunc.col("dipper.significance") > 10.)
    )
    #.write.parquet('./query_test_23.parquet')
    .write.parquet('./query_full_4.parquet')
)<jupyter_output><empty_output><jupyter_text>Convert to an axs table<jupyter_code>wtf = spark_session.read.parquet('./query_full_4.parquet')
%matplotlib inline

from ipywidgets import interact, IntSlider

def interact_lightcurve(idx, zoom=False):
    show_lightcurve(res[idx], zoom=zoom)

interact(interact_lightcurve, idx=IntSlider(0, 0, len(res) - 1))
catalog.save_axs_table(wtf, 'wtf_full_oct19_4', repartition=True)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.1725185685347738, 'fraction_numerical': 0.02650236326806212, 'mean_word_length': 3.0142276422764227, 'pattern_counts': {'":': 0, '<': 30, '<?xml version=': 0, '>': 38, '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}}
{'chain_length': '7', 'dir': 'jupyter-structured-clean-dedup', 'id': '573974', 'license': 'MIT', 'n_tokens_mistral': 2431, 'n_tokens_neox': 2286, 'n_words': 319, 'path': 'old/Dipper detection integrated.ipynb', 'repo_name': 'dirac-institute/ZTF_Boyajian', 'size': '5924'}