<jupyter_start><jupyter_text>ANTARES Filter Development Kit_<NAME>, <NAME>, ANTARES Team.__Many thanks to <NAME>, <NAME>, <NAME>, <NAME>, <NAME>._ SummaryThis Notebook demonstrates how to write filters for [ANTARES](http://antares.noao.edu) and test them against a sample of real data from [ZTF](http://ztf.caltech.edu/).This Notebook is intended to be used in NOAO DataLab's Jupyter environment. There, you will have access to ANTARES test data. If you're not running in DataLab, [sign up for DataLab](https://datalab.noao.edu), then [log in to the notebook server](https://datalab.noao.edu/devbooks).For new Data Lab accounts, this notebook will be automatically included in your `notebooks/` directory. Otherwise, you can save this `.ipynb` notebook file locally, and then upload it to your Data Lab Jupyter notebook server (use the 'Upload' button in the upper right corner).In Data Lab, you MUST use the Kernel version "Python 3 (ANTARES)". GoalsTo demonstrate:1. How to write filters using the ANTARES filter API.1. How to test filters against a small test dataset.Note: As of this writing, the test dataset is limited. It is intended to represent the format of ZTF data in ANTARES' format and API. It is not intended to represent the variety of data that is available, or to be suitable for training machine learning systems. Table of Contents* [0. Background information on ANTARES](background)* [1. Connect to test database](connect)* [2. Write a Filter](write)* [3. Test a Filter](test) * [3.1 Test against an Alert from test dataset](test-one) * [3.2 Test against an Alert from the dataset using catalog information](test-two) * [3.3 Test against an Alert from the dataset using catalog and alert information](test-three) * [3.4 Test against an user-created alert using catalog and alert information](test-four)* [4. Submit Filter to ANTARES](submit) 0. Background information on ANTARESANTARES receives alerts from surveys in real-time and sends them through a processing pipeline. The pipeline contains the following stages:1. Associate the Alert with the nearest point of known past measurements within a 1" radius. We call this a Locus.2. Discard Alerts with a high probability of being false detections.3. Discard Alerts with poor image quality.4. Look up associated objects in our catalogs.5. If the Alert's Locus has two or more measurements on it, execute the Filters.The filters are python functions which take a LocusData object as a single parameter. Functions on the LocusData provide access to the Alert's properties, the data from past Alerts on the Locus, and the associated catalog objects. The LocusData also provides functions to set new properties on the Alert, and to send it to output streams. 1. Initalize the dev kitThis will configure the `antares` package to connect to the test database<jupyter_code># Imports
import antares.dev_kit as dk
dk.init()
# You should see a happy message that says that "ANTARES DevKit is ready!"<jupyter_output><empty_output><jupyter_text>2. Write a FilterThe filter `example_filter` below does nothing of scientific interest, but it demonstrates the most basic use of the filter API.Further down, there are other example filters -  `high_snr`, `extragalactic`, etc. are examples of our current science filters. Each demonstrates one new feature of the API that you might want to build into your own super awesome science filter.The Filter API consists of the LocusData object, which is passed to the Filter as the single parameter. The `example_filter` shows examples of how to use the LocusData.<jupyter_code>def example_filter(locus_data):
    """
    A test Filter for demonstration.
    """
    print('`example_filter` is running...')

    # Get a dict of all properties on the new alert.
    print('locus_data.get_properties()')
    print('-->')
    print(locus_data.get_properties())
    print("\nYikes! You can see there are lots of properties!")

    # Any properties from the ZTF Alert are prefixed with 'ztf_'.
    # See here for ZTF's documentation of their properties:
    # https://github.com/ZwickyTransientFacility/ztf-avro-alert/blob/master/schema/candidate.avsc

    # Get a numpy array of values for particular properties.
    # Rows for 'alert_id' and 'mjd' are always included at the top of the array.
    # For example, in the following examples, the rows of the array will be:
    # - alert_id
    # - mjd
    # - ra
    # - dec
    # - ztf_fid (1=g, 2=r, 3=i)
    # - ztf_magpsf
    print("Most times, it's easier to work with the light curve which you can do with the get_time_series function")
    print("locus_data.get_time_series('ra', 'dec', 'ztf_fid', 'ztf_magpsf')")
    print('-->')
    print(locus_data.get_time_series('ra', 'dec', 'ztf_fid', 'ztf_magpsf'))
    print("\nBy default, the function returns alerts in the filter/passband that match the current alert's passband.")
    print("You can request a specific filter instead.\n")
    # In the following example, we specify only to include columns where ztf_fid == 2.
    print("locus_data.get_time_series('ra', 'dec', 'ztf_fid', 'ztf_magpsf', filters={'ztf_fid': 2})")
    print('-->')
    print(locus_data.get_time_series('ra', 'dec', 'ztf_fid', 'ztf_magpsf', filters={'ztf_fid': 2}))

    # Send the Alert to an output stream.
    # The name of your stream must be unique. We will check this before accepting your filter.
    # All streams are directed to Kafka output topics with the same name as the stream.
    # We can also configure your stream to send notifications to a channel in Slack.
    locus_data.send_to_stream('my_stream')

    print('`example_filter` is finished.')<jupyter_output><empty_output><jupyter_text>3. Test a filter 3.1 Test against an Alert from test datasetWe have placed a sample of the ANTARES database (sourced from ZTF) in a read-only database for testing.Here, we run the `example_filter` against a particular Alert and its measurement history. The `run_stage` function takes an Alert ID and a filter, and runs the filter by constructing a LocusData object identical to what would be generated in the ANTARES production system.<jupyter_code>alert_id = 153505

# Run the `example_filter`.
# `verbose=True` prints detailed logs.
result = dk.run_stage(alert_id, example_filter, verbose=True)

# `run_stage` returns a dict with a report of what happened:
print(list(result.keys()))<jupyter_output><empty_output><jupyter_text>You can look at what streams were created or populated by this alert.<jupyter_code>print(result['new_streams'])

# You can get the LocusData object too, so that you can work with it outside the stage
ld = result['locus_data']<jupyter_output><empty_output><jupyter_text>3.2 Test against an Alert from test dataset, using catalog informationLet's create a more complex filter and use the alert's catalog information to make a decision<jupyter_code>def extragalactic(ld):
    """
    Send alert to stream 'extragalactic' if it matches any extended source catalogs.
    """
    matching_catalog_names = ld.get_astro_object_matches().keys()

    # These are the catalogs (Antares-based names) with extended sources
    xsc_cats = ['2mass_xsc', 'ned', 'nyu_valueadded_gals', 'sdss_gals', 'veron_agn_qso']

    if set(matching_catalog_names) & set(xsc_cats):
        p = ld.get_properties()
        print(p['alert_id'], ' matches an extragalactic source')
        ld.send_to_stream('extragalactic')
results   = dk.run_many(extragalactic, n=50, verbose=False) # notice we're running 50 alerts at once here<jupyter_output><empty_output><jupyter_text>3.3 Test against an Alert from test dataset, using catalog information AND alert properties to make a decisionLet's create a more complex filter, and this time, also use the alert's properties together with catalog information to make a decision<jupyter_code>def high_snr(ld):
    """
    Send high-SNR non-variable star alerts to stream 'high_snr'.

    Should flag ~2-3% of alerts.
    """ 
    snr_thresholds = {
        1: 50.0,  # For filter ID 1 (g), the threshold is 50
        2: 55.0,  # For filter ID 2 (R), the threshold is 55
    }
    
    # we'll check that a star is in a variable star catalog
    bad_cats = set(['asassn_variable_catalog', 'asassn_variable_catalog_v2_20190802'])

    # here's an example of how to get the astro object information
    astro_objects = ld.get_astro_object_matches()
    cats = set(astro_objects.keys())
    n_cats = len(cats)

    p = ld.get_properties()  # get all Alert properties as a dict
    fid = p['ztf_fid']  # filter ID
     
    sigmapsf = p.get('ztf_sigmapsf', None)  # 1-sigma uncertainty in magnitude of PSF 
    if sigmapsf is None: # only available for detections
        return
    
    snr = 1.0 / sigmapsf  # compute SNR
    snr_threshold = snr_thresholds.get(fid, None)  # SNR threshold for this filter
    
    # note that the catalog check doesn't guarantee the alert isn't from a variable star
    # only that the object it is coming from isn't in a variable star catalog we have
    # of course, most of the bright variable stars that have any chance of exceeding the S/N threshold
    # are already in the catalogs, so this is reasonable
    # what we want is for this filter to trip on new flares 
    if snr_threshold is not None and snr > snr_threshold:
        if cats.isdisjoint(bad_cats):
            print("Alert ID: ", p['alert_id'], "    SNR: ",snr,"    n:",n_cats)
            if n_cats > 0:
                print(list(astro_objects.keys()))
            ld.send_to_stream('high_snr')
        else:
            print('Alert ID ',p['alert_id'],' matched with known variable star.')
            bad_objs = [astro_objects[bad_cat] for bad_cat in bad_cats]
            for bad_obj in bad_objs:
                print(bad_obj)
        print("#####\n")<jupyter_output><empty_output><jupyter_text>This time, lets process a bunch of alerts at once. We'll put one alert (id=152476) in that we know will trip this filter, and one that could have, except is in a variable star catalog (id=31544)<jupyter_code>alert_ids = sorted(set(dk.get_alert_ids(5) + [152476, 31544 ])) # notice here we're getting 
# 5 random alerts and adding two to the list
results   = dk.run_many(high_snr, alert_ids=alert_ids, verbose=False)<jupyter_output><empty_output><jupyter_text>3.4 Test against a user-created Alert, using catalog information AND alert properties to make a decisionHere's a little more realistic of an example:<jupyter_code>def nuclear_transient(ld):
    """
    Send alert to stream 'Nuclear Transient' if it is within 0.6 arcseconds of a
    source in the ZTF reference frame. It is also required that a match within
    1" of a known Pan-STARRS galaxy (ztf_distpsnr1 < 1. and ztf_sgscore1<0.3).
    To further remove small flux fluctuaion, we also require magpsf (alert PSF
    photometry) - magnr (PSF photometry of the nearby source in the reference
    image) > 1.5. The selection criteria are from Sjoert van Velzen et al.
    (2018, arXiv:1809.02608), section 2.1.
    """
    p = ld.get_properties()
    
    alert_id = p['alert_id']
    sgscore  = p.get('ztf_sgscore1', None)
    distpsnr = p.get('ztf_distpsnr1', None)
    magpsf   = p.get('ztf_magpsf', None)
    magnr    = p.get('ztf_magnr', None)
    distnr   = p.get('ztf_distnr', None)

    if None in (distnr, distpsnr, sgscore, magpsf, magnr):
        return
    
    
    if distnr < 0.6 and distpsnr < 1. and sgscore < 0.3 and magpsf - magnr < 1.5:
        
        matching_catalog_names = ld.get_astro_object_matches().keys()
        
        if 'veron_aqn_qso' in matching_catalog_names:
            print(alert_id," AGN activity")
            ld.send_to_stream("agn_qso_activity")
        elif 'french_post_starburst_gals' in matching_catalog_names:
            print(alert_id,"TDE candidate")
            ld.send_to_stream("tde_candidate")
        else:
            print("Nuclear activity")
            ld.send_to_stream("nuclear_transient")<jupyter_output><empty_output><jupyter_text>This is a more complex filter. It may not trigger, even if we give it a random set of 20 alerts - most won't be nuclear. We could make it 500 and there's still a good chance it'll not trigger.<jupyter_code>results   = dk.run_many(nuclear_transient, n=20, verbose=False)
# even if it doesn't trigger you can still see timing information from run_many
print({key: results[key] for key in results.keys() if key.startswith('t_')})
# to deal with this sort of thing, you can also create an alert of your own designed to trigger (or not)
# this is a good way to validate your code's behavior with known good (or bad) input

alerts = [{'ztf_distnr':0.5, 'ztf_distpsnr1':0.7, 'ztf_sgscore1':0.01, 'ztf_magpsf':19.5, 'ztf_magnr':19.3},]
fake_alert_id = dk.mock_locus(10.704, 41.269, alerts)
dk.run_stage(fake_alert_id, nuclear_transient, verbose=False)
# this should trigger the stage to show "Nuclear activity"
# You can also get the LocusData object directly for tinkering with.

alert_id = 153505
ld = dk.get_locus_data(alert_id)

print(ld)
print(sorted(ld.get_properties().keys()))
def in_m31(ld):
    """
    Send alerts to stream 'in_m31' if Alert is within a 2-square-degree box
    centered on M31.
    """
    ra_max = 11.434793
    ra_min = 9.934793
    dec_max = 42.269065
    dec_min = 40.269065

    p = ld.get_properties()
    ra = p['ra']
    dec = p['dec']

    if ra_max > ra > ra_min \
    and dec_max > dec > dec_min:
        print('In M31')
        ld.send_to_stream("in_m31")
    else:
        print('Not in M31')
in_m31(ld)

# we can also use the simulated locus we created earlier to test this stage:
fake_locus = dk.get_locus_data(fake_alert_id)
in_m31(fake_locus)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.08825248392752776, 'fraction_numerical': 0.01636469900642899, 'mean_word_length': 3.8646055437100215, 'pattern_counts': {'":': 0, '<': 50, '<?xml version=': 0, '>': 54, 'https://': 3, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '9', 'dir': 'jupyter-structured-clean-dedup', 'id': '221557', 'license': 'BSD-3-Clause', 'n_tokens_mistral': 4397, 'n_tokens_neox': 4145, 'n_words': 1624, 'path': '05_Contrib/TimeDomain/AntaresDevKit/AntaresFilterDevKit.ipynb', 'repo_name': 'jacquesalice/notebooks-latest', 'size': '13739'}