🚧 OSHDB osm source

.gitignore

@@ -298,3 +298,4 @@ temp_*
 *.trace.db
 !oshdb-api/src/test/resources/test-data.mv.db
 !oshdb-util/src/test/resources/test-data.mv.db
+!oshdb-osm-source/src/test/resources/sample.pbf

oshdb-osm-source/pom.xml

@@ -0,0 +1,73 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <artifactId>oshdb-parent</artifactId>
+        <groupId>org.heigit.ohsome</groupId>
+        <version>1.0.0-SNAPSHOT</version>
+    </parent>
+    <artifactId>oshdb-osm-source</artifactId>
+
+    <properties>
+    </properties>
+
+    <dependencyManagement>
+        <dependencies>
+            <dependency>
+                <groupId>io.projectreactor</groupId>
+                <artifactId>reactor-bom</artifactId>
+                <version>2020.0.24</version>
+                <type>pom</type>
+                <scope>import</scope>
+            </dependency>
+        </dependencies>
+    </dependencyManagement>
+
+
+    <dependencies>
+        <dependency>
+            <groupId>io.projectreactor</groupId>
+            <artifactId>reactor-core</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.heigit.ohsome</groupId>
+            <artifactId>oshdb-util</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <!-- https://mvnrepository.com/artifact/com.google.protobuf/protobuf-java -->
+        <dependency>
+            <groupId>com.google.protobuf</groupId>
+            <artifactId>protobuf-java</artifactId>
+            <version>3.21.9</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.openstreetmap.pbf</groupId>
+            <artifactId>osmpbf</artifactId>
+            <version>1.5.0</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>com.google.protobuf</groupId>
+                    <artifactId>protobuf-java</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <dependency>
+            <groupId>io.projectreactor</groupId>
+            <artifactId>reactor-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <version>3.23.1</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+</project>

oshdb-osm-source/src/main/java/org/heigit/ohsome/oshdb/TagTranslator.java

@@ -0,0 +1,18 @@
+package org.heigit.ohsome.oshdb;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.Set;
+import org.heigit.ohsome.oshdb.util.tagtranslator.OSMRole;
+import org.heigit.ohsome.oshdb.util.tagtranslator.OSMTag;
+
+public interface TagTranslator {
+
+  Map<OSMTag, OSHDBTag> getOSHDBTagOf(Collection<? extends OSMTag> tags);
+
+  Map<OSHDBTag, OSMTag> lookupTag(Set<? extends OSHDBTag> tags);
+
+  Map<OSMRole, OSHDBRole> getOSHDBRoleOf(Collection<? extends OSMRole> values);
+
+  Map<OSHDBRole, OSMRole> lookupRole(Set<? extends OSHDBRole> roles);
+}

oshdb-osm-source/src/main/java/org/heigit/ohsome/oshdb/osm/OSMSource.java

@@ -0,0 +1,9 @@
+package org.heigit.ohsome.oshdb.osm;
+
+import org.heigit.ohsome.oshdb.TagTranslator;
+import reactor.core.publisher.Flux;
+
+public interface OSMSource {
+
+  Flux<OSMEntity> entities(TagTranslator translator);
+}

oshdb-osm-source/src/main/java/org/heigit/ohsome/oshdb/osm/pbf/Blob.java

@@ -0,0 +1,125 @@
+package org.heigit.ohsome.oshdb.osm.pbf;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+import crosby.binary.Fileformat;
+import crosby.binary.Osmformat;
+import crosby.binary.file.FileFormatException;
+import java.io.DataInputStream;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.NoSuchElementException;
+import java.util.zip.DataFormatException;
+import java.util.zip.Inflater;
+import reactor.core.publisher.Mono;
+
+public class Blob {
+
+  private static final int MAX_HEADER_SIZE = 64 * 1024;
+
+  public static Blob read(InputStream input) throws IOException {
+    DataInputStream dataInput = new DataInputStream(input);
+    var headerSize = dataInput.readInt();
+    if (headerSize > MAX_HEADER_SIZE) {
+      throw new FileFormatException(
+          "Unexpectedly long header " + MAX_HEADER_SIZE + " bytes. Possibly corrupt file.");
+    }
+
+    var buf = new byte[headerSize];
+    dataInput.readFully(buf);
+    var header = Fileformat.BlobHeader.parseFrom(buf);
+
+    var offset = position(input);
+
+    var data = new byte[header.getDatasize()];
+    dataInput.readFully(data);
+
+    return new Blob(header.getType(), offset, data);
+  }
+
+  private static long position(InputStream input) throws IOException {
+    if (input instanceof FileInputStream) {
+      return ((FileInputStream) input).getChannel().position();
+    }
+    return -1;
+  }
+
+  private final String type;
+  private final long offset;
+  private final byte[] data;
+
+  private Blob(String type, long offset, byte[] data) {
+    this.type = type;
+    this.offset = offset;
+    this.data = data;
+  }
+
+  public long offset() {
+    return offset;
+  }
+
+  public byte[] data() {
+    return data;
+  }
+
+  public boolean isHeader() {
+    return "OSMHeader".equals(type);
+  }
+
+  public Osmformat.HeaderBlock header() throws FileFormatException {
+    if (!isHeader()) {
+      throw new NoSuchElementException();
+    }
+
+    try {
+      return Osmformat.HeaderBlock.parseFrom(decompress());
+    } catch (InvalidProtocolBufferException e) {
+      throw new FileFormatException(e);
+    }
+  }
+
+  public boolean isData() {
+    return "OSMData".equals(type);
+  }
+
+  public Mono<Block> block() {
+    if (!isData()) {
+      return Mono.error(new NoSuchElementException());
+    }
+    return Mono.fromCallable(() -> Block.parse(this, decompress()));
+  }
+
+  private byte[] decompress() throws FileFormatException {
+    var blob = parseBlob();
+    if (blob.hasRaw()) {
+      return blob.getRaw().toByteArray();
+    }
+    if (blob.hasZlibData()) {
+      return decompress(blob);
+    }
+    throw new UnsupportedOperationException();
+  }
+
+  private static byte[] decompress(Fileformat.Blob blob) throws FileFormatException {
+    var buffer = new byte[blob.getRawSize()];
+    Inflater inflater = new Inflater();
+    try {
+      inflater.setInput(blob.getZlibData().toByteArray());
+      inflater.inflate(buffer);
+      assert (inflater.finished());
+    } catch (DataFormatException e) {
+      throw new FileFormatException(e);
+    } finally {
+      inflater.end();
+    }
+    return buffer;
+  }
+
+  private Fileformat.Blob parseBlob() throws FileFormatException {
+    try {
+      return Fileformat.Blob.parseFrom(data);
+    } catch (InvalidProtocolBufferException e) {
+      throw new FileFormatException(e);
+    }
+  }
+}