the performance increase wasn't that good actually - so split up the topology into separate files so it's easier to read, and simplifying the png tile process so it joins tiles and prototypes sooner, with less info
This commit is contained in:
Adam
parent
2c518e396b
commit
560a5d4172
13 changed files with 603 additions and 518 deletions
|
|
@ -17,20 +17,20 @@ operator fun MapTilePosition.plus(addend: Int) =
|
|||
MapTilePosition(x + addend, y + addend)
|
||||
|
||||
|
||||
fun MapTilePosition.toMapChunkPosition() =
|
||||
fun MapTilePosition.toMapChunkPosition( chunkSize: Int = MAP_CHUNK_SIZE) =
|
||||
MapChunkPosition(
|
||||
floor(x.toDouble() / MAP_CHUNK_SIZE.toDouble()).toInt(),
|
||||
floor(y.toDouble() / MAP_CHUNK_SIZE.toDouble()).toInt(),
|
||||
floor(x.toDouble() / chunkSize.toDouble()).toInt(),
|
||||
floor(y.toDouble() / chunkSize.toDouble()).toInt(),
|
||||
)
|
||||
|
||||
|
||||
fun MapEntityPosition.toMapTilePosition() =
|
||||
MapChunkPosition(floor(x).toInt(), floor(y).toInt())
|
||||
|
||||
fun MapEntityPosition.toMapChunkPosition() =
|
||||
fun MapEntityPosition.toMapChunkPosition( chunkSize: Int = MAP_CHUNK_SIZE) =
|
||||
MapChunkPosition(
|
||||
floor(x / MAP_CHUNK_SIZE.toDouble()).toInt(),
|
||||
floor(y / MAP_CHUNK_SIZE.toDouble()).toInt(),
|
||||
floor(x / chunkSize.toDouble()).toInt(),
|
||||
floor(y / chunkSize.toDouble()).toInt(),
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -44,15 +44,15 @@ operator fun MapChunkPosition.minus(subtrahend: Int) =
|
|||
MapChunkPosition(x - subtrahend, y - subtrahend)
|
||||
|
||||
|
||||
val MapChunkPosition.leftTopTile: MapTilePosition
|
||||
get() = MapTilePosition(
|
||||
x * MAP_CHUNK_SIZE,
|
||||
y * MAP_CHUNK_SIZE,
|
||||
fun MapChunkPosition.leftTopTile( chunkSize: Int = MAP_CHUNK_SIZE): MapTilePosition =
|
||||
MapTilePosition(
|
||||
x * chunkSize,
|
||||
y * chunkSize,
|
||||
)
|
||||
|
||||
val MapChunkPosition.rightBottomTile: MapTilePosition
|
||||
get() = leftTopTile + (MAP_CHUNK_SIZE - 1)
|
||||
|
||||
|
||||
val MapChunkPosition.boundingBox: MapBoundingBox
|
||||
get() = MapBoundingBox(leftTopTile, rightBottomTile)
|
||||
//val MapChunkPosition.rightBottomTile: MapTilePosition
|
||||
// get() = leftTopTile + (MAP_CHUNK_SIZE - 1)
|
||||
//
|
||||
//
|
||||
//val MapChunkPosition.boundingBox: MapBoundingBox
|
||||
// get() = MapBoundingBox(leftTopTile, rightBottomTile)
|
||||
|
|
|
|||
|
|
@ -1,64 +1,22 @@
|
|||
package dev.adamko.kafkatorio.processor
|
||||
|
||||
import com.sksamuel.scrimage.ImmutableImage
|
||||
import com.sksamuel.scrimage.color.RGBColor
|
||||
import com.sksamuel.scrimage.nio.PngWriter
|
||||
import dev.adamko.kafkatorio.events.schema.FactorioConfigurationUpdate
|
||||
import dev.adamko.kafkatorio.events.schema.FactorioEvent
|
||||
import dev.adamko.kafkatorio.events.schema.FactorioObjectData
|
||||
import dev.adamko.kafkatorio.events.schema.FactorioPrototypes
|
||||
import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
|
||||
import dev.adamko.kafkatorio.events.schema.MAP_CHUNK_SIZE
|
||||
import dev.adamko.kafkatorio.events.schema.MapChunk
|
||||
import dev.adamko.kafkatorio.events.schema.MapChunkPosition
|
||||
import dev.adamko.kafkatorio.events.schema.MapTile
|
||||
import dev.adamko.kafkatorio.events.schema.MapTilePosition
|
||||
import dev.adamko.kafkatorio.events.schema.MapTilePrototype
|
||||
import dev.adamko.kafkatorio.events.schema.MapTiles
|
||||
import dev.adamko.kafkatorio.events.schema.converters.leftTopTile
|
||||
import dev.adamko.kafkatorio.events.schema.converters.toHexadecimal
|
||||
import dev.adamko.kafkatorio.events.schema.converters.toMapChunkPosition
|
||||
import dev.adamko.kotka.extensions.consumedAs
|
||||
import dev.adamko.kotka.extensions.groupedAs
|
||||
import dev.adamko.kotka.extensions.materializedAs
|
||||
import dev.adamko.kotka.extensions.materializedWith
|
||||
import dev.adamko.kotka.extensions.producedAs
|
||||
import dev.adamko.kotka.extensions.streams.aggregate
|
||||
import dev.adamko.kotka.extensions.streams.filter
|
||||
import dev.adamko.kotka.extensions.streams.groupBy
|
||||
import dev.adamko.kotka.extensions.streams.map
|
||||
import dev.adamko.kotka.extensions.streams.mapValues
|
||||
import dev.adamko.kotka.extensions.streams.merge
|
||||
import dev.adamko.kotka.extensions.streams.reduce
|
||||
import dev.adamko.kotka.extensions.streams.to
|
||||
import dev.adamko.kotka.extensions.tables.join
|
||||
import dev.adamko.kotka.extensions.tables.mapValues
|
||||
import dev.adamko.kotka.kxs.serde
|
||||
import dev.adamko.kotka.topicdata.TopicRecord
|
||||
import dev.adamko.kotka.topicdata.flatMapTopicRecords
|
||||
import java.awt.Color
|
||||
import java.awt.image.BufferedImage
|
||||
import java.io.File
|
||||
import dev.adamko.kafkatorio.processor.topology.PrototypeName
|
||||
import dev.adamko.kafkatorio.processor.topology.WebMapTileChunkPixels
|
||||
import dev.adamko.kafkatorio.processor.topology.WebMapTileChunkPosition
|
||||
import dev.adamko.kafkatorio.processor.topology.aggregateWebMapTiles
|
||||
import dev.adamko.kafkatorio.processor.topology.allMapTilesTable
|
||||
import dev.adamko.kafkatorio.processor.topology.playerUpdatesToWsServer
|
||||
import dev.adamko.kafkatorio.processor.topology.prototypesTable
|
||||
import dev.adamko.kafkatorio.processor.topology.saveTileImages
|
||||
import dev.adamko.kafkatorio.processor.topology.splitFactorioServerLog
|
||||
import java.time.Duration
|
||||
import kotlin.math.abs
|
||||
import kotlin.math.roundToInt
|
||||
import kotlinx.serialization.Serializable
|
||||
import kotlinx.serialization.Transient
|
||||
import kotlinx.serialization.decodeFromString
|
||||
import kotlinx.serialization.encodeToString
|
||||
import org.apache.kafka.common.serialization.Serdes
|
||||
import org.apache.kafka.streams.KafkaStreams
|
||||
import org.apache.kafka.streams.StoreQueryParameters
|
||||
import org.apache.kafka.streams.StreamsBuilder
|
||||
import org.apache.kafka.streams.TopologyDescription
|
||||
import org.apache.kafka.streams.kstream.Consumed
|
||||
import org.apache.kafka.streams.kstream.KStream
|
||||
import org.apache.kafka.streams.kstream.KTable
|
||||
import org.apache.kafka.streams.kstream.Suppressed
|
||||
import org.apache.kafka.streams.processor.RecordContext
|
||||
import org.apache.kafka.streams.state.QueryableStoreTypes
|
||||
import org.apache.kafka.streams.state.ReadOnlyKeyValueStore
|
||||
import org.apache.kafka.streams.state.ValueAndTimestamp
|
||||
|
||||
|
||||
class KafkatorioTopology(
|
||||
|
|
@ -70,21 +28,18 @@ class KafkatorioTopology(
|
|||
const val sourceTopic = "factorio-server-log"
|
||||
}
|
||||
|
||||
// private lateinit var tilePrototypesStore: ReadOnlyKeyValueStore<String, ValueAndTimestamp<MapTilePrototype>>
|
||||
|
||||
fun build() {
|
||||
|
||||
parseAndSplitServerLog()
|
||||
sendPlayerUpdatesToWebSocket()
|
||||
// createMapTileImages()
|
||||
val tilePrototypesTable: KTable<SurfaceIndex, SurfacePrototypesData> = tilePrototypesTable()
|
||||
splitFactorioServerLog(builder)
|
||||
playerUpdatesToWsServer(websocketServer, builder)
|
||||
|
||||
val tilesTable: KTable<TileUpdateRecordKey, TileUpdateRecord> = buildTilesTable()
|
||||
val allMapTilesTable: KTable<MapTilePosition, MapTile> = allMapTilesTable(builder)
|
||||
val tilePrototypesTable: KTable<PrototypeName, MapTilePrototype> = prototypesTable(builder)
|
||||
|
||||
val tilesGroupedByChunk: KTable<MapChunkDataPosition, MapChunkData> =
|
||||
buildChunkedTilesTable(tilesTable)
|
||||
val webMapTiles: KTable<WebMapTileChunkPosition, WebMapTileChunkPixels> =
|
||||
aggregateWebMapTiles(allMapTilesTable, tilePrototypesTable)
|
||||
|
||||
createChunkTileImage(tilePrototypesTable, tilesGroupedByChunk)
|
||||
saveTileImages(webMapTiles)
|
||||
|
||||
val topology = builder.build()
|
||||
|
||||
|
|
@ -99,370 +54,6 @@ class KafkatorioTopology(
|
|||
|
||||
streams.start()
|
||||
|
||||
// tilePrototypesStore = streams.store(
|
||||
// StoreQueryParameters.fromNameAndType(
|
||||
// tilePrototypesTable.queryableStoreName(),
|
||||
// QueryableStoreTypes.timestampedKeyValueStore<String, MapTilePrototype>()
|
||||
// )
|
||||
// )
|
||||
}
|
||||
|
||||
private fun parseAndSplitServerLog() {
|
||||
|
||||
builder.stream(
|
||||
sourceTopic,
|
||||
consumedAs(
|
||||
"read-raw-packets-from-server",
|
||||
Serdes.String(),
|
||||
Serdes.String(),
|
||||
)
|
||||
)
|
||||
.mapValues("decode-packets") { _: String, value: String ->
|
||||
// println("Mapping $readOnlyKey:$value")
|
||||
jsonMapper.decodeFromString<KafkatorioPacket>(value)
|
||||
}
|
||||
.to(
|
||||
producedAs(
|
||||
"split-server-log",
|
||||
Serdes.String(),
|
||||
KafkatorioPacketSerde
|
||||
)
|
||||
) { _: String, value: KafkatorioPacket, _: RecordContext ->
|
||||
// println("[$key] sending event:${value.eventType} to topic:${value.data.objectName()}")
|
||||
when (value) {
|
||||
is FactorioEvent ->
|
||||
"kafkatorio.${value.packetType.name}.${value.data.objectName.name}"
|
||||
is FactorioConfigurationUpdate ->
|
||||
"kafkatorio.${value.packetType.name}.FactorioConfigurationUpdate"
|
||||
is FactorioPrototypes ->
|
||||
"kafkatorio.${value.packetType.name}.all"
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private fun sendPlayerUpdatesToWebSocket() {
|
||||
builder.stream(
|
||||
"kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.LuaPlayer}",
|
||||
Consumed.with(Serdes.String(), KafkatorioPacketSerde)
|
||||
)
|
||||
.foreach { _, value ->
|
||||
// println("sending ${value.packetType} packet to websocket")
|
||||
websocketServer.sendMessage(jsonMapper.encodeToString(value))
|
||||
}
|
||||
}
|
||||
|
||||
@Serializable
|
||||
data class SurfaceIndex(
|
||||
val surfaceIndex: Int
|
||||
) {
|
||||
override fun toString() = "$surfaceIndex"
|
||||
}
|
||||
|
||||
@Serializable
|
||||
data class SurfacePrototypesData(
|
||||
val surfaceIndex: SurfaceIndex,
|
||||
val mapTilePrototypes: Set<MapTilePrototype>,
|
||||
)
|
||||
|
||||
private fun tilePrototypesTable(): KTable<SurfaceIndex, SurfacePrototypesData> {
|
||||
|
||||
// val tilePrototypesStreamName = "kafkatorio.stream.prototypes-tiles"
|
||||
// val tilePrototypesStoreName = "kafkatorio.store.prototypes-tiles"
|
||||
|
||||
return builder.stream(
|
||||
"kafkatorio.${KafkatorioPacket.PacketType.PROTOTYPES}.all",
|
||||
consumedAs(
|
||||
"consume-all-prototypes-packets",
|
||||
Serdes.String(),
|
||||
jsonMapper.serde<FactorioPrototypes>()
|
||||
)
|
||||
)
|
||||
.map("flatten-map-tile-prototypes") { _, prototypes: FactorioPrototypes ->
|
||||
val tilePrototypes = prototypes.prototypes
|
||||
.filterIsInstance<MapTilePrototype>()
|
||||
.toSet()
|
||||
val key = SurfaceIndex(1) // TODO make surfaceIndex come from Factorio, not hard coded
|
||||
val data = SurfacePrototypesData(key, tilePrototypes)
|
||||
key to data
|
||||
}
|
||||
.peek { _, value ->
|
||||
println("tile prototypes: ${value.mapTilePrototypes.joinToString { it.name }}")
|
||||
}
|
||||
.groupByKey(
|
||||
groupedAs(
|
||||
"grouping-tile-prototypes-by-surface-index",
|
||||
jsonMapper.serde(),
|
||||
jsonMapper.serde()
|
||||
)
|
||||
)
|
||||
.reduce(
|
||||
"group-surface-prototypes",
|
||||
materializedWith(
|
||||
// "kafkatorio.surface-prototypes",
|
||||
jsonMapper.serde(),
|
||||
jsonMapper.serde()
|
||||
)
|
||||
) { value1: SurfacePrototypesData, value2: SurfacePrototypesData ->
|
||||
value1.copy(mapTilePrototypes = value1.mapTilePrototypes + value2.mapTilePrototypes)
|
||||
}
|
||||
// .toTable(
|
||||
// materializedAs(
|
||||
// "kafkatorio.prototypes.map-tiles",
|
||||
// jsonMapper.serde(),
|
||||
// jsonMapper.serde()
|
||||
// )
|
||||
// )
|
||||
// .to(
|
||||
// tilePrototypesStreamName,
|
||||
// )
|
||||
|
||||
// return builder.globalTable(
|
||||
// tilePrototypesStreamName,
|
||||
// consumedAs(
|
||||
// "send-tile-protos-to-gkt-store",
|
||||
// jsonMapper.serde(),
|
||||
// jsonMapper.serde()
|
||||
// ),
|
||||
// materializedAs(
|
||||
// storeName = tilePrototypesStoreName,
|
||||
// keySerde = jsonMapper.serde(),
|
||||
// valueSerde = jsonMapper.serde(),
|
||||
// )
|
||||
// )
|
||||
}
|
||||
|
||||
@Serializable
|
||||
data class TileUpdateRecordKey(
|
||||
val surfaceIndex: SurfaceIndex,
|
||||
val tilePosition: MapTilePosition,
|
||||
)
|
||||
|
||||
@Serializable
|
||||
data class TileUpdateRecord(
|
||||
val surfaceIndex: SurfaceIndex,
|
||||
val tilePosition: MapTilePosition,
|
||||
val tile: MapTile,
|
||||
) : TopicRecord<TileUpdateRecordKey> {
|
||||
@Transient
|
||||
override val topicKey: TileUpdateRecordKey = TileUpdateRecordKey(surfaceIndex, tilePosition)
|
||||
}
|
||||
|
||||
private fun buildTilesTable(): KTable<TileUpdateRecordKey, TileUpdateRecord> {
|
||||
|
||||
val luaTilesUpdatesStream: KStream<String, MapTiles> = builder.stream(
|
||||
"kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.LuaTiles}",
|
||||
consumedAs("consume-map-tiles-packets", Serdes.String(), jsonMapper.serde<FactorioEvent>())
|
||||
)
|
||||
.filter("events.filter.map-tiles") { _: String, event: FactorioEvent -> event.data is MapTiles }
|
||||
.mapValues("events.extract-map-tiles") { _, event: FactorioEvent -> (event.data as? MapTiles)!! }
|
||||
|
||||
val chunkTilesUpdateStream: KStream<String, MapTiles> = builder.stream(
|
||||
"kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.MapChunk}",
|
||||
consumedAs("consume-map-chunk-packets", Serdes.String(), jsonMapper.serde<FactorioEvent>())
|
||||
)
|
||||
.filter("events.filter.map-chunks") { _: String, event: FactorioEvent -> event.data is MapChunk }
|
||||
.mapValues("events.extract-map-chunks") { _, event: FactorioEvent -> (event.data as? MapChunk)!!.tiles }
|
||||
|
||||
return luaTilesUpdatesStream
|
||||
.merge("luaTilesUpdatesStream-and-chunkTilesUpdateStream", chunkTilesUpdateStream)
|
||||
.flatMapTopicRecords("all-tiles-convert-to-TileUpdateRecord") { _, mapTiles: MapTiles ->
|
||||
// println("all map tiles update ${mapTiles.tiles.size}")
|
||||
mapTiles.tiles.map { tile ->
|
||||
TileUpdateRecord(
|
||||
SurfaceIndex(mapTiles.surfaceIndex),
|
||||
tile.position,
|
||||
tile,
|
||||
)
|
||||
}
|
||||
}
|
||||
.toTable(materializedWith(jsonMapper.serde(), jsonMapper.serde()))
|
||||
// .suppress(
|
||||
// Suppressed.untilTimeLimit(
|
||||
// Duration.ofSeconds(30),
|
||||
// Suppressed.BufferConfig
|
||||
// .maxRecords(100)
|
||||
// .emitEarlyWhenFull()
|
||||
// )
|
||||
// )
|
||||
}
|
||||
|
||||
@Serializable
|
||||
data class MapChunkDataPosition(
|
||||
val position: MapChunkPosition,
|
||||
val surfaceIndex: SurfaceIndex,
|
||||
)
|
||||
|
||||
@Serializable
|
||||
data class MapChunkData(
|
||||
val chunkPosition: MapChunkDataPosition,
|
||||
val tiles: Set<MapTile>,
|
||||
)
|
||||
|
||||
private fun buildChunkedTilesTable(
|
||||
tilesTable: KTable<TileUpdateRecordKey, TileUpdateRecord>
|
||||
): KTable<MapChunkDataPosition, MapChunkData> {
|
||||
|
||||
@Serializable
|
||||
data class MapChunkDataAggregator(
|
||||
val tiles: MutableSet<MapTile> = mutableSetOf(),
|
||||
)
|
||||
|
||||
return tilesTable
|
||||
.toStream()
|
||||
// .repartition(
|
||||
// repartitionedAs(
|
||||
// "repartition-all-tiles-before-chunk-grouping",
|
||||
// jsonMapper.serde(),
|
||||
// jsonMapper.serde()
|
||||
// )
|
||||
// )
|
||||
.groupBy<TileUpdateRecordKey, TileUpdateRecord, MapChunkDataPosition>(
|
||||
groupedAs("grouping-tiles-by-chunk-pos", jsonMapper.serde(), jsonMapper.serde()),
|
||||
) { key: TileUpdateRecordKey, value: TileUpdateRecord ->
|
||||
MapChunkDataPosition(
|
||||
value.tilePosition.toMapChunkPosition(),
|
||||
key.surfaceIndex
|
||||
)
|
||||
}
|
||||
.aggregate<MapChunkDataPosition, TileUpdateRecord, MapChunkDataAggregator>(
|
||||
"reducing-tiles-grouped-by-chunk-position",
|
||||
materializedWith(jsonMapper.serde(), jsonMapper.serde()),
|
||||
{ MapChunkDataAggregator() },
|
||||
) { _: MapChunkDataPosition, value: TileUpdateRecord, aggregate: MapChunkDataAggregator ->
|
||||
aggregate.tiles.add(value.tile)
|
||||
aggregate
|
||||
}
|
||||
// TODO maybe set serdes here?
|
||||
.mapValues(
|
||||
"tiles-grouped-by-chunk-position-finalise-aggregation",
|
||||
materializedWith(jsonMapper.serde(), jsonMapper.serde()),
|
||||
) { k, v ->
|
||||
MapChunkData(k, v.tiles)
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private fun createChunkTileImage(
|
||||
tilePrototypesTable: KTable<SurfaceIndex, SurfacePrototypesData>,
|
||||
tilesGroupedByChunk: KTable<MapChunkDataPosition, MapChunkData>,
|
||||
) {
|
||||
|
||||
@Serializable
|
||||
data class ChunkTilesAndProtos(
|
||||
val chunkTiles: MapChunkData,
|
||||
val protos: SurfacePrototypesData
|
||||
)
|
||||
|
||||
// tilesGroupedByChunk
|
||||
// .join<ChunkTilesAndProtos, SurfaceIndex, SurfacePrototypesData>(
|
||||
// tilePrototypesTable,
|
||||
// { cd: MapChunkData -> cd.chunkPosition.surfaceIndex },
|
||||
// { chunkTiles: MapChunkData, protos: SurfacePrototypesData ->
|
||||
// ChunkTilesAndProtos(chunkTiles, protos)
|
||||
// },
|
||||
// namedAs("joining-chunks-tiles-prototypes"),
|
||||
// materializedAs(
|
||||
// "joined-chunked-tiles-with-prototypes",
|
||||
// jsonMapper.serde(),
|
||||
// jsonMapper.serde()
|
||||
// ),
|
||||
// )
|
||||
tilesGroupedByChunk
|
||||
.join<MapChunkDataPosition, MapChunkData, SurfaceIndex, SurfacePrototypesData, ChunkTilesAndProtos>(
|
||||
tilePrototypesTable,
|
||||
"enrich-tiles-with-prototype-colour",
|
||||
materializedAs(
|
||||
"joined-chunked-tiles-with-prototypes",
|
||||
jsonMapper.serde(),
|
||||
jsonMapper.serde()
|
||||
),
|
||||
{ it.chunkPosition.surfaceIndex },
|
||||
) { chunkTiles: MapChunkData, protos: SurfacePrototypesData ->
|
||||
ChunkTilesAndProtos(chunkTiles, protos)
|
||||
}
|
||||
.suppress(
|
||||
Suppressed.untilTimeLimit(
|
||||
Duration.ofSeconds(30),
|
||||
Suppressed.BufferConfig.maxRecords(30)
|
||||
)
|
||||
)
|
||||
.toStream()
|
||||
.foreach { _, (chunkTiles: MapChunkData, protos: SurfacePrototypesData) ->
|
||||
runCatching {
|
||||
saveMapTilesPng(chunkTiles, protos)
|
||||
}.onFailure { e ->
|
||||
println("error saving map tile png chunk:${chunkTiles.chunkPosition}")
|
||||
e.printStackTrace()
|
||||
throw e
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
private fun saveMapTilesPng(chunkTiles: MapChunkData, protos: SurfacePrototypesData) {
|
||||
|
||||
val chunkPos = chunkTiles.chunkPosition
|
||||
|
||||
val tilePrototypes: Map<String, MapTilePrototype?> =
|
||||
protos.mapTilePrototypes.associateBy { it.name }.withDefault { null }
|
||||
|
||||
val chunkOriginX = chunkTiles.chunkPosition.position.leftTopTile.x
|
||||
// val chunkOriginY = chunk.chunkPosition.position.rightBottomTile.y
|
||||
val chunkOriginY = chunkTiles.chunkPosition.position.leftTopTile.y
|
||||
|
||||
val chunkImage =
|
||||
ImmutableImage.filled(
|
||||
MAP_CHUNK_SIZE,
|
||||
MAP_CHUNK_SIZE,
|
||||
Color.BLACK,
|
||||
BufferedImage.TYPE_INT_ARGB
|
||||
)
|
||||
|
||||
chunkTiles.tiles.forEach { tile ->
|
||||
|
||||
val prototypeColour = tilePrototypes
|
||||
.getValue(tile.prototypeName)
|
||||
?.mapColour
|
||||
?.toHexadecimal()
|
||||
|
||||
val rgbColour = when (prototypeColour) {
|
||||
null -> {
|
||||
println("missing prototype: ${tile.prototypeName}")
|
||||
RGBColor(11, 11, 11, 0)
|
||||
}
|
||||
else -> RGBColor(
|
||||
prototypeColour.red.roundToInt(),
|
||||
prototypeColour.green.roundToInt(),
|
||||
prototypeColour.blue.roundToInt(),
|
||||
prototypeColour.alpha.roundToInt(),
|
||||
)
|
||||
}
|
||||
|
||||
val pixelX = abs(abs(tile.position.x) - abs(chunkOriginX))
|
||||
val pixelY = abs(abs(tile.position.y) - abs(chunkOriginY))
|
||||
|
||||
chunkImage.setColor(
|
||||
pixelX,
|
||||
pixelY,
|
||||
rgbColour
|
||||
)
|
||||
|
||||
}
|
||||
|
||||
val zoom = 1u
|
||||
val file =
|
||||
File(
|
||||
"src/main/resources/kafkatorio-web-map/s${chunkPos.surfaceIndex}/z$zoom/x${chunkOriginX}/y${chunkOriginY}.png"
|
||||
)
|
||||
|
||||
if (file.parentFile.mkdirs()) {
|
||||
println("created new map tile parentFile directory ${file.absolutePath}")
|
||||
}
|
||||
|
||||
println("saving map tile $file")
|
||||
|
||||
chunkImage.output(PngWriter.MaxCompression, file)
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -10,12 +10,8 @@ import kotlinx.coroutines.flow.withIndex
|
|||
import kotlinx.coroutines.job
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import kotlinx.coroutines.supervisorScope
|
||||
import org.http4k.core.Request
|
||||
import org.http4k.core.Response
|
||||
import org.http4k.core.Status.Companion.OK
|
||||
import org.http4k.routing.bind
|
||||
import org.http4k.routing.websockets
|
||||
import org.http4k.server.PolyHandler
|
||||
import org.http4k.websocket.Websocket
|
||||
import org.http4k.websocket.WsHandler
|
||||
import org.http4k.websocket.WsMessage
|
||||
|
|
|
|||
|
|
@ -3,7 +3,6 @@ package dev.adamko.kafkatorio.processor
|
|||
|
||||
import dev.adamko.kafkatorio.processor.config.ApplicationProperties
|
||||
import dev.adamko.kafkatorio.processor.tileserver.WebMapTileServer
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import org.http4k.server.PolyHandler
|
||||
import org.http4k.server.Undertow
|
||||
|
|
@ -24,9 +23,9 @@ fun main() = runBlocking<Unit> {
|
|||
wsServer.build(),
|
||||
).asServer(Undertow(9073))
|
||||
|
||||
// Runtime.getRuntime().addShutdownHook(Thread {
|
||||
// webServer.stop()
|
||||
// })
|
||||
Runtime.getRuntime().addShutdownHook(Thread {
|
||||
webServer.stop()
|
||||
})
|
||||
// launch {
|
||||
webServer.start().block()
|
||||
// }
|
||||
|
|
|
|||
|
|
@ -1,68 +0,0 @@
|
|||
package dev.adamko.kafkatorio.processor.topology
|
||||
|
||||
import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
|
||||
import dev.adamko.kafkatorio.events.schema.MapChunkPosition
|
||||
import dev.adamko.kafkatorio.events.schema.MapTilePosition
|
||||
import dev.adamko.kafkatorio.events.schema.MapTilePrototype
|
||||
import dev.adamko.kafkatorio.processor.jsonMapper
|
||||
import kotlinx.serialization.Serializable
|
||||
import kotlinx.serialization.decodeFromString
|
||||
import org.apache.kafka.common.serialization.Serdes
|
||||
import org.apache.kafka.streams.StreamsBuilder
|
||||
import org.apache.kafka.streams.kstream.Consumed
|
||||
import org.apache.kafka.streams.kstream.KGroupedStream
|
||||
|
||||
class FactorioGameState (
|
||||
private val builder: StreamsBuilder = StreamsBuilder(),
|
||||
) {
|
||||
private val sourceTopic = "factorio-server-log"
|
||||
|
||||
|
||||
private fun gatherGameState() {
|
||||
|
||||
val serverPackets: KGroupedStream<String, KafkatorioPacket> =
|
||||
builder.stream(
|
||||
sourceTopic,
|
||||
Consumed.with(Serdes.String(), Serdes.String())
|
||||
)
|
||||
.mapValues { readOnlyKey, value ->
|
||||
println("Mapping $readOnlyKey:$value")
|
||||
jsonMapper.decodeFromString<KafkatorioPacket>(value)
|
||||
}
|
||||
.groupByKey()
|
||||
|
||||
@Serializable
|
||||
data class TileSurfaceData(
|
||||
val surfaceIndex: Int,
|
||||
val position: MapTilePosition,
|
||||
val prototype: MapTilePrototype,
|
||||
)
|
||||
|
||||
@Serializable
|
||||
data class MapChunkData(
|
||||
val position: MapChunkPosition,
|
||||
val tiles: List<TileSurfaceData>,
|
||||
)
|
||||
|
||||
@Serializable
|
||||
data class FactorioSurfaceData(
|
||||
val surfaceIndex: Int,
|
||||
val chunks: List<MapChunkData>,
|
||||
)
|
||||
|
||||
@Serializable
|
||||
data class FactorioServerState(
|
||||
val surfaces: Map<Int, FactorioSurfaceData>,
|
||||
val tilePrototypes: List<MapTilePrototype>,
|
||||
)
|
||||
|
||||
|
||||
|
||||
// serverPackets
|
||||
// .aggregate<FactorioServerState>(
|
||||
// Initializer { FactorioServerState("", emptyMap(), emptyList()) }
|
||||
// )
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,74 @@
|
|||
package dev.adamko.kafkatorio.processor.topology
|
||||
|
||||
import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
|
||||
import dev.adamko.kafkatorio.events.schema.MapChunkPosition
|
||||
import dev.adamko.kafkatorio.events.schema.MapTilePosition
|
||||
import dev.adamko.kafkatorio.events.schema.MapTilePrototype
|
||||
import dev.adamko.kafkatorio.processor.jsonMapper
|
||||
import kotlinx.serialization.Serializable
|
||||
import kotlinx.serialization.decodeFromString
|
||||
import org.apache.kafka.common.serialization.Serdes
|
||||
import org.apache.kafka.streams.StreamsBuilder
|
||||
import org.apache.kafka.streams.kstream.Consumed
|
||||
import org.apache.kafka.streams.kstream.KGroupedStream
|
||||
|
||||
private const val sourceTopic = "factorio-server-log"
|
||||
|
||||
@JvmInline
|
||||
@Serializable
|
||||
value class FactorioServerId(val id: String) {
|
||||
override fun toString() = id
|
||||
}
|
||||
|
||||
@Serializable
|
||||
data class FactorioServerState(
|
||||
val serverId: FactorioServerId,
|
||||
val mapTilePrototypes: Set<MapTilePrototype>,
|
||||
val surfaceIndexes: Set<SurfaceIndex>,
|
||||
)
|
||||
|
||||
|
||||
private fun gatherGameState(builder: StreamsBuilder = StreamsBuilder()) {
|
||||
|
||||
val serverPackets: KGroupedStream<String, KafkatorioPacket> =
|
||||
builder.stream(
|
||||
sourceTopic,
|
||||
Consumed.with(Serdes.String(), Serdes.String())
|
||||
)
|
||||
.mapValues { readOnlyKey, value ->
|
||||
println("Mapping $readOnlyKey:$value")
|
||||
jsonMapper.decodeFromString<KafkatorioPacket>(value)
|
||||
}
|
||||
.groupByKey()
|
||||
|
||||
@Serializable
|
||||
data class TileSurfaceData(
|
||||
val surfaceIndex: Int,
|
||||
val position: MapTilePosition,
|
||||
val prototype: MapTilePrototype,
|
||||
)
|
||||
|
||||
@Serializable
|
||||
data class MapChunkData(
|
||||
val position: MapChunkPosition,
|
||||
val tiles: List<TileSurfaceData>,
|
||||
)
|
||||
|
||||
@Serializable
|
||||
data class FactorioSurfaceData(
|
||||
val surfaceIndex: Int,
|
||||
val chunks: List<MapChunkData>,
|
||||
)
|
||||
|
||||
@Serializable
|
||||
data class FactorioServerState(
|
||||
val surfaces: Map<Int, FactorioSurfaceData>,
|
||||
val tilePrototypes: List<MapTilePrototype>,
|
||||
)
|
||||
|
||||
|
||||
// serverPackets
|
||||
// .aggregate<FactorioServerState>(
|
||||
// Initializer { FactorioServerState("", emptyMap(), emptyList()) }
|
||||
// )
|
||||
}
|
||||
|
|
@ -0,0 +1,86 @@
|
|||
package dev.adamko.kafkatorio.processor.topology
|
||||
|
||||
import dev.adamko.kafkatorio.events.schema.FactorioEvent
|
||||
import dev.adamko.kafkatorio.events.schema.FactorioObjectData
|
||||
import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
|
||||
import dev.adamko.kafkatorio.events.schema.MapChunk
|
||||
import dev.adamko.kafkatorio.events.schema.MapTile
|
||||
import dev.adamko.kafkatorio.events.schema.MapTilePosition
|
||||
import dev.adamko.kafkatorio.events.schema.MapTiles
|
||||
import dev.adamko.kafkatorio.processor.jsonMapper
|
||||
import dev.adamko.kotka.extensions.consumedAs
|
||||
import dev.adamko.kotka.extensions.materializedWith
|
||||
import dev.adamko.kotka.extensions.streams.filter
|
||||
import dev.adamko.kotka.extensions.streams.flatMap
|
||||
import dev.adamko.kotka.extensions.streams.mapValues
|
||||
import dev.adamko.kotka.extensions.streams.toTable
|
||||
import dev.adamko.kotka.kxs.serde
|
||||
import java.time.Duration
|
||||
import kotlinx.serialization.Serializable
|
||||
import org.apache.kafka.common.serialization.Serdes
|
||||
import org.apache.kafka.streams.StreamsBuilder
|
||||
import org.apache.kafka.streams.kstream.KStream
|
||||
import org.apache.kafka.streams.kstream.KTable
|
||||
import org.apache.kafka.streams.kstream.Suppressed
|
||||
|
||||
@Serializable
|
||||
data class TileUpdateRecordKey(
|
||||
val surfaceIndex: SurfaceIndex,
|
||||
val tilePosition: MapTilePosition,
|
||||
)
|
||||
|
||||
|
||||
fun allMapTilesTable(
|
||||
builder: StreamsBuilder
|
||||
): KTable<MapTilePosition, MapTile> {
|
||||
|
||||
val luaTilesUpdatesStream: KStream<String, MapTiles> = builder.stream(
|
||||
"kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.LuaTiles}",
|
||||
consumedAs("consume-map-tiles-packets", Serdes.String(), jsonMapper.serde<FactorioEvent>())
|
||||
)
|
||||
.filter("events.filter.map-tiles") { _: String, event: FactorioEvent -> event.data is MapTiles }
|
||||
.mapValues("events.extract-map-tiles") { _, event: FactorioEvent -> (event.data as? MapTiles)!! }
|
||||
|
||||
val chunkTilesUpdateStream: KStream<String, MapTiles> = builder.stream(
|
||||
"kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.MapChunk}",
|
||||
consumedAs("consume-map-chunk-packets", Serdes.String(), jsonMapper.serde<FactorioEvent>())
|
||||
)
|
||||
.filter("events.filter.map-chunks") { _: String, event: FactorioEvent -> event.data is MapChunk }
|
||||
.mapValues("events.extract-map-chunks") { _, event: FactorioEvent -> (event.data as? MapChunk)!!.tiles }
|
||||
|
||||
return luaTilesUpdatesStream
|
||||
.merge(chunkTilesUpdateStream)
|
||||
.flatMap("flat-map-all-tiles") { _, value ->
|
||||
value.tiles.map { it.position to it }
|
||||
}
|
||||
.toTable(
|
||||
"all-tiles-keyed-on-position",
|
||||
materializedWith(jsonMapper.serde(), jsonMapper.serde())
|
||||
)
|
||||
.suppress(
|
||||
Suppressed.untilTimeLimit(
|
||||
Duration.ofSeconds(30),
|
||||
Suppressed.BufferConfig
|
||||
.maxRecords(30)
|
||||
.emitEarlyWhenFull()
|
||||
)
|
||||
)
|
||||
}
|
||||
|
||||
//fun buildTilesTable(
|
||||
// builder: StreamsBuilder
|
||||
//): KTable<TileUpdateRecordKey, TileUpdateRecord> {
|
||||
//
|
||||
// return luaTilesUpdatesStream
|
||||
// .merge("luaTilesUpdatesStream-and-chunkTilesUpdateStream", chunkTilesUpdateStream)
|
||||
// .flatMapTopicRecords("all-tiles-convert-to-TileUpdateRecord") { _, mapTiles: MapTiles ->
|
||||
// mapTiles.tiles.map { tile ->
|
||||
// TileUpdateRecord(
|
||||
// SurfaceIndex(mapTiles.surfaceIndex),
|
||||
// tile.position,
|
||||
// tile,
|
||||
// )
|
||||
// }
|
||||
// }
|
||||
// .toTable(materializedWith(jsonMapper.serde(), jsonMapper.serde()))
|
||||
//}
|
||||
|
|
@ -0,0 +1,54 @@
|
|||
package dev.adamko.kafkatorio.processor.topology
|
||||
|
||||
import dev.adamko.kafkatorio.events.schema.FactorioConfigurationUpdate
|
||||
import dev.adamko.kafkatorio.events.schema.FactorioEvent
|
||||
import dev.adamko.kafkatorio.events.schema.FactorioPrototypes
|
||||
import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
|
||||
import dev.adamko.kafkatorio.processor.KafkatorioPacketSerde
|
||||
import dev.adamko.kafkatorio.processor.KafkatorioTopology
|
||||
import dev.adamko.kafkatorio.processor.jsonMapper
|
||||
import dev.adamko.kotka.extensions.consumedAs
|
||||
import dev.adamko.kotka.extensions.producedAs
|
||||
import dev.adamko.kotka.extensions.streams.mapValues
|
||||
import dev.adamko.kotka.extensions.streams.to
|
||||
import kotlinx.serialization.decodeFromString
|
||||
import org.apache.kafka.common.serialization.Serdes
|
||||
import org.apache.kafka.streams.StreamsBuilder
|
||||
import org.apache.kafka.streams.processor.RecordContext
|
||||
|
||||
|
||||
fun splitFactorioServerLog(
|
||||
builder: StreamsBuilder = StreamsBuilder(),
|
||||
) {
|
||||
|
||||
builder.stream(
|
||||
KafkatorioTopology.sourceTopic,
|
||||
consumedAs(
|
||||
"read-raw-packets-from-server",
|
||||
Serdes.String(),
|
||||
Serdes.String(),
|
||||
)
|
||||
)
|
||||
.mapValues("decode-packets") { _: String, value: String ->
|
||||
// println("Mapping $readOnlyKey:$value")
|
||||
jsonMapper.decodeFromString<KafkatorioPacket>(value)
|
||||
}
|
||||
.to(
|
||||
producedAs(
|
||||
"split-server-log",
|
||||
Serdes.String(),
|
||||
KafkatorioPacketSerde
|
||||
)
|
||||
) { _: String, value: KafkatorioPacket, _: RecordContext ->
|
||||
// println("[$key] sending event:${value.eventType} to topic:${value.data.objectName()}")
|
||||
when (value) {
|
||||
is FactorioEvent ->
|
||||
"kafkatorio.${value.packetType.name}.${value.data.objectName.name}"
|
||||
is FactorioConfigurationUpdate ->
|
||||
"kafkatorio.${value.packetType.name}.FactorioConfigurationUpdate"
|
||||
is FactorioPrototypes ->
|
||||
"kafkatorio.${value.packetType.name}.all"
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,10 @@
|
|||
package dev.adamko.kafkatorio.processor.topology
|
||||
|
||||
import kotlinx.serialization.Serializable
|
||||
|
||||
@Serializable
|
||||
data class SurfaceIndex(
|
||||
val surfaceIndex: Int
|
||||
) {
|
||||
override fun toString() = "$surfaceIndex"
|
||||
}
|
||||
|
|
@ -0,0 +1,26 @@
|
|||
package dev.adamko.kafkatorio.processor.topology
|
||||
|
||||
import dev.adamko.kafkatorio.events.schema.FactorioObjectData
|
||||
import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
|
||||
import dev.adamko.kafkatorio.processor.KafkatorioPacketSerde
|
||||
import dev.adamko.kafkatorio.processor.WebsocketServer
|
||||
import dev.adamko.kafkatorio.processor.jsonMapper
|
||||
import kotlinx.serialization.encodeToString
|
||||
import org.apache.kafka.common.serialization.Serdes
|
||||
import org.apache.kafka.streams.StreamsBuilder
|
||||
import org.apache.kafka.streams.kstream.Consumed
|
||||
|
||||
|
||||
fun playerUpdatesToWsServer(
|
||||
websocketServer: WebsocketServer,
|
||||
builder: StreamsBuilder = StreamsBuilder(),
|
||||
) {
|
||||
builder.stream(
|
||||
"kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.LuaPlayer}",
|
||||
Consumed.with(Serdes.String(), KafkatorioPacketSerde)
|
||||
)
|
||||
.foreach { _, value ->
|
||||
// println("sending ${value.packetType} packet to websocket")
|
||||
websocketServer.sendMessage(jsonMapper.encodeToString(value))
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,100 @@
|
|||
package dev.adamko.kafkatorio.processor.topology
|
||||
|
||||
import dev.adamko.kafkatorio.events.schema.FactorioPrototypes
|
||||
import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
|
||||
import dev.adamko.kafkatorio.events.schema.MapTilePrototype
|
||||
import dev.adamko.kafkatorio.processor.jsonMapper
|
||||
import dev.adamko.kotka.extensions.consumedAs
|
||||
//import dev.adamko.kotka.extensions.groupedAs
|
||||
import dev.adamko.kotka.extensions.materializedWith
|
||||
import dev.adamko.kotka.extensions.streams.flatMap
|
||||
//import dev.adamko.kotka.extensions.streams.*
|
||||
import dev.adamko.kotka.kxs.serde
|
||||
import kotlinx.serialization.Serializable
|
||||
import org.apache.kafka.common.serialization.Serdes
|
||||
import org.apache.kafka.streams.StreamsBuilder
|
||||
import org.apache.kafka.streams.kstream.KTable
|
||||
|
||||
|
||||
@Serializable
|
||||
@JvmInline
|
||||
value class PrototypeName(val name: String) {
|
||||
override fun toString() = name
|
||||
}
|
||||
|
||||
|
||||
fun prototypesTable(builder: StreamsBuilder): KTable<PrototypeName, MapTilePrototype> {
|
||||
return builder.stream(
|
||||
"kafkatorio.${KafkatorioPacket.PacketType.PROTOTYPES}.all",
|
||||
consumedAs(
|
||||
"consume-all-prototypes-packets",
|
||||
Serdes.String(),
|
||||
jsonMapper.serde<FactorioPrototypes>()
|
||||
)
|
||||
)
|
||||
.flatMap("map-MapTilePrototype") { _, prototypes: FactorioPrototypes ->
|
||||
prototypes.prototypes
|
||||
.filterIsInstance<MapTilePrototype>()
|
||||
.map { PrototypeName(it.name) to it }
|
||||
}
|
||||
.toTable(
|
||||
materializedWith(
|
||||
jsonMapper.serde(),
|
||||
jsonMapper.serde()
|
||||
)
|
||||
)
|
||||
}
|
||||
|
||||
|
||||
//@Serializable
|
||||
//data class FactorioServerPrototypesRecord(
|
||||
// val serverId: FactorioServerId,
|
||||
// val mapTilePrototypes: Map<PrototypeName, MapTilePrototype>,
|
||||
//)
|
||||
//
|
||||
//fun serverPrototypesTable(builder: StreamsBuilder): KTable<FactorioServerId, FactorioServerPrototypesRecord> {
|
||||
//
|
||||
// return builder.stream(
|
||||
// "kafkatorio.${KafkatorioPacket.PacketType.PROTOTYPES}.all",
|
||||
// consumedAs(
|
||||
// "consume-all-prototypes-packets",
|
||||
// Serdes.String(),
|
||||
// jsonMapper.serde<FactorioPrototypes>()
|
||||
// )
|
||||
// )
|
||||
// .map("build-FactorioServerPrototypesRecord", ::buildPrototypesRecord)
|
||||
// .peek { _, value ->
|
||||
// println("tile prototypes: ${value.mapTilePrototypes.keys.joinToString()}")
|
||||
// }
|
||||
// .groupByKey(
|
||||
// groupedAs(
|
||||
// "grouping-tile-prototypes-by-surface-index",
|
||||
// jsonMapper.serde(),
|
||||
// jsonMapper.serde()
|
||||
// )
|
||||
// )
|
||||
// .reduce(
|
||||
// "group-surface-prototypes",
|
||||
// materializedWith(
|
||||
//// "kafkatorio.surface-prototypes",
|
||||
// jsonMapper.serde(),
|
||||
// jsonMapper.serde()
|
||||
// )
|
||||
// ) { value1: FactorioServerPrototypesRecord, value2: FactorioServerPrototypesRecord ->
|
||||
// val allTiles = value1.mapTilePrototypes + value2.mapTilePrototypes
|
||||
// value1.copy(mapTilePrototypes = allTiles)
|
||||
// }
|
||||
//}
|
||||
//
|
||||
//private fun buildPrototypesRecord(
|
||||
// serverId: String,
|
||||
// prototypes: FactorioPrototypes
|
||||
//): Pair<FactorioServerId, FactorioServerPrototypesRecord> {
|
||||
// val tilePrototypes = prototypes.prototypes
|
||||
// .filterIsInstance<MapTilePrototype>()
|
||||
// .associateBy { PrototypeName(it.name) }
|
||||
//
|
||||
// val key = FactorioServerId(serverId)
|
||||
// val data = FactorioServerPrototypesRecord(key, tilePrototypes)
|
||||
// return key to data
|
||||
//}
|
||||
|
|
@ -0,0 +1,125 @@
|
|||
package dev.adamko.kafkatorio.processor.topology
|
||||
|
||||
import dev.adamko.kafkatorio.events.schema.Colour
|
||||
import dev.adamko.kafkatorio.events.schema.MapTile
|
||||
import dev.adamko.kafkatorio.events.schema.MapTilePosition
|
||||
import dev.adamko.kafkatorio.events.schema.MapTilePrototype
|
||||
import dev.adamko.kafkatorio.events.schema.converters.toMapChunkPosition
|
||||
import dev.adamko.kafkatorio.processor.jsonMapper
|
||||
import dev.adamko.kotka.extensions.groupedAs
|
||||
import dev.adamko.kotka.extensions.materializedAs
|
||||
import dev.adamko.kotka.extensions.namedAs
|
||||
import dev.adamko.kotka.extensions.tables.groupBy
|
||||
import dev.adamko.kotka.extensions.tables.join
|
||||
import dev.adamko.kotka.extensions.tables.mapValues
|
||||
import dev.adamko.kotka.extensions.toKeyValue
|
||||
import dev.adamko.kotka.kxs.serde
|
||||
import java.time.Duration
|
||||
import kotlinx.serialization.Serializable
|
||||
import org.apache.kafka.streams.kstream.KTable
|
||||
import org.apache.kafka.streams.kstream.Suppressed
|
||||
|
||||
/** The tiles of the webmap will be this many pixels high/wide */
|
||||
const val WEB_MAP_IMAGE_TILE_SIZE = 32
|
||||
|
||||
//@Serializable
|
||||
//data class MapChunkDataPosition(
|
||||
// val position: MapChunkPosition,
|
||||
// val surfaceIndex: SurfaceIndex,
|
||||
//)
|
||||
//
|
||||
//@Serializable
|
||||
//data class MapChunkData(
|
||||
// val chunkPosition: MapChunkDataPosition,
|
||||
// val tiles: Set<MapTile>,
|
||||
//)
|
||||
|
||||
@Serializable
|
||||
data class WebMapTileChunkPosition(
|
||||
val chunkSize: Int,
|
||||
val x: Int,
|
||||
val y: Int,
|
||||
)
|
||||
|
||||
@Serializable
|
||||
data class WebMapTilePixel(
|
||||
val mapColour: Colour,
|
||||
val tilePosition: MapTilePosition,
|
||||
)
|
||||
|
||||
@Serializable
|
||||
data class WebMapTileChunkPixels(
|
||||
val pixels: Set<WebMapTilePixel> = mutableSetOf(),
|
||||
)
|
||||
|
||||
/** Temporary mutable aggregator */
|
||||
@Serializable
|
||||
private data class WebMapTileChunkPixelsAcc(
|
||||
val pixels: MutableSet<WebMapTilePixel> = mutableSetOf(),
|
||||
)
|
||||
|
||||
fun aggregateWebMapTiles(
|
||||
allMapTilesTable: KTable<MapTilePosition, MapTile>,
|
||||
tilePrototypesTable: KTable<PrototypeName, MapTilePrototype>,
|
||||
): KTable<WebMapTileChunkPosition, WebMapTileChunkPixels> {
|
||||
|
||||
val webmapTileColours: KTable<MapTilePosition, WebMapTilePixel> =
|
||||
allMapTilesTable
|
||||
.join(
|
||||
tilePrototypesTable,
|
||||
"convert-tiles-to-colours",
|
||||
materializedAs(
|
||||
"store-web-map-tile-pixels",
|
||||
jsonMapper.serde(),
|
||||
jsonMapper.serde(),
|
||||
),
|
||||
{ PrototypeName(it.prototypeName) },
|
||||
) { tile: MapTile, proto: MapTilePrototype ->
|
||||
WebMapTilePixel(proto.mapColour, tile.position)
|
||||
}
|
||||
|
||||
|
||||
val webMapTileChunkAggregate: KTable<WebMapTileChunkPosition, WebMapTileChunkPixelsAcc> =
|
||||
webmapTileColours
|
||||
.groupBy(
|
||||
groupedAs(
|
||||
"group-webmap-tile-colours-into-chunks",
|
||||
jsonMapper.serde(),
|
||||
jsonMapper.serde(),
|
||||
)
|
||||
) { pos: MapTilePosition, px: WebMapTilePixel ->
|
||||
val (chunkX, chunkY) = pos.toMapChunkPosition(WEB_MAP_IMAGE_TILE_SIZE)
|
||||
val chunkPos = WebMapTileChunkPosition(WEB_MAP_IMAGE_TILE_SIZE, chunkX, chunkY)
|
||||
(chunkPos to px).toKeyValue()
|
||||
}
|
||||
.aggregate(
|
||||
{ WebMapTileChunkPixelsAcc() },
|
||||
/* adder */
|
||||
{ _: WebMapTileChunkPosition, newValue: WebMapTilePixel, acc: WebMapTileChunkPixelsAcc ->
|
||||
acc.apply { pixels.add(newValue) }
|
||||
},
|
||||
/* subtractor */
|
||||
{ _: WebMapTileChunkPosition, oldValue: WebMapTilePixel, acc: WebMapTileChunkPixelsAcc ->
|
||||
acc.apply { pixels.minus(oldValue) }
|
||||
},
|
||||
namedAs("web-map-tile-colours-into-$WEB_MAP_IMAGE_TILE_SIZE-chunks"),
|
||||
materializedAs("web-map-tile-colour-chunks-aggregate", jsonMapper.serde(), jsonMapper.serde())
|
||||
)
|
||||
.suppress(
|
||||
Suppressed.untilTimeLimit<WebMapTileChunkPosition>(
|
||||
Duration.ofSeconds(30),
|
||||
Suppressed.BufferConfig
|
||||
.maxRecords(2)
|
||||
// .withMaxBytes(31457280 ) // 30MB
|
||||
.emitEarlyWhenFull()
|
||||
).withName("web-map-tile-aggregate-debounce")
|
||||
)
|
||||
|
||||
return webMapTileChunkAggregate
|
||||
.mapValues(
|
||||
"finalise-web-map-tile-colour-chunk-aggregation",
|
||||
materializedAs("web-map-tile-colour-chunks", jsonMapper.serde(), jsonMapper.serde())
|
||||
) { _, v ->
|
||||
WebMapTileChunkPixels(v.pixels)
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,92 @@
|
|||
package dev.adamko.kafkatorio.processor.topology
|
||||
|
||||
import com.sksamuel.scrimage.ImmutableImage
|
||||
import com.sksamuel.scrimage.color.RGBColor
|
||||
import com.sksamuel.scrimage.nio.PngWriter
|
||||
import dev.adamko.kafkatorio.events.schema.MAP_CHUNK_SIZE
|
||||
import dev.adamko.kotka.extensions.tables.toStream
|
||||
import java.awt.Color
|
||||
import java.awt.image.BufferedImage
|
||||
import java.io.File
|
||||
import java.time.Duration
|
||||
import kotlin.math.abs
|
||||
import kotlin.math.roundToInt
|
||||
import org.apache.kafka.streams.kstream.KTable
|
||||
import org.apache.kafka.streams.kstream.Suppressed
|
||||
|
||||
|
||||
fun saveTileImages(
|
||||
webMapTiles: KTable<WebMapTileChunkPosition, WebMapTileChunkPixels>
|
||||
) {
|
||||
|
||||
webMapTiles
|
||||
.suppress(
|
||||
Suppressed.untilTimeLimit(
|
||||
Duration.ofSeconds(30),
|
||||
Suppressed.BufferConfig.maxRecords(30)
|
||||
)
|
||||
)
|
||||
.toStream("get-webmap-tile-updates")
|
||||
.foreach { position: WebMapTileChunkPosition, pixels: WebMapTileChunkPixels ->
|
||||
runCatching {
|
||||
saveMapTilesPng(position, pixels.pixels)
|
||||
}.onFailure { e ->
|
||||
println("error saving map tile png chunk: $position")
|
||||
e.printStackTrace()
|
||||
throw e
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
private fun saveMapTilesPng(
|
||||
chunkPosition: WebMapTileChunkPosition,
|
||||
pixels: Set<WebMapTilePixel>
|
||||
) {
|
||||
val surfaceIndex = 1 // TODO get surface index
|
||||
|
||||
val chunkOriginX = chunkPosition.x * chunkPosition.chunkSize
|
||||
val chunkOriginY = chunkPosition.y * chunkPosition.chunkSize
|
||||
|
||||
val chunkImage =
|
||||
ImmutableImage.filled(
|
||||
MAP_CHUNK_SIZE,
|
||||
MAP_CHUNK_SIZE,
|
||||
Color.BLACK,
|
||||
BufferedImage.TYPE_INT_ARGB
|
||||
)
|
||||
|
||||
pixels.forEach { pixel ->
|
||||
|
||||
val rgbColour = RGBColor(
|
||||
pixel.mapColour.red.roundToInt(),
|
||||
pixel.mapColour.green.roundToInt(),
|
||||
pixel.mapColour.blue.roundToInt(),
|
||||
pixel.mapColour.alpha.roundToInt(),
|
||||
)
|
||||
|
||||
val pixelX = abs(abs(pixel.tilePosition.x) - abs(chunkOriginX))
|
||||
val pixelY = abs(abs(pixel.tilePosition.y) - abs(chunkOriginY))
|
||||
|
||||
chunkImage.setColor(
|
||||
pixelX,
|
||||
pixelY,
|
||||
rgbColour
|
||||
)
|
||||
|
||||
}
|
||||
|
||||
val zoom = 1u
|
||||
val file =
|
||||
File(
|
||||
"src/main/resources/kafkatorio-web-map/s${surfaceIndex}/z$zoom/x${chunkOriginX}/y${chunkOriginY}.png"
|
||||
)
|
||||
|
||||
if (file.parentFile.mkdirs()) {
|
||||
println("created new map tile parentFile directory ${file.absolutePath}")
|
||||
}
|
||||
|
||||
println("saving map tile $file")
|
||||
|
||||
chunkImage.output(PngWriter.MaxCompression, file)
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue