Merge pull request #12 from Open-EO/danlooo/develop

Add process graph

.github/workflows/CI.yml

@@ -29,7 +29,6 @@ jobs:
           - ubuntu-latest
         arch:
           - x64
-          - x86
     steps:
       - uses: actions/checkout@v3
       - uses: julia-actions/setup-julia@v1

Project.toml

@@ -4,14 +4,16 @@ authors = ["Daniel Loos <[email protected]> and contributors"]
 version = "1.0.0-DEV"
 
 [deps]
+AbstractTrees = "1520ce14-60c1-5f80-bbc7-55ef81b5835c"
 Base64 = "2a0f44e3-6c83-55bd-87e4-b1978d98bd5f"
-DataFrames = "a93c6f00-e57d-5684-b7b6-d8193f3e46c0"
+DataStructures = "864edb3b-99cc-5e75-8d2d-829cb0a9cfe8"
 HTTP = "cd3eb016-35fb-5094-929b-558a96fad6f3"
+Infiltrator = "5903a43b-9cc3-4c30-8d17-598619ec4e9b"
 JSON3 = "0f8b85d8-7281-11e9-16c2-39a750bddbf1"
+OrderedCollections = "bac558e1-5e72-5ebc-8fee-abe8a469f55d"
 StructTypes = "856f2bd8-1eba-4b0a-8007-ebc267875bd4"
 
 [compat]
-DataFrames = "1"
 HTTP = "1"
 JSON3 = "1"
 StructTypes = "1"

src/API.jl

@@ -1,11 +1,10 @@
 using JSON3
-using DataFrames
 
 """
 Lists all predefined processes and returns detailed process descriptions, including parameters and return values.
 """
 function list_processes(connection::AbstractConnection)
-    response = fetchApi(connection, "processes"; output_type=OpenEOApiTypes.ProcessesRoot)
+    response = fetchApi(connection, "processes"; output_type=ProcessesRoot)
     return response.processes
 end
 
@@ -22,7 +21,7 @@ end
 Lists available collections with at least the required information.
 """
 function list_collections(connection::AbstractConnection)
-    response = fetchApi(connection, "collections"; output_type=OpenEOApiTypes.CollectionsRoot)
+    response = fetchApi(connection, "collections"; output_type=CollectionsRoot)
     collections = response.collections
     return collections
 end
@@ -35,19 +34,10 @@ function describe_collection(connection::AbstractConnection, id::String)
     return response
 end
 
-"""
-Process and download data synchronously
-"""
-function save_result(connection::AbstractConnection, process_graph::Dict{<:Any})
-    query = Dict(
-        "process" => Dict(
-            "process_graph" => process_graph
-        )
-    )
-    headers = [
-        "Accept" => "*",
-        "Content-Type" => "application/json"
-    ]
-    response = fetchApi(connection, "result", "POST", headers, json(query))
-    return response
-end
+Base.@kwdef struct BoundingBox{T<:Real}
+    west::T
+    south::T
+    east::T
+    north::T
+end
+StructTypes.StructType(::Type{BoundingBox}) = StructTypes.Struct()

src/Collections.jl

@@ -0,0 +1,35 @@
+import StructTypes
+
+struct Collection
+    stac_version::String
+    stac_extensions::Vector{Any}
+    type::String
+    id::String
+    title::String
+    description::String
+    license::String
+    providers::Vector{Any}
+    extent::Any
+    links::Vector{Any}
+end
+StructTypes.StructType(::Type{Collection}) = StructTypes.Struct()
+
+function Base.show(io::IO, ::MIME"text/plain", c::Collection)
+    println(io, "openEO Collection \"$(c.id)\"")
+    print(io, c.description)
+end
+
+function Base.show(io::IO, ::MIME"text/plain", v::Vector{Collection}; n=10)
+    println(io, "$(length(v))-element $(typeof(v)):")
+    for c in first(v, n)
+        println(io, " $(c.id): $(c.title)")
+    end
+    n < length(v) && print("⋮")
+end
+
+# @see https://earthengine.openeo.org/v1.0/collections
+struct CollectionsRoot
+    collections::Vector{Collection}
+    links::Vector{Any}
+end
+StructTypes.StructType(::Type{CollectionsRoot}) = StructTypes.Struct()

src/Connections.jl

@@ -25,15 +25,20 @@ const default_headers = [
 
 n_existing_connections = 0
 
-function fetchApi(url; method="GET", headers=deepcopy(default_headers), output_type::Type=Any, kw...)
-    response = HTTP.request(method, url, headers)
-    response_type = Dict(response.headers)["Content-Type"]
-    if response_type == "application/json"
-        response_string = String(response.body)
-        response_converted = JSON3.read(response_string, output_type)
-        return response_converted
-    else
-        return response
+function fetchApi(url; method="GET", headers=deepcopy(default_headers), output_type::Type=Any, body::String="", kw...)
+    try
+        response = HTTP.request(method, url, headers, body; kw...)
+        response_type = Dict(response.headers)["Content-Type"]
+        if response_type == "application/json"
+            response_string = String(response.body)
+            response_converted = JSON3.read(response_string, output_type)
+            return response_converted
+        else
+            return response
+        end
+    catch e
+        # make concise error message, not entire callstack
+        @error e
     end
 end
 

src/OpenEOClient.jl

@@ -1,8 +1,9 @@
 module OpenEOClient
 
-include("OpenEOApiTypes.jl")
 include("Connections.jl")
 include("Processes.jl")
+include("ProcessGraph.jl")
+include("Collections.jl")
 include("API.jl")
 
 export
@@ -12,7 +13,8 @@ export
     list_jobs,
     list_processes,
     register_processes,
-    ProcessCall,
+    ProcessNode,
     Processes,
-    save_result
+    BoundingBox,
+    compute_result
 end

src/ProcessGraph.jl

@@ -0,0 +1,68 @@
+using OrderedCollections
+
+using Infiltrator
+
+function flatten!(g::AbstractProcessNode, root_id, nodes=Vector{ProcessNode}())
+    processes = filter(((k, v),) -> v isa ProcessNode, g.arguments)
+
+    # post order tree traversal
+    for (key, child) in processes
+        flatten!(child, root_id, nodes)
+        g.arguments[key] = ProcessNodeReference(child.id)
+    end
+
+    append!(nodes, [v for (k, v) in processes])
+
+    if g.id == root_id
+        return vcat(nodes, [g])
+    end
+end
+
+function get_process_graph(process_call::ProcessNode)
+    g = deepcopy(process_call)
+    root_id = process_call.id
+    processes = flatten!(g, root_id)
+
+    res = OrderedDict()
+    for p in processes
+        id = p.id
+        delete!(res, id)
+        res[p.id] = p
+    end
+
+    l = last(res).second
+    l = ProcessNode(l.id, l.process_id, l.arguments, true)
+    res[l.id] = l
+
+    return res
+end
+
+
+"""
+Process and download data synchronously
+"""
+function compute_result(connection::AbstractConnection, process_call::ProcessNode, filepath::String="", kw...)
+    query = Dict(
+        :process => Dict(
+            :process_graph => get_process_graph(process_call),
+            :parameters => []
+        )
+    )
+
+    headers = [
+        "Accept" => "*",
+        "Content-Type" => "application/json"
+    ]
+
+    response = fetchApi(connection, "result"; method="POST", headers=headers, body=JSON3.write(query))
+
+    if isempty(filepath)
+        file_extension = split(Dict(response.headers)["Content-Type"], "/")[2]
+        filepath = "out." * file_extension
+    end
+
+    write(open(filepath, "w"), response.body)
+    return filepath
+end
+
+