Inject context and refactor interface (#35)

src/SimulationService.jl

@@ -14,8 +14,8 @@ import DataFrames: DataFrame
 import HTTP: Request, Response
 import JobSchedulers: scheduler_start, set_scheduler, scheduler_stop, submit!, job_query, result, generate_id, update_queue!, Job, JobSchedulers
 
-include("./SciMLInterface.jl"); import .SciMLInterface: sciml_operations, use_operation, conversions_for_valid_inputs
-include("./service/Service.jl"); import .Service.ArgIO: prepare_output, prepare_input
+include("./contracts/Interface.jl"); import .Interface: get_operation, use_operation, conversions_for_valid_inputs, Context
+include("./service/Service.jl"); import .Service.ArgIO: prepare_output, prepare_input; import .Service.Queuing: publish_to_rabbitmq
 include("./Settings.jl"); import .Settings: settings
 
 export run!
@@ -32,36 +32,42 @@ function health_check()
 end
 
 """
-Schedule a sim run
+Generate the task to run with the correct context    
 """
-function start_run!(prog::Function, req::Request)
-    # TODO(five): Spawn remote workers and run jobs on them
-    # TODO(five): Handle Python so a probabilistic case can work
-    job_id = generate_id()
-    sim_run = Job(@task(prog(req)))
-    sim_run.id = job_id
-    submit!(sim_run)
-    Response(
-        201,
-        ["Content-Type" => "application/json; charset=utf-8"],
-        body=JSON.write("id" => sim_run.id)
-    )
+function contextualize_prog(context)
+    prepare_output(context) ∘ use_operation(context) ∘ prepare_input(context)
 end
 
 """
-Find sim run and request a job with the given args    
+Schedule a sim run given an operation
 """
 function make_deterministic_run(req::Request, operation::String)
-    # TODO(five): Handle output on a less case by case basis
-    if !haskey(sciml_operations, Symbol(operation))
+    # TODO(five): Spawn remote workers and run jobs on them
+    # TODO(five): Handle Python so a probabilistic case can work
+    if isnothing(get_operation(operation))
         return Response(
             404,
             ["Content-Type" => "text/plain; charset=utf-8"],
             body="Operation not found"
         )
     end
-    prog = prepare_output ∘ use_operation(Symbol(operation)) ∘ prepare_input
-    start_run!(prog, req)
+
+    publish_hook = settings["SHOULD_LOG"] ? publish_to_rabbitmq : (args...) -> nothing
+
+    context = Context(
+        generate_id(),
+        publish_hook,  
+        Symbol(operation),
+    )
+    prog = contextualize_prog(context)
+    sim_run = Job(@task(prog(req)))
+    sim_run.id = context.job_id
+    submit!(sim_run)
+    Response(
+        201,
+        ["Content-Type" => "application/json; charset=utf-8"],
+        body=JSON.write("id" => sim_run.id)
+    )
 end
 
 """

src/contracts/Interface.jl

@@ -0,0 +1,44 @@
+"""
+SciML Operations interface for the simulation service
+"""
+module Interface
+
+import CSV
+
+include("./ProblemInputs.jl"); import .ProblemInputs: conversions_for_valid_inputs
+include("./SystemInputs.jl"); import .SystemInputs: Context
+include("../operations/Operations.jl"); import .Operations
+include("../Settings.jl"); import .Settings: settings
+
+export get_operation, conversions_for_valid_inputs, Context
+
+"""
+Sim runs that can be created using the `/runs/sciml/{operation}` endpoint.    
+"""
+function get_operation(raw_operation)
+    operation = Symbol(raw_operation)
+    if in(operation, names(Operations; all=false, imported=true))
+        return getfield(Operations, operation)
+    else
+        return nothing
+    end
+end
+
+"""
+Return an operation wrapped with necessary handlers    
+"""
+function use_operation(context::Context)
+    operation = get_operation(context.operation)
+
+    # NOTE: This runs inside the job so we can't use it to validate on request ATM
+    function coerced_operation(arglist::Dict{Symbol, Any}) 
+        # TODO(five): Fail properly on extra params
+        fixed_args = Dict(
+           name => conversions_for_valid_inputs[name](value)
+           for (name, value) in arglist 
+        )
+        operation(;fixed_args..., context=context)
+    end
+end
+
+end # module Interface

src/contracts/ProblemInputs.jl

@@ -0,0 +1,35 @@
+"""
+User-provided, problem-specific inputs
+"""
+module ProblemInputs
+
+import AlgebraicPetri: PropertyLabelledReactionNet, LabelledPetriNet, AbstractPetriNet
+import Catlab.CategoricalAlgebra: parse_json_acset
+import DataFrames: DataFrame
+
+export conversions_for_valid_inputs
+
+"""
+Transform string into dataframe before it is used as input    
+"""
+coerce_dataset(val::String) = CSV.read(IOBuffer(val), DataFrame)
+
+"""
+Act as identity since the value is already coerced    
+"""
+coerce_dataset(val::DataFrame) = val
+
+"""
+Inputs converted from payload to arguments expanded in operations.    
+"""
+conversions_for_valid_inputs = Dict{Symbol,Function}(
+    :model => (val) -> parse_json_acset(PropertyLabelledReactionNet{Number, Number, Dict}, val), # hack for mira
+    :tspan => (val) -> Tuple{Float64,Float64}(val),
+    :params => (val) -> Dict{String,Float64}(val),
+    :initials => (val) -> Dict{String,Float64}(val),
+    :dataset => coerce_dataset,
+    :feature_mappings => (val) -> Dict{String, String}(val),
+    :timesteps_column => (val) -> String(val)
+)
+
+end # module ProblemInputs

src/contracts/SystemInputs.jl

@@ -0,0 +1,26 @@
+"""
+System provided information that is made available to operations    
+"""
+module SystemInputs
+
+export Context
+
+struct Context
+    job_id::Int64
+    interactivity_hook::Function
+    operation::Symbol
+end 
+
+function Base.iterate(context::Context, state=nothing)
+    if isnothing(state)
+        state = Set(fieldnames(Context))
+    end
+    unused = intersect(Set(fieldnames(Context)), state)
+    if isempty(unused)
+        return nothing
+    end
+    chosen_field = pop!(unused)
+    (chosen_field=>getfield(context, chosen_field), unused)
+end
+
+end # module SystemInputs