1 package org.daisy.pipeline.tts.calabash.impl;
2
3 import java.math.BigDecimal;
4 import java.math.MathContext;
5 import java.util.ArrayList;
6 import java.util.Collection;
7 import java.util.Collections;
8 import java.util.HashMap;
9 import java.util.HashSet;
10 import java.util.Iterator;
11 import java.util.List;
12 import java.util.Map;
13 import java.util.Set;
14 import java.util.concurrent.BlockingQueue;
15 import java.util.concurrent.ConcurrentLinkedQueue;
16
17 import javax.sound.sampled.AudioFormat;
18 import javax.sound.sampled.AudioInputStream;
19
20 import net.sf.saxon.s9api.Axis;
21 import net.sf.saxon.s9api.QName;
22 import net.sf.saxon.s9api.XdmNode;
23 import net.sf.saxon.s9api.XdmNodeKind;
24 import net.sf.saxon.s9api.XdmSequenceIterator;
25
26 import org.daisy.common.messaging.MessageAppender;
27 import org.daisy.common.messaging.MessageBuilder;
28 import org.daisy.pipeline.audio.AudioUtils;
29 import org.daisy.pipeline.tts.AudioFootprintMonitor;
30 import org.daisy.pipeline.tts.AudioFootprintMonitor.MemoryException;
31 import org.daisy.pipeline.tts.SSMLMarkSplitter;
32 import org.daisy.pipeline.tts.Sentence;
33 import org.daisy.pipeline.tts.SSMLMarkSplitter.Chunk;
34 import org.daisy.pipeline.tts.TTSEngine;
35 import org.daisy.pipeline.tts.TTSLog;
36 import org.daisy.pipeline.tts.TTSEngine.SynthesisResult;
37 import org.daisy.pipeline.tts.TTSLog.ErrorCode;
38 import org.daisy.pipeline.tts.TTSRegistry;
39 import org.daisy.pipeline.tts.TTSRegistry.TTSResource;
40 import org.daisy.pipeline.tts.TTSService.SynthesisException;
41 import org.daisy.pipeline.tts.TTSTimeout;
42 import org.daisy.pipeline.tts.TimedTTSExecutor;
43 import org.daisy.pipeline.tts.TTSTimeout.ThreadFreeInterrupter;
44 import org.daisy.pipeline.tts.TimedTTSExecutor.TimeoutException;
45 import org.daisy.pipeline.tts.Voice;
46 import org.daisy.pipeline.tts.Voice.MarkSupport;
47 import org.daisy.pipeline.tts.VoiceManager;
48
49 import org.slf4j.Logger;
50
51 import com.google.common.collect.Iterables;
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68 public class TextToPcmThread implements FormatSpecifications {
69 private Logger mLogger;
70 private Map<TTSEngine, TTSResource> mResources = new HashMap<TTSEngine, TTSResource>();
71 private int mFileNrInSection;
72 private List<SoundFileLink> mSoundFileLinks;
73 private List<SoundFileLink> mLinksOfCurrentFile;
74 private Iterable<AudioInputStream> mAudioOfCurrentFile;
75 private int mOffsetInFile;
76 private int mMemFootprint;
77 private Thread mThread;
78 private TimedTTSExecutor mExecutor;
79 private TTSRegistry mTTSRegistry;
80 private AudioFormat mLastFormat;
81 private AudioFootprintMonitor mAudioFootprintMonitor;
82 private SSMLMarkSplitter mSSMLSplitter;
83 private VoiceManager mVoiceManager;
84 private TTSLog mTTSLog;
85 private int mErrorCounter;
86 private Throwable uncaughtException;
87
88
89
90
91 private class Mark {
92 public Mark(String name, int offset) {
93 this.offsetInAudio = offset;
94 this.name = name;
95 }
96
97
98
99 public String name;
100
101
102
103 public int offsetInAudio;
104 }
105
106
107
108
109
110 void start(final ConcurrentLinkedQueue<ContiguousText> input,
111 final BlockingQueue<ContiguousPCM> pcmOutput, TimedTTSExecutor executor,
112 TTSRegistry ttsregistry, VoiceManager voiceManager, SSMLMarkSplitter ssmlSplitter,
113 Logger logger, AudioFootprintMonitor audioFootprintMonitor, final int maxQueueEltSize,
114 TTSLog ttsLog, MessageAppender messageAppender,
115 long totalTextSize, BigDecimal portion) {
116 mSSMLSplitter = ssmlSplitter;
117 mSoundFileLinks = new ArrayList<SoundFileLink>();
118 mExecutor = executor;
119 mTTSRegistry = ttsregistry;
120 mLogger = logger;
121 mAudioFootprintMonitor = audioFootprintMonitor;
122 mVoiceManager = voiceManager;
123 mTTSLog = ttsLog;
124 mErrorCounter = 0;
125 flush(null, pcmOutput);
126
127 mThread = new Thread() {
128 @Override
129 public void run() {
130
131
132 MessageAppender messageThread = messageAppender != null
133 ? messageAppender.append(new MessageBuilder().withProgress(portion))
134 : null;
135 TTSTimeout timeout = new TTSTimeout();
136 try {
137
138
139 while (true) {
140 ContiguousText section = input.poll();
141 if (section == null) {
142 break;
143 }
144 mFileNrInSection = 0;
145 boolean breakloop = false;
146 for (Sentence sentence : section.sentences) {
147 if (breakloop) {
148 mErrorCounter++;
149 continue;
150 }
151 try {
152 if (!speak(section, sentence, pcmOutput, timeout, maxQueueEltSize)) mErrorCounter++;
153 } catch (Throwable t) {
154 mErrorCounter++;
155 mTTSLog.getWritableEntry(sentence.getID()).addError(
156 new TTSLog.Error(
157 TTSLog.ErrorCode.CRITICAL_ERROR, "the current thread is stopping because of an error", t));
158 breakloop = true;
159 }
160 }
161 flush(section, pcmOutput);
162
163
164 if (messageThread != null && portion.compareTo(BigDecimal.ZERO) > 0) {
165 MessageBuilder m = new MessageBuilder()
166 .withProgress(
167 new BigDecimal(section.getStringSize()).divide(new BigDecimal(totalTextSize), MathContext.DECIMAL128)
168 .divide(portion, MathContext.DECIMAL128)
169 .min(BigDecimal.ONE));
170 messageThread.append(m).close();
171 }
172 }
173
174
175 for (Map.Entry<TTSEngine, TTSResource> e : mResources.entrySet()) {
176 timeout.enableForCurrentThread(2);
177 try {
178 releaseResource(e.getKey(), e.getValue());
179 } catch (Exception ex) {
180 mTTSLog.addGeneralError(
181 ErrorCode.WARNING,
182 "Error while releasing resource of " + e.getKey().getProvider().getName() + ex.getMessage(),
183 ex);
184 } finally {
185 timeout.disable();
186 }
187 }
188 } finally {
189 timeout.close();
190 if (messageThread != null)
191 messageThread.close();
192 }
193 }
194 };
195 mThread.setUncaughtExceptionHandler(
196 (thread, throwable) -> { uncaughtException = throwable; }
197 );
198 mThread.start();
199 }
200
201 Collection<SoundFileLink> getSoundFragments() {
202 if (mThread != null) {
203 try {
204 mThread.join();
205 } catch (InterruptedException e) {
206
207 mLogger.warn("TextToPCMThread interruption");
208 }
209 mThread = null;
210 if (uncaughtException != null) {
211 throw new RuntimeException("unexpected error", uncaughtException);
212 }
213 }
214
215 return mSoundFileLinks;
216 }
217
218 int getErrorCount() {
219 return mErrorCounter;
220 }
221
222 private void releaseResource(TTSEngine tts, TTSResource r) {
223 if (r == null) {
224 return;
225 }
226 synchronized (r) {
227 try {
228 tts.releaseThreadResources(r);
229 } catch (Throwable t) {
230 mTTSLog.addGeneralError(
231 ErrorCode.WARNING, "error while releasing resources of " + tts.getProvider().getName(), t);
232 }
233 }
234 }
235
236 private void flush(ContiguousText section, BlockingQueue<ContiguousPCM> pcmOutput) {
237 if (section != null && mLinksOfCurrentFile.size() > 0) {
238 if (mLastFormat == null) {
239 throw new RuntimeException("coding error");
240 } else {
241 String filePrefix = String.format("part%04d_%02d_%03d", section
242 .getDocumentPosition(), section.getDocumentSplitPosition(),
243 mFileNrInSection);
244
245 ContiguousPCM pcm = new ContiguousPCM(
246 AudioUtils.concat(mAudioOfCurrentFile), section.getAudioOutputDir(), filePrefix);
247 for (SoundFileLink link : mLinksOfCurrentFile) {
248 link.clipBase = pcm.getDestinationFile();
249 }
250 try {
251 mAudioFootprintMonitor.transferToEncoding(mMemFootprint, pcm.sizeInBytes());
252 } catch (InterruptedException e) {
253
254 mLogger.warn("interruption of memory transfer");
255 }
256 pcmOutput.add(pcm);
257 pcm = null;
258 mSoundFileLinks.addAll(mLinksOfCurrentFile);
259 ++mFileNrInSection;
260 }
261 }
262 mLinksOfCurrentFile = new ArrayList<SoundFileLink>();
263 mAudioOfCurrentFile = new ArrayList<AudioInputStream>();
264 mOffsetInFile = 0;
265 mMemFootprint = 0;
266 mLastFormat = null;
267 }
268
269
270
271
272
273
274 private Iterable<AudioInputStream> synthesize(
275 TTSTimeout timeout, TTSTimeout.ThreadFreeInterrupter interrupter, TTSLog.Entry logEntry,
276 Sentence sentence, TTSEngine tts, Voice voice, TTSResource threadResources,
277 List<Mark> marks, List<String> markNames
278 ) throws SynthesisException, TimeoutException, MemoryException {
279 if (tts.handlesMarks()
280 && voice.getMarkSupport() != MarkSupport.MARK_NOT_SUPPORTED){
281 logEntry.setActualVoice(voice);
282 SynthesisResult result = mExecutor.synthesizeWithTimeout(
283 timeout, interrupter, logEntry, sentence.getText(), sentence.getSize(), tts, voice,
284 threadResources);
285 if (!AudioUtils.isPCM(result.audio.getFormat()))
286 throw new IllegalArgumentException();
287 List<Integer> markOffsets = result.marks;
288 if (markNames.size() != markOffsets.size()) {
289 mTTSLog.getWritableEntry(sentence.getID()).addError(
290 new TTSLog.Error(ErrorCode.WARNING, "wrong number of marks with "
291 + tts.getProvider().getName()
292 + ". Number of marks received: " + markOffsets.size()
293 + ", expected number: " + markNames.size()));
294 return null;
295 }
296 for (int i = 0; i < markNames.size(); i++) {
297 marks.add(new Mark(markNames.get(i), markOffsets.get(i)));
298 }
299 mAudioFootprintMonitor.acquireTTSMemory(result.audio);
300 return Collections.singletonList(result.audio);
301 } else {
302 Collection<Chunk> chunks = mSSMLSplitter.split(sentence.getText());
303 List<AudioInputStream> result = new ArrayList<>();
304 int offset = 0;
305 AudioFormat format = null;
306 for (Chunk chunk : chunks) {
307 logEntry.setActualVoice(voice);
308 try {
309 AudioInputStream stream = mExecutor.synthesizeWithTimeout(
310 timeout, interrupter, logEntry, chunk.ssml(), Sentence.computeSize(chunk.ssml()),
311 tts, voice, threadResources).audio;
312 if (format == null) {
313 format = stream.getFormat();
314 if (!AudioUtils.isPCM(format))
315 throw new IllegalArgumentException("AudioInputStream must be PCM encoded, but got: "+ format);
316 } else if (!format.matches(stream.getFormat()))
317 throw new IllegalArgumentException("All AudioInputStream must have the same format");
318 if (chunk.leftMark() != null) {
319 marks.add(new Mark(chunk.leftMark(), offset));
320 }
321 int size = AudioFootprintMonitor.getFootprint(stream);
322 offset += size;
323 mAudioFootprintMonitor.acquireTTSMemory(size);
324 result.add(stream);
325 } catch (MemoryException | SynthesisException | TimeoutException e) {
326
327 for (AudioInputStream s : result)
328 mAudioFootprintMonitor.releaseTTSMemory(s);
329 throw e;
330 } catch (Throwable t) {
331
332 for (AudioInputStream s : result)
333 mAudioFootprintMonitor.releaseTTSMemory(s);
334 throw new SynthesisException(t);
335 }
336 }
337 if (markNames.size() != marks.size()) {
338 throw new RuntimeException();
339 }
340 return result;
341 }
342 }
343
344
345
346
347 private Iterable<AudioInputStream> speakWithVoice(final Sentence sentence, Voice v,
348 final TTSEngine tts, List<Mark> marks, List<String> markNames, TTSTimeout timeout)
349 throws MemoryException {
350
351 TTSResource resource = mResources.get(tts);
352 if (resource == null) {
353 timeout.enableForCurrentThread(3);
354 try {
355 resource = mTTSRegistry.allocateResourceFor(tts);
356 } catch (SynthesisException e) {
357 mTTSLog.getWritableEntry(sentence.getID()).addError(
358 new TTSLog.Error(ErrorCode.WARNING,
359 "Error while allocating resources for "
360 + tts.getProvider().getName() + ": "
361 + e));
362
363 return null;
364 } catch (InterruptedException e) {
365 mTTSLog.getWritableEntry(sentence.getID()).addError(
366 new TTSLog.Error(ErrorCode.WARNING,
367 "Timeout while trying to allocate resources for "
368 + tts.getProvider().getName()));
369 return null;
370 } finally {
371 timeout.disable();
372 }
373 if (resource == null) {
374
375 mTTSLog.getWritableEntry(sentence.getID()).addError(
376 new TTSLog.Error(ErrorCode.WARNING, "Could not allocate resource for "
377 + tts.getProvider().getName()
378 + " (it has probably been stopped)."));
379 return null;
380 }
381 mResources.put(tts, resource);
382 }
383
384
385 final TTSResource fresource = resource;
386 TTSTimeout.ThreadFreeInterrupter interrupter = new ThreadFreeInterrupter() {
387 @Override
388 public void threadFreeInterrupt() {
389 mTTSLog.getWritableEntry(sentence.getID()).addError(
390 new TTSLog.Error(
391 ErrorCode.WARNING,
392 "Forcing interruption of the current work of " + tts.getProvider().getName() + "..."));
393 tts.interruptCurrentWork(fresource);
394 }
395 };
396 TTSLog.Entry logEntry = mTTSLog.getWritableEntry(sentence.getID());
397 try {
398 synchronized (resource) {
399 if (resource.invalid) {
400 mTTSLog.getWritableEntry(sentence.getID()).addError(
401 new TTSLog.Error(
402 ErrorCode.WARNING,
403 "Resource of " + tts.getProvider().getName()
404 + " is no longer valid. The corresponding service has probably been stopped."));
405 return null;
406 }
407 return synthesize(timeout, interrupter, logEntry,
408 sentence, tts, v, resource, marks, markNames);
409 }
410 } catch (TimeoutException e) {
411 logEntry.addError(
412 new TTSLog.Error(ErrorCode.WARNING, "timeout (" + e.getSeconds()
413 + " seconds) fired while speaking with "
414 + tts.getProvider().getName()));
415 return null;
416 } catch (SynthesisException e) {
417 logEntry.addError(
418 new TTSLog.Error(
419 ErrorCode.WARNING, "error while speaking with " + tts.getProvider().getName() + ": " + e, e));
420
421 return null;
422 }
423 }
424
425 static List<String> getMarkNames(XdmNode ssml) {
426 XdmSequenceIterator iter = ssml.axisIterator(Axis.DESCENDANT);
427 ArrayList<String> markNames = new ArrayList<String>();
428 while (iter.hasNext()){
429 XdmNode elt = (XdmNode) iter.next();
430 if (elt.getNodeKind() == XdmNodeKind.ELEMENT && "mark".equals(elt.getNodeName().getLocalName())){
431 markNames.add(elt.getAttributeValue(new QName("name")));
432 }
433 }
434 return markNames;
435 }
436
437
438
439
440 private boolean speak(ContiguousText section, Sentence sentence,
441 BlockingQueue<ContiguousPCM> pcmOutput, TTSTimeout timeout, int maxQueueEltSize) {
442
443 List<String> markNames = getMarkNames(sentence.getText());
444
445 TTSEngine tts = sentence.getTTSproc();
446 Voice originalVoice = sentence.getPreferredVoice();
447 List<Mark> marks = new ArrayList<Mark>();
448 Iterable<AudioInputStream> pcm;
449 try {
450 pcm = speakWithVoice(sentence, originalVoice, tts, marks, markNames, timeout);
451 } catch (MemoryException e) {
452 flush(section, pcmOutput);
453 printMemError(sentence, e);
454 return false;
455 }
456 if (pcm == null) {
457
458 releaseResource(tts, mResources.get(tts));
459 mResources.remove(tts);
460
461
462 Iterator<Voice> fallbackVoices = sentence.getVoices().iterator();
463 fallbackVoices.next();
464 if (!fallbackVoices.hasNext()) {
465 mTTSLog.getWritableEntry(sentence.getID()).addError(
466 new TTSLog.Error(
467 TTSLog.ErrorCode.AUDIO_MISSING,
468 "something went wrong but no fallback voice can be found for " + originalVoice));
469 return false;
470 }
471 Voice fallbackVoice = fallbackVoices.next();
472 tts = mVoiceManager.getTTS(fallbackVoice);
473
474
475 marks.clear();
476 try {
477 pcm = speakWithVoice(sentence, fallbackVoice, tts, marks, markNames, timeout);
478 } catch (MemoryException e) {
479 flush(section, pcmOutput);
480 printMemError(sentence, e);
481 return false;
482 }
483 if (pcm == null) {
484 mTTSLog.getWritableEntry(sentence.getID()).addError(
485 new TTSLog.Error(
486 TTSLog.ErrorCode.AUDIO_MISSING,
487 "something went wrong with " + originalVoice
488 + " and fallback voice " + fallbackVoice + " didn't work either"));
489 return false;
490 }
491
492 mLogger.info("something went wrong with " + originalVoice + ". Voice " + fallbackVoice
493 + " used instead to synthesize sentence");
494
495 if (mLastFormat != null && !pcm.iterator().next().getFormat().matches(mLastFormat))
496 flush(section, pcmOutput);
497 }
498 mLastFormat = pcm.iterator().next().getFormat();
499
500 int begin = mOffsetInFile;
501 addAudio(pcm);
502
503
504 if (marks.size() == 0) {
505 mLinksOfCurrentFile.add(
506 new SoundFileLink(
507 sentence.getBaseURI().resolve("#" + sentence.getID()),
508 AudioUtils.getDuration(mLastFormat, begin),
509 AudioUtils.getDuration(mLastFormat, mOffsetInFile)));
510 } else {
511 Map<String, Integer> starts = new HashMap<String, Integer>();
512 Map<String, Integer> ends = new HashMap<String, Integer>();
513 Set<String> all = new HashSet<String>();
514
515 for (Mark m : marks) {
516 String[] mark = m.name.split(FormatSpecifications.MarkDelimiter, -1);
517 if (!mark[0].isEmpty()) {
518 ends.put(mark[0], m.offsetInAudio);
519 all.add(mark[0]);
520 }
521 if (!mark[1].isEmpty()) {
522 starts.put(mark[1], m.offsetInAudio);
523 all.add(mark[1]);
524 }
525 }
526 for (String id : all) {
527 mLinksOfCurrentFile.add(
528 new SoundFileLink(
529 sentence.getBaseURI().resolve("#" + id),
530 AudioUtils.getDuration(mLastFormat,
531 starts.containsKey(id)
532 ? begin + starts.get(id)
533 : begin),
534 AudioUtils.getDuration(mLastFormat,
535 ends.containsKey(id)
536 ? begin + ends.get(id)
537 : mOffsetInFile)));
538 }
539
540
541
542
543
544
545
546 }
547
548 if (mMemFootprint > maxQueueEltSize) {
549
550
551
552
553
554 flush(section, pcmOutput);
555 }
556 return true;
557 }
558
559 private void printMemError(Sentence sentence, MemoryException e) {
560 String msg = "out of memory";
561 mTTSLog.getWritableEntry(sentence.getID()).addError(
562 new TTSLog.Error(ErrorCode.AUDIO_MISSING, msg));
563 }
564
565 private void addAudio(Iterable<AudioInputStream> toadd) {
566 for (AudioInputStream b : toadd) {
567 int size = AudioFootprintMonitor.getFootprint(b);
568 mOffsetInFile += size;
569 mMemFootprint += size;
570 }
571 mAudioOfCurrentFile = Iterables.concat(mAudioOfCurrentFile, toadd);
572 }
573 }