so close to working performantly.... just the tiles are transparent???
- bump some versions - MapTile no longer has objectName field - fix nulls/optional fields in FactorioConfigurationUpdate - use bytes to encode rgba colours - waaaayyyy smaller - increase Kafka record size to ~10mb
This commit is contained in:
Adam
parent
ea0bfb350b
commit
bdb2d9577f
24 changed files with 619 additions and 404 deletions
|
|
@ -1,16 +1,16 @@
|
|||
package dev.adamko.kafkatorio.processor
|
||||
|
||||
import dev.adamko.kafkatorio.processor.topology.buildFactorioServerMap
|
||||
import dev.adamko.kafkatorio.processor.topology.factorioServerPacketStream
|
||||
import dev.adamko.kafkatorio.processor.topology.groupTilesIntoChunksWithColours
|
||||
import dev.adamko.kafkatorio.processor.topology.playerUpdatesToWsServer
|
||||
import dev.adamko.kafkatorio.processor.topology.saveMapTiles
|
||||
import dev.adamko.kafkatorio.processor.topology.splitFactorioServerPacketStream
|
||||
import dev.adamko.kafkatorio.processor.topology.tileProtoColourDictionary
|
||||
import java.time.Duration
|
||||
import org.apache.kafka.streams.KafkaStreams
|
||||
import org.apache.kafka.streams.StreamsBuilder
|
||||
import org.apache.kafka.streams.TopologyDescription
|
||||
|
||||
|
||||
class KafkatorioTopology(
|
||||
private val websocketServer: WebsocketServer,
|
||||
private val builder: StreamsBuilder = StreamsBuilder(),
|
||||
|
|
@ -20,17 +20,27 @@ class KafkatorioTopology(
|
|||
const val sourceTopic = "factorio-server-log"
|
||||
}
|
||||
|
||||
|
||||
fun build() {
|
||||
|
||||
|
||||
val packets = factorioServerPacketStream(builder)
|
||||
|
||||
|
||||
|
||||
splitFactorioServerPacketStream(packets)
|
||||
|
||||
playerUpdatesToWsServer(websocketServer, builder)
|
||||
|
||||
val serverMapDataTable = buildFactorioServerMap(packets)
|
||||
// val serverMapDataTable = buildFactorioServerMap(packets, tilePrototypesTable)
|
||||
|
||||
saveMapTiles(serverMapDataTable)
|
||||
val tileProtoColourDict = tileProtoColourDictionary(packets)
|
||||
|
||||
val tilePrototypesTable = groupTilesIntoChunksWithColours(
|
||||
packets,
|
||||
tileProtoColourDict,
|
||||
)
|
||||
saveMapTiles(tilePrototypesTable)
|
||||
|
||||
|
||||
// the old way...
|
||||
|
|
|
|||
|
|
@ -1,127 +1,196 @@
|
|||
package dev.adamko.kafkatorio.processor.topology
|
||||
|
||||
import dev.adamko.kafkatorio.events.schema.FactorioConfigurationUpdate
|
||||
import dev.adamko.kafkatorio.events.schema.ColourHex
|
||||
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.events.schema.MapChunk
|
||||
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.events.schema.MapTiles
|
||||
import dev.adamko.kafkatorio.processor.serdes.jsonMapper
|
||||
import dev.adamko.kafkatorio.events.schema.converters.toMapChunkPosition
|
||||
import dev.adamko.kafkatorio.processor.serdes.kxsBinary
|
||||
import dev.adamko.kotka.extensions.groupedAs
|
||||
import dev.adamko.kotka.extensions.materializedAs
|
||||
import dev.adamko.kotka.extensions.materializedWith
|
||||
import dev.adamko.kotka.extensions.streams.aggregate
|
||||
import dev.adamko.kotka.extensions.tables.mapValues
|
||||
import dev.adamko.kotka.extensions.streams.flatMap
|
||||
import dev.adamko.kotka.extensions.streams.reduce
|
||||
import dev.adamko.kotka.extensions.tables.join
|
||||
import dev.adamko.kotka.kxs.serde
|
||||
import kotlinx.serialization.Serializable
|
||||
import org.apache.kafka.streams.kstream.KStream
|
||||
import org.apache.kafka.streams.kstream.KTable
|
||||
|
||||
|
||||
//@Serializable
|
||||
//data class FactorioServerMap(
|
||||
// val serverId: FactorioServerId,
|
||||
// val tilePrototypes: Map<PrototypeName, MapTilePrototype>,
|
||||
// val surfaces: Map<SurfaceIndex, FactorioMapSurface>,
|
||||
//)
|
||||
|
||||
|
||||
//@Serializable
|
||||
//data class FactorioMapSurface(
|
||||
// val index: SurfaceIndex,
|
||||
// val tiles: Map<MapTilePosition, PrototypeName>,
|
||||
//)
|
||||
|
||||
|
||||
@Serializable
|
||||
data class FactorioServerMap(
|
||||
data class ServerMapChunkId(
|
||||
val serverId: FactorioServerId,
|
||||
val tilePrototypes: Map<PrototypeName, MapTilePrototype>,
|
||||
val surfaces: Map<SurfaceIndex, FactorioMapSurface>,
|
||||
val chunkPosition: MapChunkPosition,
|
||||
val surfaceIndex: SurfaceIndex,
|
||||
val zoomLevel: ZoomLevel,
|
||||
)
|
||||
|
||||
|
||||
@Serializable
|
||||
data class FactorioMapSurface(
|
||||
val index: SurfaceIndex,
|
||||
val tiles: Map<MapTilePosition, PrototypeName>,
|
||||
)
|
||||
|
||||
/** Aggregate [KafkatorioPacket]s into a single [FactorioServerMap] (per [serverId][FactorioServerId]). */
|
||||
fun buildFactorioServerMap(
|
||||
factorioServerPacketStream: KStream<FactorioServerId, KafkatorioPacket>
|
||||
): KTable<FactorioServerId, FactorioServerMap> {
|
||||
|
||||
return factorioServerPacketStream
|
||||
.groupByKey(
|
||||
groupedAs("factorio-server-map-data-grouping", jsonMapper.serde(), jsonMapper.serde())
|
||||
)
|
||||
.aggregate<FactorioServerId, KafkatorioPacket, FactorioServerMapAggregator>(
|
||||
"factorio-server-map-data-aggregate",
|
||||
materializedAs("factorio-server-map-aggregate-store", kxsBinary.serde(), kxsBinary.serde()),
|
||||
{ FactorioServerMapAggregator() },
|
||||
) { _: FactorioServerId, packet: KafkatorioPacket, aggregate: FactorioServerMapAggregator ->
|
||||
aggregate + packet
|
||||
}
|
||||
.mapValues(
|
||||
"finalise-factorio-server-map-aggregate",
|
||||
materializedWith(kxsBinary.serde(), kxsBinary.serde())
|
||||
) { readOnlyKey, aggregate ->
|
||||
|
||||
val surfaces = aggregate.surfaces.mapValues { (_, surface) ->
|
||||
FactorioMapSurface(surface.index, surface.tiles)
|
||||
}
|
||||
|
||||
FactorioServerMap(readOnlyKey, aggregate.tilePrototypes, surfaces)
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
|
||||
@Serializable
|
||||
private data class FactorioServerMapAggregator(
|
||||
val tilePrototypes: MutableMap<PrototypeName, MapTilePrototype> = mutableMapOf(),
|
||||
val surfaces: MutableMap<SurfaceIndex, FactorioMapSurfaceAggregator> = mutableMapOf(),
|
||||
data class ServerMapChunkTiles<Data>(
|
||||
val chunkId: ServerMapChunkId,
|
||||
val map: Map<MapTilePosition, Data>,
|
||||
) {
|
||||
|
||||
operator fun plus(packet: KafkatorioPacket): FactorioServerMapAggregator {
|
||||
return when (packet) {
|
||||
is FactorioEvent -> {
|
||||
when (val data = packet.data) {
|
||||
is MapChunk -> plus(data)
|
||||
is MapTiles -> plus(data)
|
||||
// is MapTile -> {} // hmmm... no surface index
|
||||
else -> this
|
||||
}
|
||||
}
|
||||
is FactorioPrototypes -> plus(packet)
|
||||
is FactorioConfigurationUpdate -> this
|
||||
}
|
||||
}
|
||||
|
||||
operator fun plus(mapTiles: MapTiles): FactorioServerMapAggregator {
|
||||
// println("FactorioServerMapAggregator: adding ${mapTiles.tiles.size} MapTiles from surface ${mapTiles.surfaceIndex}")
|
||||
|
||||
val surfaceIndex = SurfaceIndex(mapTiles.surfaceIndex)
|
||||
|
||||
val surface = surfaces.getOrPut(surfaceIndex) { FactorioMapSurfaceAggregator(surfaceIndex) }
|
||||
|
||||
val surfaceTiles = mapTiles.tiles.associate { it.position to PrototypeName(it.prototypeName) }
|
||||
surface.tiles.putAll(surfaceTiles)
|
||||
|
||||
return this
|
||||
}
|
||||
|
||||
operator fun plus(mapChunk: MapChunk): FactorioServerMapAggregator {
|
||||
// println("FactorioServerMapAggregator: adding mapChunk ${mapChunk.position}")
|
||||
return plus(mapChunk.tiles)
|
||||
}
|
||||
|
||||
operator fun plus(packet: FactorioPrototypes): FactorioServerMapAggregator {
|
||||
// println("FactorioServerMapAggregator: adding FactorioPrototypes")
|
||||
|
||||
val packetPrototypes = packet.prototypes
|
||||
.filterIsInstance<MapTilePrototype>()
|
||||
.associateBy { PrototypeName(it.name) }
|
||||
|
||||
tilePrototypes.putAll(packetPrototypes)
|
||||
|
||||
return this
|
||||
}
|
||||
|
||||
operator fun plus(other: ServerMapChunkTiles<Data>) = copy(map = map + other.map)
|
||||
}
|
||||
|
||||
|
||||
@Serializable
|
||||
private data class FactorioMapSurfaceAggregator(
|
||||
val index: SurfaceIndex,
|
||||
val tiles: MutableMap<MapTilePosition, PrototypeName> = mutableMapOf(),
|
||||
)
|
||||
fun groupTilesIntoChunksWithColours(
|
||||
factorioServerPacketStream: KStream<FactorioServerId, KafkatorioPacket>,
|
||||
tileProtoColourDict: KTable<FactorioServerId, TileColourDict>,
|
||||
): KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> {
|
||||
|
||||
// group tiles by server & surface & chunk
|
||||
val chunkedTilesTable: KTable<ServerMapChunkId, ServerMapChunkTiles<TileProtoHashCode>> =
|
||||
factorioServerPacketStream
|
||||
.flatMap("server-map-data.tiles.flatMapByChunk") { key: FactorioServerId, packet: KafkatorioPacket ->
|
||||
|
||||
val mapTileData = when (packet) {
|
||||
is FactorioEvent ->
|
||||
when (val data = packet.data) {
|
||||
is MapChunk -> data.tiles
|
||||
is MapTiles -> data
|
||||
else -> null
|
||||
}
|
||||
else -> null
|
||||
}
|
||||
|
||||
val zoomLevel = ZoomLevel.ZOOM_0
|
||||
|
||||
//
|
||||
// ZoomLevel
|
||||
// .values
|
||||
// .mapNotNull { zoomLevel ->
|
||||
|
||||
mapTileData
|
||||
?.tiles
|
||||
?.groupBy { tile ->
|
||||
tile.position.toMapChunkPosition(zoomLevel.chunkSize)
|
||||
}
|
||||
?.map { (chunkPos, tiles) ->
|
||||
|
||||
val chunkId = ServerMapChunkId(
|
||||
key,
|
||||
chunkPos,
|
||||
SurfaceIndex(mapTileData.surfaceIndex),
|
||||
zoomLevel,
|
||||
)
|
||||
|
||||
val tilesMap = tiles.associate { it.position to TileProtoHashCode(it) }
|
||||
|
||||
val chunkTiles = ServerMapChunkTiles(chunkId, tilesMap)
|
||||
|
||||
chunkId to chunkTiles
|
||||
} ?: listOf()
|
||||
// }.flatten()
|
||||
}
|
||||
.groupByKey(
|
||||
groupedAs("server-map-data.tiles.group", kxsBinary.serde(), kxsBinary.serde())
|
||||
)
|
||||
.reduce(
|
||||
"server-map-data.tiles.reduce",
|
||||
materializedAs("server-map-data.tiles.store", kxsBinary.serde(), kxsBinary.serde()),
|
||||
) { a, b -> a + b }
|
||||
|
||||
|
||||
val colourisedChunkTable: KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
|
||||
chunkedTilesTable
|
||||
.join(
|
||||
other = tileProtoColourDict,
|
||||
name = "server-map-data.join-tiles-with-prototypes",
|
||||
materialized = materializedWith(kxsBinary.serde(), kxsBinary.serde()),
|
||||
foreignKeyExtractor = { chunkTiles: ServerMapChunkTiles<TileProtoHashCode> -> chunkTiles.chunkId.serverId }
|
||||
) { tiles: ServerMapChunkTiles<TileProtoHashCode>, colourDict: TileColourDict ->
|
||||
|
||||
ServerMapChunkTiles(
|
||||
tiles.chunkId,
|
||||
tiles.map.mapValues { (_, code) ->
|
||||
colourDict.map.getOrElse(code) {
|
||||
println("missing prototype $code")
|
||||
ColourHex.TRANSPARENT
|
||||
}
|
||||
}
|
||||
)
|
||||
}
|
||||
|
||||
return colourisedChunkTable
|
||||
|
||||
}
|
||||
|
||||
//@Serializable
|
||||
//private data class FactorioServerMapAggregator(
|
||||
// val tilePrototypes: SortedMap<PrototypeName, Colour> = TreeMap(),
|
||||
// val surfaces: MutableMap<SurfaceIndex, FactorioMapSurfaceAggregator> = mutableMapOf(),
|
||||
//) {
|
||||
//
|
||||
// operator fun plus(packet: KafkatorioPacket): FactorioServerMapAggregator {
|
||||
// return when (packet) {
|
||||
// is FactorioEvent -> {
|
||||
// when (val data = packet.data) {
|
||||
// is MapChunk -> plus(data)
|
||||
// is MapTiles -> plus(data)
|
||||
// else -> this
|
||||
// }
|
||||
// }
|
||||
// is FactorioPrototypes -> plus(packet)
|
||||
// is FactorioConfigurationUpdate -> this
|
||||
// }
|
||||
// }
|
||||
//
|
||||
// operator fun plus(mapTiles: MapTiles): FactorioServerMapAggregator {
|
||||
//// println("FactorioServerMapAggregator: adding ${mapTiles.tiles.size} MapTiles from surface ${mapTiles.surfaceIndex}")
|
||||
//
|
||||
// val surfaceIndex = SurfaceIndex(mapTiles.surfaceIndex)
|
||||
//
|
||||
// val surface = surfaces.getOrPut(surfaceIndex) { FactorioMapSurfaceAggregator(surfaceIndex) }
|
||||
//
|
||||
// val surfaceTiles = mapTiles.tiles.associate { it.position to PrototypeName(it.prototypeName) }
|
||||
// surface.tiles.putAll(surfaceTiles)
|
||||
//
|
||||
// return this
|
||||
// }
|
||||
//
|
||||
// operator fun plus(mapChunk: MapChunk): FactorioServerMapAggregator {
|
||||
//// println("FactorioServerMapAggregator: adding mapChunk ${mapChunk.position}")
|
||||
// return plus(mapChunk.tiles)
|
||||
// }
|
||||
//
|
||||
// operator fun plus(packet: FactorioPrototypes): FactorioServerMapAggregator {
|
||||
//// println("FactorioServerMapAggregator: adding FactorioPrototypes")
|
||||
//
|
||||
// val packetPrototypes = packet.prototypes
|
||||
// .filterIsInstance<MapTilePrototype>()
|
||||
// .associate { PrototypeName(it.name) to it.mapColour }
|
||||
//
|
||||
// tilePrototypes.putAll(packetPrototypes)
|
||||
//
|
||||
// return this
|
||||
// }
|
||||
//
|
||||
//}
|
||||
|
||||
//
|
||||
//@Serializable
|
||||
//private data class FactorioMapSurfaceAggregator(
|
||||
// val index: SurfaceIndex,
|
||||
// val tiles: MutableMap<MapTilePosition, PrototypeName> = mutableMapOf(),
|
||||
//)
|
||||
|
|
|
|||
|
|
@ -3,10 +3,9 @@ package dev.adamko.kafkatorio.processor.topology
|
|||
import com.sksamuel.scrimage.ImmutableImage
|
||||
import com.sksamuel.scrimage.ScaleMethod
|
||||
import com.sksamuel.scrimage.nio.PngWriter
|
||||
import dev.adamko.kafkatorio.events.schema.Colour
|
||||
import dev.adamko.kafkatorio.events.schema.ColourHex
|
||||
import dev.adamko.kafkatorio.events.schema.MapChunkPosition
|
||||
import dev.adamko.kafkatorio.events.schema.MapTilePosition
|
||||
import dev.adamko.kafkatorio.events.schema.converters.toMapChunkPosition
|
||||
import java.awt.image.BufferedImage
|
||||
import java.io.File
|
||||
import kotlin.math.abs
|
||||
|
|
@ -15,17 +14,17 @@ import kotlinx.coroutines.CoroutineName
|
|||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.SupervisorJob
|
||||
import kotlinx.coroutines.delay
|
||||
import kotlinx.coroutines.flow.MutableStateFlow
|
||||
import kotlinx.coroutines.flow.asFlow
|
||||
import kotlinx.coroutines.channels.BufferOverflow
|
||||
import kotlinx.coroutines.flow.MutableSharedFlow
|
||||
import kotlinx.coroutines.flow.collect
|
||||
import kotlinx.coroutines.flow.conflate
|
||||
import kotlinx.coroutines.flow.filterNotNull
|
||||
import kotlinx.coroutines.flow.flatMapMerge
|
||||
import kotlinx.coroutines.flow.launchIn
|
||||
import kotlinx.coroutines.flow.mapLatest
|
||||
import kotlinx.coroutines.flow.onEach
|
||||
import kotlinx.coroutines.flow.debounce
|
||||
import kotlinx.coroutines.flow.distinctUntilChanged
|
||||
import kotlinx.coroutines.flow.distinctUntilChangedBy
|
||||
import kotlinx.coroutines.flow.map
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import kotlinx.coroutines.supervisorScope
|
||||
import org.apache.kafka.streams.kstream.KTable
|
||||
|
||||
|
||||
|
|
@ -36,106 +35,84 @@ private val cScope = CoroutineScope(
|
|||
)
|
||||
|
||||
/** A bridge between Kafka and Kotlin */
|
||||
private val serverMapDataFlow: MutableStateFlow<FactorioServerMap?> = MutableStateFlow(null)
|
||||
private val chunkedTiles: MutableSharedFlow<ServerMapChunkTiles<ColourHex>> =
|
||||
MutableSharedFlow(
|
||||
0,
|
||||
1,
|
||||
BufferOverflow.SUSPEND
|
||||
)
|
||||
|
||||
|
||||
fun saveMapTiles(
|
||||
serverMapTable: KTable<FactorioServerId, FactorioServerMap>
|
||||
serverMapTable: KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>>
|
||||
) {
|
||||
|
||||
serverMapTable
|
||||
.toStream()
|
||||
.foreach { _, value ->
|
||||
runBlocking(Dispatchers.Default) {
|
||||
serverMapDataFlow.emit(value)
|
||||
if (value != null) {
|
||||
runBlocking(Dispatchers.Default) {
|
||||
println("emitting chunkedTiles ${value.chunkId} tiles: ${value.map.size}")
|
||||
chunkedTiles.emit(value)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
cScope.launch {
|
||||
println("launching serverMapDataFlow merging")
|
||||
|
||||
|
||||
serverMapDataFlow
|
||||
.filterNotNull()
|
||||
|
||||
serverMapDataFlow
|
||||
.filterNotNull()
|
||||
.conflate()
|
||||
// .transform { serverMapData ->
|
||||
.flatMapMerge(10) { serverMapData ->
|
||||
println("flatMapMerge serverMapData ${serverMapData.serverId}, surfaces ${serverMapData.surfaces.keys}")
|
||||
|
||||
val prototypes = serverMapData.tilePrototypes
|
||||
|
||||
serverMapData
|
||||
.surfaces
|
||||
.values
|
||||
// .forEach { surfaceData ->
|
||||
.asFlow()
|
||||
.flatMapMerge(10) { surfaceData ->
|
||||
println("flatMapMerge surfaceData ${surfaceData.index}")
|
||||
|
||||
// get all tile colours for the surface
|
||||
val surfaceTileColours = surfaceData.tiles
|
||||
.entries
|
||||
.mapNotNull { (tilePosition, prototypeName) ->
|
||||
prototypes[prototypeName]?.let { prototype ->
|
||||
tilePosition to prototype.mapColour
|
||||
}
|
||||
}
|
||||
println("surfaceTileColours.size ${surfaceTileColours.size}")
|
||||
|
||||
// split a surface into chunks, per zoom-level
|
||||
ZoomLevel.values
|
||||
// .forEach { zoom ->
|
||||
.asFlow()
|
||||
.flatMapMerge(10) { zoom ->
|
||||
println("flatMapMerge zoom $zoom")
|
||||
|
||||
surfaceTileColours
|
||||
.groupBy { (tilePosition, _) ->
|
||||
tilePosition.toMapChunkPosition(zoom.chunkSize)
|
||||
}
|
||||
.entries
|
||||
// .forEach { (chunkPosition, tileColours) ->
|
||||
.asFlow()
|
||||
.mapLatest { (chunkPosition, tileColours) ->
|
||||
println("mapLatest chunkPosition $chunkPosition")
|
||||
|
||||
// mapLatest + delay = an alternative to debounce-per-key?
|
||||
delay(30.seconds)
|
||||
|
||||
println("emitting ChunkData $chunkPosition")
|
||||
|
||||
ChunkData(
|
||||
surfaceData.index,
|
||||
zoom,
|
||||
chunkPosition,
|
||||
tileColours.toMap()
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
.onEach { chunkData ->
|
||||
|
||||
println("starting to save chunkData ${chunkData.chunkPosition}...")
|
||||
|
||||
val chunkImageFile = File(chunkData.filename.value)
|
||||
if (chunkImageFile.parentFile.mkdirs()) {
|
||||
println("created new map tile parentFile directory ${chunkImageFile.absolutePath}")
|
||||
supervisorScope {
|
||||
chunkedTiles
|
||||
.conflate()
|
||||
.distinctUntilChanged()
|
||||
.debounce(5.seconds)
|
||||
.map {
|
||||
println("createMapTileImage ${it.chunkId}")
|
||||
createMapTileImage(it)
|
||||
}
|
||||
|
||||
val image =
|
||||
createMapTileImage(chunkData.chunkPosition, chunkData.tileColours, chunkData.zoomLevel)
|
||||
|
||||
println("saving map tile $chunkImageFile")
|
||||
val savedTile = image.output(PngWriter.NoCompression, chunkImageFile)
|
||||
println("savedTile: $savedTile")
|
||||
|
||||
}
|
||||
.launchIn(this)
|
||||
.collect()
|
||||
}
|
||||
}
|
||||
|
||||
// cScope.launch {
|
||||
// println("launching serverMapDataFlow merging")
|
||||
//
|
||||
// chunkedTiles
|
||||
// .runningFold(
|
||||
// mapOf<ServerMapChunkId, MutableSharedFlow<ServerMapChunkTiles<ColourHex>>>()
|
||||
// ) { accumulator, value ->
|
||||
//
|
||||
// println("folding chunkedTiles ${value.chunkId}, current flows: ${accumulator.keys}")
|
||||
//
|
||||
// val chunkFlow = accumulator.getOrElse(value.chunkId) {
|
||||
//
|
||||
// println("creating new MutableSharedFlow for ${value.chunkId}")
|
||||
//
|
||||
// val input = MutableSharedFlow<ServerMapChunkTiles<ColourHex>>(
|
||||
// replay = 0,
|
||||
// extraBufferCapacity = 1,
|
||||
// onBufferOverflow = BufferOverflow.DROP_OLDEST,
|
||||
// )
|
||||
//
|
||||
// supervisorScope {
|
||||
// println("launching MutableSharedFlow for ${value.chunkId}")
|
||||
// input
|
||||
// .debounce(30.seconds)
|
||||
// .mapLatest {
|
||||
// println("createMapTileImage ${it.chunkId}")
|
||||
// createMapTileImage(it)
|
||||
// }.launchIn(this)
|
||||
// }
|
||||
//
|
||||
// input
|
||||
// }
|
||||
//
|
||||
// chunkFlow.emit(value)
|
||||
//
|
||||
// accumulator + (value.chunkId to chunkFlow)
|
||||
// }
|
||||
// .collect()
|
||||
//
|
||||
// println("end of cScope.launch")
|
||||
// }
|
||||
}
|
||||
|
||||
|
||||
|
|
@ -145,35 +122,58 @@ private value class TilePngFilename(
|
|||
) {
|
||||
|
||||
constructor(
|
||||
surfaceIndex: SurfaceIndex,
|
||||
zoom: Int,
|
||||
chunkX: Int,
|
||||
chunkY: Int,
|
||||
) : this("src/main/resources/kafkatorio-web-map/s${surfaceIndex.index}/z$zoom/x${chunkX}/y${chunkY}.png")
|
||||
|
||||
}
|
||||
|
||||
|
||||
private data class ChunkData(
|
||||
val surfaceIndex: SurfaceIndex,
|
||||
val zoomLevel: ZoomLevel,
|
||||
val chunkPosition: MapChunkPosition,
|
||||
val tileColours: Map<MapTilePosition, Colour>,
|
||||
) {
|
||||
val filename = TilePngFilename(surfaceIndex, zoomLevel.level, chunkPosition.x, chunkPosition.y)
|
||||
chunkId: ServerMapChunkId
|
||||
) : this(buildString {
|
||||
append("src/main/resources/kafkatorio-web-map/s")
|
||||
append(chunkId.surfaceIndex.index)
|
||||
append("/z")
|
||||
append(chunkId.zoomLevel.level)
|
||||
append("/x")
|
||||
append(chunkId.chunkPosition.x)
|
||||
append("/y")
|
||||
append(chunkId.chunkPosition.y)
|
||||
append(".png")
|
||||
})
|
||||
}
|
||||
|
||||
private fun createMapTileImage(
|
||||
chunkTiles: ServerMapChunkTiles<ColourHex>
|
||||
) {
|
||||
|
||||
println("emitting ChunkData ${chunkTiles.chunkId.chunkPosition}")
|
||||
|
||||
println("starting to save chunkData ${chunkTiles.chunkId.chunkPosition}...")
|
||||
|
||||
val filename = TilePngFilename(chunkTiles.chunkId)
|
||||
|
||||
val chunkImageFile = File(filename.value)
|
||||
if (chunkImageFile.parentFile.mkdirs()) {
|
||||
println("created new map tile parentFile directory ${chunkImageFile.absolutePath}")
|
||||
}
|
||||
|
||||
val image =
|
||||
createMapTileImage(
|
||||
chunkTiles.chunkId.chunkPosition,
|
||||
chunkTiles.map,
|
||||
chunkTiles.chunkId.zoomLevel
|
||||
)
|
||||
|
||||
println("saving map tile $chunkImageFile")
|
||||
val savedTile = image.output(PngWriter.NoCompression, chunkImageFile)
|
||||
println("savedTile: $savedTile")
|
||||
}
|
||||
|
||||
|
||||
private fun createMapTileImage(
|
||||
chunkPosition: MapChunkPosition,
|
||||
chunkColours: Map<MapTilePosition, Colour>,
|
||||
chunkColours: Map<MapTilePosition, ColourHex>,
|
||||
z: ZoomLevel
|
||||
): ImmutableImage {
|
||||
|
||||
|
||||
val chunkImage = ImmutableImage.filled(
|
||||
z.chunkSize,
|
||||
z.chunkSize,
|
||||
COLOUR_TRANSPARENT.awt(),
|
||||
ColourHex.TRANSPARENT.toRgbColor().awt(),
|
||||
BufferedImage.TYPE_INT_ARGB
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@ package dev.adamko.kafkatorio.processor.topology
|
|||
|
||||
import com.sksamuel.scrimage.color.RGBColor
|
||||
import dev.adamko.kafkatorio.events.schema.Colour
|
||||
import dev.adamko.kafkatorio.events.schema.ColourHex
|
||||
import kotlin.math.pow
|
||||
import kotlin.math.roundToInt
|
||||
import kotlinx.serialization.Serializable
|
||||
|
|
@ -23,12 +24,12 @@ value class FactorioServerId(val id: String) {
|
|||
}
|
||||
|
||||
|
||||
fun Colour.toRgbColor(): RGBColor {
|
||||
fun ColourHex.toRgbColor(): RGBColor {
|
||||
return RGBColor(
|
||||
red.roundToInt(),
|
||||
green.roundToInt(),
|
||||
blue.roundToInt(),
|
||||
alpha.roundToInt(),
|
||||
red.toInt(),
|
||||
green.toInt(),
|
||||
blue.toInt(),
|
||||
alpha.toInt(),
|
||||
)
|
||||
}
|
||||
|
||||
|
|
@ -36,15 +37,14 @@ fun Colour.toRgbColor(): RGBColor {
|
|||
enum class ZoomLevel(
|
||||
val level: Int
|
||||
) {
|
||||
ZOOM_0(0),
|
||||
ZOOM_1(1),
|
||||
ZOOM_2(2),
|
||||
ZOOM_3(3),
|
||||
ZOOM_4(4),
|
||||
ZOOM_5(5),
|
||||
ZOOM_0(0), // 512
|
||||
ZOOM_1(1), // 256
|
||||
ZOOM_2(2), // 128
|
||||
ZOOM_3(3), // 64
|
||||
ZOOM_4(4), // 32
|
||||
;
|
||||
|
||||
val chunkSize: Int = 2f.pow(10 - level).roundToInt()
|
||||
val chunkSize: Int = 2f.pow(9 - level).roundToInt()
|
||||
|
||||
init {
|
||||
require(chunkSize > 0) { "chunkSize must be positive" }
|
||||
|
|
|
|||
|
|
@ -1,19 +1,17 @@
|
|||
package dev.adamko.kafkatorio.processor.topology
|
||||
|
||||
import com.sksamuel.scrimage.color.RGBColor
|
||||
import dev.adamko.kafkatorio.events.schema.Colour
|
||||
import dev.adamko.kafkatorio.events.schema.ColourHex
|
||||
import dev.adamko.kafkatorio.events.schema.FactorioPrototypes
|
||||
import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
|
||||
import dev.adamko.kafkatorio.events.schema.MapTile
|
||||
import dev.adamko.kafkatorio.events.schema.MapTilePrototype
|
||||
import dev.adamko.kafkatorio.processor.serdes.jsonMapper
|
||||
import dev.adamko.kafkatorio.processor.serdes.kxsBinary
|
||||
import dev.adamko.kotka.extensions.consumedAs
|
||||
import dev.adamko.kotka.extensions.materializedWith
|
||||
import dev.adamko.kotka.extensions.streams.flatMap
|
||||
import dev.adamko.kotka.extensions.materializedAs
|
||||
import dev.adamko.kotka.extensions.streams.mapValues
|
||||
import dev.adamko.kotka.extensions.streams.toTable
|
||||
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.KStream
|
||||
import org.apache.kafka.streams.kstream.KTable
|
||||
|
||||
|
||||
|
|
@ -24,50 +22,45 @@ value class PrototypeName(val name: String) {
|
|||
}
|
||||
|
||||
|
||||
//fun prototypesTable(builder: StreamsBuilder): KTable<PrototypeName, MapTilePrototype> {
|
||||
// return builder.stream<FactorioPacketKey, FactorioPrototypes>(
|
||||
// "kafkatorio.${KafkatorioPacket.PacketType.PROTOTYPES}.all",
|
||||
// consumedAs(
|
||||
// "consume-all-prototypes-packets",
|
||||
// jsonMapper.serde(),
|
||||
// jsonMapper.serde()
|
||||
// )
|
||||
// )
|
||||
// .flatMap("map-MapTilePrototype") { _, prototypes: FactorioPrototypes ->
|
||||
// prototypes.prototypes
|
||||
// .filterIsInstance<MapTilePrototype>()
|
||||
// .map { PrototypeName(it.name) to it }
|
||||
// }
|
||||
// .toTable(
|
||||
// materializedWith(
|
||||
// kxsBinary.serde(),
|
||||
// kxsBinary.serde()
|
||||
// )
|
||||
// )
|
||||
//}
|
||||
@JvmInline
|
||||
@Serializable
|
||||
value class TileProtoHashCode(val code: Int) {
|
||||
constructor(prototype: MapTilePrototype) : this(prototype.name.hashCode())
|
||||
constructor(tile: MapTile) : this(tile.prototypeName.hashCode())
|
||||
}
|
||||
|
||||
|
||||
/** Get the [Colour] of each [MapTilePrototype] */
|
||||
fun tilePrototypeColourTable(builder: StreamsBuilder): KTable<PrototypeName, Colour> {
|
||||
return builder.stream<FactorioServerId, FactorioPrototypes>(
|
||||
"kafkatorio.${KafkatorioPacket.PacketType.PROTOTYPES}.all",
|
||||
consumedAs(
|
||||
"consume-all-prototypes-packets",
|
||||
jsonMapper.serde(),
|
||||
jsonMapper.serde()
|
||||
)
|
||||
)
|
||||
.flatMap("map-MapTilePrototype") { _, prototypes: FactorioPrototypes ->
|
||||
prototypes.prototypes
|
||||
.filterIsInstance<MapTilePrototype>()
|
||||
.map {
|
||||
PrototypeName(it.name) to it.mapColour
|
||||
}
|
||||
@JvmInline
|
||||
@Serializable
|
||||
value class TileColourDict(val map: Map<TileProtoHashCode, ColourHex>) {
|
||||
operator fun plus(other: TileColourDict) = TileColourDict(this.map + other.map)
|
||||
}
|
||||
|
||||
|
||||
/** Aggregate [KafkatorioPacket]s into a single [FactorioServerMap] (per [serverId][FactorioServerId]). */
|
||||
fun tileProtoColourDictionary(
|
||||
factorioServerPacketStream: KStream<FactorioServerId, KafkatorioPacket>
|
||||
): KTable<FactorioServerId, TileColourDict> {
|
||||
|
||||
// get all the prototypes
|
||||
return factorioServerPacketStream
|
||||
.mapValues("server-map-data.tile-prototypes.mapValues") { _, packet ->
|
||||
val map = when (packet) {
|
||||
is FactorioPrototypes -> packet
|
||||
.prototypes
|
||||
.filterIsInstance<MapTilePrototype>()
|
||||
.associate { TileProtoHashCode(it) to it.mapColour.toHex() }
|
||||
else -> mapOf()
|
||||
}
|
||||
TileColourDict(map)
|
||||
}
|
||||
.toTable(
|
||||
materializedWith(
|
||||
kxsBinary.serde(),
|
||||
kxsBinary.serde()
|
||||
"server-map-data.tile-prototypes",
|
||||
materializedAs(
|
||||
"server-map-data.tile-prototypes.store",
|
||||
jsonMapper.serde(),
|
||||
jsonMapper.serde()
|
||||
)
|
||||
)
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -13,89 +13,89 @@ import kotlin.math.abs
|
|||
import kotlin.math.roundToInt
|
||||
import org.apache.kafka.streams.kstream.KTable
|
||||
|
||||
|
||||
val COLOUR_TRANSPARENT = RGBColor.fromAwt(Color(0f, 0f, 0f, 0f))
|
||||
|
||||
fun saveTileImages(
|
||||
webMapTiles: KTable<WebMapTileChunkPosition, WebMapTileChunkPixels>
|
||||
) {
|
||||
|
||||
webMapTiles
|
||||
.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
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
//val saveMapTilesContext = Dispatchers.IO.limitedParallelism(1)
|
||||
|
||||
|
||||
private fun saveMapTilesPng(
|
||||
chunkPosition: WebMapTileChunkPosition,
|
||||
pixels: Map<MapTilePosition, Colour>,
|
||||
) {
|
||||
val chunkOriginX: Int = chunkPosition.x * chunkPosition.chunkSize // - 1
|
||||
val chunkOriginY: Int = chunkPosition.y * chunkPosition.chunkSize // - 1
|
||||
val surfaceIndex: Int = chunkPosition.surfaceIndex.index
|
||||
|
||||
val chunkImage =
|
||||
ImmutableImage.filled(
|
||||
chunkPosition.chunkSize,
|
||||
chunkPosition.chunkSize,
|
||||
COLOUR_TRANSPARENT.awt(),
|
||||
BufferedImage.TYPE_INT_ARGB
|
||||
)
|
||||
|
||||
pixels.forEach { (tilePosition, colour) ->
|
||||
|
||||
val rgbColour = colour.toRgbColor()
|
||||
|
||||
val pixelX = abs(abs(tilePosition.x) - abs(chunkOriginX))
|
||||
val pixelY = abs(abs(tilePosition.y) - abs(chunkOriginY))
|
||||
|
||||
chunkImage.setColor(
|
||||
pixelX,
|
||||
pixelY,
|
||||
rgbColour
|
||||
)
|
||||
|
||||
}
|
||||
|
||||
val zoom = 1u
|
||||
val file =
|
||||
File(tileFilename("$surfaceIndex", "$zoom", "${chunkPosition.x}", "${chunkPosition.y}"))
|
||||
|
||||
if (file.parentFile.mkdirs()) {
|
||||
println("created new map tile parentFile directory ${file.absolutePath}")
|
||||
}
|
||||
|
||||
println("saving map tile $file")
|
||||
val savedTile = chunkImage.output(PngWriter.NoCompression, file)
|
||||
println("savedTile: $savedTile")
|
||||
}
|
||||
|
||||
fun tileFilename(
|
||||
surfaceIndex: String,
|
||||
zoom: String,
|
||||
chunkX: String,
|
||||
chunkY: String,
|
||||
) = "src/main/resources/kafkatorio-web-map/s${surfaceIndex}/z$zoom/x${chunkX}/y${chunkY}.png"
|
||||
|
||||
//fun subdivide(tile: ImmutableImage, size: Int) {
|
||||
//
|
||||
// val halfSize = size / 2
|
||||
//val COLOUR_TRANSPARENT: RGBColor = RGBColor.fromAwt(Color(0f, 0f, 0f, 0f))
|
||||
//
|
||||
// val bl = tile.resizeTo(halfSize, halfSize, Position.BottomLeft)
|
||||
// val savedTile = bl.output(PngWriter.NoCompression, file)
|
||||
//
|
||||
// tile.resizeTo(halfSize, halfSize, Position.BottomRight)
|
||||
// tile.resizeTo(halfSize, halfSize, Position.TopLeft)
|
||||
// tile.resizeTo(halfSize, halfSize, Position.TopRight)
|
||||
//fun saveTileImages(
|
||||
// webMapTiles: KTable<WebMapTileChunkPosition, WebMapTileChunkPixels>
|
||||
//) {
|
||||
//
|
||||
// webMapTiles
|
||||
// .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
|
||||
// }
|
||||
// }
|
||||
//}
|
||||
//
|
||||
////val saveMapTilesContext = Dispatchers.IO.limitedParallelism(1)
|
||||
//
|
||||
//
|
||||
//private fun saveMapTilesPng(
|
||||
// chunkPosition: WebMapTileChunkPosition,
|
||||
// pixels: Map<MapTilePosition, Colour>,
|
||||
//) {
|
||||
// val chunkOriginX: Int = chunkPosition.x * chunkPosition.chunkSize // - 1
|
||||
// val chunkOriginY: Int = chunkPosition.y * chunkPosition.chunkSize // - 1
|
||||
// val surfaceIndex: Int = chunkPosition.surfaceIndex.index
|
||||
//
|
||||
// val chunkImage =
|
||||
// ImmutableImage.filled(
|
||||
// chunkPosition.chunkSize,
|
||||
// chunkPosition.chunkSize,
|
||||
// COLOUR_TRANSPARENT.awt(),
|
||||
// BufferedImage.TYPE_INT_ARGB
|
||||
// )
|
||||
//
|
||||
// pixels.forEach { (tilePosition, colour) ->
|
||||
//
|
||||
// val rgbColour = colour.toRgbColor()
|
||||
//
|
||||
// val pixelX = abs(abs(tilePosition.x) - abs(chunkOriginX))
|
||||
// val pixelY = abs(abs(tilePosition.y) - abs(chunkOriginY))
|
||||
//
|
||||
// chunkImage.setColor(
|
||||
// pixelX,
|
||||
// pixelY,
|
||||
// rgbColour
|
||||
// )
|
||||
//
|
||||
// }
|
||||
//
|
||||
// val zoom = 1u
|
||||
// val file =
|
||||
// File(tileFilename("$surfaceIndex", "$zoom", "${chunkPosition.x}", "${chunkPosition.y}"))
|
||||
//
|
||||
// if (file.parentFile.mkdirs()) {
|
||||
// println("created new map tile parentFile directory ${file.absolutePath}")
|
||||
// }
|
||||
//
|
||||
// println("saving map tile $file")
|
||||
// val savedTile = chunkImage.output(PngWriter.NoCompression, file)
|
||||
// println("savedTile: $savedTile")
|
||||
//}
|
||||
//
|
||||
//fun tileFilename(
|
||||
// surfaceIndex: String,
|
||||
// zoom: String,
|
||||
// chunkX: String,
|
||||
// chunkY: String,
|
||||
//) = "src/main/resources/kafkatorio-web-map/s${surfaceIndex}/z$zoom/x${chunkX}/y${chunkY}.png"
|
||||
//
|
||||
////fun subdivide(tile: ImmutableImage, size: Int) {
|
||||
////
|
||||
//// val halfSize = size / 2
|
||||
////
|
||||
//// val bl = tile.resizeTo(halfSize, halfSize, Position.BottomLeft)
|
||||
//// val savedTile = bl.output(PngWriter.NoCompression, file)
|
||||
////
|
||||
//// tile.resizeTo(halfSize, halfSize, Position.BottomRight)
|
||||
//// tile.resizeTo(halfSize, halfSize, Position.TopLeft)
|
||||
//// tile.resizeTo(halfSize, halfSize, Position.TopRight)
|
||||
////
|
||||
////}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,62 @@
|
|||
package dev.adamko.kafkatorio.processor.misc
|
||||
|
||||
import dev.adamko.kafkatorio.events.schema.ColourHex
|
||||
import dev.adamko.kafkatorio.events.schema.MapChunkPosition
|
||||
import dev.adamko.kafkatorio.events.schema.MapTilePosition
|
||||
import dev.adamko.kafkatorio.processor.serdes.kxsBinary
|
||||
import dev.adamko.kafkatorio.processor.topology.FactorioServerId
|
||||
import dev.adamko.kafkatorio.processor.topology.ServerMapChunkId
|
||||
import dev.adamko.kafkatorio.processor.topology.ServerMapChunkTiles
|
||||
import dev.adamko.kafkatorio.processor.topology.SurfaceIndex
|
||||
import dev.adamko.kafkatorio.processor.topology.ZoomLevel
|
||||
import java.util.UUID
|
||||
import kotlinx.serialization.encodeToByteArray
|
||||
import org.junit.jupiter.api.Test
|
||||
|
||||
class Blah {
|
||||
|
||||
@Test
|
||||
fun asdad() {
|
||||
|
||||
val tiles = ServerMapChunkTiles<ColourHex>(
|
||||
chunkId = ServerMapChunkId(
|
||||
serverId = FactorioServerId(UUID.randomUUID().toString()),
|
||||
chunkPosition = MapChunkPosition(Int.MAX_VALUE, Int.MAX_VALUE),
|
||||
surfaceIndex = SurfaceIndex(Int.MAX_VALUE),
|
||||
zoomLevel = ZoomLevel.ZOOM_0
|
||||
),
|
||||
buildMap {
|
||||
repeat(512) { x ->
|
||||
repeat(512) { y ->
|
||||
put(MapTilePosition(x, y), ColourHex(255u, 255u, 255u, 255u))
|
||||
}
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
val encode = kxsBinary.encodeToByteArray(tiles)
|
||||
|
||||
encode.inputStream()
|
||||
val size = encode.size * Byte.SIZE_BYTES
|
||||
|
||||
println(size)
|
||||
|
||||
|
||||
val map2 = buildList {
|
||||
repeat(512) { x ->
|
||||
repeat(512) { y ->
|
||||
val element = (x to y) to Int.MAX_VALUE
|
||||
add(element)
|
||||
}
|
||||
}
|
||||
}
|
||||
// .toTypedArray()
|
||||
val encode2 = kxsBinary.encodeToByteArray(map2)
|
||||
|
||||
encode2.inputStream()
|
||||
val size2 = encode2.size * Byte.SIZE_BYTES
|
||||
|
||||
println(size2)
|
||||
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue