-
Notifications
You must be signed in to change notification settings - Fork 510
/
annotation.py
executable file
·1588 lines (1316 loc) · 60.8 KB
/
annotation.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
#!/usr/bin/env python
# -*- Mode: Python; tab-width: 4; indent-tabs-mode: nil; coding: utf-8; -*-
# vim:set ft=python ts=4 sw=4 sts=4 autoindent:
from __future__ import with_statement
'''
Functionality related to the annotation file format.
Author: Pontus Stenetorp <pontus is s u-tokyo ac jp>
Version: 2011-01-25
'''
# TODO: Major re-work, cleaning up and conforming with new server paradigm
from logging import info as log_info
from codecs import open as codecs_open
from functools import partial
from itertools import chain, takewhile
from os import close as os_close, utime
from time import time
from os.path import join as path_join
from os.path import basename, splitext
from re import match as re_match
from re import compile as re_compile
from common import ProtocolError
from filelock import file_lock
from message import Messager
### Constants
# The only suffix we allow to write to, which is the joined annotation file
JOINED_ANN_FILE_SUFF = 'ann'
# These file suffixes indicate partial annotations that can not be written to
# since they depend on multiple files for completeness
PARTIAL_ANN_FILE_SUFF = ['a1', 'a2', 'co', 'rel']
KNOWN_FILE_SUFF = [JOINED_ANN_FILE_SUFF]+PARTIAL_ANN_FILE_SUFF
TEXT_FILE_SUFFIX = 'txt'
# String used to catenate texts of discontinuous annotations in reference text
DISCONT_SEP = ' '
###
# If True, use BioNLP Shared Task 2013 compatibilty mode, allowing
# normalization annotations to be parsed using the BioNLP Shared Task
# 2013 format in addition to the brat format and allowing relations to
# reference event triggers. NOTE: Alternate format supported only for
# reading normalization annotations, not for creating new ones. Don't
# change this setting unless you are sure you know what you are doing.
BIONLP_ST_2013_COMPATIBILITY = True
BIONLP_ST_2013_NORMALIZATION_RES = [
(re_compile(r'^(Reference) Annotation:(\S+) Referent:(\S+)'), r'\1 \2 \3'),
(re_compile(r'^(Reference) Referent:(\S+) Annotation:(\S+)'), r'\1 \3 \2'),
]
class AnnotationLineSyntaxError(Exception):
def __init__(self, line, line_num, filepath):
self.line = line
self.line_num = line_num
self.filepath = filepath
def __str__(self):
u'Syntax error on line %d: "%s"' % (self.line_num, self.line)
class IdedAnnotationLineSyntaxError(AnnotationLineSyntaxError):
def __init__(self, id, line, line_num, filepath):
AnnotationLineSyntaxError.__init__(self, line, line_num, filepath)
self.id = id
def __str__(self):
u'Syntax error on line %d (id %s): "%s"' % (self.line_num, self.id, self.line)
class AnnotationNotFoundError(Exception):
def __init__(self, id):
self.id = id
def __str__(self):
return u'Could not find an annotation with id: %s' % (self.id, )
class AnnotationFileNotFoundError(ProtocolError):
def __init__(self, fn):
self.fn = fn
def __str__(self):
return u'Could not find any annotations for %s' % (self.fn, )
def json(self, json_dic):
json_dic['exception'] = 'annotationFileNotFound'
return json_dic
class AnnotationCollectionNotFoundError(ProtocolError):
def __init__(self, cn):
self.cn = cn
def __str__(self):
return u'Error accessing collection %s' % (self.cn, )
def json(self, json_dic):
# TODO: more specific error?
json_dic['exception'] = 'annotationCollectionNotFound'
return json_dic
class EventWithoutTriggerError(ProtocolError):
def __init__(self, event):
self.event = event
def __str__(self):
return u'Event "%s" lacks a trigger' % (self.event, )
def json(self, json_dic):
json_dic['exception'] = 'eventWithoutTrigger'
return json_dic
class EventWithNonTriggerError(ProtocolError):
def __init__(self, event, non_trigger):
self.event = event
self.non_trigger = non_trigger
def __str__(self):
return u'Non-trigger "%s" used by "%s" as trigger' % (
self.non_trigger, self.event, )
def json(self, json_dic):
json_dic['exception'] = 'eventWithNonTrigger'
return json_dic
class TriggerReferenceError(ProtocolError):
def __init__(self, trigger, referencer):
self.trigger = trigger
self.referencer = referencer
def __str__(self):
return u'Trigger "%s" referenced by non-event "%s"' % (self.trigger,
self.referencer, )
def json(self, json_dic):
json_dic['exception'] = 'triggerReference'
return json_dic
class AnnotationTextFileNotFoundError(AnnotationFileNotFoundError):
def __str__(self):
return u'Could not read text file for %s' % (self.fn, )
class AnnotationsIsReadOnlyError(ProtocolError):
def __init__(self, fn):
self.fn = fn
def __str__(self):
# No extra message; the client is doing a fine job of reporting this
#return u'Annotations read-only for %s' % (self.fn, )
return ''
def json(self, json_dic):
json_dic['exception'] = 'annotationIsReadOnly'
return json_dic
class DuplicateAnnotationIdError(AnnotationLineSyntaxError):
def __init__(self, id, line, line_num, filepath):
AnnotationLineSyntaxError.__init__(self, line, line_num, filepath)
self.id = id
def __str__(self):
return (u'Duplicate id: %s on line %d (id %s): "%s"'
) % (self.id, self.line_num, self.id, self.line)
class InvalidIdError(Exception):
def __init__(self, id):
self.id = id
def __str__(self):
return u'Invalid id: %s' % (self.id, )
class DependingAnnotationDeleteError(Exception):
def __init__(self, target, dependants):
self.target = target
self.dependants = dependants
def __str__(self):
return u'%s can not be deleted due to depending annotations %s' % (
unicode(self.target).rstrip(), ",".join([unicode(d).rstrip() for d in self.dependants]))
def html_error_str(self, response=None):
return u'''Annotation:
%s
Has depending annotations attached to it:
%s''' % (unicode(self.target).rstrip(), ",".join([unicode(d).rstrip() for d in self.dependants]))
class SpanOffsetOverlapError(ProtocolError):
def __init__(self, offsets):
self.offsets = offsets
def __str__(self):
return u'The offsets [%s] overlap' % (', '.join(unicode(e)
for e in self.offsets, ))
def json(self, json_dic):
json_dic['exception'] = 'spanOffsetOverlapError'
return json_dic
# Open function that enforces strict, utf-8, and universal newlines for reading
# TODO: Could have another wrapping layer raising an appropriate ProtocolError
def open_textfile(filename, mode='rU'):
# enforce universal newline support ('U') in read modes
if len(mode) != 0 and mode[0] == 'r' and 'U' not in mode:
mode = mode + 'U'
return codecs_open(filename, mode, encoding='utf8', errors='strict')
def __split_annotation_id(id):
m = re_match(r'^([A-Za-z]+|#[A-Za-z]*)([0-9]+)(.*?)$', id)
if m is None:
raise InvalidIdError(id)
pre, num_str, suf = m.groups()
return pre, num_str, suf
def annotation_id_prefix(id):
pre = ''.join(c for c in takewhile(lambda x : not x.isdigit(), id))
if not pre:
raise InvalidIdError(id)
return pre
def annotation_id_number(id):
return __split_annotation_id(id)[1]
def is_valid_id(id):
# special case: '*' is acceptable as an "ID"
if id == '*':
return True
try:
# currently accepting any ID that can be split.
# TODO: consider further constraints
__split_annotation_id(id)[1]
return True
except InvalidIdError:
return False
class Annotations(object):
"""
Basic annotation storage. Not concerned with conformity of
annotations to text; can be created without access to the
text file to which the annotations apply.
"""
def get_document(self):
return self._document
def _select_input_files(self, document):
"""
Given a document name (path), returns a list of the names of
specific annotation files relevant to the document, or the
empty list if none found. For example, given "1000", may
return ["1000.a1", "1000.a2"]. May set self._read_only flag to
True.
"""
from os.path import isfile
from os import access, W_OK
try:
# Do we have a valid suffix? If so, it is probably best to the file
suff = document[document.rindex('.') + 1:]
if suff == JOINED_ANN_FILE_SUFF:
# It is a joined file, let's load it
input_files = [document]
# Do we lack write permissions?
if not access(document, W_OK):
#TODO: Should raise an exception or warning
self._read_only = True
elif suff in PARTIAL_ANN_FILE_SUFF:
# It is only a partial annotation, we will most likely fail
# but we will try opening it
input_files = [document]
self._read_only = True
else:
input_files = []
except ValueError:
# The document lacked a suffix
input_files = []
if not input_files:
# Our first attempts at finding the input by checking suffixes
# failed, so we try to attach know suffixes to the path.
sugg_path = document + '.' + JOINED_ANN_FILE_SUFF
if isfile(sugg_path):
# We found a joined file by adding the joined suffix
input_files = [sugg_path]
# Do we lack write permissions?
if not access(sugg_path, W_OK):
#TODO: Should raise an exception or warning
self._read_only = True
else:
# Our last shot, we go for as many partial files as possible
input_files = [sugg_path for sugg_path in
(document + '.' + suff
for suff in PARTIAL_ANN_FILE_SUFF)
if isfile(sugg_path)]
self._read_only = True
return input_files
#TODO: DOC!
def __init__(self, document, read_only=False):
# this decides which parsing function is invoked by annotation
# ID prefix (first letter)
self._parse_function_by_id_prefix = {
'T': self._parse_textbound_annotation,
'M': self._parse_modifier_annotation,
'A': self._parse_attribute_annotation,
'N': self._parse_normalization_annotation,
'R': self._parse_relation_annotation,
'*': self._parse_equiv_annotation,
'E': self._parse_event_annotation,
'#': self._parse_comment_annotation,
}
#TODO: DOC!
#TODO: Incorparate file locking! Is the destructor called upon inter crash?
from collections import defaultdict
from os.path import basename, getmtime, getctime
#from fileinput import FileInput, hook_encoded
# we should remember this
self._document = document
self.failed_lines = []
self.externally_referenced_triggers = set()
### Here be dragons, these objects need constant updating and syncing
# Annotation for each line of the file
self._lines = []
# Mapping between annotation objects and which line they occur on
# Range: [0, inf.) unlike [1, inf.) which is common for files
self._line_by_ann = {}
# Maximum id number used for each id prefix, to speed up id generation
#XXX: This is effectively broken by the introduction of id suffixes
self._max_id_num_by_prefix = defaultdict(lambda : 1)
# Annotation by id, not includid non-ided annotations
self._ann_by_id = {}
###
## We use some heuristics to find the appropriate annotation files
self._read_only = read_only
input_files = self._select_input_files(document)
if not input_files:
with open('{}.{}'.format(document, JOINED_ANN_FILE_SUFF), 'w'):
pass
input_files = self._select_input_files(document)
if not input_files:
raise AnnotationFileNotFoundError(document)
# We then try to open the files we got using the heuristics
#self._file_input = FileInput(openhook=hook_encoded('utf-8'))
self._input_files = input_files
# Finally, parse the given annotation file
try:
self._parse_ann_file()
# Sanity checking that can only be done post-parse
self._sanity()
except UnicodeDecodeError:
Messager.error('Encoding error reading annotation file: '
'nonstandard encoding or binary?', -1)
# TODO: more specific exception
raise AnnotationFileNotFoundError(document)
#XXX: Hack to get the timestamps after parsing
if (len(self._input_files) == 1 and
self._input_files[0].endswith(JOINED_ANN_FILE_SUFF)):
self.ann_mtime = getmtime(self._input_files[0])
self.ann_ctime = getctime(self._input_files[0])
else:
# We don't have a single file, just set to epoch for now
self.ann_mtime = -1
self.ann_ctime = -1
def _sanity(self):
# Beware, we ONLY do format checking, leave your semantics hat at home
# Check that referenced IDs are defined
for ann in self:
for rid in chain(*ann.get_deps()):
try:
self.get_ann_by_id(rid)
except AnnotationNotFoundError:
# TODO: do more than just send a message for this error?
Messager.error('ID '+rid+' not defined, referenced from annotation '+str(ann))
# Check that each event has a trigger
for e_ann in self.get_events():
try:
tr_ann = self.get_ann_by_id(e_ann.trigger)
# If the annotation is not text-bound or of different type
if (not isinstance(tr_ann, TextBoundAnnotation) or
tr_ann.type != e_ann.type):
raise EventWithNonTriggerError(e_ann, tr_ann)
except AnnotationNotFoundError:
raise EventWithoutTriggerError(e_ann)
# Check that every trigger is only referenced by events
# Create a map for non-event references
referenced_to_referencer = {}
for non_e_ann in (a for a in self
if not isinstance(a, EventAnnotation)
and isinstance(a, IdedAnnotation)):
for ref in chain(*non_e_ann.get_deps()):
try:
referenced_to_referencer[ref].add(non_e_ann.id)
except KeyError:
referenced_to_referencer[ref] = set((non_e_ann.id, ))
# Ensure that no non-event references a trigger
for tr_ann in self.get_triggers():
if tr_ann.id in referenced_to_referencer:
conflict_ann_ids = referenced_to_referencer[tr_ann.id]
if BIONLP_ST_2013_COMPATIBILITY:
# Special-case processing for BioNLP ST 2013: allow
# Relations to reference event triggers (#926).
remaining_confict_ann_ids = set()
for rid in conflict_ann_ids:
referencer = self.get_ann_by_id(rid)
if not isinstance(referencer, BinaryRelationAnnotation):
remaining_confict_ann_ids.add(rid)
else:
self.externally_referenced_triggers.add(tr_ann.id)
conflict_ann_ids = remaining_confict_ann_ids
# Note: Only reporting one of the conflicts (TODO)
if conflict_ann_ids:
referencer = self.get_ann_by_id(list(conflict_ann_ids)[0])
raise TriggerReferenceError(tr_ann, referencer)
def get_events(self):
return (a for a in self if isinstance(a, EventAnnotation))
def get_attributes(self):
return (a for a in self if isinstance(a, AttributeAnnotation))
def get_equivs(self):
return (a for a in self if isinstance(a, EquivAnnotation))
def get_textbounds(self):
return (a for a in self if isinstance(a, TextBoundAnnotation))
def get_relations(self):
return (a for a in self if isinstance(a, BinaryRelationAnnotation))
def get_normalizations(self):
return (a for a in self if isinstance(a, NormalizationAnnotation))
def get_entities(self):
# Entities are textbounds that are not triggers
triggers = [t for t in self.get_triggers()]
return (a for a in self if (isinstance(a, TextBoundAnnotation) and
not a in triggers))
def get_oneline_comments(self):
#XXX: The status exception is for the document status protocol
# which is yet to be formalised
return (a for a in self if isinstance(a, OnelineCommentAnnotation)
and a.type != 'STATUS')
def get_statuses(self):
return (a for a in self if isinstance(a, OnelineCommentAnnotation)
and a.type == 'STATUS')
def get_triggers(self):
# Triggers are text-bounds referenced by events
# TODO: this omits entity triggers that lack a referencing event
# (for one reason or another -- brat shouldn't define any.)
return (self.get_ann_by_id(e.trigger) for e in self.get_events())
# TODO: getters for other categories of annotations
#TODO: Remove read and use an internal and external version instead
def add_annotation(self, ann, read=False):
#log_info(u'Will add: ' + unicode(ann).rstrip('\n') + ' ' + unicode(type(ann)))
#TODO: DOC!
#TODO: Check read only
if not read and self._read_only:
raise AnnotationsIsReadOnlyError(self.get_document())
# Equivs have to be merged with other equivs
try:
# Bail as soon as possible for non-equivs
ann.entities # TODO: what is this?
merge_cand = ann
for eq_ann in self.get_equivs():
try:
# Make sure that this Equiv duck quacks
eq_ann.entities
except AttributeError, e:
assert False, 'got a non-entity from an entity call'
# Do we have an entitiy in common with this equiv?
for ent in merge_cand.entities:
if ent in eq_ann.entities:
for m_ent in merge_cand.entities:
if m_ent not in eq_ann.entities:
eq_ann.entities.append(m_ent)
# Don't try to delete ann since it never was added
if merge_cand != ann:
try:
self.del_annotation(merge_cand)
except DependingAnnotationDeleteError:
assert False, ('Equivs lack ids and should '
'never have dependent annotations')
merge_cand = eq_ann
# We already merged it all, break to the next ann
break
if merge_cand != ann:
# The proposed annotation was simply merged, no need to add it
# Update the modification time
from time import time
self.ann_mtime = time()
return
except AttributeError:
#XXX: This can catch a ton more than we want to! Ugly!
# It was not an Equiv, skip along
pass
# Register the object id
try:
self._ann_by_id[ann.id] = ann
pre, num = annotation_id_prefix(ann.id), annotation_id_number(ann.id)
self._max_id_num_by_prefix[pre] = max(num, self._max_id_num_by_prefix[pre])
except AttributeError:
# The annotation simply lacked an id which is fine
pass
# Add the annotation as the last line
self._lines.append(ann)
self._line_by_ann[ann] = len(self) - 1
# Update the modification time
from time import time
self.ann_mtime = time()
def del_annotation(self, ann, tracker=None):
#TODO: Check read only
#TODO: Flag to allow recursion
#TODO: Sampo wants to allow delet of direct deps but not indirect, one step
#TODO: needed to pass tracker to track recursive mods, but use is too
# invasive (direct modification of ModificationTracker.deleted)
#TODO: DOC!
if self._read_only:
raise AnnotationsIsReadOnlyError(self.get_document())
try:
ann.id
except AttributeError:
# If it doesn't have an id, nothing can depend on it
if tracker is not None:
tracker.deletion(ann)
self._atomic_del_annotation(ann)
# Update the modification time
from time import time
self.ann_mtime = time()
return
# collect annotations dependending on ann
ann_deps = []
for other_ann in self:
soft_deps, hard_deps = other_ann.get_deps()
if unicode(ann.id) in soft_deps | hard_deps:
ann_deps.append(other_ann)
# If all depending are AttributeAnnotations or EquivAnnotations,
# delete all modifiers recursively (without confirmation) and remove
# the annotation id from the equivs (and remove the equiv if there is
# only one id left in the equiv)
# Note: this assumes AttributeAnnotations cannot have
# other dependencies depending on them, nor can EquivAnnotations
if all((False for d in ann_deps if (
not isinstance(d, AttributeAnnotation)
and not isinstance(d, EquivAnnotation)
and not isinstance(d, OnelineCommentAnnotation)
and not isinstance(d, NormalizationAnnotation)
))):
for d in ann_deps:
if isinstance(d, AttributeAnnotation):
if tracker is not None:
tracker.deletion(d)
self._atomic_del_annotation(d)
elif isinstance(d, EquivAnnotation):
if len(d.entities) <= 2:
# An equiv has to have more than one member
self._atomic_del_annotation(d)
if tracker is not None:
tracker.deletion(d)
else:
if tracker is not None:
before = unicode(d)
d.entities.remove(unicode(ann.id))
if tracker is not None:
tracker.change(before, d)
elif isinstance(d, OnelineCommentAnnotation):
#TODO: Can't anything refer to comments?
self._atomic_del_annotation(d)
if tracker is not None:
tracker.deletion(d)
elif isinstance(d, NormalizationAnnotation):
# Nothing should be able to reference normalizations
self._atomic_del_annotation(d)
if tracker is not None:
tracker.deletion(d)
else:
# all types we allow to be deleted along with
# annotations they depend on should have been
# covered above.
assert False, "INTERNAL ERROR"
ann_deps = []
if ann_deps:
raise DependingAnnotationDeleteError(ann, ann_deps)
if tracker is not None:
tracker.deletion(ann)
self._atomic_del_annotation(ann)
def _atomic_del_annotation(self, ann):
#TODO: DOC
# Erase the ann by id shorthand
try:
del self._ann_by_id[ann.id]
except AttributeError:
# So, we did not have id to erase in the first place
pass
ann_line = self._line_by_ann[ann]
# Erase the main annotation
del self._lines[ann_line]
# Erase the ann by line shorthand
del self._line_by_ann[ann]
# Update the line shorthand of every annotation after this one
# to reflect the new self._lines
for l_num in xrange(ann_line, len(self)):
self._line_by_ann[self[l_num]] = l_num
# Update the modification time
from time import time
self.ann_mtime = time()
def get_ann_by_id(self, id):
#TODO: DOC
try:
return self._ann_by_id[id]
except KeyError:
raise AnnotationNotFoundError(id)
def get_new_id(self, prefix, suffix=None):
'''
Return a new valid unique id for this annotation file for the given
prefix. No ids are re-used for traceability over time for annotations,
but this only holds for the lifetime of the annotation object. If the
annotation file is parsed once again into an annotation object the
next assigned id will be the maximum seen for a given prefix plus one
which could have been deleted during a previous annotation session.
Warning: get_new_id('T') == get_new_id('T')
Just calling this method does not reserve the id, you need to
add the annotation with the returned id to the annotation object in
order to reserve it.
Argument(s):
id_pre - an annotation prefix on the format [A-Za-z]+
Returns:
An id that is guaranteed to be unique for the lifetime of the
annotation.
'''
#XXX: We have changed this one radically!
#XXX: Stupid and linear
if suffix is None:
suffix = ''
#XXX: Arbitrary constant!
for suggestion in (prefix + unicode(i) + suffix for i in xrange(1, 2**15)):
# This is getting more complicated by the minute, two checks since
# the developers no longer know when it is an id or string.
if suggestion not in self._ann_by_id:
return suggestion
# XXX: This syntax is subject to change
def _parse_attribute_annotation(self, id, data, data_tail, input_file_path):
match = re_match(r'(.+?) (.+?) (.+?)$', data)
if match is None:
# Is it an old format without value?
match = re_match(r'(.+?) (.+?)$', data)
if match is None:
raise IdedAnnotationLineSyntaxError(id, self.ann_line,
self.ann_line_num + 1, input_file_path)
_type, target = match.groups()
value = True
else:
_type, target, value = match.groups()
# Verify that the ID is indeed valid
try:
annotation_id_number(target)
except InvalidIdError:
raise IdedAnnotationLineSyntaxError(id, self.ann_line,
self.ann_line_num + 1, input_file_path)
return AttributeAnnotation(target, id, _type, '', value, source_id=input_file_path)
def _parse_event_annotation(self, id, data, data_tail, input_file_path):
#XXX: A bit nasty, we require a single space
try:
type_delim = data.index(' ')
type_trigger, type_trigger_tail = (data[:type_delim], data[type_delim:])
except ValueError:
type_trigger = data.rstrip('\r\n')
type_trigger_tail = None
try:
type, trigger = type_trigger.split(':')
except ValueError:
# TODO: consider accepting events without triggers, e.g.
# BioNLP ST 2011 Bacteria task
raise IdedAnnotationLineSyntaxError(id, self.ann_line, self.ann_line_num+1, input_file_path)
if type_trigger_tail is not None:
args = [tuple(arg.split(':')) for arg in type_trigger_tail.split()]
else:
args = []
return EventAnnotation(trigger, args, id, type, data_tail, source_id=input_file_path)
def _parse_relation_annotation(self, id, data, data_tail, input_file_path):
try:
type_delim = data.index(' ')
type, type_tail = (data[:type_delim], data[type_delim:])
except ValueError:
# cannot have a relation with just a type (contra event)
raise IdedAnnotationLineSyntaxError(id, self.ann_line, self.ann_line_num+1, input_file_path)
try:
args = [tuple(arg.split(':')) for arg in type_tail.split()]
except ValueError:
raise IdedAnnotationLineSyntaxError(id, self.ann_line, self.ann_line_num+1, input_file_path)
if len(args) != 2:
Messager.error('Error parsing relation: must have exactly two arguments')
raise IdedAnnotationLineSyntaxError(id, self.ann_line, self.ann_line_num+1, input_file_path)
if args[0][0] == args[1][0]:
Messager.error('Error parsing relation: arguments must not be identical')
raise IdedAnnotationLineSyntaxError(id, self.ann_line, self.ann_line_num+1, input_file_path)
return BinaryRelationAnnotation(id, type,
args[0][0], args[0][1],
args[1][0], args[1][1],
data_tail, source_id=input_file_path)
def _parse_equiv_annotation(self, dummy, data, data_tail, input_file_path):
# NOTE: first dummy argument to have a uniform signature with other
# parse_* functions
# TODO: this will split on any space, which is likely not correct
try:
type, type_tail = data.split(None, 1)
except ValueError:
# no space: Equiv without arguments?
raise AnnotationLineSyntaxError(self.ann_line, self.ann_line_num+1, input_file_path)
equivs = type_tail.split(None)
return EquivAnnotation(type, equivs, data_tail, source_id=input_file_path)
# Parse an old modifier annotation for backwards compatibility
def _parse_modifier_annotation(self, id, data, data_tail, input_file_path):
type, target = data.split()
return AttributeAnnotation(target, id, type, data_tail, True, source_id=input_file_path)
def _split_textbound_data(self, id, data, input_file_path):
try:
# first space-separated string is type
type, rest = data.split(' ', 1)
# rest should be semicolon-separated list of "START END"
# pairs, where START and END are integers
spans = []
for span_str in rest.split(';'):
start_str, end_str = span_str.split(' ', 2)
# ignore trailing whitespace
end_str = end_str.rstrip()
if any((c.isspace() for c in end_str)):
Messager.error('Error parsing textbound "%s\t%s". (Using space instead of tab?)' % (id, data))
raise IdedAnnotationLineSyntaxError(id, self.ann_line, self.ann_line_num+1, input_file_path)
start, end = (int(start_str), int(end_str))
spans.append((start, end))
except:
raise IdedAnnotationLineSyntaxError(id, self.ann_line, self.ann_line_num+1, input_file_path)
return type, spans
def _parse_textbound_annotation(self, _id, data, data_tail, input_file_path):
_type, spans = self._split_textbound_data(_id, data, input_file_path)
return TextBoundAnnotation(spans, _id, _type, data_tail, source_id=input_file_path)
def _parse_normalization_annotation(self, _id, data, data_tail, input_file_path):
# special-case processing for BioNLP ST 2013 variant of
# normalization format
if BIONLP_ST_2013_COMPATIBILITY:
for r, s in BIONLP_ST_2013_NORMALIZATION_RES:
d = r.sub(s, data, count=1)
if d != data:
data = d
break
match = re_match(r'(\S+) (\S+) (\S+?):(\S+)', data)
if match is None:
raise IdedAnnotationLineSyntaxError(_id, self.ann_line, self.ann_line_num + 1, input_file_path)
_type, target, refdb, refid = match.groups()
return NormalizationAnnotation(_id, _type, target, refdb, refid, data_tail, source_id=input_file_path)
def _parse_comment_annotation(self, _id, data, data_tail, input_file_path):
try:
_type, target = data.split()
except ValueError:
raise IdedAnnotationLineSyntaxError(_id, self.ann_line, self.ann_line_num+1, input_file_path)
return OnelineCommentAnnotation(target, _id, _type, data_tail, source_id=input_file_path)
def _parse_ann_file(self):
self.ann_line_num = -1
for input_file_path in self._input_files:
with open_textfile(input_file_path) as input_file:
#for self.ann_line_num, self.ann_line in enumerate(self._file_input):
for self.ann_line in input_file:
self.ann_line_num += 1
try:
# ID processing
try:
id, id_tail = self.ann_line.split('\t', 1)
except ValueError:
raise AnnotationLineSyntaxError(self.ann_line, self.ann_line_num+1, input_file_path)
pre = annotation_id_prefix(id)
if id in self._ann_by_id and pre != '*':
raise DuplicateAnnotationIdError(id,
self.ann_line, self.ann_line_num+1,
input_file_path)
# if the ID is not valid, need to fail with
# AnnotationLineSyntaxError (not
# IdedAnnotationLineSyntaxError).
if not is_valid_id(id):
raise AnnotationLineSyntaxError(self.ann_line, self.ann_line_num+1, input_file_path)
# Cases for lines
try:
data_delim = id_tail.index('\t')
data, data_tail = (id_tail[:data_delim],
id_tail[data_delim:])
except ValueError:
data = id_tail
# No tail at all, although it should have a \t
data_tail = ''
new_ann = None
#log_info('Will evaluate prefix: ' + pre)
assert len(pre) >= 1, "INTERNAL ERROR"
pre_first = pre[0]
try:
parse_func = self._parse_function_by_id_prefix[pre_first]
new_ann = parse_func(id, data, data_tail, input_file_path)
except KeyError:
raise IdedAnnotationLineSyntaxError(id, self.ann_line, self.ann_line_num+1, input_file_path)
assert new_ann is not None, "INTERNAL ERROR"
self.add_annotation(new_ann, read=True)
except IdedAnnotationLineSyntaxError, e:
# Could parse an ID but not the whole line; add UnparsedIdedAnnotation
self.add_annotation(UnparsedIdedAnnotation(e.id,
e.line, source_id=e.filepath), read=True)
self.failed_lines.append(e.line_num - 1)
except AnnotationLineSyntaxError, e:
# We could not parse even an ID on the line, just add it as an unknown annotation
self.add_annotation(UnknownAnnotation(e.line,
source_id=e.filepath), read=True)
# NOTE: For access we start at line 0, not 1 as in here
self.failed_lines.append(e.line_num - 1)
def __str__(self):
s = u'\n'.join(unicode(ann).rstrip(u'\r\n') for ann in self)
if not s:
return u''
else:
return s if s[-1] == u'\n' else s + u'\n'
def __it__(self):
for ann in self._lines:
yield ann
def __getitem__(self, val):
try:
# First, try to use it as a slice object
return self._lines[val.start, val.stop, val.step]
except AttributeError:
# It appears not to be a slice object, try it as an index
return self._lines[val]
def __len__(self):
return len(self._lines)
def __enter__(self):
# No need to do any handling here, the constructor handles that
return self
def __exit__(self, type, value, traceback):
#self._file_input.close()
if not self._read_only:
assert len(self._input_files) == 1, 'more than one valid outfile'
# We are hitting the disk a lot more than we should here, what we
# should have is a modification flag in the object but we can't
# due to how we change the annotations.
out_str = unicode(self)
with open_textfile(self._input_files[0], 'r') as old_ann_file:
old_str = old_ann_file.read()
# Was it changed?
if out_str == old_str:
# Then just return
return
from config import WORK_DIR
# Protect the write so we don't corrupt the file
with file_lock(path_join(WORK_DIR,
str(hash(self._input_files[0].replace('/', '_')))
+ '.lock')
) as lock_file:
#from tempfile import NamedTemporaryFile
from tempfile import mkstemp
# TODO: XXX: Is copyfile really atomic?
from shutil import copyfile
# XXX: NamedTemporaryFile only supports encoding for Python 3
# so we hack around it.
#with NamedTemporaryFile('w', suffix='.ann') as tmp_file:
# Grab the filename, but discard the handle
tmp_fh, tmp_fname = mkstemp(suffix='.ann')
os_close(tmp_fh)
try:
with open_textfile(tmp_fname, 'w') as tmp_file:
#XXX: Temporary hack to make sure we don't write corrupted
# files, but the client will already have the version
# at this stage leading to potential problems upon
# the next change to the file.
tmp_file.write(out_str)
tmp_file.flush()
try:
with Annotations(tmp_file.name) as ann:
# Move the temporary file onto the old file
copyfile(tmp_file.name, self._input_files[0])
# As a matter of convention we adjust the modified
# time of the data dir when we write to it. This
# helps us to make back-ups
now = time()
#XXX: Disabled for now!
#utime(DATA_DIR, (now, now))
except Exception, e:
Messager.error('ERROR writing changes: generated annotations cannot be read back in!\n(This is almost certainly a system error, please contact the developers.)\n%s' % e, -1)
raise
finally:
try:
from os import remove
remove(tmp_fname)
except Exception, e: