diff --git a/docs/wiki/config/config.md b/docs/wiki/config/config.md index cd027ae..7275dd7 100644 --- a/docs/wiki/config/config.md +++ b/docs/wiki/config/config.md @@ -3,7 +3,7 @@ ## Simulation * `duration`:`int` - number of simulation ticks, after which everything will be reset to the initial configuration -* `tickfactor`:`int` - used to artificially add ticks by repeating the result of a simulation tick by the given value +* `tickfactor`:`int` - used to artificially add ticks by repeating the calculation of a simulation tick by the given value * `loop`:`bool` - loop the simulation * `loopcount`:`int` - amount of loops, `0` for infinite @@ -59,7 +59,7 @@ Elements: * `end`:`int` _[optional]_ - last simulation tick before this function is inactive Alternative condensed of `timeline` notation as `array`: `[start, end]` Alternative condensed of `value` notation as `array`: `[start, value or None, ...]` -* `delta`:`array` _[optional]_ - function to calculate `value` of the next simulation tick, only called if no value is defined for the next tick. Result is saved as new `value` and passed again to the function for the next tick. If no function is given, the `value` remains +* `delta`:`array` _[optional]_ - function to calculate `value` of the next simulation tick, overwrites any existing value. Result is saved as new `value` and passed again to the function for the next tick. If no function is given, the `value` remains * `name`:`String` - name of the function. Needs to exist at compile time in a map: `["name" -> (value: double, parentValue: double, args: ...) => double]` * `args`:`array` - arguments passed to the function in addition to the `value` * `timeline`:`array` - sets simulation tick when this function is active and turns inactive. diff --git a/docs/wiki/config/config.yml b/docs/wiki/config/config.yml index db3e2c8..0389a10 100644 --- a/docs/wiki/config/config.yml +++ b/docs/wiki/config/config.yml @@ -1,8 +1,14 @@ simulation: duration: 500 tickfactor: 10 - loop: true - loopcount: 20 + runs: 20 + parCell: 0 + parCellThreads: 5 + parGroup: 0 + parGroupThreads: 5 + parSensor: 0 + parSensorThreads: 5 + sensors: - group: temperature @@ -16,27 +22,27 @@ sensors: end: [3, 3] count: 30 -groundtruth: + - group: faultyTemp + parent: temperature + cells: + - [1, 2, 10] + +groundtruths: - group: temperature truth: - - value: 11 + - value: 11.0 timeline: - start: 0 end: 120 - [240, 360] delta: - name: factor - args: - - 1.1 + args: {factor: 1.1} timeline: - - start: 60 - end: 120 - cells: - - [0, 2, 10] - blocks: - - start: [8, 1] - end: [3, 3] - count: 50 + - start: 20 + end: 95 + - [170, randomValue, {arg: true}] + - group: humidity parent: temperature truth: @@ -44,27 +50,18 @@ groundtruth: - [30, 0.6] - [50, None] delta: - - [10, None, []] - - [30, addValuePercent, [0.03, 0.8]] - - [50, humidFromTemp, []] - blocks: - - start: [15, 15] - end: [17, 17] - percent: 0.8 + - [10, None] + - [30, addValuePercent, {jitterA: 0.03, jitterB: 0.8}] + - [50, humidFromTemp] -behaviour: +behaviours: - group: temperature sharedObject: objectTemp functions: - name: randomValue - args: - - 40 - - 80 - duration: - - start: 60 - - [100, normalValue, [2]] - - [120, None, []] - blocks: - - start: [5,6] - end: [8,9] - percent: 1.0 \ No newline at end of file + args: {a: 40, b: 80} + timeline: + - start: 44 + end: 166 + - [100, normalValue, {v: 2}] + - [120, addValuePercent, {}] \ No newline at end of file diff --git a/src/main/scala/scim/Library.scala b/src/main/scala/scim/Library.scala index 94e15d8..5f135c4 100644 --- a/src/main/scala/scim/Library.scala +++ b/src/main/scala/scim/Library.scala @@ -4,6 +4,7 @@ package lib // java imports import java.io.{File, FileInputStream} + // scala imports import scala.collection.convert.wrapAll._ import scala.collection.mutable @@ -13,13 +14,13 @@ import scala.collection.JavaConverters._ import org.yaml.snakeyaml.Yaml package object printUtil { - def indentCollection(s: String) : String = { + def indentCollection(s: String): String = { var indent = 0 var skip = false var bracket = false var output = "" - def closureIndentCollection(c: Char) : Unit = { + def closureIndentCollection(c: Char): Unit = { if(c == '(') { indent += 1 output += "(\n" + "\t" * indent @@ -44,8 +45,39 @@ package object printUtil { } } +package object evalUtil { + def time[A](name: String, f: => A): A = { + val t0 = System.nanoTime() + val result = f + val t1 = System.nanoTime() + println(name + " Elapsed time: " + (t1 - t0) + "ns") + result + } +} + +package object debugUtil { + sealed case class DebugType(name: String, level: Int) + + object DebugType { + object ERROR extends DebugType("ERROR", 0) + object WARNING extends DebugType("WARNING", 1) + object MESSAGE extends DebugType("MESSAGE", 2) + val values = Seq(ERROR, WARNING, MESSAGE) + } + val level: Int = DebugType.ERROR.level + + def currentMethodName: String = Thread.currentThread.getStackTrace()(2).getMethodName + def currentMethodCallerName: String = Thread.currentThread.getStackTrace()(3).getMethodName + private def debugPrefix(dbg: DebugType): String = { + dbg.name + " [" + Thread.currentThread.getStackTrace()(4).getMethodName + "]: " + } + def debug(dbg: DebugType, x: Any): Unit = if(level <= dbg.level) print(debugPrefix(dbg) + x) + + def debugf(dbg: DebugType, text: String, xs: Any*): Unit = if(level <= dbg.level) printf(debugPrefix(dbg) + text, xs: _*) +} + package object convertUtil { - def javaArrayListToSeq(javaArray: Any) : Option[Seq[Any]] = { + def javaArrayListToSeq(javaArray: Any): Option[Seq[Any]] = { javaArray match { /* While LinearSeq(List) for array.length == 2 is faster than than IndexedSeq(Vector) when accessed via tail/head * this would convert a lot of elements to List that are accessed via foreach and not tail/head lookup @@ -61,13 +93,19 @@ package object convertUtil { package object mapUtil { - def javaHashMapToScalaMap[K, V](javaMap: java.util.LinkedHashMap[_, _]) : Map[K, V] = { + def traversableToMap[A, B](traversable: Traversable[Any], getTuple: (Any) => (A, B)): Map[A, B] = { + // Traversable[Tuple[A, B]] can be converted via .map + // .foldLeft used in favor of .map().toMap as the latter needs two traversals + // traversable.map(e => getKey(e) -> getValue(e)).toMap + traversable.foldLeft(Map[A, B]()) { (m, e) => m + getTuple(e) } + } + def javaHashMapToScalaMap[K, V](javaMap: java.util.LinkedHashMap[_, _]): Map[K, V] = { mapAsScalaMap(javaMap).toMap.asInstanceOf[Map[K, V]] } - def anyToNestedMap[K](map: Any, depthLimit: Int = 100) : Map[K, Any] = { + def anyToNestedMap[K](map: Any, depthLimit: Int = 100): Map[K, Any] = { - def closureAnyToNestedMap(nestedValue: Any, depth: Int = 0) : Any = { + def closureAnyToNestedMap(nestedValue: Any, depth: Int = 0): Any = { if(depth > depthLimit) return nestedValue // crude way to avoid a circular map anyToMap[K, Any](nestedValue) match { case Some(nestedMap) => nestedMap.map(kv => kv._1 -> closureAnyToNestedMap(kv._2, depth + 1)) @@ -85,7 +123,7 @@ package object mapUtil { } } - def anyToMutableMap[K, V](mapAny: Any) : Option[mutable.Map[K, V]] = { + def anyToMutableMap[K, V](mapAny: Any): Option[mutable.Map[K, V]] = { // TODO: use TypeTag to work around type erasure to check for T[K, V] // http://squidarth.com/scala/types/2019/01/11/type-erasure-scala.html mapAny match { @@ -96,7 +134,7 @@ package object mapUtil { } } - def anyToMap[K, V](mapAny: Any) : Option[Map[K, V]] = { + def anyToMap[K, V](mapAny: Any): Option[Map[K, V]] = { // TODO: use TypeTag to work around type erasure to check for T[K, V] // http://squidarth.com/scala/types/2019/01/11/type-erasure-scala.html mapAny match { @@ -112,9 +150,14 @@ package object mapUtil { package object yml { val yaml = new Yaml // TODO: handle errors (invalid file/format) - def parse(file:String) : Map[String, Any] = { + def parse(file:String): Map[String, Any] = { val fis = new FileInputStream(new File(file)) mapUtil.anyToNestedMap[String](yaml.load(fis)) } } +package object simulation { + import scim.datastruct._ + // maybe change to by-name parameters? have to check performance gain + type SimulationBehaviour = (Cell, Time, Option[Double], Option[FunctionArgs]) => Option[Double] +} \ No newline at end of file diff --git a/src/main/scala/scim/Main.scala b/src/main/scala/scim/Main.scala index cc0462f..6516877 100644 --- a/src/main/scala/scim/Main.scala +++ b/src/main/scala/scim/Main.scala @@ -1,12 +1,48 @@ package scim +import scim.components.{Parser, SimulationConfigParser, Output, Simulation} +import scim.datastruct.{Cell, FunctionArgs, SharedMemory, SimulationBehaviours, Time} import scim.lib._ - +import scim.lib.simulation.SimulationBehaviour +//package scim.components._ object Main { + + val func: (Cell, Time, Option[Double], Option[FunctionArgs]) => Option[Double] = (cell, time, truth, args) => { truth } + val behaviours: Map[String, SimulationBehaviour] = Map[String, SimulationBehaviour]( + "factor" -> func, + "None" -> func, "addValuePercent" -> func, + "humidFromTemp" -> func, + "normalValue" -> func, + "randomValue" -> func) + val simBehavirous = SimulationBehaviours(behaviours) + def main(args: Array[String]): Unit = { val config = "docs/wiki/config/config.yml" val list: Map[String, Any] = yml.parse(config) - println(printUtil.indentCollection(list.toString()).toString) + new SimulationConfigParser(list, simBehavirous).parse() + // println(printUtil.indentCollection(list.toString()).toString) + //new Test2().run() + val parlist: Map[String, Int] = Map("a" -> 1, "b" -> 1, "c" -> 1, "d" -> 1) + val v: Vector[Int] = Vector(1,2,3,4,5,6,7) + + //lib.evalUtil.time("par", parlist.par.map({_ => Thread.sleep(10); 2})) + //lib.evalUtil.time("ser", parlist.map({_ => Thread.sleep(10); 2})) + + } + + def startSimulation(): Unit = { + val config = new Parser().parse("docs/wiki/config/config.yml", simBehavirous).get + val sharedMemory = config.sharedMemory + val output = new Output(sharedMemory) + val outputThread = new Thread(output) + val simulator = new Simulation(config) + + for(_ <- 0 to sharedMemory.config.runs) { + outputThread.start() + simulator.start() + outputThread.join() + } + } } \ No newline at end of file diff --git a/src/main/scala/scim/archive/Elements.scala b/src/main/scala/scim/archive/Elements.scala new file mode 100644 index 0000000..edb8b7e --- /dev/null +++ b/src/main/scala/scim/archive/Elements.scala @@ -0,0 +1,19 @@ +package scim.archive + +// child elements level 2 +case class Coordinate(x: Int, y: Int) +case class FunctionArgs(args: Map[String, Any]) +case class FunctionId(id: Int, name: String) +case class TimeLine(start: Int, end: Option[Int]) + +// child elements level 1 +case class Group(id: Int, name: String) +case class Block(start: Coordinate, end: Coordinate, count: Int) +case class Truth(value: Option[Double], timeline: TimeLine) +case class Function(function: FunctionId, args: Option[FunctionArgs], timeLine: TimeLine) + +// root elements +case class Simulation(duration: Int, tickfactor: Int, loop: Boolean, loopcount: Int) +case class Sensor(group: Group, blocks: Vector[Block]) +case class GroundTruth(group: Group, parent: Option[GroundTruth], truths: Vector[Truth], deltas: Option[Vector[Function]]) +case class Behaviour(group: Group, parent: Option[Behaviour], functions: Option[Vector[Function]]) diff --git a/src/main/scala/scim/archive/GroundTruth.scala b/src/main/scala/scim/archive/GroundTruth.scala new file mode 100644 index 0000000..258472e --- /dev/null +++ b/src/main/scala/scim/archive/GroundTruth.scala @@ -0,0 +1,27 @@ +package scim.archive + +case class Handler(config: Configuration, data: Data) +case class BufferHandler() { + +} +case class Data() { + //val buffer = Map[Position, (BufferHandler, ArraySeq[(Time, Truth)])]() +} + + +//object SimBuffer { +// def apply[A](v: A*): SimBuffer[A] = new SimBuffer(v) +// def apply[A](): SimBuffer[A] = new SimBuffer[A]() +//} + +//class SimBuffer[A] (buffer: ArraySeq[A], handler: BufferHandler) { +// var bufferd: Int = 1 +// def get(time: Time): A = { +// +// } +//} +case class Configuration() + +case class GroundTruth(handler: Handler, config: Configuration, data: Data) + +case class GroundTruths(groundTruths: Map[Group, GroundTruth]) \ No newline at end of file diff --git a/src/main/scala/scim/archive/Simulation.scala b/src/main/scala/scim/archive/Simulation.scala new file mode 100644 index 0000000..74c6983 --- /dev/null +++ b/src/main/scala/scim/archive/Simulation.scala @@ -0,0 +1,21 @@ +package scim.datastruct + +import scala.collection.mutable + +//case class Coordinate(x: Int, y: Int) +//case class World(size: Int) // implements trait def getPosition(coord: Coordinate): Position; def getCoordinate(pos: Position): Coordinate +//case class Position(pos: Int) +//case class Time(tick: Int) +//case class Truth(truth: Option[Double]) +//case class FunctionArgs(args: Map[String, Any]) +//case class Function(function: (Truth, Position, Time, FunctionArgs) => Option[Double], args: FunctionArgs) +//case class Group(id: Int, name: String) +// +//case class GroundTruths(truth: Map[Group, mutable.ArraySeq[(Time, Truth)]]) // implements trait def getTruth(group: Group, pos: Position): Option[Double] +//case class Result(truth: Map[Group, List[(Time, Truth)]]) // implements trait def getTruth(group: Group, pos: Position): Option[Double] +//case class Behaviours(function: Map[Group, List[(Time, Option[Function])]]) // implement trait def run(truth: Truth, pos: Position, time: Time): Option[Double] +//case class Sensor(group: Group, count: Int) +//case class Cells(cells: Map[Position, Vector[Sensor]]) +//case class Sensors(sensors: Map[Group, Vector[(Position, Int)]]) // implement trait def sim(time: Time): Option[Double] +//case class Simulation(groundTruths: GroundTruths, behaviours: Behaviours, cells: Cells) + diff --git a/src/main/scala/scim/components/Output.scala b/src/main/scala/scim/components/Output.scala new file mode 100644 index 0000000..39abb0c --- /dev/null +++ b/src/main/scala/scim/components/Output.scala @@ -0,0 +1,13 @@ +package scim.components + +import scim.datastruct.SharedMemory + +class Output(sharedMemory: SharedMemory) extends Runnable { + def run(): Unit = { + outputData() + } + + def outputData(): Unit = { + + } +} diff --git a/src/main/scala/scim/components/Parser.scala b/src/main/scala/scim/components/Parser.scala new file mode 100644 index 0000000..b3c39f8 --- /dev/null +++ b/src/main/scala/scim/components/Parser.scala @@ -0,0 +1,226 @@ +package scim.components + +import scim.datastruct.{Behaviour, BehaviourVector, Cell, Configuration, FunctionArgs, Group, SensorCount, SharedMemory, SimulationBehaviours, SimulationConfig, Time, Truth} +import scim.lib.ConcurrentBroadcast.BroadcastObject +import scim.lib.debugUtil.{DebugType, debugf} +import scim.lib.mapUtil.traversableToMap +import scim.lib.simulation.SimulationBehaviour +import scim.lib.yml + +import scala.collection.mutable + +class Parser() { + def parse(file: String, simulationBehaviours: SimulationBehaviours): Option[Configuration] = { + val parser = new SimulationConfigParser(yml.parse(file), simulationBehaviours) + parser.parse() + None // FIXME: this should return something + } + +} + + +// when creating (Group, Sensor, Option[GroundTruth]), keep ref in Map[Group, Sensor, Option[GroundTruth]] for all, to find +// root group via group.parent (while !isempty get next) (maybe tree instead) + +// for cells s-> create Map[Group, Map[Cell, Count]] +// for gt g-> create Map[Group, GroundTruth] +// for behaviour b-> create Map[Group, Behaviour] +// sensors = s.map(kv => (kv._1, kv._2.map(kv2 => (kv2._1, SENSOR, TRUTH))) +// sensors.foreach(kv => kv._1 + +class SimulationConfigParser(input: Map[String, Any], simulationBehaviours: SimulationBehaviours) { + private val config: SimulationConfig = parseSimulationConfig(input("simulation").asInstanceOf[Map[String, Any]]) + private val groups = mutable.Map[String, Group]().empty + private val sensorList = mutable.Map[Group, Map[Cell, SensorCount]]().empty + private val behaviourListSensor = mutable.Map[Group, Behaviour]().empty + private val behaviourListGroundTruth = mutable.Map[Group, Behaviour]().empty + private val groundTruthValues = mutable.Map[Time, mutable.Map[Group, Truth]]() // FIXME: are not parsed correctly + // this will throw if the map doesn't contain what it should + @throws(classOf[Exception]) + def parse(): Unit = { + + val sensors = input("sensors").asInstanceOf[Vector[Map[String, Any]]] + val groundTruths = input("groundtruths").asInstanceOf[Vector[Map[String, Any]]] + val behaviours = input("behaviours").asInstanceOf[Vector[Map[String, Any]]] + + sensors.foreach(sensor => parseSensor(sensor)) + behaviours.foreach(behaviour => parseBehaviour(behaviour)) + groundTruths.foreach(groundTruth => parseGroundTruth(groundTruth)) + // TODO: insert dummy sensors if none exists at cells of the parent group where this group exists + Unit + } + + private def parseSimulationConfig(configInput: Map[String, Any]): SimulationConfig = { + val config = configInput.asInstanceOf[Map[String, Int]] + val duration = config("duration") + val runs = config("runs") + val tickfactor = if(config.contains("tickfactor")) config("tickfactor") else 1 + val parCell = if(config.contains("parCell")) config("parCell") else 0 + val parCellThreads = if(config.contains("parCellThreads")) config("parCellThreads") else 5 + val parGroup = if(config.contains("parGroup")) config("parGroup") else 0 + val parGroupThreads = if(config.contains("parGroupThreads")) config("parGroupThreads") else 5 + val parSensor = if(config.contains("parSensor")) config("parSensor") else 0 + val parSensorThreads = if(config.contains("parSensorThreads")) config("parSensorThreads") else 5 + SimulationConfig(duration, runs, tickfactor, parCell, parCellThreads, parGroup, parGroupThreads, parSensor, parSensorThreads) + } + + private def parseSensor(sensor: Map[String, Any]): Unit = { + val groupString = sensor("group").asInstanceOf[String] + val parentString: Option[String] = if(sensor.contains("parent")) Some(sensor("parent").asInstanceOf[String]) else None + val cells: Map[Cell, SensorCount] = parseCells(sensor("cells").asInstanceOf[Vector[Any]]) + val group = Group(groupString, parentString) + if(!groups.contains(groupString)) { + groups(groupString) = group + sensorList(group) = cells + } else { + debugf(DebugType.WARNING, "Duplicate Sensors in Group [%s]. Ignoring Sensor: %s", groupString, sensor.toString()) + } + } + + private def parseCells(cells: Vector[Any]): Map[Cell, SensorCount] = { + val parseCell = (cell: Any) => cell match { + case cell: Vector[_] => val cellVector = cell.asInstanceOf[Vector[Int]] + (Cell(cellVector(0), cellVector(1)), SensorCount(cellVector(2))) + case cell: Map[_, _] => val cellMap = cell.asInstanceOf[Map[String, Int]] + (Cell(cellMap("x"), cellMap("y")), SensorCount(cellMap("count"))) + case _ => throw new Exception("Invalid Cell: " + cell.toString) + } + traversableToMap[Cell, SensorCount](cells, parseCell) + } + + private def parseTimeLineElement(timeLineElement: Any): (Int, Option[Int]) = { + timeLineElement match { + case timeLineElement: Vector[_] => val elementVector = timeLineElement.asInstanceOf[Vector[Int]] + (elementVector(1), if(timeLineElement.contains(2)) Some(elementVector(2)) else None) + case timeLineElement: Map[_, _] => val elementMap = timeLineElement.asInstanceOf[Map[String, Int]] + (elementMap("start"), if(elementMap.contains("end")) Some(elementMap("end")) else None) + } + } + + private def parseFunction(functions: Vector[Any]): Behaviour = { + val parseFunctionHelper = (function: Any) => { + var funcPointer: Option[(SimulationBehaviour, Option[FunctionArgs])] = None + function match { + case function: Vector[_] => { + val start: Int = function(0).asInstanceOf[Int] + // val end: Option[Int] = None // Not used + val name: String = function(1).asInstanceOf[String] + val args: Option[FunctionArgs] = if (function.contains(3)) Some(FunctionArgs(function(2).asInstanceOf[Map[String, Any]])) else None + if (!simulationBehaviours.get(name).isEmpty) funcPointer = Some((simulationBehaviours.get(name).get, args)) + Map[Int, Option[(SimulationBehaviour, Option[FunctionArgs])]](start -> funcPointer) + } + case function: Map[_, _] => { + val functionMap = function.asInstanceOf[Map[String, Any]] + val timeline: Vector[Any] = functionMap("timeline").asInstanceOf[Vector[Any]] + val timeLineVector: Vector[(Int, Option[Int])] = timeline.map(e => parseTimeLineElement(e)) + val name: String = functionMap("name").asInstanceOf[String] + val args: Option[FunctionArgs] = if (functionMap.contains("args")) Some(FunctionArgs(functionMap("args").asInstanceOf[Map[String, Any]])) else None + if (!simulationBehaviours.get(name).isEmpty) funcPointer = Some((simulationBehaviours.get(name).get, args)) + val output: mutable.Map[Int, Option[(SimulationBehaviour, Option[FunctionArgs])]] = mutable.Map() + timeLineVector.map(tuple => { + output(tuple._1) = funcPointer + if (!tuple._2.isEmpty) output(tuple._2.get) = None + }) + output.toMap + } + case _ => throw new Exception("Invalid Function: " + function.toString) + } + } + val parsedFunction: mutable.Map[Int, Option[(SimulationBehaviour, Option[FunctionArgs])]] = mutable.Map().empty + // TODO: This will currently overwrite already assigned values; output warning + functions.map(function => parseFunctionHelper(function)).foreach(e => e.foreach(kv => parsedFunction(kv._1) = kv._2)) + val timeVector: mutable.ArraySeq[Option[(SimulationBehaviour, Option[FunctionArgs])]] = new mutable.ArraySeq[Option[(SimulationBehaviour, Option[FunctionArgs])]](config.duration).map(_ => None) + val it = parsedFunction.toList.sortBy(_._1).iterator + println(parsedFunction.toList.sortBy(_._1)) + var current = if(it.hasNext) Some(it.next) else None + while(!current.isEmpty) { + val next = if(it.hasNext) Some(it.next) else None + for(time <- current.get._1 to (if(!next.isEmpty) next.get._1 else config.duration - 1)) { + timeVector(time) = current.get._2 + } + current = next + } + Behaviour(BehaviourVector(timeVector.toVector)) + } + + private def parseBehaviour(behaviourMap: Map[String, Any]): Unit = { + val groupString = behaviourMap("group").asInstanceOf[String] + // only continue parsing if sensors in this group exist + if(groups.contains(groupString)) { + val group: Group = groups(groupString) + if(!behaviourListSensor.contains(group)) { + behaviourListSensor(group) = parseFunction(behaviourMap("functions").asInstanceOf[Vector[Any]]) + } else { + debugf(DebugType.WARNING, "Duplicate Sensor Behaviour in Group [%s]. Ignoring Behaviour: %s", groupString, behaviourMap.toString()) + } + } else { + debugf(DebugType.WARNING, "Unused Sensor Behaviour, no sensor for group [%s] exists. Ignoring Behaviour: %s", groupString, behaviourMap.toString()) + } + } + + private def parseGroundTruth(groundTruthMap: Map[String, Any]): Unit = { + val groupString = groundTruthMap("group").asInstanceOf[String] + // only continue parsing if sensors in this group exist + if(groups.contains(groupString)) { + val group: Group = groups(groupString) + if(group.parent.isEmpty) { + if(!behaviourListGroundTruth.contains(group)) { + group.parent = if(groundTruthMap.contains("parent")) Some(groundTruthMap("parent").asInstanceOf[String]) else None + behaviourListGroundTruth(group) = parseFunction(groundTruthMap("delta").asInstanceOf[Vector[Any]]) + } else { + debugf(DebugType.WARNING, "Duplicate Ground Truth in Group [%s]. Ignoring Ground Truth: %s", groupString, groundTruthMap.toString()) + } + parseTruth(groundTruthMap("truth").asInstanceOf[Vector[Any]]).map(kv => { + if(groundTruthValues.contains(kv._1)) { + groundTruthValues(kv._1)(group) = kv._2 + } else { + groundTruthValues(kv._1) = mutable.Map[Group, Truth](group -> kv._2) + } + }) + } else { + debugf(DebugType.WARNING, "Unused Ground Truth in Group [%s] because the Sensor has Parent [%s]. Ignoring Ground Truth: %s", groupString, group.parent.get, groundTruthMap.toString()) + } + } else { + debugf(DebugType.WARNING, "Unused Ground Truth, no sensor for group [%s] exists. Ignoring Ground Truth: %s", groupString, groundTruthMap.toString()) + } + } + + private def parseTruth(truthElements: Vector[Any]): Map[Time, Truth] = { + val parseTruthHelper = (truthElement: Any) => { + truthElement match { + case truthObject: Vector[_] => { + val truth: Truth = Truth(new BroadcastObject()) + println(truthElement) + val start: Int = truthObject(0).asInstanceOf[Int] + // val end: Option[Int] = None // Not used + truth.set(Some(truthObject(1).asInstanceOf[Double])) + Map[Time, Truth](Time(start) -> truth) + } + case truthObject: Map[_, _] => { + val truth: Truth = Truth(new BroadcastObject()) + lazy val endTruth: Truth = Truth(new BroadcastObject()) + val truthObjectMap = truthObject.asInstanceOf[Map[String, Any]] + val timeline: Vector[Any] = truthObjectMap("timeline").asInstanceOf[Vector[Any]] + val timeLineVector: Vector[(Int, Option[Int])] = timeline.map(e => parseTimeLineElement(e)) + truth.set(Some(truthObjectMap("value").asInstanceOf[Double])) + + val output: mutable.Map[Time, Truth] = mutable.Map() + timeLineVector.map(tuple => { + output(Time(tuple._1)) = truth + if (!tuple._2.isEmpty) { + endTruth.set(None) + output(Time(tuple._2.get)) = endTruth + } + }) + output.toMap + } + case _ => throw new Exception("Invalid Truth: " + truthElement.toString) + } + } + val parsedTruth: mutable.Map[Time,Truth] = mutable.Map().empty + // TODO: This will currently overwrite already assigned values; output warning + truthElements.map(truth => parseTruthHelper(truth)).foreach(e => e.foreach(kv => parsedTruth(kv._1) = kv._2)) + parsedTruth.toMap + } + +} \ No newline at end of file diff --git a/src/main/scala/scim/components/Simulation.scala b/src/main/scala/scim/components/Simulation.scala new file mode 100644 index 0000000..5914e21 --- /dev/null +++ b/src/main/scala/scim/components/Simulation.scala @@ -0,0 +1,56 @@ +package scim.components + +import scim.datastruct.{Behaviour, BehaviourVector, Cell, Configuration, FunctionArgs, GroundTruth, Group, Sensor, Time} +import scim.lib.simulation.SimulationBehaviour + +class Simulation(config: Configuration) { + private val sharedMemory = config.sharedMemory + private val cells = config.cells + val timeVector = Vector.tabulate(sharedMemory.config.duration)(t => Time(t)) + + def start(): Unit = { + if(sharedMemory.config.parCell != 0 && cells.size >= sharedMemory.config.parCell ) { + val cellList = cells.par + cellList.tasksupport = sharedMemory.config.cellThreadPool + timeVector.foreach(time => cellList.foreach(cell => simGroups(time, cell._1, cell._2))) + } else { + timeVector.foreach(time => cells.foreach(cell => simGroups(time, cell._1, cell._2))) + } + } + + private def simGroups(time: Time, cell: Cell, groups: Map[Group, List[(Group, Sensor, Option[GroundTruth])]]): Unit = { + // TODO: possible optimisation (see text below) + // this can be further optimised as a child is only dependent on its parents ground truth and could run + // at the same time as the parent in a different thread. This would allow the child to start running as soon + // as the ground truth is calculated in the parent and doesn't have have to wait for the sensors to finish. + // Any subsequent children can start in the same manner, when their parent is done with the ground truth. + // -- + // other possibility: + // run simulateGT foreach first, then run simulateSensor foreach (dont know if there is much benefit tho) + if(sharedMemory.config.parGroup != 0 && cells.size >= sharedMemory.config.parGroup ) { + val groupList = groups.par + groupList.tasksupport = sharedMemory.config.groupThreadPool + groupList.foreach(group => group._2.foreach(element => simulate(time, cell, element))) + } else { + groups.foreach(group => group._2.foreach(element => simulate(time, cell, element))) + } + } + + private def simulate(time: Time, cell: Cell, element: (Group, Sensor, Option[GroundTruth])): Unit = { + val group = element._1 + val sensor = element._2 + val groundTruth = element._3 + // simulate the ground truth if required + if(!groundTruth.isEmpty) { + groundTruth.get.simulate(cell,group, time, sharedMemory) + } else if (group.parent.isEmpty) { + // calc default ground truth if none was defined and no parent given which is None + // this ensures a value when the sensor accesses the gt and prevents a deadlock + val v: Vector[Option[(SimulationBehaviour, Option[FunctionArgs])]] = Vector.fill(sharedMemory.config.duration)(None) + GroundTruth(Some(Behaviour(BehaviourVector(v)))).simulate(cell, group, time, sharedMemory) + } + // simulate the sensor + sensor.simulate(cell, group, time, sharedMemory) + } + +} \ No newline at end of file diff --git a/src/main/scala/scim/datastruct/Configuration.scala b/src/main/scala/scim/datastruct/Configuration.scala new file mode 100644 index 0000000..c18daf8 --- /dev/null +++ b/src/main/scala/scim/datastruct/Configuration.scala @@ -0,0 +1,132 @@ +package scim.datastruct + +import scim.lib.ConcurrentBroadcast.BroadcastObject +import scim.lib.simulation._ +import scala.collection.parallel.ForkJoinTaskSupport + +// configuration +//case class Configuration(simulation: Simulation, sensors: Vector[Sensor], groundTruths: Vector[GroundTruth], behaviours: Vector[Behaviour]) + +case class Time(time: Int){ + def prev: Time = { Time(time - 1) } + def start: Boolean = { time == 0 } + def get: Int = { time } +} + +case class SharedMemory( sensor: Map[Time, Map[SensorID, SensorResult]], + groundTruth: Map[Cell, Map[Time, Map[Group, Truth]]], + config: SimulationConfig) { + def getTruth(cell: Cell, time: Time, group: Group): Truth = { + groundTruth(cell)(time)(group) + } + def getResult(time: Time): Map[SensorID, SensorResult] = { + sensor(time) + } +} + +abstract class BroadcastHandler[A](broadcastObject: BroadcastObject[A]) { + def get: A = { + broadcastObject.watch() + } + def set(newValue: A): Unit = { + broadcastObject.broadcast(newValue) + } + def exists: Boolean = { + !broadcastObject.peak().isEmpty + } +} +case class SimulationBehaviours(functions: Map[String, SimulationBehaviour]) { + def get(name: String): Option[SimulationBehaviour] = { + if(functions.contains(name)) Some(functions(name)) else None + } +} + +case class SimulationConfig(duration: Int, runs: Int, tickfactor: Int = 1, + parCell: Int = 0, parCellThreads: Int = 5, + parGroup: Int = 0, parGroupThreads: Int = 5, + parSensor: Int = 0, parSensorThreads: Int = 5) { + val cellThreadPool: ForkJoinTaskSupport = new ForkJoinTaskSupport(new scala.concurrent.forkjoin.ForkJoinPool(parCellThreads)) + val groupThreadPool: ForkJoinTaskSupport = new ForkJoinTaskSupport(new scala.concurrent.forkjoin.ForkJoinPool(parGroupThreads)) + val sensorThreadPool: ForkJoinTaskSupport = new ForkJoinTaskSupport(new scala.concurrent.forkjoin.ForkJoinPool(parSensorThreads)) + // TODO: give truth their own threadpool and parallize the calculation +} +case class SensorResult(result: BroadcastObject[(Group, Cell, Option[Double])]) extends BroadcastHandler(result) +case class SensorID(group: Group, Id: Int) +case class Duration(simDuration: Int) +case class TickFactor(factor: Int) // processed when parsing configuration, artifically place timings +case class Cell(x: Int, y: Int) +case class Group(name: String, parentIn: Option[String] = None){ + // FIXME: use parent as get function and parent.set as setter to make var private + var parent: Option[String] = parentIn // have to make this mutable to reassign it during parsing + def getParentOrThis: Group = { + if(parent.isEmpty) { + this + } else { + Group(parent.get) + } + } + +} +case class Truth(value: BroadcastObject[Option[Double]]) extends BroadcastHandler(value) +case class FunctionArgs(args: Map[String, Any]) +// passing the group and therefore the behaviour during the simulation adds additional processing costs during runtime. +// Benefit would be fewer sensor objects. + +case class BehaviourVector(function: Vector[Option[(SimulationBehaviour, Option[FunctionArgs])]], + defaultFunc: SimulationBehaviour = (_: Cell, _: Time, v: Option[Double], _: Option[FunctionArgs]) => v, + defaultArgs: Option[FunctionArgs] = None){ + def get(t: Time): (SimulationBehaviour, Option[FunctionArgs]) = { + function(t.time).getOrElse((defaultFunc, defaultArgs)) + } +} + +case class Behaviour(behaviour: BehaviourVector){ + def run(time: Time, cell: Cell, truth: Truth): Option[Double] = { + val functionTuple = behaviour.get(time) + val function = functionTuple._1 + val args = functionTuple._2 + // TODO: add try catch, to prevent crash in case the args are wrong + function(cell, time, truth.get, args) + } +} + +case class GroundTruth(behaviour: Option[Behaviour]) { + def simulate(cell: Cell, group: Group, time: Time, sharedMemory: SharedMemory): Unit = { + val result = sharedMemory.getTruth(cell, time, group.getParentOrThis) + // only calculate a new ground truth if a behaviour is defined + if(!behaviour.isEmpty) { + val prevTruth = sharedMemory.getTruth(cell, time.prev, group) + result.set(behaviour.get.run(time, cell, prevTruth)) + } else { + result.set(None) + } + } +} + +case class Sensor(behaviour: Behaviour, count: SensorCount) { + def simulate(cell: Cell, group: Group, time: Time, sharedMemory: SharedMemory): Unit = { + val truth = sharedMemory.getTruth(cell, time, group.getParentOrThis) + val result = sharedMemory.getResult(time) + if(sharedMemory.config.parSensor != 0 && count.count >= sharedMemory.config.parSensor ) { + val sensorList = count.getVector.par + sensorList.tasksupport = sharedMemory.config.sensorThreadPool + sensorList.foreach(i => result(SensorID(group, i)).set((group, cell, behaviour.run(time, cell, truth)))) + } else { + count.getVector.foreach(i => result(SensorID(group, i)).set((group, cell, behaviour.run(time, cell, truth)))) + } + } +} + +// calculate 1 full tick at a time +// a group may depend on others within a cell +// a cell does not depend on others within a group +// a sensor and ground truth are unique within a group in a cell +case class SensorCount(count: Int) { + def getVector: Vector[Int] = { + Vector.tabulate(count)(e => e) + } +} + + +case class Configuration(cells: Map[Cell, Map[Group, List[(Group, Sensor, Option[GroundTruth])]]], sharedMemory: SharedMemory) + diff --git a/src/main/scala/scim/lib/Concurrency.scala b/src/main/scala/scim/lib/Concurrency.scala new file mode 100644 index 0000000..caa90a4 --- /dev/null +++ b/src/main/scala/scim/lib/Concurrency.scala @@ -0,0 +1,229 @@ +package scim.lib + +import java.util.Calendar +import java.util.concurrent.Executors +import java.util.concurrent.locks.{Condition, Lock, ReentrantReadWriteLock} + +import scala.collection.mutable + +object ScalaLocks { + + class ScalaReadLock(lock: Lock) { + def apply[A](read: => A): A = { + lock.lock() + try read finally { + lock.unlock() + } + } + } + + class ScalaWriteLock(lock: Lock) { + def apply[A](write: () => A): Unit = { + lock.lock() + try write() finally { + lock.unlock() + } + } + } + + class ScalaReadWriteLock(lock: Lock) { + def apply[A](readWrite: () => A): A = { + lock.lock() + try readWrite() finally { + lock.unlock() + } + } + } + + class ScalaConditionalReadLock(readLock: ReentrantReadWriteLock.ReadLock, writeLock: ReentrantReadWriteLock.WriteLock, signal: Condition) { + private val scalaWriteLock = new ScalaWriteLock(writeLock) + + // condition and read should not change any mutable state + def apply[A](condition: => Boolean, read: => A): A = { + readLock.lock() // block writer + if(!condition) { + readLock.unlock() // unblock writer + // acquire write lock, await signal and release it + scalaWriteLock (() => { + while(!condition) { + signal.await() + } + // block the writer before releasing write lock + readLock.lock() + }) + } + try read finally { + readLock.unlock() + } + } + } + +} + +object ConcurrentBroadcast { + + class BroadcastObject[A]() { + + import ScalaLocks._ + + private val javaLock = new ReentrantReadWriteLock() + private val javaReadLock = javaLock.readLock() + private val javaWriteLock = javaLock.writeLock() + + private val signal = javaWriteLock.newCondition() + private val conditionalReadLock = new ScalaConditionalReadLock(javaReadLock, javaWriteLock, signal) + private val readWriteLock = new ScalaReadWriteLock(javaWriteLock) + private val writeLock = new ScalaWriteLock(javaWriteLock) + private val readLock = new ScalaReadLock(javaReadLock) + + private var value: Option[A] = None + + // NON-BLOCKING: check if value has been set, write new value and return if it was overwritten + def broadcast(newValue: A): Unit = { + writeLock(() => { + value = Some(newValue) + signal.signalAll() + }) + } + + // NON-BLOCKING: check if value has been set, write new value and return if it was overwritten + def broadcastOverwrite(newValue: A): Boolean = { + readWriteLock(() => { + val result = value.isEmpty + value = Some(newValue) + signal.signalAll() + result + }) + } + + // BLOCKING: watch value, block until it is set, then return it + def watch(): A = { + conditionalReadLock(!value.isEmpty, value.get) + } + + // NON-BLOCKING: returns value as Option + def peak(): Option[A] = { + readLock { + value + } + } + + } + + abstract class Broadcast[A](stream: mutable.ArraySeq[BroadcastObject[A]]) extends Runnable { + override def run(): Unit = { + access(stream) + } + def access(stream: mutable.ArraySeq[BroadcastObject[A]]) + } + + // Trait is just used to rename the function (maybe a bit redundant) + trait Watcher[A] extends Broadcast[A] { + override def access(stream: mutable.ArraySeq[BroadcastObject[A]]): Unit = { watch(stream) } + def watch(stream: mutable.ArraySeq[BroadcastObject[A]]) + } + + trait Broadcaster[A] extends Broadcast[A] { + override def access(stream: mutable.ArraySeq[BroadcastObject[A]]): Unit = { broadcast(stream) } + def broadcast(stream: mutable.ArraySeq[BroadcastObject[A]]) + } + +} + + +class Test2 { + import scim.lib.ConcurrentBroadcast._ + + private val stream: mutable.ArraySeq[BroadcastObject[Int]] = new mutable.ArraySeq[Int](5).map(_ => new BroadcastObject[Int]) + + class Producer(stream: mutable.ArraySeq[BroadcastObject[Int]]) extends Broadcast[Int](stream) with Broadcaster[Int] { + def a(): Unit = { + stream.foreach(element => {element.broadcast(1); Thread.sleep(1000)}) + } + + override def broadcast(stream: mutable.ArraySeq[BroadcastObject[Int]]): Unit = { + val f: (Int, Int) => Int = (x, y) => x + y + var previous: Int = 0 + stream.foreach(element => { + previous = f(previous, 2) + element.broadcast(previous) + }) + } + } + + class Watcher(id: Int, stream: mutable.ArraySeq[BroadcastObject[Int]]) extends Runnable { + def run(): Unit = { + stream.foreach(element => { Thread.sleep(id); println(id + ": " + element.watch() + " - " + Calendar.getInstance().getTime()) }) + } + } + + def run(): Unit = { + val threadPool = Executors.newFixedThreadPool(4) + val producer = new Producer(stream) + val watcher = new Watcher(0,stream) + val watcher2 = new Watcher(3000,stream) + + val p = new Thread(producer) + + val t = new Thread(watcher) + val t2 = new Thread(watcher2) + println("start producer") + p.start() + println("start watcher") + t.start() + t2.start() + println("join producer") + p.join() + println("join watcher") + t.join() + t2.join() + println("done") + + + } + +} + +class Test { + import scim.lib.ConcurrentBroadcast.BroadcastObject + + private val stream: mutable.ArraySeq[BroadcastObject[Int]] = new mutable.ArraySeq[Int](5).map(_ => new BroadcastObject[Int]) + + class Producer(stream: mutable.ArraySeq[BroadcastObject[Int]]) extends Runnable { + def run(): Unit = { + stream.foreach(element => {element.broadcast(1); Thread.sleep(1000)}) + } + } + + class Watcher(id: Int, stream: mutable.ArraySeq[BroadcastObject[Int]]) extends Runnable { + def run(): Unit = { + stream.foreach(element => { Thread.sleep(id); println(id + ": " + element.watch() + " - " + Calendar.getInstance().getTime()) }) + } + } + + def run(): Unit = { + val threadPool = Executors.newFixedThreadPool(4) + val producer = new Producer(stream) + val watcher = new Watcher(0,stream) + val watcher2 = new Watcher(3000,stream) + + val p = new Thread(producer) + + val t = new Thread(watcher) + val t2 = new Thread(watcher2) + println("start producer") + p.start() + println("start watcher") + t.start() + t2.start() + println("join producer") + p.join() + println("join watcher") + t.join() + t2.join() + println("done") + + + } + +}