From e3127503ec863fb5781f6efb4ac93e77aab5658f Mon Sep 17 00:00:00 2001 From: Alexander Nozik Date: Fri, 29 Mar 2019 12:16:24 +0300 Subject: [PATCH] Workspace is working --- .../kotlin/hep/dataforge/data/DataNode.kt | 2 + .../kotlin/hep/dataforge/data/JoinAction.kt | 2 +- .../kotlin/hep/dataforge/data/PipeAction.kt | 2 +- .../kotlin/hep/dataforge/data/SplitAction.kt | 2 +- .../hep/dataforge/workspace/Dependency.kt | 32 +++-- .../dataforge/workspace/SimpleWorkspace.kt | 27 ++++ .../hep/dataforge/workspace/TaskModel.kt | 14 +- .../hep/dataforge/workspace/Workspace.kt | 54 ++++---- .../dataforge/workspace/WorkspaceBuilder.kt | 109 +++++++++++---- .../hep/dataforge/workspace/GenericTask.kt | 124 ++++++++++-------- .../workspace/SimpleWorkspaceTest.kt | 48 +++++++ 11 files changed, 295 insertions(+), 121 deletions(-) create mode 100644 dataforge-workspace/src/commonMain/kotlin/hep/dataforge/workspace/SimpleWorkspace.kt create mode 100644 dataforge-workspace/src/jvmTest/kotlin/hep/dataforge/workspace/SimpleWorkspaceTest.kt diff --git a/dataforge-data/src/commonMain/kotlin/hep/dataforge/data/DataNode.kt b/dataforge-data/src/commonMain/kotlin/hep/dataforge/data/DataNode.kt index b463ba88..6081a13e 100644 --- a/dataforge-data/src/commonMain/kotlin/hep/dataforge/data/DataNode.kt +++ b/dataforge-data/src/commonMain/kotlin/hep/dataforge/data/DataNode.kt @@ -206,4 +206,6 @@ fun DataNode.filter(predicate: (Name, Data) -> Boolean): DataNod } } +fun DataNode.first(): Data = data().first().second + //fun DataNode.filterIsInstance(type: KClass): DataNode = filter{_,data -> type.} \ No newline at end of file diff --git a/dataforge-data/src/jvmMain/kotlin/hep/dataforge/data/JoinAction.kt b/dataforge-data/src/jvmMain/kotlin/hep/dataforge/data/JoinAction.kt index 5529a287..6f2cf3f9 100644 --- a/dataforge-data/src/jvmMain/kotlin/hep/dataforge/data/JoinAction.kt +++ b/dataforge-data/src/jvmMain/kotlin/hep/dataforge/data/JoinAction.kt @@ -16,7 +16,7 @@ class JoinGroup(var name: String, internal val node: DataNode< var meta: MetaBuilder = MetaBuilder() - var result: suspend ActionEnv.(Map) -> R = TODO("Action not implemented") + lateinit var result: suspend ActionEnv.(Map) -> R fun result(f: suspend ActionEnv.(Map) -> R) { this.result = f; diff --git a/dataforge-data/src/jvmMain/kotlin/hep/dataforge/data/PipeAction.kt b/dataforge-data/src/jvmMain/kotlin/hep/dataforge/data/PipeAction.kt index 4e2f9380..268fafec 100644 --- a/dataforge-data/src/jvmMain/kotlin/hep/dataforge/data/PipeAction.kt +++ b/dataforge-data/src/jvmMain/kotlin/hep/dataforge/data/PipeAction.kt @@ -14,7 +14,7 @@ class ActionEnv(val name: Name, val meta: Meta) * Action environment */ class PipeBuilder(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 diff --git a/dataforge-data/src/jvmMain/kotlin/hep/dataforge/data/SplitAction.kt b/dataforge-data/src/jvmMain/kotlin/hep/dataforge/data/SplitAction.kt index 6ffa8881..38740124 100644 --- a/dataforge-data/src/jvmMain/kotlin/hep/dataforge/data/SplitAction.kt +++ b/dataforge-data/src/jvmMain/kotlin/hep/dataforge/data/SplitAction.kt @@ -14,7 +14,7 @@ import kotlin.reflect.full.isSuperclassOf class FragmentRule(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; diff --git a/dataforge-workspace/src/commonMain/kotlin/hep/dataforge/workspace/Dependency.kt b/dataforge-workspace/src/commonMain/kotlin/hep/dataforge/workspace/Dependency.kt index b1f3e90a..1aa4bbb1 100644 --- a/dataforge-workspace/src/commonMain/kotlin/hep/dataforge/workspace/Dependency.kt +++ b/dataforge-workspace/src/commonMain/kotlin/hep/dataforge/workspace/Dependency.kt @@ -19,35 +19,49 @@ sealed class Dependency : MetaRepr { class DataDependency(val filter: DataFilter, val placement: Name = EmptyName) : Dependency() { override fun apply(workspace: Workspace): DataNode { - val result = workspace.data.filter(filter) + val result = workspace.data.filter(filter) return if (placement.isEmpty()) { result } else { - DataNode.build(Any::class){ this[placement] = result } + DataNode.build(Any::class) { this[placement] = result } } } - 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 = 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 { 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 { - DataNode.build(Any::class){ this[placement] = result } + DataNode.build(Any::class) { this[placement] = result } } } override fun toMeta(): Meta = buildMeta { - "name" to name + "task" to name "meta" to meta + "to" to placement } } \ No newline at end of file diff --git a/dataforge-workspace/src/commonMain/kotlin/hep/dataforge/workspace/SimpleWorkspace.kt b/dataforge-workspace/src/commonMain/kotlin/hep/dataforge/workspace/SimpleWorkspace.kt new file mode 100644 index 00000000..16ab8f6f --- /dev/null +++ b/dataforge-workspace/src/commonMain/kotlin/hep/dataforge/workspace/SimpleWorkspace.kt @@ -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, + override val targets: Map, + tasks: Collection> +) : Workspace { + override val tasks: Map> by lazy { + (context.members>(Task.TYPE) + tasks).associate { it.name to it } + } + + companion object { + fun build(parent: Context = Global, block: SimpleWorkspaceBuilder.() -> Unit): SimpleWorkspace = + SimpleWorkspaceBuilder(parent).apply(block).build() + } +} \ No newline at end of file diff --git a/dataforge-workspace/src/commonMain/kotlin/hep/dataforge/workspace/TaskModel.kt b/dataforge-workspace/src/commonMain/kotlin/hep/dataforge/workspace/TaskModel.kt index 25ec2e0f..0e07147c 100644 --- a/dataforge-workspace/src/commonMain/kotlin/hep/dataforge/workspace/TaskModel.kt +++ b/dataforge-workspace/src/commonMain/kotlin/hep/dataforge/workspace/TaskModel.kt @@ -68,15 +68,21 @@ class TaskModelBuilder(val name: String, meta: Meta = EmptyMeta) { var meta: MetaBuilder = meta.builder() val dependencies = HashSet() - var target: String by this.meta.string(key = MODEL_TARGET_KEY,default = "") + var target: String by this.meta.string(key = MODEL_TARGET_KEY, default = "") /** * 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) diff --git a/dataforge-workspace/src/commonMain/kotlin/hep/dataforge/workspace/Workspace.kt b/dataforge-workspace/src/commonMain/kotlin/hep/dataforge/workspace/Workspace.kt index e9a2914d..89af04a2 100644 --- a/dataforge-workspace/src/commonMain/kotlin/hep/dataforge/workspace/Workspace.kt +++ b/dataforge-workspace/src/commonMain/kotlin/hep/dataforge/workspace/Workspace.kt @@ -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 Task.invoke(config: Meta): DataNode { + + /** + * Invoke a task in the workspace utilizing caching if possible + */ + fun run(task: Task, config: Meta): DataNode { 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 Task.invoke(targetName: String): DataNode { - 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 Task.invoke(targetName: String): DataNode { +// 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, - override val targets: Map, - tasks: Collection> -) : Workspace { - - override val tasks: Map> by lazy { - (context.members>(Task.TYPE) + tasks).associate { it.name to it } - } - +fun Workspace.run(task: Task<*>, target: String): DataNode { + 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)) diff --git a/dataforge-workspace/src/commonMain/kotlin/hep/dataforge/workspace/WorkspaceBuilder.kt b/dataforge-workspace/src/commonMain/kotlin/hep/dataforge/workspace/WorkspaceBuilder.kt index 9f8eb08f..99348682 100644 --- a/dataforge-workspace/src/commonMain/kotlin/hep/dataforge/workspace/WorkspaceBuilder.kt +++ b/dataforge-workspace/src/commonMain/kotlin/hep/dataforge/workspace/WorkspaceBuilder.kt @@ -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 + var tasks: MutableSet> + var targets: MutableMap + + 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() - val tasks = HashSet>() +fun WorkspaceBuilder.context(name: String, block: ContextBuilder.() -> Unit) { + context = ContextBuilder(name, parentContext).apply(block).build() +} - fun context(action: ContextBuilder.() -> Unit) { - this.context = ContextBuilder().apply(action).build() +fun WorkspaceBuilder.data(name: Name, data: Data) { + this.data[name] = data +} + +fun WorkspaceBuilder.data(name: String, data: Data) = data(name.toName(), data) + +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) { + this.data[name] = node +} + +fun WorkspaceBuilder.data(name: String, node: DataNode) = data(name.toName(), node) + +fun WorkspaceBuilder.data(name: Name, block: DataTreeBuilder.() -> Unit) { + this.data[name] = DataNode.build(Any::class, block) +} + +fun WorkspaceBuilder.data(name: String, block: DataTreeBuilder.() -> 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) { + 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> = HashSet() + override var targets: MutableMap = HashMap() + + override fun build(): SimpleWorkspace { + return SimpleWorkspace(context, data.build(), targets, tasks) } - - fun data(action: DataTreeBuilder.() -> Unit) = data.apply(action) - - fun target(name: String, meta: Meta) { - targets[name] = meta - } - - fun target(name: String, action: MetaBuilder.() -> Unit) = target(name, buildMeta(action)) - - fun task(task: Task<*>) { - tasks.add(task) - } - - fun build(): Workspace = SimpleWorkspace( - context, - data.build(), - targets, - tasks - ) } \ No newline at end of file diff --git a/dataforge-workspace/src/jvmMain/kotlin/hep/dataforge/workspace/GenericTask.kt b/dataforge-workspace/src/jvmMain/kotlin/hep/dataforge/workspace/GenericTask.kt index e6ecf690..e2236c6a 100644 --- a/dataforge-workspace/src/jvmMain/kotlin/hep/dataforge/workspace/GenericTask.kt +++ b/dataforge-workspace/src/jvmMain/kotlin/hep/dataforge/workspace/GenericTask.kt @@ -75,13 +75,13 @@ class KTaskBuilder(val name: String) { val to: String = "", val transform: Workspace.() -> TaskModel.(DataNode) -> DataNode ) { - operator fun Workspace.invoke(model: TaskModel, node: DataNode): DataNode? { + operator fun invoke(workspace: Workspace, model: TaskModel, node: DataNode): DataNode? { 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 transform( inputType: KClass, from: String = "", to: String = "", transform: Workspace.() -> TaskModel.(DataNode) -> DataNode ) { - dataTransforms += DataTransformation(from, to) { data: DataNode -> - transform.invoke(this, data.cast(inputType)) + dataTransforms += DataTransformation(from, to) { + { data: DataNode -> + 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 action(action: Action, from: String = "", to: String = "") { - transform(from, to) { data: DataNode -> - action(data, model.meta) + inline fun action( + from: String = "", + to: String = "", + crossinline actionBuilder: Workspace.() -> Action + ) { + transform(from, to) { + val res: TaskModel.(DataNode) -> DataNode = { data: DataNode -> + actionBuilder().invoke(data, meta) + } + res } } inline fun pipeAction( from: String = "", to: String = "", - noinline action: PipeBuilder.() -> Unit + crossinline block: Workspace.() -> PipeBuilder.() -> Unit ) { - val pipe: Action = PipeAction( - inputType = T::class, - outputType = R::class, - block = action - ) - action(pipe, from, to); + action(from, to) { + PipeAction( + inputType = T::class, + outputType = R::class, + block = block() + ) + } } inline fun pipe( from: String = "", to: String = "", - noinline action: suspend ActionEnv.(T) -> R + crossinline block: Workspace.() -> suspend ActionEnv.(T) -> R ) { - val pipe: Action = PipeAction( - inputType = T::class, - outputType = R::class - ) { result(action) } - action(pipe, from, to); + action(from, to) { + PipeAction( + inputType = T::class, + outputType = R::class + ) { result(block()) } + } } inline fun joinAction( from: String = "", to: String = "", - noinline action: JoinGroupBuilder.() -> Unit + crossinline block: Workspace.() -> JoinGroupBuilder.() -> Unit ) { - val join: Action = JoinAction( - inputType = T::class, - outputType = R::class, - action = action - ) - action(join, from, to); + action(from, to) { + JoinAction( + inputType = T::class, + outputType = R::class, + action = block() + ) + } } inline fun join( from: String = "", to: String = "", - noinline action: suspend ActionEnv.(Map) -> R + crossinline block: Workspace.() -> suspend ActionEnv.(Map) -> R ) { - val join: Action = JoinAction( - inputType = T::class, - outputType = R::class, - action = { - result(actionMeta[TaskModel.MODEL_TARGET_KEY]?.string ?: "@anonimous", action) - } - ) - action(join, from, to); + action(from, to) { + JoinAction( + inputType = T::class, + outputType = R::class, + action = { + result(actionMeta[TaskModel.MODEL_TARGET_KEY]?.string ?: "@anonimous", block()) + } + ) + } } inline fun splitAction( from: String = "", to: String = "", - noinline action: SplitBuilder.() -> Unit + crossinline block: Workspace.() -> SplitBuilder.() -> Unit ) { - val split: Action = SplitAction( - inputType = T::class, - outputType = R::class, - action = action - ) - action(split, from, to); + action(from, to) { + SplitAction( + inputType = T::class, + outputType = R::class, + action = block() + ) + } } /** @@ -193,22 +209,24 @@ class KTaskBuilder(val name: String) { this.descriptor = NodeDescriptor.build(transform) } - fun build(): GenericTask { - val transform: Workspace.() -> TaskModel.(DataNode) -> DataNode = { + fun build(): GenericTask = + 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 { - 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()) } \ No newline at end of file diff --git a/dataforge-workspace/src/jvmTest/kotlin/hep/dataforge/workspace/SimpleWorkspaceTest.kt b/dataforge-workspace/src/jvmTest/kotlin/hep/dataforge/workspace/SimpleWorkspaceTest.kt new file mode 100644 index 00000000..62b72505 --- /dev/null +++ b/dataforge-workspace/src/jvmTest/kotlin/hep/dataforge/workspace/SimpleWorkspaceTest.kt @@ -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 { + { data -> + context.logger.info { "Starting square on $data" } + data * data + } + } + } + + task("sum"){ + model { + dependsOn("square") + } + join { + { 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()) + } +} \ No newline at end of file