API Reference

class SOCS_Xray.pipeline.Pipeline(email, password, root='./main/')[source]

Bases: object

authorize(email, password)[source]

initialize()[source]: [matched] stores all candidates passed your filters. The unique id is pair of X-ray name and optical name tuple, i.e. (EP240506a, AT2024ofs), (A85_1, SN 025ujd)

request_obs_time(ids)[source]

run(dt=[-5, 30], show_progress=True, wxt_radii: float = 3.5, fxt_radii: float = 20, update_result=False, fxt_search_max: float = 3000)[source]

Run pipeline! The pipeline will performing cross-matching with TNS and ZTF based on your tme window. And it will store the result in self.uniform_match; and automatically parse email content in self.uniform_html.

Parameters:

dt (list, optional) – time offset threshold. Defaults to [-5,30].
show_progress (bool, optional) – Whether to show progress. Defaults to True.
wxt_radii (float, optional) – Searching radius for WXT sources (in arcmin). Defaults to 3.5.
fxt_radii (float, optional) – Searching radius for WXT sources (in arcsec). Defaults to 20.
update_result (bool, optional) – Whether to merge new identified candidates to previous result. Defaults to False.
fxt_search_max (float, optional) – Maximum cross-matched FXT sources. Defaults to 3000.

update_LSST(ndays=10)[source]

update_TNS(ndays=5)[source]

update_ZTF(ndays=10)[source]

SOCS_Xray.fetch.base_alerce_query(ndet, mjdfirst, types=['SN', 'AGN'], classifier='stamp_classifier')[source]

_summary_

Parameters:

ndet (float) – number of detections
mjdfirst (float) – first detection MJD
types (list, optional) – Classification. Defaults to [“SN”, “AGN”].
classifier (str, optional) – Classifier name. Defaults to ‘stamp_classifier’.

Returns:

Alerce query

Return type:

str

SOCS_Xray.fetch.download_fxt_data(username: str, password: str, ra: float, dec: float, start_time: str, end_time: str, destination_path: str, radius: float = 0.01, headless: bool = True, state_path: str = 'ep_oauth_state.json')[source]

Download Einstein Probe FXT observation data for given coordinates and time range.

Parameters:

username (str) – EP account username.
password (str) – EP account password.
ra (float) – Target right ascension (deg).
dec (float) – Target declination (deg).
start_time (str) – Start time in “YYYY-MM-DD HH:MM:SS” format.
end_time (str) – End time in “YYYY-MM-DD HH:MM:SS” format.
destination_path (str) – Directory where data will be saved.
radius (str, optional) – Search radius in degrees. Default = 0.01.
headless (bool, optional) – Run browser in headless mode (default True).
state_path (str, optional) – Path to save the Playwright state for login persistence.

SOCS_Xray.fetch.download_wxt_data(username: str, password: str, local_path: str, start_time: str, end_time: str, ra: float, dec: float, radius: float = 0.05, host='101.201.57.201')[source]

Download WXT level 2-3 data.

Parameters:

username (str) – user name
password (str) – pasword
local_path (str) – local path to store data
start_time (str) – start observation time for searching
end_time (str) – end observation time for searching
ra (float) – RA in degree
dec (float) – Dec in degree
radius (float, optional) – cone searhing radius in degree. Defaults to 0.05.
host (str, optional) – host ip. Defaults to ‘101.201.57.201’.

SOCS_Xray.fetch.get_Alerce(query)[source]

_summary_

Parameters:: query (str) – Alerce query
Returns:: Table of filtered alerts
Return type:: Table

SOCS_Xray.fetch.get_Lasair(ndays, survey='ztf', lsst_ncut=5)[source]

_summary_

Parameters:: ndays (int) – How many days to look back.
Returns:: Table of filtered alerts
Return type:: Table

SOCS_Xray.fetch.get_TNS(filename='tns_public_objects.csv.zip', save_dir='./')[source]

Download TNS object list.

Args: filename (str): saved filename save_dir (str): saving path.

Returns: Table: Table of TNS public objects.

SOCS_Xray.fetch.request_obs_time(EMAIL, PASSWORD, ids)[source]

Parameters:

EMAIL (str) – user email
PASSWORD (str) – password
ids (list) – source ids

Returns:

list observation time

Return type:

list

SOCS_Xray.fetch.search_TNS(user_id, user_name, url_parameters={'discovered_period_units': 'days', 'discovered_period_value': '10', 'format': 'csv', 'num_page': '200'}, save_dir='./')[source]

_summary_

Parameters:

user_id (str) – ID
user_name (str) – Usr Name
url_parameters (dict, optional) – url parameters. Defaults to {“discovered_period_value” : “10”, “discovered_period_units” : “days”, “format” : “csv”, “num_page” : “200”,}.
save_dir (str, optional) – saving path. Defaults to ‘./’.

Returns:

Table of filtered TNS objects

Return type:

Table

SOCS_Xray.fetch.search_TNS_api(api_key, bot_id, bot_name, search_params, pause=1, save_dir='./')[source]

Search TNS objects via API and fetch detailed information for each object.

Parameters:

api_key (str) – Your TNS API key.
bot_id (str or int) – Your bot ID.
bot_name (str) – Your bot name.
search_params (dict) – Parameters for TNS search API (see TNS API manual).
pause (float) – Pause in seconds between individual object detail requests to avoid rate limit.

Returns:

Astropy Table containing objname, ra, dec, discovery_date, and other fields.

Return type:

Table

SOCS_Xray.fetch.update_WXT_source_list(EMAIL, PASSWORD, save_dir=None)[source]

Download WXT source list.

Parameters:

EMAIL (str) – user email
PASSWORD (str) – password
save_dir (str, optional) – Saving path. Defaults to None.

Returns:

A Table of all sources detected by EP

Return type:

Table

SOCS_Xray.mail.send_email(smtp_server: str, smtp_port: float, sender_email: str, receiver_emails: list, password: str, html_body: str, title='EP Counterpart Searching Notice')[source]

Send email to a list of address.

Parameters:

smtp_server (str) – SMTP server.
smtp_port (float) – SMTP port.
sender_email (str) – Sender email.
receiver_emails (list) – List of receiver emails.
password (str) – password.
html_body (str) – content.
title (str, optional) – Email title. Defaults to ‘EP Counterpart Searching Notice’.