D-Net

package eu.dnetlib.data.transform;

import com.google.common.collect.Lists;

import com.google.common.collect.Maps;

import com.googlecode.protobuf.format.JsonFormat;

import eu.dnetlib.data.graph.model.DNGFRowKeyDecoder;

import eu.dnetlib.data.mapreduce.util.dao.HBaseTableDAO;

import eu.dnetlib.data.proto.DNGFProtos.DNGF;

import eu.dnetlib.data.proto.TypeProtos.Type;

import eu.dnetlib.data.proto.dli.Scholix2ObjectProtos;

import eu.dnetlib.dli.proto.DNGF2Scholix2Converter;

import eu.dnetlib.dli.proto.DNGFDLISummaryConverter;

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.junit.Before;

import org.junit.Test;

import java.util.Iterator;

import java.util.List;

import java.util.Map;

import java.util.Map.Entry;

import static eu.dnetlib.data.graph.utils.RelDescriptor.QUALIFIER_SEPARATOR;

import static eu.dnetlib.data.proto.dli.Scholix2ObjectProtos.*;

/**

 * Created by sandro on 2/13/17.

 */

public class DLIConvertertTest extends AbstractTransformerTest {

	private static final Log log = LogFactory.getLog(DLIConvertertTest.class);

	private Ontologies ontologies;

    @Before

    public void setUp() throws Exception {

        factory = new XsltRowTransformerFactory();

        ontologies = OntologyLoader.loadOntologiesFromCp();

    }

    @Test

    public void testLinkPangaeaDLI() throws Exception {

        final List<Row> rows = Lists.newArrayList();

        rows.addAll(asRows(loadFromTransformationProfile("dmfdli2hbase.xml"), load("record_dli_dmf.xml")));

        rows.addAll(asRows(loadFromTransformationProfile("dmfdli2hbase.xml"), load("record_dli_dmf2.xml")));

//        rows.addAll(asRows(loadFromTransformationProfile("pmfdli2hbase.xml"), load("record_dli_pmf.xml")));

	    printAllSummary(mapAllSummary(buildTable(rows)));

    }

	@Test

	public void testLinkPangaeaDLIScholix() throws Exception {

		final List<Row> rows = Lists.newArrayList();

		rows.addAll(asRows(loadFromTransformationProfile("dmfdli2hbase.xml"), load("record_dli_dmf2.xml")));

		rows.addAll(asRows(loadFromTransformationProfile("dmfdli2hbase.xml"), load("record_dli_dmf.xml")));

//		rows.forEach(row ->

//				row.getColumns().forEach(

//						result -> {

//							if (result != null) {

//

//

//								final DNGFDecoder decoder =

//										DNGFDecoder.decode(result.getValue(), DliFieldTypeProtos.completionStatus, DliProtos.completionStatus, DliProtos.resolvedfrom, DliProtos.typedIdentifier);

//								System.out.println(row.getKey());

//								Put put = HBaseTableDAO.asPutByCollectedFrom(decoder.getDNGF());

//								System.out.println("put.getTimeStamp() = " + put.getTimeStamp());

//

//								System.out.println("decoder.getDNGF().toString() = " + decoder.getDNGF().toString());

//							}

//						}

//				));

//		//rows.addAll(asRows(loadFromTransformationProfile("pmfdli2hbase.xml"), load("record_dli_pmf.xml")));

		reduceScholix(mapAllScholix(buildTable(rows)));

	}

	private void reduceScholix(final Map<String, List<Scholix>> scholix) {

    	scholix.forEach((id, list) -> {

			    final Iterator<Scholix> it = list.iterator();

			    final Scholix source = it.next();

			    if (source.hasSource()) {

				    it.forEachRemaining(s -> log.info(JsonFormat.printToString(

						    Scholix.newBuilder(s)

								    .setSource(source.getSource())

								    .build())));

			    } else {

			    	log.warn("missing source in scholix: " + id);

			    }

		    }

	    );

    }

	private void printAllSummary(final Map<String, DNGFDLISummaryConverter> converters) {

		converters.forEach((id, converter) -> log.info(JsonFormat.printToString(converter.convert())));

	}

	private void printAllScholix(final Map<String, List<Scholix>> converters) {

		converters.values().forEach(converter -> converter.forEach(s -> log.info(JsonFormat.printToString(s))));

	}

	private Map<String, DNGFDLISummaryConverter> mapAllSummary(final Map<String, Map<String, Map<String, byte[]>>> table) throws Exception {

		final Map<String, DNGFDLISummaryConverter> builders = Maps.newHashMap();

		for (final Entry<String, Map<String, Map<String, byte[]>>> e : table.entrySet()) {

			mapSummary(builders, e.getKey(), e.getValue());

	    }

	    return builders;

    }

	private Map<String, List<Scholix>> mapAllScholix(final Map<String, Map<String, Map<String, byte[]>>> table) throws Exception {

		final Map<String, List<Scholix>> builders = Maps.newHashMap();

		for (final Entry<String, Map<String, Map<String, byte[]>>> e : table.entrySet()) {

			mapScholix(builders, e.getKey(), e.getValue());

		}

		return builders;

	}

	private void mapScholix(final Map<String, List<Scholix>> builders, final String rowKey, final Map<String, Map<String, byte[]>> row) {

		final Type type = DNGFRowKeyDecoder.decode(rowKey).getType();

		final Map<String, byte[]> familyMap = row.get("metadata");

		if (familyMap == null) return;

		final byte[] bodyB = familyMap.get(type.toString());

		if (bodyB != null) {

			ensureScholixBuilder(builders, rowKey);

			final DNGF entity = HBaseTableDAO.parseProto(familyMap.get(type.toString()));

			final Scholix.Builder source = DNGF2Scholix2Converter.withSource(entity.getEntity());

			builders.get(rowKey).add(source.build());

			final Map<String, byte[]> rels = row.get("rels");

			rels.forEach((q, v) -> {

				final DNGF.Builder r = DNGF.newBuilder(HBaseTableDAO.parseProto(v));

                final Scholix.Builder target = DNGF2Scholix2Converter.withTarget(entity.getEntity(), r.getRel());

                ensureScholixBuilder(builders, r.getRel().getTarget());

				builders.get(r.getRel().getTarget()).add(target.build());

			});

		}

	}

	private void mapSummary(final Map<String, DNGFDLISummaryConverter> converters, final String rowKey, final Map<String, Map<String, byte[]>> row) throws Exception {

		final Type type = DNGFRowKeyDecoder.decode(rowKey).getType();

		final Map<String, byte[]> familyMap = row.get("metadata");

		if (familyMap == null) return;

		final byte[] bodyB = familyMap.get(type.toString());

		if (bodyB != null) {

			ensureSummaryBuilder(converters, rowKey);

			final DNGF mainEntity = HBaseTableDAO.parseProto(familyMap.get(type.toString()));

			converters.get(rowKey).setMainEntity(mainEntity);

			final Map<String, byte[]> rels = row.get("rels");

			rels.forEach((q, v) -> {

				final DNGF.Builder rel = DNGF.newBuilder(HBaseTableDAO.parseProto(v));

				rel.getRelBuilder().setCachedTarget(mainEntity.getEntity());

				final String source = rel.getRel().getSource();

				final Type sourceType = rel.getRel().getSourceType();

				rel.getRelBuilder().setSource(rel.getRel().getTarget());

				rel.getRelBuilder().setSourceType(rel.getRel().getTargetType());

				rel.getRelBuilder().setTarget(source);

				rel.getRelBuilder().setTargetType(sourceType);

				final Ontology o = ontologies.get(rel.getRel().getRelType().getSchemeid());

				final String inverse = o.inverseOf(rel.getRel().getRelType().getClassid());

				rel.getRelBuilder().getRelTypeBuilder().setClassid(inverse).setClassname(inverse);

				final String targetId = StringUtils.substringAfter(q, QUALIFIER_SEPARATOR);

				ensureSummaryBuilder(converters, targetId);

				converters.get(targetId).addRelation(rel.build());

			});

		}

	}

	private void ensureSummaryBuilder(final Map<String, DNGFDLISummaryConverter> builders, final String rowKey) {

		if (!builders.containsKey(rowKey)) {

			builders.put(rowKey, new DNGFDLISummaryConverter());

		}

	}

	private void ensureScholixBuilder(final Map<String, List<Scholix>> builders, final String rowKey) {

		if (!builders.containsKey(rowKey)) {

			builders.put(rowKey, Lists.newLinkedList());

		}

	}

}