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)
10 collapsed lines
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
# Define output for one of the sorting examples
16
for job in islice(jobs_sorted, 0, 5): # Print first 5 results
17
    print(f"- Job ID:       {job.id}")
18
    print(f"  Price:        {job.credits} credits")
19
    print(f"  Status:       {job.status.value}")
20
    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"Price:              {job.credits} credits")
11
print(f"Status:             {job.status.value}")
12
print(f"Created at:         {job.created}")
13
print(f"Started at:         {job.started}")
14
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"Process ID:         {job.process_id}")
10
print(f"Price:              {job.credits} credits")
11
print(f"Status:             {job.status.value}")
12
print(f"Created at:         {job.created}")
13
print(f"Started at:         {job.started}")
14
print(f"Finished at:        {job.finished or 'Not finished'}")

1
from itertools import islice
2

3
# Search for jobs
4
jobs = up42.Job.all(
5
    max_duration=500,
6
    sort_by=up42.JobSorting.process_id,
7
)
8

9
# Define output
10
for job in islice(jobs, 0, 5): # Print first 5 results
11
    print(f"- Job ID:         {job.id}")
12
    print(f"  Process ID:     {job.process_id}")
13
    print(f"  Price:          {job.credits} credits")
14
    print(f"  Status:         {job.status.value}")
15
    print(f"  Created at:     {job.created}")
16
    print(f"  Started at:     {job.started}")
17
    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
job_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 job_template.cost <= acceptable_cost:
20
  print(f"Template cost:       {job_template.cost.credits} credits")
21
  print(f"Template strategy:   {job_template.cost.strategy}")
22
  print(f"STAC item size:      {job_template.cost.size}")
23
  print(f"Unit of measurement: {job_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_template = 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_template.is_valid:
18
  for error in job_template.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}")