Home / Python SDK / Sample notebooks

Notify on new catalog imagery

Automate the process of checking for newly added catalog imagery over a specified AOI by setting up notifications to your email address.

Open notebook

Automate the process of checking for newly added catalog imagery over a specified AOI by setting up notifications to your email address.

Step 1. Set up the notebook

1. Install dependencies

Python
!pip install up42-py --upgrade -q
!pip install google-api-python-client google-auth-oauthlib google-auth-httplib2 --upgrade -q


import up42, time, base64, pathlib
from datetime import date, datetime, timedelta
from itertools import islice
from email.mime.text import MIMEText
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build

2. Configure credentials

Run the cell below to create a credentials.json file in a directory named .up42 in your home folder.

Python
# Defines the credentials file path if it doesn't exist
credentials_file_path = pathlib.Path.home().joinpath(".up42/credentials.json")
credentials_file_path.parent.mkdir(parents=True, exist_ok=True)
credentials_file_path.touch(exist_ok=True)


# Prints the path to the file
print(f"Credentials file is located at: {credentials_file_path}")
  1. Click the link above to the created file and paste the following code:
    JSON
    {
        "username": "<your-email-address>",
        "password": "<your-password>"
    }
  2. Retrieve the email address and password used for logging into the console. Use them as values for username and password.
  3. Save the credentials.json file.

3. Authenticate and create a PySTAC client connection

Python
up42.authenticate(cfg_file=credentials_file_path)


# Use region if you're logging with the NSG service
# up42.authenticate(cfg_file=credentials_file_path, region="sa")


catalog = up42.catalog_client()

4. Set up sender credentials

The script sends mail through the Gmail API, so the sending account must be a Google account.

  1. Go to Google Console and create a new project or select an existing one.
  2. Open your project and go to APIs and services → Credentials.
  3. Click + Create credentials → OAuth client ID.
  4. Select Desktop app as the application type and click Create.
  5. Open the newly created client ID. In the Client secrets section, download the JSON file.
  6. Go to Audience. Click Publish app → Confirm.

Run the cell below to create a google_credentials.json file in a directory named .up42 in your home folder.

Python
# Defines the Google credentials file path if it doesn't exist
google_credentials_file_path = pathlib.Path.home().joinpath(
    ".up42/google_credentials.json"
)
google_credentials_file_path.parent.mkdir(parents=True, exist_ok=True)
google_credentials_file_path.touch(exist_ok=True)


# Prints the path to the file
print(f"Credentials file is located at: {google_credentials_file_path}")


GOOGLE_CREDENTIALS_PATH = google_credentials_file_path

Paste the contents of the downloaded client ID JSON into the google_credentials.json file and save it.

Specify the email address associated with your Google Cloud account:

Python
SENDER_EMAIL = "[email protected]"

5. Specify the receiving email address

Specify the email address where notifications will be sent. This address doesn’t have to be a Google account.

Python
RECEIVER_EMAIL = "[email protected]"

Step 2. Define search parameters

Python
# Specify the maximum cloud cover percentage from 0 to 100
MAX_CLOUD_COVER_PERCENTAGE = 30


# Define the time period
start_date = date.today() - timedelta(days=90)


# Define search geometry
geometry = {
    "type": "Polygon",
    "coordinates": [
        [
            [13.369713, 52.452327],
            [13.369713, 52.470760],
            [13.339159, 52.470760],
            [13.339159, 52.452327],
            [13.369713, 52.452327],
        ]
    ],
}


cloud_cover_filter = {
    "op": "<",
    "args": [{"property": "eo:cloud_cover"}, MAX_CLOUD_COVER_PERCENTAGE],
}

The first search establishes a baseline of existing imagery. The monitoring loop compares later results with this baseline.

The search filters on acquisition date. A scene acquired before the start date is not included, even if it was added to the catalog later.

Python
def scene_key(item):
    return (item.collection_id, item.id)




def search_scenes():
    # Fetch every match. A max_items cap would hide scenes and break the comparison.
    return list(
        catalog.search(
            intersects=geometry,
            datetime=f"{start_date}/{date.today()}",
            filter=cloud_cover_filter,
        ).items()
    )




previous_items = search_scenes()
items = previous_items
print(f"\nFound {len(items)} scenes matching the criteria:\n")
for item in islice(items, 3):  # Print first 3
    print(f"- Collection:       {item.collection_id}")
    print(f"  Scene ID:         {item.id}")
    print(f"  Acquisition date: {item.datetime}")
    print(f"  Cloud coverage:   {item.properties.get('eo:cloud_cover')}%")
    print(f"  Resolution:       {item.properties.get('gsd')} m\n")

Step 4. Set up the notification system

1. Define the email function

This function sends the notification through the Gmail API.

Python
# OAuth scope for sending email
SCOPES = ["https://www.googleapis.com/auth/gmail.send"]
google_token_path = pathlib.Path.home().joinpath(".up42/google_token.json")




def gmail_credentials():
    creds = None
    if google_token_path.exists() and google_token_path.stat().st_size:
        creds = Credentials.from_authorized_user_file(str(google_token_path), SCOPES)
    if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token:
            creds.refresh(Request())
        else:
            flow = InstalledAppFlow.from_client_secrets_file(
                str(GOOGLE_CREDENTIALS_PATH), SCOPES
            )
            creds = flow.run_local_server(port=0)
        google_token_path.write_text(creds.to_json())
    return creds




def gmail_authenticate():
    return build("gmail", "v1", credentials=gmail_credentials())




def send_email_notification(service, sender_email, receiver_email, subject, body):
    # Email configuration
    message = MIMEText(body)
    message["to"] = receiver_email
    message["from"] = sender_email
    message["subject"] = subject
    raw = base64.urlsafe_b64encode(message.as_bytes()).decode()
    payload = {"raw": raw}
    message = service.users().messages().send(userId="me", body=payload).execute()
    print(f'Message ID: {message["id"]}')

2. Define the checking function

The cell below performs a new search with the same parameters as the initial search. It then compares the scene IDs from the new search with the IDs from the previous search. If new IDs are found, it calls the send_email_notification function for each new image.

Python
def check_for_new_imagery(previous_items, service, sender_email, receiver_email):
    recent_items = search_scenes()


    # If no scenes are returned at all, we can stop early
    if not recent_items:
        print(f"No imagery found over AOI at {datetime.now()}")
        return previous_items


    previous_ids = {scene_key(item) for item in previous_items}
    new_items = [item for item in recent_items if scene_key(item) not in previous_ids]


    if new_items:
        for item in new_items:
            now_time = str(datetime.now())
            subject = "New UP42 imagery alert"
            body = (
                "New catalog image available over your AOI.\n\n"
                f"Collection: {item.collection_id}\n"
                f"Scene ID: {item.id}\n"
                f"Time found: {now_time}"
            )
            print(
                f"New imagery found: {item.collection_id}/{item.id}. Sending a notification."
            )
            send_email_notification(
                service, sender_email, receiver_email, subject, body
            )
    else:
        print(f"No new imagery found over AOI at {datetime.now()}")


    return recent_items

Step 5. Run the monitoring loop

This final cell starts an infinite loop to monitor for new imagery. It refreshes the saved Gmail token, runs check_for_new_imagery, waits, and repeats.

To stop the script, interrupt the kernel.

Python
# Authenticate with Gmail. The first run opens a browser window and saves the token.
print("Authenticating with Gmail...")
gmail_service = gmail_authenticate()
print("Authentication successful.")


# Check for new imagery periodically
while True:
    try:
        gmail_service = gmail_authenticate()
        previous_items = check_for_new_imagery(
            previous_items,
            gmail_service,
            SENDER_EMAIL,
            RECEIVER_EMAIL,
        )


        # Sleep for 3600 seconds (1 hour)
        print("Check complete. Waiting for 1 hour before the next check.")
        time.sleep(3600)


    except KeyboardInterrupt:
        print("\nMonitoring stopped by user.")
        break
    except Exception as e:
        print(f"An error occurred: {e}")
        # Add a shorter sleep time on error before retrying
        time.sleep(300)