Skip to content

stac, vector and retries #317

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
AM1729 merged 4 commits into from
Aug 21, 2026
Merged

stac, vector and retries #317

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
e6dd4b5
stac, vector and retries
AM1729 Aug 20, 2026
4b831ad
clean
AM1729 Aug 20, 2026
ae1d7d6
vector
AM1729 Aug 20, 2026
21c1f3e
final fixes to observe vectors from STAC
AM1729 Aug 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
255 changes: 189 additions & 66 deletions adapters/klab.ogc/src/main/java/org/integratedmodelling/klab/stac/STACEncoder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Date;
import java.util.List;
import java.util.Map;
Expand Down Expand Up @@ -224,37 +225,55 @@ public void getEncodedData(IResource resource, Map<String, String> urnParameters
Space space = (Space) geometry.getDimensions().stream().filter(d -> d instanceof Space).findFirst().orElseThrow();
IEnvelope envelope = space.getEnvelope();
List<Double> bbox = List.of(envelope.getMinX(), envelope.getMaxX(), envelope.getMinY(), envelope.getMaxY());
IGrid grid = space.getGrid();

RegionMap region = RegionMap.fromBoundsAndGrid(space.getEnvelope().getMinX(), space.getEnvelope().getMaxX(),
space.getEnvelope().getMinY(), space.getEnvelope().getMaxY(), (int) grid.getXCells(), (int) grid.getYCells());

ReferencedEnvelope regionEnvelope = new ReferencedEnvelope(region.toEnvelope(),
space.getProjection().getCoordinateReferenceSystem());
RegionMap regionTransformed = RegionMap.fromEnvelopeAndGrid(regionEnvelope, (int) grid.getXCells(),
(int) grid.getYCells());

// Only for Backward Compatiability
// A new COG Adapter would be added
if (resource.getParameters().get("cog") != null) {
COGURL = resource.getParameters().get("cog", String.class);
scope.getMonitor().info("Getting requested extent from the COG Asset from url" + COGURL);
GridCoverage2D coverage = COGAssetExtension.getCOGWindowCoverage(bbox, COGURL);

String rcrs = geometry.getDimension(IGeometry.Dimension.Type.SPACE).getParameters()
.get(org.integratedmodelling.klab.common.Geometry.PARAMETER_SPACE_PROJECTION, String.class);

Projection crs = Projection.create(rcrs);
org.locationtech.jts.geom.Envelope requestedExtend = new org.locationtech.jts.geom.Envelope(bbox.get(0), bbox.get(1),
bbox.get(2), bbox.get(3));

HMRaster raster = HMRaster.fromGridCoverage(coverage);
HMRaster outRaster = new HMRasterWritableBuilder()
.setRegion(RegionMap.fromEnvelopeAndGrid(requestedExtend, (int) space.shape()[0], (int) space.shape()[1]))
.setCrs(crs.getCoordinateReferenceSystem()).setNoValue(raster.getNovalue()).build();

GridCoverage2D adjCoverage = null;
try {
outRaster.mapRaster(null, raster, null);
adjCoverage = outRaster.buildCoverage();
try {
COGURL = resource.getParameters().get("cog", String.class);
scope.getMonitor().info("Getting requested extent from the COG Asset from url" + COGURL);
GridCoverage2D coverage = COGAssetExtension.getCOGWindowCoverage(bbox, COGURL);

String rcrs = geometry.getDimension(IGeometry.Dimension.Type.SPACE).getParameters()
.get(org.integratedmodelling.klab.common.Geometry.PARAMETER_SPACE_PROJECTION, String.class);

Projection crs = Projection.create(rcrs);
org.locationtech.jts.geom.Envelope requestedExtend = new org.locationtech.jts.geom.Envelope(bbox.get(0), bbox.get(1),
bbox.get(2), bbox.get(3));

HMRaster raster = HMRaster.fromGridCoverage(coverage);


CoordinateReferenceSystem targetCRS = HMCrsRegistry.INSTANCE.getCrs("4326");
if (!HMCrsRegistry.crsEquals(raster.getCrs(),targetCRS)) {
var transformer = new HMCrsTransformer(raster.getCrs(), targetCRS);
transformer.setAcceptLenientDatumShift(true);
raster = transformer.transform(raster);
}


HMRaster paddedRaster = new HMRasterWritableBuilder()
.setName("padded")
.setRegion(RegionMap.fromEnvelopeAndGrid(requestedExtend, (int) space.shape()[0], (int) space.shape()[1]))
.setCrs(targetCRS)
.setNoValue(raster.getNovalue()).build();
paddedRaster.mapRaster(null, raster, null);
coverage = paddedRaster.buildCoverage();
encoder = new RasterEncoder();
((RasterEncoder) encoder).encodeFromCoverage(resource, urnParameters, coverage, geometry, builder, scope);
return;
} catch (Exception e) {
throw new KlabResourceAccessException("Cannot build COG Output " + e.getMessage());
}

encoder = new RasterEncoder();
((RasterEncoder) encoder).encodeFromCoverage(resource, urnParameters, adjCoverage, geometry, builder, scope);
return;
}

String collectionUrl = resource.getParameters().get("collection", String.class);
Expand Down Expand Up @@ -345,14 +364,6 @@ public boolean test(HMStacAsset asset) { // Assuming for now that "eo:bands" wou
}
var time2 = effectiveTime;
// TODO merge with similar code from below
IGrid grid = space.getGrid();
RegionMap region = RegionMap.fromBoundsAndGrid(space.getEnvelope().getMinX(), space.getEnvelope().getMaxX(),
space.getEnvelope().getMinY(), space.getEnvelope().getMaxY(), (int) grid.getXCells(), (int) grid.getYCells());

ReferencedEnvelope regionEnvelope = new ReferencedEnvelope(region.toEnvelope(),
space.getProjection().getCoordinateReferenceSystem());
RegionMap regionTransformed = RegionMap.fromEnvelopeAndGrid(regionEnvelope, (int) grid.getXCells(),
(int) grid.getYCells());
// end //TODO
List<HMStacItem> items = features.stream().map(f -> {
try {
Expand All @@ -368,14 +379,12 @@ public boolean test(HMStacAsset asset) { // Assuming for now that "eo:bands" wou
GridCoverage2D coverage = null;

try {
// TODO see if we can access to the same readRasterBandOnRegion without using a
// collection
LogProgressMonitor lpm = new LogProgressMonitor();
try (HMStacManager manager = new HMStacManager(catalogUrl, lpm)) {
HMStacCollection collection = null;
try {
manager.open();
collection = manager.getCollectionById(collectionId);
collection = manager.getCollectionByURL(collectionUrl);
} catch (Exception e1) {
throw new KlabResourceAccessException("Cannot access to STAC collection " + collectionUrl);
}
Expand Down Expand Up @@ -420,7 +429,7 @@ public boolean test(HMStacAsset asset) { // Assuming for now that "eo:bands" wou
HMStacCollection collection = null;
try {
manager.open();
collection = manager.getCollectionById(collectionId);
collection = manager.getCollectionByURL(collectionUrl);

if (collection == null) {
scope.getMonitor()
Expand Down Expand Up @@ -448,28 +457,51 @@ public boolean test(HMStacAsset asset) { // Assuming for now that "eo:bands" wou
bbox.get(1),
bbox.get(3)
});

// Allow transform ensures the process to finish, but I would not bet on the resulting
// data
if (assetPredicate == null) {
if (assetPredicate == null ||
resource.getType() != org.integratedmodelling.klab.api.provenance.IArtifact.Type.NUMBER) {
// If it was raster, then it necessarily should have been number, object means certainly that it's a vector
// NO JSONSelector and JSONValue found, NO assetID was passed as well
scope.getMonitor().debug("Query STAC " + collectionUrl + "to get the features");
// Only get the features from STAC Collection, no need to interact with Rasters
FeatureSource<SimpleFeatureType, SimpleFeature> source;
try {
source = STACFeatureExtension.getFeatures(catalogData, collectionId, bbox, effectiveTime.getStart(),
effectiveTime.getEnd());
} catch (Exception e) {
if (assetPredicate == null) {
try {
source = STACFeatureExtension.getFeatures(catalogData, collectionId, bbox, effectiveTime.getStart(),
effectiveTime.getEnd());
} catch (Exception e) {
manager.close();
throw new KlabResourceAccessException("Cannot extract features from STAC Collection - " + e.getMessage());
}
encoder = new VectorEncoder();
((VectorEncoder) encoder).encodeFromFeatures(source, resource, urnParameters, geometry, builder, scope);
manager.close();
} else {
List<HMStacItem> items = collection.searchItems();
if (items.isEmpty()) {
manager.close();
throw new KlabIllegalStateException("No STAC items found for this context, check the Spatial/ Temporal bounds of items and the Context");
}

var assets = items.stream()
.flatMap(item -> item.getAssets().stream())
.filter(assetPredicate)
.toList();
scope.getMonitor().debug("Found " + assets.size() + " Assets to get the requested geometry");
try {
source = STACFeatureExtension.getFeatures(assets);
encoder = new VectorEncoder();
((VectorEncoder) encoder).encodeFromFeatures(source, resource, urnParameters, geometry, builder, scope);
} catch (Exception e) {
manager.close();
throw new KlabResourceAccessException("Cannot extract features from STAC Collection - " + e.getMessage());
}
manager.close();
throw new KlabResourceAccessException("Cannot extract features from STAC Collection - " + e.getMessage());
}
encoder = new VectorEncoder();
((VectorEncoder) encoder).encodeFromFeatures(source, resource, urnParameters, geometry, builder, scope);
manager.close();
return;
return;
}

List<HMStacItem> items = collection.searchItems();
List<HMStacItem> items = searchItemsWithRetry(collection, scope.getMonitor());
if (items.isEmpty()) {
manager.close();
throw new KlabIllegalStateException("No STAC items found for this context, check the Spatial/ Temporal bounds of items and the Context");
Expand All @@ -479,16 +511,6 @@ public boolean test(HMStacAsset asset) { // Assuming for now that "eo:bands" wou
sortByDate(items, scope.getMonitor());
}

IGrid grid = space.getGrid();

RegionMap region = RegionMap.fromBoundsAndGrid(space.getEnvelope().getMinX(), space.getEnvelope().getMaxX(),
space.getEnvelope().getMinY(), space.getEnvelope().getMaxY(), (int) grid.getXCells(), (int) grid.getYCells());

ReferencedEnvelope regionEnvelope = new ReferencedEnvelope(region.toEnvelope(),
space.getProjection().getCoordinateReferenceSystem());
RegionMap regionTransformed = RegionMap.fromEnvelopeAndGrid(regionEnvelope, (int) grid.getXCells(),
(int) grid.getYCells());

if (resource.getParameters().contains("s3EndpointUrl")) {
String s3EndpointURL = resource.getParameters().get("s3EndpointUrl", String.class);
Client s3Client = buildS3Client(s3EndpointURL);
Expand All @@ -497,16 +519,48 @@ public boolean test(HMStacAsset asset) { // Assuming for now that "eo:bands" wou
var time = effectiveTime;
// Filter here based on time, since in some STAC collections they don't yet support
// temporal filtering :( like ECDC
items = items.stream()
.filter(item -> isWithinRange(item, time.getStart().getMilliseconds(), time.getEnd().getMilliseconds()))
.collect(Collectors.toList());

var pred2 = assetPredicate;

List<HMStacItem> itemsWithinTime = items.stream()
.filter(item -> isWithinRange(
item,
time.getStart().getMilliseconds(),
time.getEnd().getMilliseconds()))
.filter(item -> item.getAssets().stream()
.anyMatch(pred2))
.collect(Collectors.toList());

if (itemsWithinTime.isEmpty()) {
scope.getMonitor().debug("Couldn't find items satisfying temporal and asset based constraints"
+ " within the specified time range, Applying Temporal Mediation");

Map<String, List<HMStacItem>> grouped = items.stream()
.filter(item -> item.getAssets().stream()
.anyMatch(pred2))
.collect(Collectors.groupingBy(
item -> geometryKey(item.getGeometry())
));

items = grouped.values().stream()
.map(group -> group.stream()
.min(Comparator.comparingLong(
groupItem -> checkDuration(
groupItem,
time.getEnd().getMilliseconds())))
.orElseThrow())
.collect(Collectors.toList());

} else {
items = itemsWithinTime;
}

if (items.size() == 0) {
manager.close();
throw new KlabIllegalStateException(
"No STAC items found covering the entire time duration of the context requested");
"No STAC items were satifying constraints and Couldn't apply Temporal Mediation");
} else {
scope.getMonitor().debug("Found " + items.size() + " STAC items satisfying the temporal constraint.");
scope.getMonitor().debug("Found " + items.size() + " STAC items for generating the observation.");
}

// Once the support for customized predicate is added, we can apply for features as well
Expand Down Expand Up @@ -604,6 +658,12 @@ public boolean test(HMStacAsset asset) { // Assuming for now that "eo:bands" wou
throw new KlabInternalErrorException("Cannot build STAC raster output. Reason " + e.getMessage());
}
}

private static String geometryKey(Geometry geom) {
Geometry normalized = geom.copy();
normalized.normalize();
return normalized.toText();
}

private Predicate<HMStacAsset> getAssetPredicateFromJSONSelector(IResource resource) {
String jsonSelector = resource.getParameters().get("jsonSelector", String.class);
Expand Down Expand Up @@ -647,7 +707,32 @@ private boolean isWithinRange(HMStacItem item, long startMillis, long endMillis)
return false;
}
}


private long checkDuration(HMStacItem item, long endMillis) {

DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
if (item.getEndTimestamp() == null) { // assume like it's the correct data
return 0;
}

try {
long itemEnd = LocalDateTime.parse(item.getEndTimestamp(), formatter).atZone(ZoneOffset.UTC).toInstant()
.toEpochMilli();
long itemStart = LocalDateTime.parse(item.getEndTimestamp(), formatter).atZone(ZoneOffset.UTC).toInstant()
.toEpochMilli();

if (itemStart > endMillis) { // Deprioritize future items since it doesn't make a lot of sense
return Long.MAX_VALUE;
}

return Math.abs(itemEnd - endMillis);
} catch(Exception e) {
e.printStackTrace();
return Long.MAX_VALUE; // return an arbitrary large number
}
}


private List<SimpleFeature> getFeaturesFromStaticCollection(String collectionUrl, JSONObject collectionData,
String collectionId) {
List<JSONObject> links = collectionData.getJSONArray("links").toList().stream()
Expand All @@ -662,6 +747,44 @@ private List<SimpleFeature> getFeaturesFromStaticCollection(String collectionUrl
}
}).toList();
}


private List<HMStacItem> searchItemsWithRetry(HMStacCollection collection, IMonitor monitor) throws IOException {
int maxRetries = 5;
long backoffMillis = 10000; // start at 10s

for (int attempt = 0; attempt <= maxRetries; attempt++) {
try {
return collection.searchItems();
} catch (Exception e) {
boolean isRateLimited = isRateLimitError(e);
if (!isRateLimited || attempt == maxRetries) {
throw e instanceof IOException ? (IOException) e : new IOException(e);
}

try {
Thread.sleep(backoffMillis);
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
throw new IOException("Interrupted while backing off after 429", ie);
}

backoffMillis *= 2; // exponential: 10s, 20s, 40s, 80s, 160s
monitor.debug("Search Request throttled; Retrying Search after: " + backoffMillis + " Miliseconds");
}
}
throw new IOException("Unreachable");
}

private boolean isRateLimitError(Throwable e) {
while (e != null) {
if (e.getMessage() != null && e.getMessage().contains("429")) {
return true;
}
e = e.getCause();
}
return false;
}

@Override
public void listDetail(IResource resource, OutputStream stream, boolean verbose, IMonitor monitor) {
Expand Down
Loading
Loading