Project

General

Profile

Download (44.7 KB) Statistics
| Branch: | Tag: | Revision:
1
// $Id$
2
/**
3
* Copyright (C) 2013 EDIT
4
* European Distributed Institute of Taxonomy
5
* http://www.e-taxonomy.eu
6
*
7
* The contents of this file are subject to the Mozilla Public License Version 1.1
8
* See LICENSE.TXT at the top of this package for the full license terms.
9
*/
10
package eu.etaxonomy.cdm.api.service.description;
11

    
12
import java.util.ArrayList;
13
import java.util.Arrays;
14
import java.util.HashMap;
15
import java.util.HashSet;
16
import java.util.Iterator;
17
import java.util.LinkedList;
18
import java.util.List;
19
import java.util.Map;
20
import java.util.Set;
21
import java.util.UUID;
22

    
23
import org.apache.log4j.Logger;
24
import org.hibernate.FlushMode;
25
import org.hibernate.HibernateException;
26
import org.hibernate.Session;
27
import org.hibernate.engine.spi.SessionFactoryImplementor;
28
import org.hibernate.search.Search;
29
import org.springframework.beans.factory.annotation.Autowired;
30
import org.springframework.orm.hibernate5.HibernateTransactionManager;
31
import org.springframework.stereotype.Service;
32
import org.springframework.transaction.TransactionDefinition;
33
import org.springframework.transaction.TransactionStatus;
34
import org.springframework.transaction.support.DefaultTransactionDefinition;
35

    
36
import eu.etaxonomy.cdm.api.service.IClassificationService;
37
import eu.etaxonomy.cdm.api.service.IDescriptionService;
38
import eu.etaxonomy.cdm.api.service.INameService;
39
import eu.etaxonomy.cdm.api.service.ITaxonService;
40
import eu.etaxonomy.cdm.api.service.ITermService;
41
import eu.etaxonomy.cdm.common.DynamicBatch;
42
import eu.etaxonomy.cdm.common.JvmLimitsException;
43
import eu.etaxonomy.cdm.common.monitor.IProgressMonitor;
44
import eu.etaxonomy.cdm.common.monitor.NullProgressMonitor;
45
import eu.etaxonomy.cdm.common.monitor.SubProgressMonitor;
46
import eu.etaxonomy.cdm.model.common.CdmBase;
47
import eu.etaxonomy.cdm.model.common.DefinedTermBase;
48
import eu.etaxonomy.cdm.model.common.Extension;
49
import eu.etaxonomy.cdm.model.common.ExtensionType;
50
import eu.etaxonomy.cdm.model.common.Marker;
51
import eu.etaxonomy.cdm.model.common.MarkerType;
52
import eu.etaxonomy.cdm.model.common.OrderedTermBase;
53
import eu.etaxonomy.cdm.model.description.DescriptionElementBase;
54
import eu.etaxonomy.cdm.model.description.DescriptionElementSource;
55
import eu.etaxonomy.cdm.model.description.Distribution;
56
import eu.etaxonomy.cdm.model.description.PresenceAbsenceTerm;
57
import eu.etaxonomy.cdm.model.description.TaxonDescription;
58
import eu.etaxonomy.cdm.model.location.NamedArea;
59
import eu.etaxonomy.cdm.model.name.Rank;
60
import eu.etaxonomy.cdm.model.taxon.Classification;
61
import eu.etaxonomy.cdm.model.taxon.Taxon;
62
import eu.etaxonomy.cdm.model.taxon.TaxonBase;
63
import eu.etaxonomy.cdm.persistence.dto.ClassificationLookupDTO;
64

    
65
/**
66
 *
67
 * <h2>GENERAL NOTES </h2>
68
 * <em>TODO: These notes are directly taken from original Transmission Engine Occurrence
69
 * version 14 written in Visual Basic and still need to be
70
 * adapted to the java version of the transmission engine!</em>
71
 *
72
 * <h3>summaryStatus</h3>
73
 *
74
 *   Each distribution information has a summaryStatus, this is an summary of the status codes
75
 *   as stored in the fields of emOccurrence native, introduced, cultivated, ...
76
 *   The summaryStatus seems to be equivalent to  the CDM DistributionStatus
77
 *
78
 * <h3>map generation</h3>
79
 *
80
 *   When generating maps from the accumulated distribution information some special cases have to be handled:
81
 * <ol>
82
 *   <li>if a entered or imported status information exist for the same area for which calculated (accumulated)
83
 *       data is available, the calculated data has to be given preference over other data.
84
 *   </li>
85
 *   <li>If there is an area with a sub area and both areas have the same calculated status only the subarea
86
 *       status should be shown in the map, whereas the super area should be ignored.
87
 *   </li>
88
 * </ol>
89
 *
90
 * @author Anton Güntsch (author of original Transmission Engine Occurrence version 14 written in Visual Basic)
91
 * @author Andreas Kohlbecker (2013, porting Transmission Engine Occurrence to Java)
92
 * @date Feb 22, 2013
93
 */
94
@Service
95
public class TransmissionEngineDistribution { //TODO extends IoBase?
96

    
97

    
98
    public static final String EXTENSION_VALUE_PREFIX = "transmissionEngineDistribution.priority:";
99

    
100
    public static final Logger logger = Logger.getLogger(TransmissionEngineDistribution.class);
101

    
102
    /**
103
     * only used for performance testing
104
     */
105
    final boolean ONLY_FISRT_BATCH = false;
106

    
107

    
108
    protected static final List<String> TAXONDESCRIPTION_INIT_STRATEGY = Arrays.asList(new String [] {
109
            "description.markers.markerType",
110
            "description.elements.markers.markerType",
111
            "description.elements.area",
112
            "description.elements.status",
113
            "description.elements.sources.citation.authorship",
114
//            "description.elements.sources.nameUsedInSource",
115
//            "description.elements.multilanguageText",
116
//            "name.status.type",
117
    });
118

    
119

    
120
    /**
121
     * A map which contains the status terms as key and the priority as value
122
     * The map will contain both, the PresenceTerms and the AbsenceTerms
123
     */
124
    private Map<PresenceAbsenceTerm, Integer> statusPriorityMap = null;
125

    
126
    @Autowired
127
    private IDescriptionService descriptionService;
128

    
129
    @Autowired
130
    private ITermService termService;
131

    
132
    @Autowired
133
    private ITaxonService taxonService;
134

    
135
    @Autowired
136
    private IClassificationService classificationService;
137

    
138
    @Autowired
139
    private INameService mameService;
140

    
141
    @Autowired
142
    private HibernateTransactionManager transactionManager;
143

    
144
    private List<PresenceAbsenceTerm> byAreaIgnoreStatusList = null;
145

    
146
    private List<PresenceAbsenceTerm> byRankIgnoreStatusList = null;
147

    
148
    private final Map<NamedArea, Set<NamedArea>> subAreaMap = new HashMap<NamedArea, Set<NamedArea>>();
149

    
150
    int byRankTicks = 300;
151
    int byAreasTicks = 100;
152

    
153

    
154
    private static final long BATCH_MIN_FREE_HEAP = 800  * 1024 * 1024;
155
    /**
156
     * ratio of the initially free heap which should not be used
157
     * during the batch processing. This amount of the heap is reserved
158
     * for the flushing of the session and to the index
159
     */
160
    private static final double BATCH_FREE_HEAP_RATIO = 0.9;
161
    private static final int BATCH_SIZE_BY_AREA = 1000;
162
    private static final int BATCH_SIZE_BY_RANK = 500;
163

    
164

    
165

    
166
    /**
167
     * byAreaIgnoreStatusList contains by default:
168
     *  <ul>
169
     *    <li>AbsenceTerm.CULTIVATED_REPORTED_IN_ERROR()</li>
170
     *    <li>AbsenceTerm.INTRODUCED_REPORTED_IN_ERROR()</li>
171
     *    <li>AbsenceTerm.INTRODUCED_FORMERLY_INTRODUCED()</li>
172
     *    <li>AbsenceTerm.NATIVE_REPORTED_IN_ERROR()</li>
173
     *    <li>AbsenceTerm.NATIVE_FORMERLY_NATIVE()</li>
174
     *  </ul>
175
     *
176
     * @return the byAreaIgnoreStatusList
177
     */
178
    public List<PresenceAbsenceTerm> getByAreaIgnoreStatusList() {
179
        if(byAreaIgnoreStatusList == null ){
180
            byAreaIgnoreStatusList = Arrays.asList(
181
                    new PresenceAbsenceTerm[] {
182
                    		PresenceAbsenceTerm.CULTIVATED_REPORTED_IN_ERROR(),
183
                    		PresenceAbsenceTerm.INTRODUCED_REPORTED_IN_ERROR(),
184
                    		PresenceAbsenceTerm.NATIVE_REPORTED_IN_ERROR(),
185
                    		PresenceAbsenceTerm.INTRODUCED_FORMERLY_INTRODUCED(),
186
                    		PresenceAbsenceTerm.NATIVE_FORMERLY_NATIVE()
187
                            // TODO what about PresenceAbsenceTerm.ABSENT() also ignore?
188
                    });
189
        }
190
        return byAreaIgnoreStatusList;
191
    }
192

    
193
    /**
194
     * @param byAreaIgnoreStatusList the byAreaIgnoreStatusList to set
195
     */
196
    public void setByAreaIgnoreStatusList(List<PresenceAbsenceTerm> byAreaIgnoreStatusList) {
197
        this.byAreaIgnoreStatusList = byAreaIgnoreStatusList;
198
    }
199

    
200
    /**
201
     * byRankIgnoreStatusList contains by default
202
     *  <ul>
203
     *    <li>PresenceTerm.ENDEMIC_FOR_THE_RELEVANT_AREA()</li>
204
     *  </ul>
205
     *
206
     * @return the byRankIgnoreStatusList
207
     */
208
    public List<PresenceAbsenceTerm> getByRankIgnoreStatusList() {
209

    
210
        if (byRankIgnoreStatusList == null) {
211
            byRankIgnoreStatusList = Arrays.asList(
212
                    new PresenceAbsenceTerm[] {
213
                    		PresenceAbsenceTerm.ENDEMIC_FOR_THE_RELEVANT_AREA()
214
                    });
215
        }
216
        return byRankIgnoreStatusList;
217
    }
218

    
219
    /**
220
     * @param byRankIgnoreStatusList the byRankIgnoreStatusList to set
221
     */
222
    public void setByRankIgnoreStatusList(List<PresenceAbsenceTerm> byRankIgnoreStatusList) {
223
        this.byRankIgnoreStatusList = byRankIgnoreStatusList;
224
    }
225

    
226
    /**
227
     *
228
     * @param superAreas
229
     */
230
    public TransmissionEngineDistribution() {
231
    }
232

    
233
    /**
234
     * initializes the map which contains the status terms as key and the priority as value
235
     * The map will contain both, the PresenceTerms and the AbsenceTerms
236
     */
237
    private void initializeStatusPriorityMap() {
238

    
239
        statusPriorityMap = new HashMap<PresenceAbsenceTerm, Integer>();
240
        Integer priority;
241

    
242
        // PresenceTerms
243
        for(PresenceAbsenceTerm term : termService.list(PresenceAbsenceTerm.class, null, null, null, null)){
244
            priority = getPriorityFor(term);
245
            if(priority != null){
246
                statusPriorityMap.put(term, priority);
247
            }
248
        }
249
    }
250

    
251
    /**
252
     * Compares the PresenceAbsenceTermBase terms contained in <code>a.status</code> and <code>b.status</code> after
253
     * the priority as stored in the statusPriorityMap. The StatusAndSources object with
254
     * the higher priority is returned. In the case of <code>a == b</code> the sources of b will be added to the sources
255
     * of a.
256
     *
257
     * If either a or b or the status are null b or a is returned.
258
     *
259
     * @see initializeStatusPriorityMap()
260
     *
261
     * @param a
262
     * @param b
263
     * @param sourcesForWinnerB
264
     *  In the case when <code>b</code> is preferred over <code>a</code> these Set of sources will be added to the sources of <code>b</code>
265
     * @return
266
     */
267
    private StatusAndSources choosePreferred(StatusAndSources a, StatusAndSources b, Set<DescriptionElementSource> sourcesForWinnerB){
268

    
269
        if (statusPriorityMap == null) {
270
            initializeStatusPriorityMap();
271
        }
272

    
273
        if (b == null || b.status == null) {
274
            return a;
275
        }
276
        if (a == null || a.status == null) {
277
            return b;
278
        }
279

    
280
        if (statusPriorityMap.get(a.status) == null) {
281
            logger.warn("No priority found in map for " + a.status.getLabel());
282
            return b;
283
        }
284
        if (statusPriorityMap.get(b.status) == null) {
285
            logger.warn("No priority found in map for " + b.status.getLabel());
286
            return a;
287
        }
288
        if(statusPriorityMap.get(a.status) < statusPriorityMap.get(b.status)){
289
            if(sourcesForWinnerB != null) {
290
                b.addSources(sourcesForWinnerB);
291
            }
292
            return b;
293
        } else if (statusPriorityMap.get(a.status) == statusPriorityMap.get(b.status)){
294
            a.addSources(b.sources);
295
            return a;
296
        } else {
297
            return a;
298
        }
299
    }
300

    
301
    /**
302
     * reads the priority for the given status term from the extensions.
303
     *
304
     * @param term
305
     * @return the priority value
306
     */
307
    private Integer getPriorityFor(DefinedTermBase<?> term) {
308
        Set<Extension> extensions = term.getExtensions();
309
        for(Extension extension : extensions){
310
            if(!extension.getType().equals(ExtensionType.ORDER())) {
311
                continue;
312
            }
313
            int pos = extension.getValue().indexOf(EXTENSION_VALUE_PREFIX);
314
            if(pos == 0){ // if starts with EXTENSION_VALUE_PREFIX
315
                try {
316
                    Integer priority = Integer.valueOf(extension.getValue().substring(EXTENSION_VALUE_PREFIX.length()));
317
                    return priority;
318
                } catch (NumberFormatException e) {
319
                    logger.warn("Invalid number format in Extension:" + extension.getValue());
320
                }
321
            }
322
        }
323
        logger.warn("no priority defined for '" + term.getLabel() + "'");
324
        return null;
325
    }
326

    
327
    /**
328
     * runs both steps
329
     * <ul>
330
     * <li>Step 1: Accumulate occurrence records by area</li>
331
     * <li>Step 2: Accumulate by ranks starting from lower rank to upper rank,
332
     * the status of all children are accumulated on each rank starting from
333
     * lower rank to upper rank.</li>
334
     * </ul>
335
     *
336
     * @param superAreas
337
     *            the areas to which the subordinate areas should be projected.
338
     * @param lowerRank
339
     * @param upperRank
340
     * @param classification
341
     * @param classification
342
     *            limit the accumulation process to a specific classification
343
     *            (not yet implemented)
344
     * @param monitor
345
     *            the progress monitor to use for reporting progress to the
346
     *            user. It is the caller's responsibility to call done() on the
347
     *            given monitor. Accepts null, indicating that no progress
348
     *            should be reported and that the operation cannot be cancelled.
349
     */
350
    public void accumulate(AggregationMode mode, List<NamedArea> superAreas, Rank lowerRank, Rank upperRank,
351
            Classification classification, IProgressMonitor monitor) throws JvmLimitsException {
352

    
353
        if (monitor == null) {
354
            monitor = new NullProgressMonitor();
355
        }
356

    
357
        // only for debugging:
358
        //logger.setLevel(Level.TRACE); // TRACE will slow down a lot since it forces loading all term representations
359
        //Logger.getLogger("org.hibernate.SQL").setLevel(Level.DEBUG);
360

    
361
        logger.info("Hibernate JDBC Batch size: "
362
                + ((SessionFactoryImplementor) getSession().getSessionFactory()).getSettings().getJdbcBatchSize());
363

    
364
        Set<Classification> classifications = new HashSet<Classification>();
365
        if(classification == null) {
366
            classifications.addAll(classificationService.listClassifications(null, null, null, null));
367
        } else {
368
            classifications.add(classification);
369
        }
370

    
371
        int aggregationWorkTicks;
372
        switch(mode){
373
        case byAreasAndRanks:
374
            aggregationWorkTicks = byAreasTicks + byRankTicks;
375
            break;
376
        case byAreas:
377
            aggregationWorkTicks = byAreasTicks;
378
            break;
379
        case byRanks:
380
            aggregationWorkTicks = byRankTicks;
381
            break;
382
        default:
383
            aggregationWorkTicks = 0;
384
            break;
385
        }
386

    
387
        // take start time for performance testing
388
        // NOTE: use ONLY_FISRT_BATCH = true to measure only one batch
389
        double start = System.currentTimeMillis();
390

    
391
        monitor.beginTask("Accumulating distributions", (classifications.size() * aggregationWorkTicks) + 1 );
392

    
393
        updatePriorities();
394

    
395
        List<Rank> ranks = rankInterval(lowerRank, upperRank);
396

    
397
        monitor.worked(1);
398

    
399

    
400
        for(Classification _classification : classifications) {
401

    
402
            ClassificationLookupDTO classificationLookupDao = classificationService.classificationLookup(_classification);
403
            classificationLookupDao.filterInclude(ranks);
404

    
405
            double end1 = System.currentTimeMillis();
406
            logger.info("Time elapsed for classificationLookup() : " + (end1 - start) / (1000) + "s");
407
            double start2 = System.currentTimeMillis();
408

    
409
            monitor.subTask("Accumulating distributions to super areas for " + _classification.getTitleCache());
410
            if (mode.equals(AggregationMode.byAreas) || mode.equals(AggregationMode.byAreasAndRanks)) {
411
                accumulateByArea(superAreas, classificationLookupDao, new SubProgressMonitor(monitor, byAreasTicks), true);
412
            }
413
            monitor.subTask("Accumulating distributions to higher ranks for " + _classification.getTitleCache());
414

    
415
            double end2 = System.currentTimeMillis();
416
            logger.info("Time elapsed for accumulateByArea() : " + (end2 - start2) / (1000) + "s");
417

    
418
            double start3 = System.currentTimeMillis();
419
            if (mode.equals(AggregationMode.byRanks) || mode.equals(AggregationMode.byAreasAndRanks)) {
420
                accumulateByRank(ranks, classificationLookupDao, new SubProgressMonitor(monitor, byRankTicks), mode.equals(AggregationMode.byRanks));
421
            }
422

    
423
            double end3 = System.currentTimeMillis();
424
            logger.info("Time elapsed for accumulateByRank() : " + (end3 - start3) / (1000) + "s");
425
            logger.info("Time elapsed for accumulate(): " + (end3 - start) / (1000) + "s");
426

    
427
            if(ONLY_FISRT_BATCH) {
428
                monitor.done();
429
                break;
430
            }
431
        }
432
        monitor.done();
433
    }
434

    
435

    
436
    /**
437
     * Step 1: Accumulate occurrence records by area
438
     * <ul>
439
     * <li>areas are projected to super areas e.g.:  HS <-- HS(A), HS(G), HS(S)</li>
440
     * <li>super areas do initially not have a status set ==> Prerequisite to check in CDM</li>
441
     * <li>areas having a summary status of summary value different from {@link #getByAreaIgnoreStatusList()} are ignored</li>
442
     * <li>areas have a priority value, the status of the area with highest priority determines the status of the super area</li>
443
     * <li>the source references of the accumulated distributions are also accumulated into the new distribution,,</li>
444
     * <li>this has been especially implemented for the EuroMed Checklist Vol2 and might not be a general requirement</li>
445
     * </ul>
446
     *
447
     * @param superAreas
448
     *      the areas to which the subordinate areas should be projected
449
     * @param classificationLookupDao
450
     * @throws JvmLimitsException
451
     *
452
     */
453
    protected void accumulateByArea(List<NamedArea> superAreas, ClassificationLookupDTO classificationLookupDao,  IProgressMonitor subMonitor, boolean doClearDescriptions) throws JvmLimitsException {
454

    
455
        DynamicBatch batch = new DynamicBatch(BATCH_SIZE_BY_AREA, BATCH_MIN_FREE_HEAP);
456
        batch.setRequiredFreeHeap(BATCH_FREE_HEAP_RATIO);
457

    
458
        TransactionStatus txStatus = startTransaction(false);
459

    
460
        // reload superAreas TODO is it faster to getSession().merge(object) ??
461
        Set<UUID> superAreaUuids = new HashSet<UUID>(superAreas.size());
462
        for (NamedArea superArea : superAreas){
463
            superAreaUuids.add(superArea.getUuid());
464
        }
465

    
466
        // visit all accepted taxa
467
        subMonitor.beginTask("Accumulating by area ",  classificationLookupDao.getTaxonIds().size());
468
        Iterator<Integer> taxonIdIterator = classificationLookupDao.getTaxonIds().iterator();
469

    
470
        while (taxonIdIterator.hasNext() || batch.hasUnprocessedItems()) {
471

    
472
            if(txStatus == null) {
473
                // transaction has been comitted at the end of this batch, start a new one
474
                txStatus = startTransaction(false);
475
            }
476

    
477
            // the session is cleared after each batch, so load the superAreaList for each batch
478
            List<NamedArea> superAreaList = (List)termService.find(superAreaUuids);
479

    
480
            // load taxa for this batch
481
            List<Integer> taxonIds = batch.nextItems(taxonIdIterator);
482
//            logger.debug("accumulateByArea() - taxon " + taxonPager.getFirstRecord() + " to " + taxonPager.getLastRecord() + " of " + taxonPager.getCount() + "]");
483
            List<TaxonBase> taxa = taxonService.loadByIds(taxonIds, TAXONDESCRIPTION_INIT_STRATEGY);
484

    
485
            // iterate over the taxa and accumulate areas
486
            // start processing the new batch
487

    
488
            for(TaxonBase taxonBase : taxa) {
489
                if(logger.isDebugEnabled()){
490
                    logger.debug("accumulateByArea() - taxon :" + taxonToString(taxonBase));
491
                }
492

    
493
                batch.incementCounter();
494

    
495
                Taxon taxon = (Taxon)taxonBase;
496
                TaxonDescription description = findComputedDescription(taxon, doClearDescriptions);
497
                List<Distribution> distributions = distributionsFor(taxon);
498

    
499
                // Step through superAreas for accumulation of subAreas
500
                for (NamedArea superArea : superAreaList){
501

    
502
                    // accumulate all sub area status
503
                    StatusAndSources accumulatedStatusAndSources = null;
504
                    // TODO consider using the TermHierarchyLookup (only in local branch a.kohlbecker)
505
                    Set<NamedArea> subAreas = getSubAreasFor(superArea);
506
                    for(NamedArea subArea : subAreas){
507
                        if(logger.isTraceEnabled()){
508
                            logger.trace("accumulateByArea() - \t\t" + termToString(subArea));
509
                        }
510
                        // step through all distributions for the given subArea
511
                        for(Distribution distribution : distributions){
512
                            if(distribution.getArea() != null && distribution.getArea().equals(subArea) && distribution.getStatus() != null) {
513
                                PresenceAbsenceTerm status = distribution.getStatus();
514
                                if(logger.isTraceEnabled()){
515
                                    logger.trace("accumulateByArea() - \t\t" + termToString(subArea) + ": " + termToString(status));
516
                                }
517
                                // skip all having a status value different of those in byAreaIgnoreStatusList
518
                                if (getByAreaIgnoreStatusList().contains(status)){
519
                                    continue;
520
                                }
521
                                StatusAndSources subStatusAndSources = new StatusAndSources(status, distribution.getSources());
522
                                accumulatedStatusAndSources = choosePreferred(accumulatedStatusAndSources, subStatusAndSources, null);
523
                            }
524
                        }
525
                    } // next sub area
526
                    if (accumulatedStatusAndSources != null) {
527
                        if(logger.isDebugEnabled()){
528
                            logger.debug("accumulateByArea() - \t >> " + termToString(superArea) + ": " + termToString(accumulatedStatusAndSources.status));
529
                        }
530
                        // store new distribution element for superArea in taxon description
531
                        Distribution newDistribitionElement = Distribution.NewInstance(superArea, accumulatedStatusAndSources.status);
532
                        newDistribitionElement.getSources().addAll(accumulatedStatusAndSources.sources);
533
                        newDistribitionElement.addMarker(Marker.NewInstance(MarkerType.COMPUTED(), true));
534
                        description.addElement(newDistribitionElement);
535
                    }
536

    
537
                } // next super area ....
538

    
539
                descriptionService.saveOrUpdate(description);
540
                taxonService.saveOrUpdate(taxon);
541
                subMonitor.worked(1);
542
                if(!batch.isWithinJvmLimits()) {
543
                    break; // flushAndClear and start with new batch
544
                }
545

    
546
            } // next taxon
547

    
548
            flushAndClear();
549

    
550
            // commit for every batch, otherwise the persistent context
551
            // may grow too much and eats up all the heap
552
            commitTransaction(txStatus);
553
            txStatus = null;
554

    
555
            if(ONLY_FISRT_BATCH) {
556
                break;
557
            }
558

    
559
        } // next batch of taxa
560

    
561
        subMonitor.done();
562
    }
563

    
564
   /**
565
    * Step 2: Accumulate by ranks starting from lower rank to upper rank, the status of all children
566
    * are accumulated on each rank starting from lower rank to upper rank.
567
    * <ul>
568
    * <li>aggregate distribution of included taxa of the next lower rank for any rank level starting from the lower rank (e.g. sub species)
569
    *    up to upper rank (e.g. Genus)</li>
570
    *  <li>the accumulation id done for each distribution area found in the included taxa</li>
571
    *  <li>areas of subtaxa with status endemic are ignored</li>
572
    *  <li>the status with the highest priority determines the value for the accumulated distribution</li>
573
    *  <li>the source reference of the accumulated distributions are also accumulated into the new distribution,
574
    *    this has been especially implemented for the EuroMed Checklist Vol2 and might not be a general requirement</li>
575
    *</ul>
576
 * @throws JvmLimitsException
577
    */
578
    protected void accumulateByRank(List<Rank> rankInterval, ClassificationLookupDTO classificationLookupDao,  IProgressMonitor subMonitor, boolean doClearDescriptions) throws JvmLimitsException {
579

    
580
        DynamicBatch batch = new DynamicBatch(BATCH_SIZE_BY_RANK, BATCH_MIN_FREE_HEAP);
581
        batch.setRequiredFreeHeap(BATCH_FREE_HEAP_RATIO);
582
        batch.setMaxAllowedGcIncreases(10);
583

    
584
        int ticksPerRank = 100;
585

    
586
        TransactionStatus txStatus = startTransaction(false);
587

    
588
        // the loadRankSpecificRootNodes() method not only finds
589
        // taxa of the specified rank but also taxa of lower ranks
590
        // if no taxon of the specified rank exists, so we need to
591
        // remember which taxa have been processed already
592
        Set<Integer> taxaProcessedIds = new HashSet<Integer>();
593
        List<TaxonBase> taxa = null;
594
        List<TaxonBase> childTaxa = null;
595

    
596
        List<Rank> ranks = rankInterval;
597

    
598
        subMonitor.beginTask("Accumulating by rank", ranks.size() * ticksPerRank);
599

    
600
        for (Rank rank : ranks) {
601

    
602
            if(logger.isDebugEnabled()){
603
                logger.debug("accumulateByRank() - at Rank '" + termToString(rank) + "'");
604
            }
605

    
606
            Set<Integer> taxonIdsPerRank = classificationLookupDao.getTaxonIdByRank().get(rank);
607

    
608
            int taxonCountperRank = taxonIdsPerRank != null ? taxonIdsPerRank.size() : 0;
609

    
610
            SubProgressMonitor taxonSubMonitor = new SubProgressMonitor(subMonitor, ticksPerRank);
611
            taxonSubMonitor.beginTask("Accumulating by rank " + termToString(rank), taxonCountperRank);
612

    
613
            if(taxonCountperRank == 0) {
614
                taxonSubMonitor.done();
615
                continue;
616
            }
617

    
618

    
619
            Iterator<Integer> taxonIdIterator = taxonIdsPerRank.iterator();
620
            while (taxonIdIterator.hasNext() || batch.hasUnprocessedItems()) {
621

    
622
                if(txStatus == null) {
623
                    // transaction has been committed at the end of this batch, start a new one
624
                    txStatus = startTransaction(false);
625
                }
626

    
627
                // load taxa for this batch
628
                List<Integer> taxonIds = batch.nextItems(taxonIdIterator);
629

    
630
                taxa = taxonService.loadByIds(taxonIds, null);
631

    
632
//                if(logger.isDebugEnabled()){
633
//                           logger.debug("accumulateByRank() - taxon " + taxonPager.getFirstRecord() + " to " + taxonPager.getLastRecord() + " of " + taxonPager.getCount() + "]");
634
//                }
635

    
636
                for(TaxonBase taxonBase : taxa) {
637

    
638
                    batch.incementCounter();
639

    
640
                    Taxon taxon = (Taxon)taxonBase;
641
                    if (taxaProcessedIds.contains(taxon.getId())) {
642
                        if(logger.isDebugEnabled()){
643
                            logger.debug("accumulateByRank() - skipping already processed taxon :" + taxonToString(taxon));
644
                        }
645
                        continue;
646
                    }
647
                    taxaProcessedIds.add(taxon.getId());
648
                    if(logger.isDebugEnabled()){
649
                        logger.debug("accumulateByRank() [" + rank.getLabel() + "] - taxon :" + taxonToString(taxon));
650
                    }
651

    
652
                    // Step through direct taxonomic children for accumulation
653
                    Map<NamedArea, StatusAndSources> accumulatedStatusMap = new HashMap<NamedArea, StatusAndSources>();
654

    
655
                    List<Integer> childTaxonIds = new ArrayList<>();
656
                    Set<Integer> childSet = classificationLookupDao.getChildTaxonMap().get(taxon.getId());
657
                    if(childSet != null) {
658
                        childTaxonIds.addAll(childSet);
659
                    }
660
                    if(!childTaxonIds.isEmpty()) {
661
                        childTaxa = taxonService.loadByIds(childTaxonIds, TAXONDESCRIPTION_INIT_STRATEGY);
662
                        LinkedList<TaxonBase> childStack = new LinkedList<TaxonBase>(childTaxa);
663
                        childTaxa = null; // allow to be garbage collected
664

    
665
                        while(childStack.size() > 0){
666

    
667
                            TaxonBase childTaxonBase = childStack.pop();
668
                            getSession().setReadOnly(childTaxonBase, true);
669

    
670
                            Taxon childTaxon = (Taxon) childTaxonBase;
671
                            getSession().setReadOnly(childTaxon, true);
672
                            if(logger.isTraceEnabled()){
673
                                logger.trace("                   subtaxon :" + taxonToString(childTaxon));
674
                            }
675

    
676
                            for(Distribution distribution : distributionsFor(childTaxon) ) {
677
                                PresenceAbsenceTerm status = distribution.getStatus();
678
                                NamedArea area = distribution.getArea();
679
                                if (status == null || getByRankIgnoreStatusList().contains(status)){
680
                                  continue;
681
                                }
682

    
683
                                StatusAndSources subStatusAndSources = new StatusAndSources(status, distribution.getSources());
684
                                accumulatedStatusMap.put(area, choosePreferred(accumulatedStatusMap.get(area), subStatusAndSources, null));
685
                             }
686

    
687
                            // evict all initialized entities of the childTaxon
688
                            // TODO consider using cascade="evict" in the model classes
689
//                            for( TaxonDescription description : ((Taxon)childTaxonBase).getDescriptions()) {
690
//                                for (DescriptionElementBase deb : description.getElements()) {
691
//                                    getSession().evict(deb);
692
//                                }
693
//                                getSession().evict(description); // this causes in some cases the taxon object to be detached from the session
694
//                            }
695
                            getSession().evict(childTaxonBase); // no longer needed, save heap
696
                        }
697

    
698
                        if(accumulatedStatusMap.size() > 0) {
699
                            TaxonDescription description = findComputedDescription(taxon, doClearDescriptions);
700
                            for (NamedArea area : accumulatedStatusMap.keySet()) {
701
                                Distribution distribition = findDistribution(description, area, accumulatedStatusMap.get(area).status);
702
                                if(distribition == null) {
703
                                    // create a new distribution element
704
                                    distribition = Distribution.NewInstance(area, accumulatedStatusMap.get(area).status);
705
                                    distribition.addMarker(Marker.NewInstance(MarkerType.COMPUTED(), true));
706
                                }
707
                                addSourcesDeduplicated(distribition.getSources(), accumulatedStatusMap.get(area).sources);
708

    
709
                                description.addElement(distribition);
710
                            }
711
                            taxonService.saveOrUpdate(taxon);
712
                            descriptionService.saveOrUpdate(description);
713
                        }
714

    
715
                    }
716
                    taxonSubMonitor.worked(1); // one taxon worked
717
                    if(!batch.isWithinJvmLimits()) {
718
                        break; // flushAndClear and start with new batch
719
                    }
720

    
721
                } // next taxon ....
722

    
723
                flushAndClear();
724

    
725
                // commit for every batch, otherwise the persistent context
726
                // may grow too much and eats up all the heap
727
                commitTransaction(txStatus);
728
                txStatus = null;
729

    
730
                // flushing the session and to the index (flushAndClear() ) can impose a
731
                // massive heap consumption. therefore we explicitly do a check after the
732
                // flush to detect these situations and to reduce the batch size.
733
                if(batch.getJvmMonitor().getGCRateSiceLastCheck() > 0.05) {
734
                    batch.reduceSize(0.5);
735
                }
736

    
737
                if(ONLY_FISRT_BATCH) {
738
                    break;
739
                }
740
            } // next batch
741

    
742
            taxonSubMonitor.done();
743
            subMonitor.worked(1);
744

    
745
            if(ONLY_FISRT_BATCH) {
746
                break;
747
            }
748
        } // next Rank
749

    
750
        logger.info("accumulateByRank() - done");
751
        subMonitor.done();
752
    }
753

    
754
/**
755
 * @param description
756
 * @param area
757
 * @param status
758
 * @return
759
 */
760
private Distribution findDistribution(TaxonDescription description, NamedArea area, PresenceAbsenceTerm status) {
761
    for(DescriptionElementBase item : description.getElements()) {
762
        if(!(item instanceof Distribution)) {
763
            continue;
764
        }
765
        Distribution distribution = ((Distribution)item);
766
        if(distribution.getArea().equals(area) && distribution.getStatus().equals(status)) {
767
            return distribution;
768
        }
769
    }
770
    return null;
771
}
772

    
773
/**
774
 * @param lowerRank
775
 * @param upperRank
776
 * @return
777
 */
778
private List<Rank> rankInterval(Rank lowerRank, Rank upperRank) {
779

    
780
    TransactionStatus txStatus = startTransaction(false);
781
    Rank currentRank = lowerRank;
782
    List<Rank> ranks = new ArrayList<Rank>();
783
    ranks.add(currentRank);
784
    while (!currentRank.isHigher(upperRank)) {
785
        currentRank = findNextHigherRank(currentRank);
786
        ranks.add(currentRank);
787
    }
788
    commitTransaction(txStatus);
789
    txStatus = null;
790
    return ranks;
791
}
792

    
793
    /**
794
     * @return
795
     */
796
    private Session getSession() {
797
        return descriptionService.getSession();
798
    }
799

    
800
    /**
801
     *
802
     */
803
    private void flush() {
804
        logger.debug("flushing session ...");
805
        getSession().flush();
806
        try {
807
            logger.debug("flushing to indexes ...");
808
            Search.getFullTextSession(getSession()).flushToIndexes();
809
        } catch (HibernateException e) {
810
            /* IGNORE - Hibernate Search Event listeners not configured ... */
811
            if(!e.getMessage().startsWith("Hibernate Search Event listeners not configured")){
812
                throw e;
813
            }
814
        }
815
    }
816

    
817
    /**
818
    *
819
    */
820
   private void flushAndClear() {
821
       flush();
822
       logger.debug("clearing session ...");
823
       getSession().clear();
824
   }
825

    
826

    
827
    // TODO merge with CdmApplicationDefaultConfiguration#startTransaction() into common base class
828
    public TransactionStatus startTransaction(Boolean readOnly) {
829

    
830
        DefaultTransactionDefinition defaultTxDef = new DefaultTransactionDefinition();
831
        defaultTxDef.setReadOnly(readOnly);
832
        TransactionDefinition txDef = defaultTxDef;
833

    
834
        // Log some transaction-related debug information.
835
        if (logger.isTraceEnabled()) {
836
            logger.trace("Transaction name = " + txDef.getName());
837
            logger.trace("Transaction facets:");
838
            logger.trace("Propagation behavior = " + txDef.getPropagationBehavior());
839
            logger.trace("Isolation level = " + txDef.getIsolationLevel());
840
            logger.trace("Timeout = " + txDef.getTimeout());
841
            logger.trace("Read Only = " + txDef.isReadOnly());
842
            // org.springframework.orm.hibernate5.HibernateTransactionManager
843
            // provides more transaction/session-related debug information.
844
        }
845

    
846
        TransactionStatus txStatus = transactionManager.getTransaction(txDef);
847

    
848
        getSession().setFlushMode(FlushMode.COMMIT);
849

    
850
        return txStatus;
851
    }
852

    
853
    // TODO merge with CdmApplicationDefaultConfiguration#startTransaction() into common base class
854
    public void commitTransaction(TransactionStatus txStatus){
855
        logger.debug("commiting transaction ...");
856
        transactionManager.commit(txStatus);
857
        return;
858
    }
859

    
860
    /**
861
     * returns the next higher rank
862
     *
863
     * TODO better implement OrderedTermBase.getNextHigherTerm() and OrderedTermBase.getNextLowerTerm()?
864
     *
865
     * @param rank
866
     * @return
867
     */
868
    private Rank findNextHigherRank(Rank rank) {
869
        rank = (Rank) termService.load(rank.getUuid());
870
        return rank.getNextHigherTerm();
871
//        OrderedTermVocabulary<Rank> rankVocabulary = mameService.getRankVocabulary();;
872
//        return rankVocabulary.getNextHigherTerm(rank);
873
    }
874

    
875
    /**
876
     * Either finds an existing taxon description of the given taxon or creates a new one.
877
     * If the doClear is set all existing description elements will be cleared.
878
     *
879
     * @param taxon
880
     * @param doClear will remove all existing Distributions if the taxon already
881
     * has a MarkerType.COMPUTED() TaxonDescription
882
     * @return
883
     */
884
    private TaxonDescription findComputedDescription(Taxon taxon, boolean doClear) {
885

    
886
        String descriptionTitle = this.getClass().getSimpleName();
887

    
888
        // find existing one
889
        for (TaxonDescription description : taxon.getDescriptions()) {
890
            if (description.hasMarker(MarkerType.COMPUTED(), true)) {
891
                logger.debug("reusing computed description for " + taxon.getTitleCache());
892
                if (doClear) {
893
                    int deleteCount = 0;
894
                    Set<DescriptionElementBase> deleteCandidates = new HashSet<DescriptionElementBase>();
895
                    for (DescriptionElementBase descriptionElement : description.getElements()) {
896
                        if(descriptionElement instanceof Distribution) {
897
                            deleteCandidates.add(descriptionElement);
898
                        }
899
                    }
900
                    if(deleteCandidates.size() > 0){
901
                        for(DescriptionElementBase descriptionElement : deleteCandidates) {
902
                            description.removeElement(descriptionElement);
903
                            descriptionService.deleteDescriptionElement(descriptionElement);
904
                            descriptionElement = null;
905
                            deleteCount++;
906
                        }
907
                        descriptionService.saveOrUpdate(description);
908
                        logger.debug("\t" + deleteCount +" distributions cleared");
909
                    }
910

    
911
                }
912
                return description;
913
            }
914
        }
915

    
916
        // create a new one
917
        logger.debug("creating new description for " + taxon.getTitleCache());
918
        TaxonDescription description = TaxonDescription.NewInstance(taxon);
919
        description.setTitleCache(descriptionTitle, true);
920
        description.addMarker(Marker.NewInstance(MarkerType.COMPUTED(), true));
921
        return description;
922
    }
923

    
924
    /**
925
     * @param superArea
926
     * @return
927
     */
928
    private Set<NamedArea> getSubAreasFor(NamedArea superArea) {
929

    
930
        if(!subAreaMap.containsKey(superArea)) {
931
            if(logger.isDebugEnabled()){
932
                logger.debug("loading included areas for " + superArea.getLabel());
933
            }
934
            subAreaMap.put(superArea, superArea.getIncludes());
935
        }
936
        return subAreaMap.get(superArea);
937
    }
938

    
939
    /**
940
     * @param taxon
941
     * @return
942
     */
943
    private List<Distribution> distributionsFor(Taxon taxon) {
944
        List<Distribution> distributions = new ArrayList<Distribution>();
945
        for(TaxonDescription description: taxon.getDescriptions()) {
946
            readOnlyIfInSession(description);
947
            for(DescriptionElementBase deb : description.getElements()) {
948
                if(deb instanceof Distribution) {
949
                    readOnlyIfInSession(deb);
950
                    distributions.add((Distribution)deb);
951
                }
952
            }
953
        }
954
        return distributions;
955
    }
956

    
957
    /**
958
     * This method avoids problems when running the TransmissionEngineDistribution test.
959
     * For some unknown reason entities are not in the PersitenceContext even if they are
960
     * loaded by a service method. Setting these entities to readonly would raise a
961
     * TransientObjectException("Instance was not associated with this persistence context")
962
     *
963
     * @param entity
964
     */
965
    private void readOnlyIfInSession(CdmBase entity) {
966
        if(getSession().contains(entity)) {
967
            getSession().setReadOnly(entity, true);
968
        }
969
    }
970

    
971
    /**
972
     * @param taxon
973
     * @param logger2
974
     * @return
975
     */
976
    private String taxonToString(TaxonBase taxon) {
977
        if(logger.isTraceEnabled()) {
978
            return taxon.getTitleCache();
979
        } else {
980
            return taxon.toString();
981
        }
982
    }
983

    
984
    /**
985
     * @param taxon
986
     * @param logger2
987
     * @return
988
     */
989
    private String termToString(OrderedTermBase<?> term) {
990
        if(logger.isTraceEnabled()) {
991
            return term.getLabel() + " [" + term.getIdInVocabulary() + "]";
992
        } else {
993
            return term.getIdInVocabulary();
994
        }
995
    }
996

    
997
    /**
998
     * Sets the priorities for presence and absence terms, the priorities are stored in extensions.
999
     * This method will start a new transaction and commits it after the work is done.
1000
     */
1001
    public void updatePriorities() {
1002

    
1003
        TransactionStatus txStatus = startTransaction(false);
1004

    
1005
        Map<PresenceAbsenceTerm, Integer> priorityMap = new HashMap<PresenceAbsenceTerm, Integer>();
1006

    
1007
        priorityMap.put(PresenceAbsenceTerm.CULTIVATED_REPORTED_IN_ERROR(), 1);
1008
        priorityMap.put(PresenceAbsenceTerm.INTRODUCED_UNCERTAIN_DEGREE_OF_NATURALISATION(), 2);
1009
        priorityMap.put(PresenceAbsenceTerm.INTRODUCED_FORMERLY_INTRODUCED(), 3);
1010
        priorityMap.put(PresenceAbsenceTerm.INTRODUCED_REPORTED_IN_ERROR(), 20);
1011
        priorityMap.put(PresenceAbsenceTerm.NATIVE_REPORTED_IN_ERROR(), 30);
1012
        priorityMap.put(PresenceAbsenceTerm.CULTIVATED(), 45);
1013
        priorityMap.put(PresenceAbsenceTerm.NATIVE_FORMERLY_NATIVE(), 40);
1014
        priorityMap.put(PresenceAbsenceTerm.NATIVE_PRESENCE_QUESTIONABLE(), 60);
1015
        priorityMap.put(PresenceAbsenceTerm.INTRODUCED_PRESENCE_QUESTIONABLE(), 50);
1016
        priorityMap.put(PresenceAbsenceTerm.INTRODUCED_DOUBTFULLY_INTRODUCED(), 80);
1017
        priorityMap.put(PresenceAbsenceTerm.INTRODUCED(), 90);
1018
        priorityMap.put(PresenceAbsenceTerm.INTRODUCED_ADVENTITIOUS(), 100);
1019
        priorityMap.put(PresenceAbsenceTerm.INTRODUCED_NATURALIZED(), 110);
1020
        priorityMap.put(PresenceAbsenceTerm.NATIVE_DOUBTFULLY_NATIVE(), 120); // null
1021
        priorityMap.put(PresenceAbsenceTerm.NATIVE(), 130); // null
1022
        priorityMap.put(PresenceAbsenceTerm.ENDEMIC_FOR_THE_RELEVANT_AREA(), 999);
1023

    
1024
        for(PresenceAbsenceTerm term : priorityMap.keySet()) {
1025
            // load the term
1026
            term = (PresenceAbsenceTerm) termService.load(term.getUuid());
1027
            // find the extension
1028
            Extension priorityExtension = null;
1029
            Set<Extension> extensions = term.getExtensions();
1030
            for(Extension extension : extensions){
1031
                if (!extension.getType().equals(ExtensionType.ORDER())) {
1032
                    continue;
1033
                }
1034
                int pos = extension.getValue().indexOf(EXTENSION_VALUE_PREFIX);
1035
                if(pos == 0){ // if starts with EXTENSION_VALUE_PREFIX
1036
                    priorityExtension = extension;
1037
                    break;
1038
                }
1039
            }
1040
            if(priorityExtension == null) {
1041
                priorityExtension = Extension.NewInstance(term, null, ExtensionType.ORDER());
1042
            }
1043
            priorityExtension.setValue(EXTENSION_VALUE_PREFIX + priorityMap.get(term));
1044

    
1045
            // save the term
1046
            termService.saveOrUpdate(term);
1047
            if (logger.isDebugEnabled()) {
1048
                logger.debug("Priority updated for " + term.getLabel());
1049
            }
1050
        }
1051

    
1052
        commitTransaction(txStatus);
1053
    }
1054

    
1055
    public static void addSourcesDeduplicated(Set<DescriptionElementSource> target, Set<DescriptionElementSource> sources) {
1056
        for(DescriptionElementSource source : sources) {
1057
            boolean contained = false;
1058
            for(DescriptionElementSource existingSource: target) {
1059
                if(existingSource.equalsByShallowCompare(source)) {
1060
                    contained = true;
1061
                    break;
1062
                }
1063
            }
1064
            if(!contained) {
1065
                try {
1066
                    target.add((DescriptionElementSource)source.clone());
1067
                } catch (CloneNotSupportedException e) {
1068
                    // should never happen
1069
                    throw new RuntimeException(e);
1070
                }
1071
            }
1072
        }
1073
    }
1074

    
1075
    public enum AggregationMode {
1076
        byAreas,
1077
        byRanks,
1078
        byAreasAndRanks
1079

    
1080
    }
1081

    
1082
    private class StatusAndSources {
1083

    
1084
        private final PresenceAbsenceTerm status;
1085

    
1086
        private final Set<DescriptionElementSource> sources = new HashSet<>();
1087

    
1088
        public StatusAndSources(PresenceAbsenceTerm status, Set<DescriptionElementSource> sources) {
1089
            this.status = status;
1090
            addSourcesDeduplicated(this.sources, sources);
1091
        }
1092

    
1093
        /**
1094
         * @param sources
1095
         */
1096
        public void addSources(Set<DescriptionElementSource> sources) {
1097
            addSourcesDeduplicated(this.sources, sources);
1098
        }
1099

    
1100
    }
1101
}
    (1-1/1)