pittgoogle.alert

Classes for working with astronomical alerts.

Alert(*[, dict, attributes, schema_name, ...])

Container for an astronomical alert.

Container for an astronomical alert.

To create an Alert, use one of the from_* methods like pittgoogle.Alert.from_dict(). Instances of this class are also returned by other calls like pittgoogle.pubsub.Subscription.pull_batch().

Parameters:

dict (dict, optional) – The alert data as a dictionary. If not provided, it will be loaded from the
attributes (dict, optional) – Attributes or custom metadata for the alert.
schema_name (str) – Name of the schema for the alert. This is use to deserialize the alert bytes. See pittgoogle.registry.Schemas.names() for a list of options. If not provided, some properties of the Alert may not be available.
msg (PubsubMessageLike or google.cloud.pubsub_v1.types.PubsubMessage, optional) – The incoming Pub/Sub message object. This class is documented at https://cloud.google.com/python/docs/reference/pubsub/latest/google.cloud.pubsub_v1.types.PubsubMessage.
path (pathlib.Path, optional) – Path to a file containing the alert data.

property alertid: str | int

Return the alert ID. Convenience wrapper around Alert.get.

If the survey does not define an alert ID, this returns the sourceid.

property attributes: Mapping

Return the alert’s custom metadata.

If this was not provided (typical case), this attribute will contain a copy of the incoming Alert.msg.attributes.

You may update this dictionary as desired. If you publish this alert using pittgoogle.Topic.publish, this dictionary will be sent as the outgoing message’s Pub/Sub attributes.

property dataframe: pd.DataFrame: Return a pandas DataFrame containing the source detections.

property dict: Mapping

Alert data as a dictionary.

If this was not provided (typical case), this attribute will contain the deserialized alert bytes from Alert.msg.data.

You may update this dictionary as desired. If you publish this alert using pittgoogle.Topic.publish, this dictionary will be sent as the outgoing Pub/Sub message’s data payload.

Returns:: The alert data as a dictionary.
Return type:: dict
Raises:: SchemaError – If unable to deserialize the alert bytes.

classmethod from_cloud_run(envelope: Mapping, schema_name: str | None = None) → Alert[source]

Create an Alert from an HTTP request envelope containing a Pub/Sub message, as received by a Cloud Run module.

Parameters:

envelope (dict) – The HTTP request envelope containing the Pub/Sub message.
schema_name (str, optional) – The name of the schema to use. Defaults to None.

Returns:

An instance of the Alert class.

Return type:

Alert

Raises:

BadRequest – If the Pub/Sub message is invalid or missing.

Example

Code for a Cloud Run module that uses this method to open a ZTF alert:

import pittgoogle
# flask is used to work with HTTP requests, which trigger Cloud Run modules
# the request contains the Pub/Sub message, which contains the alert packet
import flask

app = flask.Flask(__name__)

# function that receives the request
@app.route("/", methods=["POST"])
def index():

    try:
        # unpack the alert
        # if the request does not contain a valid message, this raises a `BadRequest`
        alert = pittgoogle.Alert.from_cloud_run(envelope=flask.request.get_json(), schema_name="ztf")

    except pittgoogle.exceptions.BadRequest as exc:
        # return the error text and an HTTP 400 Bad Request code
        return str(exc), 400

    # continue processing the alert
    # when finished, return an empty string and an HTTP success code
    return "", 204

classmethod from_dict(payload: Mapping, attributes: Mapping[str, str] | None = None, schema_name: str | None = None) → Alert[source]

Create an Alert object from the given payload dictionary.

Parameters:

payload (dict) – The dictionary containing the data for the Alert object.
attributes (Mapping[str, str], None) – Additional attributes for the Alert object. Defaults to None.
schema_name (str, None) – The name of the schema. Defaults to None.

Returns:

An instance of the Alert class.

Return type:

Alert

classmethod from_msg(msg: PubsubMessage, schema_name: str | None = None) → Alert[source]

Create an Alert object from a google.cloud.pubsub_v1.types.PubsubMessage.

Parameters:

msg (google.cloud.pubsub_v1.types.PubsubMessage) – The PubsubMessage object to create the Alert from.
schema_name (str, optional) – The name of the schema to use for the Alert. Defaults to None.

Returns:

The created Alert object.

Return type:

Alert

classmethod from_path(path: str | Path, schema_name: str | None = None) → Alert[source]

Creates an Alert object from the file at the specified path.

Parameters:

path (str or Path) – The path to the file containing the alert data.
schema_name (str, optional) – The name of the schema to use for the alert. Defaults to None.

Returns:

An instance of the Alert class.

Return type:

Alert

Raises:

FileNotFoundError – If the file at the specified path does not exist.
IOError – If there is an error reading the file.

get(field: str, default: Any = None) → Any[source]

Return the value of a field from the alert data.

Parameters:

field (str) – Name of a field. This must be one of the generic field names used by Pitt-Google (keys in Alert.schema.map). To use a survey-specific field name instead, use Alert.dict.get.
default (str, optional) – The default value to be returned if the field is not found.

Returns:

The value in the Alert.dict corresponding to the field.

Return type:

any

get_key(field: str, name_only: bool = False, default: str | None = None) → str | list[str] | None[source]

Return the survey-specific field name.

Parameters:

field (str) – Generic field name whose survey-specific name is to be returned. This must be one of the keys in the dict self.schema.map.
name_only (bool) – In case the survey-specific field name is nested below the top level, whether to return just the single final name as a str (True) or the full path as a list[str] (False).
default (str or None) – Default value to be returned if the field is not found.

Returns:

Survey-specific name for the field, or default if the field is not found. list[str] if this is a nested field and name_only is False, else str with the final field name only.

Return type:

str or list[str])

property objectid: str | int

Return the object ID. Convenience wrapper around Alert.get.

The “object” represents a collection of sources, as determined by the survey.

property schema: Schema

Return the schema from the pittgoogle.registry.Schemas registry.

Raises:: SchemaError – If the schema_name is not supplied or a schema with this name is not found.

property sourceid: str | int

Return the source ID. Convenience wrapper around Alert.get.

The “source” is the detection that triggered the alert.