app-import/src/main/java/eu/etaxonomy/cdm/io/algaterra/AlgaTerraSpecimenImport.java

   1 /**
   2 * Copyright (C) 2007 EDIT
   3 * European Distributed Institute of Taxonomy
   4 * http://www.e-taxonomy.eu
   5 *
   6 * The contents of this file are subject to the Mozilla Public License Version 1.1
   7 * See LICENSE.TXT at the top of this package for the full license terms.
   8 */
   9
  10 package eu.etaxonomy.cdm.io.algaterra;
  11
  12 import java.sql.ResultSet;
  13 import java.sql.SQLException;
  14 import java.util.HashMap;
  15 import java.util.HashSet;
  16 import java.util.Map;
  17 import java.util.Set;
  18 import java.util.UUID;
  19
  20 import org.apache.commons.lang.StringUtils;
  21 import org.apache.log4j.Logger;
  22 import org.springframework.stereotype.Component;
  23
  24 import eu.etaxonomy.cdm.api.facade.DerivedUnitFacade;
  25 import eu.etaxonomy.cdm.api.facade.DerivedUnitFacade.DerivedUnitType;
  26 import eu.etaxonomy.cdm.api.facade.DerivedUnitFacadeNotSupportedException;
  27 import eu.etaxonomy.cdm.io.algaterra.validation.AlgaTerraSpecimenImportValidator;
  28 import eu.etaxonomy.cdm.io.berlinModel.in.BerlinModelImportConfigurator;
  29 import eu.etaxonomy.cdm.io.berlinModel.in.BerlinModelImportState;
  30 import eu.etaxonomy.cdm.io.berlinModel.in.BerlinModelTaxonImport;
  31 import eu.etaxonomy.cdm.io.common.IOValidator;
  32 import eu.etaxonomy.cdm.io.common.ResultSetPartitioner;
  33 import eu.etaxonomy.cdm.model.common.CdmBase;
  34 import eu.etaxonomy.cdm.model.common.DefinedTermBase;
  35 import eu.etaxonomy.cdm.model.common.Language;
  36 import eu.etaxonomy.cdm.model.common.Marker;
  37 import eu.etaxonomy.cdm.model.common.MarkerType;
  38 import eu.etaxonomy.cdm.model.common.TermVocabulary;
  39 import eu.etaxonomy.cdm.model.description.CategoricalData;
  40 import eu.etaxonomy.cdm.model.description.DescriptionBase;
  41 import eu.etaxonomy.cdm.model.description.Feature;
  42 import eu.etaxonomy.cdm.model.description.IndividualsAssociation;
  43 import eu.etaxonomy.cdm.model.description.MeasurementUnit;
  44 import eu.etaxonomy.cdm.model.description.Modifier;
  45 import eu.etaxonomy.cdm.model.description.QuantitativeData;
  46 import eu.etaxonomy.cdm.model.description.State;
  47 import eu.etaxonomy.cdm.model.description.StatisticalMeasure;
  48 import eu.etaxonomy.cdm.model.description.StatisticalMeasurementValue;
  49 import eu.etaxonomy.cdm.model.description.TaxonDescription;
  50 import eu.etaxonomy.cdm.model.description.TextData;
  51 import eu.etaxonomy.cdm.model.name.BotanicalName;
  52 import eu.etaxonomy.cdm.model.name.Rank;
  53 import eu.etaxonomy.cdm.model.occurrence.Collection;
  54 import eu.etaxonomy.cdm.model.occurrence.DerivedUnit;
  55 import eu.etaxonomy.cdm.model.occurrence.FieldObservation;
  56 import eu.etaxonomy.cdm.model.reference.Reference;
  57 import eu.etaxonomy.cdm.model.taxon.Taxon;
  58 import eu.etaxonomy.cdm.model.taxon.TaxonBase;
  59
  60
  61 /**
  62  * @author a.mueller
  63  * @created 01.09.2012
  64  */
  65 @Component
  66 public class AlgaTerraSpecimenImport  extends AlgaTerraSpecimenImportBase {
  67         private static final Logger logger = Logger.getLogger(AlgaTerraSpecimenImport.class);
  68
  69
  70         private static int modCount = 5000;
  71         private static final String pluralString = "specimen and observation";
  72         private static final String dbTableName = "Fact";  //??
  73
  74
  75         public AlgaTerraSpecimenImport(){
  76                 super();
  77         }
  78
  79
  80
  81         /* (non-Javadoc)
  82          * @see eu.etaxonomy.cdm.io.berlinModel.in.BerlinModelImportBase#getIdQuery()
  83          */
  84         @Override
  85         protected String getIdQuery(BerlinModelImportState state) {
  86                 String result = " SELECT factId " +
  87                                 " FROM Fact " +
  88                                         " INNER JOIN EcoFact ON Fact.ExtensionFk = EcoFact.EcoFactId " +
  89                                         "INNER JOIN PTaxon ON Fact.PTNameFk = PTaxon.PTNameFk AND Fact.PTRefFk = PTaxon.PTRefFk "
  90                                 + " WHERE FactCategoryFk = 202 "
  91                                 + " ORDER BY EcoFact.EcoFactId, PTaxon.RIdentifier, Fact.FactId ";
  92                 return result;
  93         }
  94
  95         /* (non-Javadoc)
  96          * @see eu.etaxonomy.cdm.io.berlinModel.in.BerlinModelImportBase#getRecordQuery(eu.etaxonomy.cdm.io.berlinModel.in.BerlinModelImportConfigurator)
  97          */
  98         @Override
  99         protected String getRecordQuery(BerlinModelImportConfigurator config) {
 100                         String strQuery =
 101             " SELECT PTaxon.RIdentifier as taxonId, Fact.FactId, Fact.RecordBasis, EcoFact.*, EcoFact.EcoFactId as unitId, " +
 102                " tg.ID AS GazetteerId, tg.L2Code, tg.L3Code, tg.L4Code, tg.Country, tg.ISOCountry, " +
 103                " ec.UUID as climateUuid, eh.UUID as habitatUuid, elf.UUID as lifeFormUuid " +
 104             " FROM Fact " +
 105                  " INNER JOIN EcoFact ON Fact.ExtensionFk = EcoFact.EcoFactId " +
 106                  " INNER JOIN PTaxon ON dbo.Fact.PTNameFk = dbo.PTaxon.PTNameFk AND dbo.Fact.PTRefFk = dbo.PTaxon.PTRefFk " +
 107                  " LEFT OUTER JOIN TDWGGazetteer tg ON EcoFact.TDWGGazetteerFk = tg.ID " +
 108                  " LEFT OUTER JOIN EcoClimate  ec  ON EcoFact.ClimateFk  = ec.ClimateId " +
 109                  " LEFT OUTER JOIN EcoHabitat  eh  ON EcoFact.HabitatFk  = eh.HabitatId " +
 110                  " LEFT OUTER JOIN EcoLifeForm elf ON EcoFact.LifeFormFk = elf.LifeFormId " +
 111               " WHERE Fact.FactCategoryFk = 202 AND (Fact.FactId IN (" + ID_LIST_TOKEN + ")  )"
 112             + " ORDER BY EcoFact.EcoFactId, PTaxon.RIdentifier, Fact.FactId "
 113             ;
 114                 return strQuery;
 115         }
 116
 117         /* (non-Javadoc)
 118          * @see eu.etaxonomy.cdm.io.berlinModel.in.IPartitionedIO#doPartition(eu.etaxonomy.cdm.io.berlinModel.in.ResultSetPartitioner, eu.etaxonomy.cdm.io.berlinModel.in.BerlinModelImportState)
 119          */
 120         public boolean doPartition(ResultSetPartitioner partitioner, BerlinModelImportState bmState) {
 121                 boolean success = true;
 122
 123                 AlgaTerraImportState state = (AlgaTerraImportState)bmState;
 124                 try {
 125                         makeVocabulariesAndFeatures(state);
 126                 } catch (SQLException e1) {
 127                         logger.warn("Exception occurred when trying to create Ecofact vocabularies: " + e1.getMessage());
 128                         e1.printStackTrace();
 129                 }
 130                 Set<TaxonBase> taxaToSave = new HashSet<TaxonBase>();
 131
 132                 Map<String, TaxonBase> taxonMap = (Map<String, TaxonBase>) partitioner.getObjectMap(BerlinModelTaxonImport.NAMESPACE);
 133                 Map<String, DerivedUnit> ecoFactDerivedUnitMap = (Map<String, DerivedUnit>) partitioner.getObjectMap(ECO_FACT_DERIVED_UNIT_NAMESPACE);
 134
 135                 ResultSet rs = partitioner.getResultSet();
 136
 137                 try {
 138
 139                         int i = 0;
 140
 141                         //for each reference
 142             while (rs.next()){
 143
 144                         if ((i++ % modCount) == 0 && i!= 1 ){ logger.info("Specimen facts handled: " + (i-1));}
 145
 146                                 int newTaxonId = rs.getInt("taxonId");
 147                                 int factId = rs.getInt("FactId");
 148                                 int ecoFactId = rs.getInt("EcoFactId");
 149                                 String recordBasis = rs.getString("RecordBasis");
 150
 151                                 try {
 152
 153                                         //source ref
 154                                         Reference<?> sourceRef = state.getTransactionalSourceReference();
 155
 156                                         //facade
 157                                         DerivedUnitType type = makeDerivedUnitType(recordBasis);
 158                                         DerivedUnitFacade facade = getDerivedUnit(state, ecoFactId, ecoFactDerivedUnitMap, type);
 159
 160                                         //field observation
 161                                         handleSingleSpecimen(rs, facade, state, partitioner);
 162
 163                                         handleEcoFactSpecificSpecimen(rs,facade, state);
 164
 165                                         state.setCurrentFieldObservationNotNew(false);
 166
 167                                         //description element
 168                                         TaxonDescription taxonDescription = getTaxonDescription(state, newTaxonId, taxonMap, factId, sourceRef);
 169                                         IndividualsAssociation indAssociation = IndividualsAssociation.NewInstance();
 170                                         Feature feature = makeFeature(type);
 171                                         indAssociation.setAssociatedSpecimenOrObservation(facade.innerDerivedUnit());
 172                                         indAssociation.setFeature(feature);
 173                                         taxonDescription.addElement(indAssociation);
 174
 175                                         taxaToSave.add(taxonDescription.getTaxon());
 176
 177
 178                                 } catch (Exception e) {
 179                                         logger.warn("Exception in ecoFact: FactId " + factId + ". " + e.getMessage());
 180 //                                      e.printStackTrace();
 181                                 }
 182
 183             }
 184
 185 //            logger.warn("Specimen: " + countSpecimen + ", Descriptions: " + countDescriptions );
 186
 187                         logger.warn("Taxa to save: " + taxaToSave.size());
 188                         getTaxonService().save(taxaToSave);
 189
 190                         return success;
 191                 } catch (SQLException e) {
 192                         logger.error("SQLException:" +  e);
 193                         return false;
 194                 }
 195         }
 196
 197         protected String getDerivedUnitNameSpace(){
 198                 return ECO_FACT_DERIVED_UNIT_NAMESPACE;
 199         }
 200
 201
 202
 203         private void handleEcoFactSpecificSpecimen(ResultSet rs, DerivedUnitFacade facade, AlgaTerraImportState state) throws SQLException {
 204
 205                 Object alkalinityFlag = rs.getBoolean("AlkalinityFlag");
 206
 207                 //alkalinity marker
 208                 if (alkalinityFlag != null){
 209                         MarkerType alkalinityMarkerType = getMarkerType(state, uuidMarkerAlkalinity, "Alkalinity", "Alkalinity", null);
 210                         boolean alkFlag = Boolean.valueOf(alkalinityFlag.toString());
 211                         Marker alkalinityMarker = Marker.NewInstance(alkalinityMarkerType, alkFlag);
 212                         facade.getFieldObservation(true).addMarker(alkalinityMarker);
 213                 }
 214
 215
 216                 DescriptionBase<?> fieldDescription = getFieldObservationDescription(facade);
 217
 218                 //habitat, ecology, community, etc.
 219                 String habitat = rs.getString("HabitatExplanation");
 220
 221                 if (isNotBlank(habitat)){
 222                         Feature habitatExplanation = getFeature(state, uuidFeatureHabitatExplanation, "Habitat Explanation", "HabitatExplanation", null, null);
 223                         TextData textData = TextData.NewInstance(habitatExplanation);
 224                         textData.putText(Language.DEFAULT(), habitat);
 225                         fieldDescription.addElement(textData);
 226                 }
 227
 228                 String community = rs.getString("Comunity");
 229                 if (isNotBlank(community)){
 230                         Feature communityFeature = getFeature(state, uuidFeatureSpecimenCommunity, "Community", "The community of a specimen (e.g. other algae in the same sample)", null, null);
 231                         TextData textData = TextData.NewInstance(communityFeature);
 232                         textData.putText(Language.DEFAULT(), community);
 233                         fieldDescription.addElement(textData);
 234                 }
 235
 236                 String additionalData = rs.getString("AdditionalData");
 237                 if (isNotBlank(additionalData)){  //or handle it as Annotation ??
 238                         Feature additionalDataFeature = getFeature(state, uuidFeatureAdditionalData, "Additional Data", "Additional Data", null, null);
 239                         TextData textData = TextData.NewInstance(additionalDataFeature);
 240                         textData.putText(Language.DEFAULT(), additionalData);
 241                         fieldDescription.addElement(textData);
 242                 }
 243
 244                 String climateUuid = rs.getString("climateUuid");
 245                 String habitatUuid = rs.getString("habitatUuid");
 246                 String lifeFormUuid = rs.getString("lifeFormUuid");
 247
 248                 addCategoricalValue(state, fieldDescription, climateUuid, uuidFeatureAlgaTerraClimate);
 249                 addCategoricalValue(state, fieldDescription, habitatUuid, Feature.HABITAT().getUuid());
 250                 addCategoricalValue(state, fieldDescription, lifeFormUuid, uuidFeatureAlgaTerraLifeForm);
 251
 252
 253                 //collection
 254                 String voucher = rs.getString("Voucher");
 255                 if (StringUtils.isNotBlank(voucher)){
 256                         facade.setAccessionNumber(voucher);
 257                 }
 258
 259                 //parameters
 260                 makeParameter(state, rs, getFieldObservationDescription(facade));
 261
 262         }
 263
 264
 265
 266
 267         private void addCategoricalValue(AlgaTerraImportState importState, DescriptionBase description, String uuidTerm, UUID featureUuid) {
 268                 if (uuidTerm != null){
 269                         State state = this.getStateTerm(importState, UUID.fromString(uuidTerm));
 270                         Feature feature = getFeature(importState, featureUuid);
 271                         CategoricalData categoricalData = CategoricalData.NewInstance(state, feature);
 272                         description.addElement(categoricalData);
 273                 }
 274         }
 275
 276         private void makeParameter(AlgaTerraImportState state, ResultSet rs, DescriptionBase<?> descriptionBase) throws SQLException {
 277                 for (int i = 1; i <= 10; i++){
 278                         String valueStr = rs.getString(String.format("P%dValue", i));
 279                         String unitStr = rs.getString(String.format("P%dUnit", i));
 280                         String parameter = rs.getString(String.format("P%dParameter", i));
 281                         String method = rs.getString(String.format("P%dMethod", i));
 282
 283                         //method
 284                         if (StringUtils.isNotBlank(method)){
 285                                 logger.warn("Methods not yet handled: " + method);
 286                         }
 287                         //parameter
 288                         TermVocabulary<Feature> vocParameter = getVocabulary(uuidVocParameter, "Feature vocabulary for AlgaTerra measurement parameters", "Parameters", null, null, false, Feature.COMMON_NAME());
 289                         if (StringUtils.isNotBlank(parameter)){
 290                                 UUID featureUuid = getParameterFeatureUuid(state, parameter);
 291                                 Feature feature = getFeature(state, featureUuid, parameter, parameter, null, vocParameter);
 292                                 QuantitativeData quantData = QuantitativeData.NewInstance(feature);
 293
 294                                 //unit
 295                                 MeasurementUnit unit = getMeasurementUnit(state, unitStr);
 296                                 quantData.setUnit(unit);
 297                                 try {
 298
 299                                         Set<Modifier> valueModifier = new HashSet<Modifier>();
 300                                         valueStr = normalizeAndModifyValue(state, valueStr, valueModifier);
 301                                         //value
 302                                         Float valueFlt = Float.valueOf(valueStr);  //TODO maybe change model to Double ??
 303
 304                                         StatisticalMeasure measureSingleValue = getStatisticalMeasure(state, uuidStatMeasureSingleValue, "Value", "Single measurement value", null, null);
 305                                         StatisticalMeasurementValue value = StatisticalMeasurementValue.NewInstance(measureSingleValue, valueFlt);
 306                                         quantData.addStatisticalValue(value);
 307                                         descriptionBase.addElement(quantData);
 308
 309                                 } catch (NumberFormatException e) {
 310                                         logger.warn(String.format("Value '%s' can't be converted to double. Parameter %s not imported.", valueStr, parameter));
 311                                 }
 312                         }else if (isNotBlank(valueStr) || isNotBlank(unitStr) ){
 313                                 logger.warn("There is value or unit without parameter: " + i);
 314                         }
 315
 316
 317                 }
 318
 319         }
 320
 321         private String normalizeAndModifyValue(AlgaTerraImportState state, String valueStr, Set<Modifier> valueModifier) {
 322                 valueStr = valueStr.replace(",", ".");
 323                 if (valueStr.startsWith("<")){
 324                         TermVocabulary<Modifier> measurementValueModifierVocabulary = getVocabulary(uuidMeasurementValueModifier, "Measurement value modifier", "Measurement value modifier", null, null, false, Modifier.NewInstance());
 325                         Modifier modifier = getModifier(state, uuidModifierLowerThan, "Lower", "Lower than the given measurement value", "<", measurementValueModifierVocabulary);
 326                         valueModifier.add(modifier);
 327                         valueStr = valueStr.replace("<", "");
 328                 }
 329                 if (valueStr.startsWith(">")){
 330                         TermVocabulary<Modifier> measurementValueModifierVocabulary = getVocabulary(uuidMeasurementValueModifier, "Measurement value modifier", "Measurement value modifier", null, null, false, Modifier.NewInstance());
 331                         Modifier modifier = getModifier(state, uuidModifierGreaterThan, "Lower", "Lower than the given measurement value", "<", measurementValueModifierVocabulary);
 332                         valueModifier.add(modifier);
 333                         valueStr = valueStr.replace(">", "");
 334                 }
 335                 return valueStr;
 336         }
 337
 338
 339
 340         private UUID getParameterFeatureUuid(AlgaTerraImportState state, String key) {
 341                 //TODO define some UUIDs in Transformer
 342                 UUID uuid = state.getParameterFeatureUuid(key);
 343                 if (uuid == null){
 344                         uuid = UUID.randomUUID();
 345                         state.putParameterFeatureUuid(key, uuid);
 346                 }
 347                 return uuid;
 348         }
 349
 350
 351
 352         /**
 353          * TODO move to InputTransformerBase
 354          * @param state
 355          * @param unitStr
 356          * @return
 357          */
 358         private MeasurementUnit getMeasurementUnit(AlgaTerraImportState state, String unitStr) {
 359                 if (StringUtils.isNotBlank(unitStr)){
 360                         UUID uuidMeasurementUnitMgL = UUID.fromString("7ac302c5-3cbd-4334-964a-bf5d11eb9ead");
 361                         UUID uuidMeasurementUnitMolMol = UUID.fromString("96b78d78-3e49-448f-8100-e7779b71dd53");
 362                         UUID uuidMeasurementUnitMicroMolSiL = UUID.fromString("2cb8bc85-a4af-42f1-b80b-34c36c9f75d4");
 363                         UUID uuidMeasurementUnitMicroMolL = UUID.fromString("a631f62e-377e-405c-bd1a-76885b13a72b");
 364                         UUID uuidMeasurementUnitDegreeC = UUID.fromString("55222aec-d5be-413e-8db7-d9a48c316c6c");
 365                         UUID uuidMeasurementUnitPercent = UUID.fromString("3ea3110e-f048-4bed-8bfe-33c60f63626f");
 366                         UUID uuidMeasurementUnitCm = UUID.fromString("3ea3110e-f048-4bed-8bfe-33c60f63626f");
 367                         UUID uuidMeasurementUnitMicroSiCm = UUID.fromString("3ea3110e-f048-4bed-8bfe-33c60f63626f");
 368
 369
 370                         if (unitStr.equalsIgnoreCase("mg/L")){
 371                                 return getMeasurementUnit(state, uuidMeasurementUnitMgL, unitStr, unitStr, unitStr, null);
 372                         }else if (unitStr.equalsIgnoreCase("mol/mol")){
 373                                 return getMeasurementUnit(state, uuidMeasurementUnitMolMol, unitStr, unitStr, unitStr, null);
 374                         }else if (unitStr.equalsIgnoreCase("\u00B5mol Si/L")){   //µmol Si/L
 375                                 return getMeasurementUnit(state, uuidMeasurementUnitMicroMolSiL, unitStr, unitStr, unitStr, null);
 376                         }else if (unitStr.equalsIgnoreCase("\u00B5mol/L")){             //µmol/L
 377                                 return getMeasurementUnit(state, uuidMeasurementUnitMicroMolL, unitStr, unitStr, unitStr, null);
 378                         }else if (unitStr.equalsIgnoreCase("\u00B0C")){               //°C
 379                                 return getMeasurementUnit(state, uuidMeasurementUnitDegreeC, unitStr, unitStr, unitStr, null);
 380                         }else if (unitStr.equalsIgnoreCase("%")){
 381                                 return getMeasurementUnit(state, uuidMeasurementUnitPercent, unitStr, unitStr, unitStr, null);
 382                         }else if (unitStr.equalsIgnoreCase("cm")){
 383                                 return getMeasurementUnit(state, uuidMeasurementUnitCm, unitStr, unitStr, unitStr, null);
 384                         }else if (unitStr.equalsIgnoreCase("\u00B5S/cm")){   //µS/cm
 385                                 return getMeasurementUnit(state, uuidMeasurementUnitMicroSiCm, unitStr, unitStr, unitStr, null);
 386                         }else{
 387                                 logger.warn("MeasurementUnit was not recognized");
 388                                 return null;
 389                         }
 390                 }else{
 391                         return null;
 392                 }
 393         }
 394
 395
 396
 397
 398
 399         /**
 400          * @param state
 401          * @param ecoFactId
 402          * @param derivedUnitMap
 403          * @param type
 404          * @return
 405          */
 406         private DerivedUnitFacade getDerivedUnit(AlgaTerraImportState state, int ecoFactId, Map<String, DerivedUnit> derivedUnitMap, DerivedUnitType type) {
 407                 String key = String.valueOf(ecoFactId);
 408                 DerivedUnit derivedUnit = derivedUnitMap.get(key);
 409                 DerivedUnitFacade facade;
 410                 if (derivedUnit == null){
 411                         facade = DerivedUnitFacade.NewInstance(type);
 412                         derivedUnitMap.put(key, derivedUnit);
 413                 }else{
 414                         try {
 415                                 facade = DerivedUnitFacade.NewInstance(derivedUnit);
 416                         } catch (DerivedUnitFacadeNotSupportedException e) {
 417                                 logger.error(e.getMessage());
 418                                 facade = DerivedUnitFacade.NewInstance(type);
 419                         }
 420                 }
 421
 422                 return facade;
 423         }
 424
 425         private Feature makeFeature(DerivedUnitType type) {
 426                 if (type.equals(DerivedUnitType.DerivedUnit)){
 427                         return Feature.INDIVIDUALS_ASSOCIATION();
 428                 }else if (type.equals(DerivedUnitType.FieldObservation) || type.equals(DerivedUnitType.Observation) ){
 429                         return Feature.OBSERVATION();
 430                 }else if (type.equals(DerivedUnitType.Fossil) || type.equals(DerivedUnitType.LivingBeing) || type.equals(DerivedUnitType.Specimen )){
 431                         return Feature.SPECIMEN();
 432                 }
 433                 logger.warn("No feature defined for derived unit type: " + type);
 434                 return null;
 435         }
 436
 437
 438         private DerivedUnitType makeDerivedUnitType(String recordBasis) {
 439                 DerivedUnitType result = null;
 440                 if (StringUtils.isBlank(recordBasis)){
 441                         result = DerivedUnitType.DerivedUnit;
 442                 } else if (recordBasis.equalsIgnoreCase("FossileSpecimen")){
 443                         result = DerivedUnitType.Fossil;
 444                 }else if (recordBasis.equalsIgnoreCase("HumanObservation")){
 445                         result = DerivedUnitType.Observation;
 446                 }else if (recordBasis.equalsIgnoreCase("Literature")){
 447                         logger.warn("Literature record basis not yet supported");
 448                         result = DerivedUnitType.DerivedUnit;
 449                 }else if (recordBasis.equalsIgnoreCase("LivingSpecimen")){
 450                         result = DerivedUnitType.LivingBeing;
 451                 }else if (recordBasis.equalsIgnoreCase("MachineObservation")){
 452                         logger.warn("MachineObservation record basis not yet supported");
 453                         result = DerivedUnitType.Observation;
 454                 }else if (recordBasis.equalsIgnoreCase("PreservedSpecimen")){
 455                         result = DerivedUnitType.Specimen;
 456                 }
 457                 return result;
 458         }
 459
 460         /* (non-Javadoc)
 461          * @see eu.etaxonomy.cdm.io.berlinModel.in.IPartitionedIO#getRelatedObjectsForPartition(java.sql.ResultSet)
 462          */
 463         public Map<Object, Map<String, ? extends CdmBase>> getRelatedObjectsForPartition(ResultSet rs) {
 464                 String nameSpace;
 465                 Class cdmClass;
 466                 Set<String> idSet;
 467                 Map<Object, Map<String, ? extends CdmBase>> result = new HashMap<Object, Map<String, ? extends CdmBase>>();
 468
 469                 try{
 470                         Set<String> taxonIdSet = new HashSet<String>();
 471                         Set<String> fieldObservationIdSet = new HashSet<String>();
 472                         Set<String> termsIdSet = new HashSet<String>();
 473                         Set<String> collectionIdSet = new HashSet<String>();
 474
 475                         while (rs.next()){
 476                                 handleForeignKey(rs, taxonIdSet, "taxonId");
 477                                 handleForeignKey(rs, fieldObservationIdSet, "ecoFactId");
 478                                 handleForeignKey(rs, termsIdSet, "ClimateFk");
 479                                 handleForeignKey(rs, termsIdSet, "HabitatFk");
 480                                 handleForeignKey(rs, termsIdSet, "LifeFormFk");
 481                                 handleForeignKey(rs, collectionIdSet, "CollectionFk");
 482                         }
 483
 484                         //taxon map
 485                         nameSpace = BerlinModelTaxonImport.NAMESPACE;
 486                         cdmClass = TaxonBase.class;
 487                         idSet = taxonIdSet;
 488                         Map<String, TaxonBase> objectMap = (Map<String, TaxonBase>)getCommonService().getSourcedObjectsByIdInSource(cdmClass, idSet, nameSpace);
 489                         result.put(nameSpace, objectMap);
 490
 491                         //field observation map map
 492                         nameSpace = AlgaTerraSpecimenImport.ECO_FACT_FIELD_OBSERVATION_NAMESPACE;
 493                         cdmClass = FieldObservation.class;
 494                         idSet = fieldObservationIdSet;
 495                         Map<String, FieldObservation> fieldObservationMap = (Map<String, FieldObservation>)getCommonService().getSourcedObjectsByIdInSource(cdmClass, idSet, nameSpace);
 496                         result.put(nameSpace, fieldObservationMap);
 497
 498                         //collections
 499                         nameSpace = AlgaTerraCollectionImport.NAMESPACE_COLLECTION;
 500                         cdmClass = Collection.class;
 501                         idSet = collectionIdSet;
 502                         Map<String, Collection> collectionMap = (Map<String, Collection>)getCommonService().getSourcedObjectsByIdInSource(cdmClass, idSet, nameSpace);
 503                         result.put(nameSpace, collectionMap);
 504
 505                         //sub-collections
 506                         nameSpace = AlgaTerraCollectionImport.NAMESPACE_SUBCOLLECTION;
 507                         cdmClass = Collection.class;
 508                         idSet = collectionIdSet;
 509                         Map<String, Collection> subCollectionMap = (Map<String, Collection>)getCommonService().getSourcedObjectsByIdInSource(cdmClass, idSet, nameSpace);
 510                         result.put(nameSpace, subCollectionMap);
 511
 512                         //terms
 513                         nameSpace = AlgaTerraSpecimenImport.TERMS_NAMESPACE;
 514                         cdmClass = FieldObservation.class;
 515                         idSet = termsIdSet;
 516                         Map<String, DefinedTermBase> termMap = (Map<String, DefinedTermBase>)getCommonService().getSourcedObjectsByIdInSource(cdmClass, idSet, nameSpace);
 517                         result.put(nameSpace, termMap);
 518
 519                 } catch (SQLException e) {
 520                         throw new RuntimeException(e);
 521                 }
 522                 return result;
 523         }
 524
 525
 526         /**
 527          * Use same TaxonDescription if two records belong to the same taxon
 528          * @param state
 529          * @param newTaxonId
 530          * @param oldTaxonId
 531          * @param oldDescription
 532          * @param taxonMap
 533          * @return
 534          */
 535         private TaxonDescription getTaxonDescription(AlgaTerraImportState state, int newTaxonId, Map<String, TaxonBase> taxonMap, int factId, Reference<?> sourceSec){
 536                 TaxonDescription result = null;
 537                 TaxonBase<?> taxonBase = taxonMap.get(String.valueOf(newTaxonId));
 538
 539                 //TODO for testing
 540                 if (taxonBase == null && ! state.getConfig().isDoTaxa()){
 541                         taxonBase = Taxon.NewInstance(BotanicalName.NewInstance(Rank.SPECIES()), null);
 542                 }
 543
 544                 Taxon taxon;
 545                 if ( taxonBase instanceof Taxon ) {
 546                         taxon = (Taxon) taxonBase;
 547                 } else if (taxonBase != null) {
 548                         logger.warn("TaxonBase for Fact(Specimen) with factId" + factId + " was not of type Taxon but: " + taxonBase.getClass().getSimpleName());
 549                         return null;
 550                 } else {
 551                         logger.warn("TaxonBase for Fact(Specimen) " + factId + " is null.");
 552                         return null;
 553                 }
 554                 Set<TaxonDescription> descriptionSet= taxon.getDescriptions();
 555                 if (descriptionSet.size() > 0) {
 556                         result = descriptionSet.iterator().next();
 557                 }else{
 558                         result = TaxonDescription.NewInstance();
 559                         result.setTitleCache(sourceSec.getTitleCache(), true);
 560                         taxon.addDescription(result);
 561                 }
 562                 return result;
 563         }
 564
 565
 566         /* (non-Javadoc)
 567          * @see eu.etaxonomy.cdm.io.common.CdmIoBase#doCheck(eu.etaxonomy.cdm.io.common.IoStateBase)
 568          */
 569         @Override
 570         protected boolean doCheck(BerlinModelImportState state){
 571                 IOValidator<BerlinModelImportState> validator = new AlgaTerraSpecimenImportValidator();
 572                 return validator.validate(state);
 573         }
 574
 575         /* (non-Javadoc)
 576          * @see eu.etaxonomy.cdm.io.berlinModel.in.BerlinModelImportBase#getTableName()
 577          */
 578         @Override
 579         protected String getTableName() {
 580                 return dbTableName;
 581         }
 582
 583         /* (non-Javadoc)
 584          * @see eu.etaxonomy.cdm.io.berlinModel.in.BerlinModelImportBase#getPluralString()
 585          */
 586         @Override
 587         public String getPluralString() {
 588                 return pluralString;
 589         }
 590
 591         /* (non-Javadoc)
 592          * @see eu.etaxonomy.cdm.io.common.CdmIoBase#isIgnore(eu.etaxonomy.cdm.io.common.IImportConfigurator)
 593          */
 594         protected boolean isIgnore(BerlinModelImportState state){
 595                 return ! ((AlgaTerraImportState)state).getAlgaTerraConfigurator().isDoEcoFacts();
 596         }
 597
 598 }