Processing in SDK

1
from itertools import islice
2

3
# Sort by cost, from the least expensive to the most expensive
4
jobs_sorted = up42.Job.all(sort_by=up42.JobSorting.credits.asc)
5

6
# Sort by creation date, from the most recent to the earliest
7
# jobs_sorted = up42.Job.all(sort_by=up42.JobSorting.created.desc)
8

9
# Sort by job status, in descending alphabetical order
10
# jobs_sorted = up42.Job.all(sort_by=up42.JobSorting.status.desc)
11

12
# Sort by process ID in ascending alphabetical order
13
# jobs_sorted = up42.Job.all(sort_by=up42.JobSorting.process_id.asc)
14

15
for job in islice(jobs_sorted, 0, 5): # Print first 5 results
16
  print(f"- Job ID:       {job.id}")
17
  print(f"  Price:        {job.credits} credits")
18
  print(f"  Created at:   {job.created}")
19
  print(f"  Status:       {job.status.value}")
20
  print(f"  Started at:   {job.started}")
21
  print(f"  Finished at:  {job.finished or 'Not finished'}\n")

1
# Select a job
2
job_id = "a0d443a2-41e8-4995-8b54-a5cc4c448227"
3

4
# Fetch job info
5
job = up42.Job.get(job_id=job_id)
6

7
# Define output
8
print(f"Job details for ID: {job.id}")
9
print(f"Process ID:         {job.process_id}")
10
print(f"Definition:         {job.definition}")
11
print(f"Collection URL:     {job.collection_url}")
12
print(f"Price:              {job.credits} credits")
13
print(f"Created at:         {job.created}")
14
print(f"Status:             {job.status.value}")
15
print(f"Started at:         {job.started}")
16
print(f"Finished at:        {job.finished or 'Not finished'}")

1
# Select a job
2
job_id = "a0d443a2-41e8-4995-8b54-a5cc4c448227"
3

4
# Fetch job info
5
job = up42.Job.get(job_id=job_id)
6

7
# Define output
8
print(f"STAC collection ID: {job.collection.id}")

1
# Select a job
2
job_id = "a0d443a2-41e8-4995-8b54-a5cc4c448227"
3

4
# Fetch job info
5
job = up42.Job.get(job_id=job_id)
6
print(f"Initial job status is '{job.status.value}'. Starting to track...\n")
7

8
# This will print a status log to your console every 20 seconds
9
# and will PAUSE the script here until the job is done
10
job.track(wait=20)
11

12
# This line will only run AFTER the job has finished tracking
13
print("\nTracking complete!")
14
print(f"The final job status is: {job.status.value}")

1
# Select a job
2
job_id = "a0d443a2-41e8-4995-8b54-a5cc4c448227"
3

4
# Fetch job info
5
job = up42.Job.get(job_id=job_id)
6

7
# Define output
8
print(f"Job details for ID: {job.id}")
9
print(f"Price:              {job.credits} credits")
10
print(f"Created at:         {job.created}")
11
print(f"Status:             {job.status.value}")
12
print(f"Started at:         {job.started}")
13
print(f"Finished at:        {job.finished or 'Not finished'}")

1
# Fetch all jobs, filtered by max duration and sorted by process ID
2
jobs = up42.Job.all(
3
    max_duration=500,
4
    sort_by=up42.JobSorting.process_id,
5
)
6

7
# Define output
8
for job in jobs:
9
  print(f"- Job ID:         {job.id}")
10
  print(f"  Process ID:     {job.process_id}")
11
  print(f"  Price:          {job.credits} credits")
12
  print(f"  Status:         {job.status.value}")
13
  print(f"  Created at:     {job.created}")
14
  print(f"  Started at:     {job.started}")
15
  print(f"  Finished at:    {job.finished or 'Not finished'}\n")

1
from up42 import processing_templates
2

3
# Select an item
11 collapsed lines
4
stac_item_id = "68567134-27ad-7bd7-4b65-d61adb11fc78"
5

6
# Get the item from the STAC client
7
stac_client = up42.stac_client()
8
stac_item = next(stac_client.get_items(stac_item_id))
9

10
# Instanitate a DetectionTreesSpacept template
11
template = processing_templates.DetectionTreesSpacept(
12
  title="Detect trees over UK",
13
  item=stac_item,
14
)
15

16
# Verify that the template cost is less than or equal to 100 credits
17
acceptable_cost = 100
18

19
if template.cost <= acceptable_cost:
20
  print(f"Template cost:       {template.cost.credits} credits")
21
  print(f"Template strategy:   {template.cost.strategy}")
22
  print(f"STAC item size:      {template.cost.size}")
23
  print(f"Unit of measurement: {template.cost.unit}")
24

25
else:
26
  print(f"Template cost is higher than {acceptable cost} credits")

1
from up42 import processing_templates
2

3
# Select an item
4
stac_item_id = "68567134-27ad-7bd7-4b65-d61adb11fc78"
5

6
# Get the item from the STAC client
7
stac_client = up42.stac_client()
8
stac_item = next(stac_client.get_items(stac_item_id))
9

10
# Instanitate a DetectionTreesSpacept template
11
job = processing_templates.DetectionTreesSpacept(
12
    title="Detect trees over UK",
13
    item=stac_item,
14
)
15

16
# Assert that the job is valid, print errors if not
17
if not job.is_valid:
18
  for error in job.errors:
19
    print(f"{error}\n")

1
from up42 import processing_templates
2

3
# Select an item
11 collapsed lines
4
stac_item_id = "68567134-27ad-7bd7-4b65-d61adb11fc78"
5

6
# Get the item from the STAC client
7
stac_client = up42.stac_client()
8
stac_item = next(stac_client.get_items(stac_item_id))
9

10
# Instanitate a DetectionTreesSpacept template
11
job_template = processing_templates.DetectionTreesSpacept(
12
    title="Detect trees over UK",
13
    item=stac_item,
14
)
15

16
# Execute the templated job
17
job = job_template.execute()
18

19
# Define output
20
print(f"Job details for ID: {job.id}")
21
print(f"Process ID:         {job.process_id}")
22
print(f"Status:             {job.status.value}")
23
print(f"Price:              {job.credits or 'N/A'} credits")
24
print(f"Created at:         {job.created}")