kscience/dataforge-core
8
0
Fork 0
Code Issues Pull Requests 1 Packages Projects Releases 12 Wiki Activity

Workspace is working

Browse Source
This commit is contained in:
Alexander Nozik 2019-03-29 12:16:24 +03:00
parent 929f1b9d69
commit e3127503ec
11 changed files with 295 additions and 121 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
dataforge-data/src/commonMain/kotlin/hep/dataforge/data/DataNode.kt
View File

@ -206,4 +206,6 @@ fun <T : Any> DataNode<T>.filter(predicate: (Name, Data<T>) -> Boolean): DataNod
}
}
fun <T: Any> DataNode<T>.first(): Data<T> = data().first().second
//fun <T : Any, R: T> DataNode<T>.filterIsInstance(type: KClass<R>): DataNode<R> = filter{_,data -> type.}

2
dataforge-data/src/jvmMain/kotlin/hep/dataforge/data/JoinAction.kt
View File

@ -16,7 +16,7 @@ class JoinGroup<T : Any, R : Any>(var name: String, internal val node: DataNode<
var meta: MetaBuilder = MetaBuilder()
var result: suspend ActionEnv.(Map<Name, T>) -> R = TODO("Action not implemented")
lateinit var result: suspend ActionEnv.(Map<Name, T>) -> R
fun result(f: suspend ActionEnv.(Map<Name, T>) -> R) {
this.result = f;

2
dataforge-data/src/jvmMain/kotlin/hep/dataforge/data/PipeAction.kt
View File

@ -14,7 +14,7 @@ class ActionEnv(val name: Name, val meta: Meta)
* Action environment
*/
class PipeBuilder<T, R>(var name: Name, var meta: MetaBuilder) {
var result: suspend ActionEnv.(T) -> R = TODO("Action not implemented")
lateinit var result: suspend ActionEnv.(T) -> R
/**
* Calculate the result of goal

2
dataforge-data/src/jvmMain/kotlin/hep/dataforge/data/SplitAction.kt
View File

@ -14,7 +14,7 @@ import kotlin.reflect.full.isSuperclassOf
class FragmentRule<T : Any, R : Any>(val name: Name, var meta: MetaBuilder) {
var result: suspend (T) -> R = TODO("Action not implemented")
lateinit var result: suspend (T) -> R
fun result(f: suspend (T) -> R) {
result = f;

26
dataforge-workspace/src/commonMain/kotlin/hep/dataforge/workspace/Dependency.kt
View File

@ -27,18 +27,31 @@ class DataDependency(val filter: DataFilter, val placement: Name = EmptyName) :
}
}
override fun toMeta(): Meta = filter.config
companion object {
val all: DataDependency = DataDependency(DataFilter.build { })
override fun toMeta(): Meta = buildMeta {
"data" to filter.config
"to" to placement
}
}
class AllDataDependency(val placement: Name = EmptyName) : Dependency() {
override fun apply(workspace: Workspace): DataNode<Any> = if (placement.isEmpty()) {
workspace.data
} else {
DataNode.build(Any::class) { this[placement] = workspace.data }
}
override fun toMeta() = buildMeta {
"data" to "*"
"to" to placement
}
}
class TaskModelDependency(val name: String, val meta: Meta, val placement: Name = EmptyName) : Dependency() {
override fun apply(workspace: Workspace): DataNode<Any> {
val task = workspace.tasks[name] ?: error("Task with name ${name} is not found in the workspace")
if (task.isTerminal) TODO("Support terminal task")
val result = with(workspace) { task(meta) }
val result = workspace.run(task, meta)
return if (placement.isEmpty()) {
result
} else {
@ -47,7 +60,8 @@ class TaskModelDependency(val name: String, val meta: Meta, val placement: Name
}
override fun toMeta(): Meta = buildMeta {
"name" to name
"task" to name
"meta" to meta
"to" to placement
}
}

27
dataforge-workspace/src/commonMain/kotlin/hep/dataforge/workspace/SimpleWorkspace.kt Normal file
View File

@ -0,0 +1,27 @@
package hep.dataforge.workspace
import hep.dataforge.context.Context
import hep.dataforge.context.Global
import hep.dataforge.context.members
import hep.dataforge.data.DataNode
import hep.dataforge.meta.Meta
/**
* A simple workspace without caching
*/
class SimpleWorkspace(
override val context: Context,
override val data: DataNode<Any>,
override val targets: Map<String, Meta>,
tasks: Collection<Task<Any>>
) : Workspace {
override val tasks: Map<String, Task<*>> by lazy {
(context.members<Task<*>>(Task.TYPE) + tasks).associate { it.name to it }
}
companion object {
fun build(parent: Context = Global, block: SimpleWorkspaceBuilder.() -> Unit): SimpleWorkspace =
SimpleWorkspaceBuilder(parent).apply(block).build()
}
}

12
dataforge-workspace/src/commonMain/kotlin/hep/dataforge/workspace/TaskModel.kt
View File

@ -73,10 +73,16 @@ class TaskModelBuilder(val name: String, meta: Meta = EmptyMeta) {
/**
* Add dependency for
*/
fun dependsOn(name: String, meta: Meta, placement: Name = EmptyName) {
fun dependsOn(name: String, meta: Meta = this.meta, placement: Name = EmptyName) {
dependencies.add(TaskModelDependency(name, meta, placement))
}
fun dependsOn(task: Task<*>, meta: Meta = this.meta, placement: Name = EmptyName) =
dependsOn(task.name, meta, placement)
fun dependsOn(task: Task<*>, placement: Name = EmptyName, metaBuilder: MetaBuilder.() -> Unit) =
dependsOn(task.name, buildMeta(metaBuilder), placement)
/**
* Add custom data dependency
*/
@ -96,8 +102,8 @@ class TaskModelBuilder(val name: String, meta: Meta = EmptyMeta) {
/**
* Add all data as root node
*/
fun allData() {
dependencies.add(DataDependency.all)
fun allData(to: Name = EmptyName) {
dependencies.add(AllDataDependency(to))
}
fun build(): TaskModel = TaskModel(name, meta.seal(), dependencies)

52
dataforge-workspace/src/commonMain/kotlin/hep/dataforge/workspace/Workspace.kt
View File

@ -1,11 +1,11 @@
package hep.dataforge.workspace
import hep.dataforge.context.Context
import hep.dataforge.context.ContextAware
import hep.dataforge.context.members
import hep.dataforge.data.Data
import hep.dataforge.data.DataNode
import hep.dataforge.meta.Meta
import hep.dataforge.meta.MetaBuilder
import hep.dataforge.meta.buildMeta
import hep.dataforge.names.Name
import hep.dataforge.names.toName
import hep.dataforge.provider.Provider
@ -49,40 +49,46 @@ interface Workspace : ContextAware, Provider {
}
}
operator fun <R : Any> Task<R>.invoke(config: Meta): DataNode<R> {
/**
* Invoke a task in the workspace utilizing caching if possible
*/
fun <R : Any> run(task: Task<R>, config: Meta): DataNode<R> {
context.activate(this)
try {
val model = build(this@Workspace, config)
validate(model)
return run(this@Workspace, model)
val model = task.build(this, config)
task.validate(model)
return task.run(this, model)
} finally {
context.deactivate(this)
}
}
/**
* Invoke a task in the workspace utilizing caching if possible
*/
operator fun <R : Any> Task<R>.invoke(targetName: String): DataNode<R> {
val target = targets[targetName] ?: error("A target with name $targetName not found in ${this@Workspace}")
return invoke(target)
}
// /**
// * Invoke a task in the workspace utilizing caching if possible
// */
// operator fun <R : Any> Task<R>.invoke(targetName: String): DataNode<R> {
// val target = targets[targetName] ?: error("A target with name $targetName not found in ${this@Workspace}")
// context.logger.info { "Running ${this.name} on $target" }
// return invoke(target)
// }
companion object {
const val TYPE = "workspace"
}
}
class SimpleWorkspace(
override val context: Context,
override val data: DataNode<Any>,
override val targets: Map<String, Meta>,
tasks: Collection<Task<Any>>
) : Workspace {
override val tasks: Map<String, Task<*>> by lazy {
(context.members<Task<*>>(Task.TYPE) + tasks).associate { it.name to it }
fun Workspace.run(task: Task<*>, target: String): DataNode<Any> {
val meta = targets[target] ?: error("A target with name $target not found in ${this}")
return run(task, meta)
}
}
fun Workspace.run(task: String, target: String) =
tasks[task]?.let { run(it, target) } ?: error("Task with name $task not found")
fun Workspace.run(task: String, meta: Meta) =
tasks[task]?.let { run(it, meta) } ?: error("Task with name $task not found")
fun Workspace.run(task: String, block: MetaBuilder.() -> Unit = {}) =
run(task, buildMeta(block))

99
dataforge-workspace/src/commonMain/kotlin/hep/dataforge/workspace/WorkspaceBuilder.kt
View File

@ -2,39 +2,90 @@ package hep.dataforge.workspace
import hep.dataforge.context.Context
import hep.dataforge.context.ContextBuilder
import hep.dataforge.data.Data
import hep.dataforge.data.DataNode
import hep.dataforge.data.DataTreeBuilder
import hep.dataforge.meta.Meta
import hep.dataforge.meta.MetaBuilder
import hep.dataforge.meta.buildMeta
import hep.dataforge.meta.*
import hep.dataforge.names.Name
import hep.dataforge.names.toName
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.GlobalScope
interface WorkspaceBuilder {
val parentContext: Context
var context: Context
var data: DataTreeBuilder<Any>
var tasks: MutableSet<Task<Any>>
var targets: MutableMap<String, Meta>
fun build(): Workspace
}
/**
* A builder for a workspace
* Set the context for future workspcace
*/
class WorkspaceBuilder(var context: Context) {
val data = DataTreeBuilder(Any::class)
val targets = HashMap<String, Meta>()
val tasks = HashSet<Task<Any>>()
fun context(action: ContextBuilder.() -> Unit) {
this.context = ContextBuilder().apply(action).build()
fun WorkspaceBuilder.context(name: String, block: ContextBuilder.() -> Unit) {
context = ContextBuilder(name, parentContext).apply(block).build()
}
fun data(action: DataTreeBuilder<Any>.() -> Unit) = data.apply(action)
fun target(name: String, meta: Meta) {
targets[name] = meta
fun WorkspaceBuilder.data(name: Name, data: Data<Any>) {
this.data[name] = data
}
fun target(name: String, action: MetaBuilder.() -> Unit) = target(name, buildMeta(action))
fun WorkspaceBuilder.data(name: String, data: Data<Any>) = data(name.toName(), data)
fun task(task: Task<*>) {
tasks.add(task)
fun WorkspaceBuilder.static(name: Name, data: Any, scope: CoroutineScope = GlobalScope, meta: Meta = EmptyMeta) =
data(name, Data.static(scope, data, meta))
fun WorkspaceBuilder.static(name: Name, data: Any, scope: CoroutineScope = GlobalScope, block: MetaBuilder.() -> Unit = {}) =
data(name, Data.static(scope, data, buildMeta(block)))
fun WorkspaceBuilder.static(name: String, data: Any, scope: CoroutineScope = GlobalScope, block: MetaBuilder.() -> Unit = {}) =
data(name, Data.static(scope, data, buildMeta(block)))
fun WorkspaceBuilder.data(name: Name, node: DataNode<Any>) {
this.data[name] = node
}
fun build(): Workspace = SimpleWorkspace(
context,
data.build(),
targets,
tasks
)
fun WorkspaceBuilder.data(name: String, node: DataNode<Any>) = data(name.toName(), node)
fun WorkspaceBuilder.data(name: Name, block: DataTreeBuilder<Any>.() -> Unit) {
this.data[name] = DataNode.build(Any::class, block)
}
fun WorkspaceBuilder.data(name: String, block: DataTreeBuilder<Any>.() -> Unit) = data(name.toName(), block)
fun WorkspaceBuilder.target(name: String, block: MetaBuilder.() -> Unit) {
targets[name] = buildMeta(block).seal()
}
/**
* Use existing target as a base updating it with the block
*/
fun WorkspaceBuilder.target(name: String, base: String, block: MetaBuilder.() -> Unit) {
val parentTarget = targets[base] ?: error("Base target with name $base not found")
targets[name] = parentTarget.builder()
.apply { "@baseTarget" to base }
.apply(block)
.seal()
}
fun WorkspaceBuilder.task(task: Task<Any>) {
this.tasks.add(task)
}
/**
* A builder for a simple workspace
*/
class SimpleWorkspaceBuilder(override val parentContext: Context) : WorkspaceBuilder {
override var context: Context = parentContext
override var data = DataTreeBuilder(Any::class)
override var tasks: MutableSet<Task<Any>> = HashSet()
override var targets: MutableMap<String, Meta> = HashMap()
override fun build(): SimpleWorkspace {
return SimpleWorkspace(context, data.build(), targets, tasks)
}
}

92
dataforge-workspace/src/jvmMain/kotlin/hep/dataforge/workspace/GenericTask.kt
View File

@ -75,13 +75,13 @@ class KTaskBuilder(val name: String) {
val to: String = "",
val transform: Workspace.() -> TaskModel.(DataNode<Any>) -> DataNode<Any>
) {
operator fun Workspace.invoke(model: TaskModel, node: DataNode<Any>): DataNode<Any>? {
operator fun invoke(workspace: Workspace, model: TaskModel, node: DataNode<Any>): DataNode<Any>? {
val localData = if (from.isEmpty()) {
node
} else {
node.getNode(from.toName()) ?: return null
}
return transform(this).invoke(model, localData)
return transform(workspace).invoke(model, localData)
}
}
@ -91,14 +91,18 @@ class KTaskBuilder(val name: String) {
this.modelTransform = modelTransform
}
//class TaskEnv(val workspace: Workspace, val model: TaskModel)
fun <T : Any> transform(
inputType: KClass<T>,
from: String = "",
to: String = "",
transform: Workspace.() -> TaskModel.(DataNode<T>) -> DataNode<Any>
) {
dataTransforms += DataTransformation(from, to) { data: DataNode<Any> ->
transform.invoke(this, data.cast(inputType))
dataTransforms += DataTransformation(from, to) {
{ data: DataNode<Any> ->
transform().invoke(this, data.cast(inputType))
}
}
}
@ -113,77 +117,89 @@ class KTaskBuilder(val name: String) {
/**
* Perform given action on data elements in `from` node in input and put the result to `to` node
*/
inline fun <reified T : Any, reified R : Any> action(action: Action<T, R>, from: String = "", to: String = "") {
transform(from, to) { data: DataNode<T> ->
action(data, model.meta)
inline fun <reified T : Any, reified R : Any> action(
from: String = "",
to: String = "",
crossinline actionBuilder: Workspace.() -> Action<T, R>
) {
transform(from, to) {
val res: TaskModel.(DataNode<T>) -> DataNode<Any> = { data: DataNode<T> ->
actionBuilder().invoke(data, meta)
}
res
}
}
inline fun <reified T : Any, reified R : Any> pipeAction(
from: String = "",
to: String = "",
noinline action: PipeBuilder<T, R>.() -> Unit
crossinline block: Workspace.() -> PipeBuilder<T, R>.() -> Unit
) {
val pipe: Action<T, R> = PipeAction(
action(from, to) {
PipeAction(
inputType = T::class,
outputType = R::class,
block = action
block = block()
)
action(pipe, from, to);
}
}
inline fun <reified T : Any, reified R : Any> pipe(
from: String = "",
to: String = "",
noinline action: suspend ActionEnv.(T) -> R
crossinline block: Workspace.() -> suspend ActionEnv.(T) -> R
) {
val pipe: Action<T, R> = PipeAction(
action(from, to) {
PipeAction(
inputType = T::class,
outputType = R::class
) { result(action) }
action(pipe, from, to);
) { result(block()) }
}
}
inline fun <reified T : Any, reified R : Any> joinAction(
from: String = "",
to: String = "",
noinline action: JoinGroupBuilder<T, R>.() -> Unit
crossinline block: Workspace.() -> JoinGroupBuilder<T, R>.() -> Unit
) {
val join: Action<T, R> = JoinAction(
action(from, to) {
JoinAction(
inputType = T::class,
outputType = R::class,
action = action
action = block()
)
action(join, from, to);
}
}
inline fun <reified T : Any, reified R : Any> join(
from: String = "",
to: String = "",
noinline action: suspend ActionEnv.(Map<Name, T>) -> R
crossinline block: Workspace.() -> suspend ActionEnv.(Map<Name, T>) -> R
) {
val join: Action<T, R> = JoinAction(
action(from, to) {
JoinAction(
inputType = T::class,
outputType = R::class,
action = {
result(actionMeta[TaskModel.MODEL_TARGET_KEY]?.string ?: "@anonimous", action)
result(actionMeta[TaskModel.MODEL_TARGET_KEY]?.string ?: "@anonimous", block())
}
)
action(join, from, to);
}
}
inline fun <reified T : Any, reified R : Any> splitAction(
from: String = "",
to: String = "",
noinline action: SplitBuilder<T, R>.() -> Unit
crossinline block: Workspace.() -> SplitBuilder<T, R>.() -> Unit
) {
val split: Action<T, R> = SplitAction(
action(from, to) {
SplitAction(
inputType = T::class,
outputType = R::class,
action = action
action = block()
)
action(split, from, to);
}
}
/**
@ -193,22 +209,24 @@ class KTaskBuilder(val name: String) {
this.descriptor = NodeDescriptor.build(transform)
}
fun build(): GenericTask<Any> {
val transform: Workspace.() -> TaskModel.(DataNode<Any>) -> DataNode<Any> = {
fun build(): GenericTask<Any> =
GenericTask(name, Any::class, descriptor ?: NodeDescriptor.empty(), modelTransform) {
val workspace = this
{ data ->
val model = this
if (dataTransforms.isEmpty()) {
//return data node as is
logger.warn("No transformation present, returning input data")
data
} else {
val builder = DataTreeBuilder(Any::class)
dataTransforms.forEach { transform ->
val res = transform(this, data)
dataTransforms.forEach { transformation ->
val res = transformation(workspace, model, data)
if (res != null) {
if (transform.to.isEmpty()) {
if (transformation.to.isEmpty()) {
builder.update(res)
} else {
builder[transform.to.toName()] = res
builder[transformation.to.toName()] = res
}
}
}
@ -216,10 +234,12 @@ class KTaskBuilder(val name: String) {
}
}
}
return GenericTask(name, Any::class, descriptor ?: NodeDescriptor.empty(), modelTransform, transform);
}
}
fun task(name: String, builder: KTaskBuilder.() -> Unit): GenericTask<Any> {
return KTaskBuilder(name).apply(builder).build();
return KTaskBuilder(name).apply(builder).build()
}
fun WorkspaceBuilder.task(name: String, builder: KTaskBuilder.() -> Unit) {
task(KTaskBuilder(name).apply(builder).build())
}

48
dataforge-workspace/src/jvmTest/kotlin/hep/dataforge/workspace/SimpleWorkspaceTest.kt Normal file
View File

@ -0,0 +1,48 @@
package hep.dataforge.workspace
import hep.dataforge.data.first
import hep.dataforge.data.get
import org.junit.Test
import kotlin.test.assertEquals
class SimpleWorkspaceTest {
val workspace = SimpleWorkspace.build {
repeat(100) {
static("myData[$it]", it)
}
task("square") {
model {
allData()
}
pipe<Int, Int> {
{ data ->
context.logger.info { "Starting square on $data" }
data * data
}
}
}
task("sum"){
model {
dependsOn("square")
}
join<Int,Int> {
{ data ->
context.logger.info { "Starting sum" }
data.values.sum()
}
}
}
}
@Test
fun testWorkspace(){
val node = workspace.run("sum")
val res = node.first()
assertEquals(328350, res.get())
}
}