UMR2AMR.py

#!/usr/bin/env python3

# This library is under the 3-Clause BSD License
#
# Copyright (c) 2025,  Orange
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#    * Redistributions of source code must retain the above copyright
#      notice, this list of conditions and the following disclaimer.
#
#    * Redistributions in binary form must reproduce the above copyright
#      notice, this list of conditions and the following disclaimer in the
#      documentation and/or other materials provided with the distribution.
#
#    * Neither the name of Orange nor the
#      names of its contributors may be used to endorse or promote products
#      derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
# DISCLAIMED. IN NO EVENT SHALL ORANGE BE LIABLE FOR ANY
# DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
# ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#
# SPDX-License-Identifier: BSD-3-Clause
# Software Name: MetAMoRphosED AMR-Editor
# Author: Johannes Heinecke

import os
import re
import sys

import penman
#from penman.models.noop import NoOpModel
import metamorphosed.umrdoc as umrdoc

# undo silencing of umrdoc
# import logging
# logging.getLogger('penman').setLevel(logging.INFO)


class UMR2AMR:
    def __init__(self, infile, outfile, first=0, last=0, filterid=None, doalignments=True, adddoclevel=None, prefix=None, gen=True):
        udoc = umrdoc.UMRdoc(infile)

        if outfile:
            ofp = open(outfile, "w")
        else:
            ofp = sys.stdout

        self.varname = re.compile(r"s(\d+)[a-z]\d?")
        if gen:
            print("# generated by:", " ".join(sys.argv), file=ofp)
        for ct, sent in enumerate(udoc.sentences, 1):
            if ct < first:
                continue
            if last != 0 and ct > last:
                continue

            if len(sent.amr.split("\n")) == 1 and "umr-empty" in sent.amr:
                print("Ignore empty sentence", sent.id, file=sys.stderr)
                continue

            try:
                pg = penman.decode(sent.amr)
                # we need this since penman does not stop at warnings:
                # things like
                #  (t / 12
                #       :00
                #        :name (n / name
                #                :op1 "Noon")))
                # output a "missing target" in the logger, but adds a triple with a None object
                for s, p, o in pg.triples:
                    if o is None:
                        raise Exception("bad triple <%s %s %s>" % (s, p, o))
            except Exception as e:
                print("ERROR: Invalid penman format in %s: %s" % (sent.id, e), file=sys.stderr)
                continue

            if "sent_id" in sent.meta:
                if filterid:
                    if not filterid.match(sent.meta["sent_id"]):
                        continue
                if prefix:
                    print("# ::id", prefix + sent.meta["sent_id"], file=ofp)
                else:
                    print("# ::id", sent.meta["sent_id"], file=ofp)
            else:
                if filterid:
                    if not filterid.match(sent.id):
                        continue
                if prefix:
                    print("# ::id", prefix + sent.id, file=ofp)
                else:
                    print("# ::id", sent.id, file=ofp)
            if sent.text:
                print("# ::snt", sent.text, file=ofp)
            elif sent.words:
                print("# ::snt", " ".join(sent.words), file=ofp)
            if sent.words:
                print("# ::tok", " ".join(sent.words), file=ofp)
            else:
                print("ERROR: No sentence in " + sent.id, file=sys.stderr)

            #newtriples = []
            # pg = penman.decode(sent.amr)
            newtriples = []
            for s, p, o in pg.triples:
                if p.startswith(":refer-"):
                    p = p[7:]
                if p != ":instance" and o not in pg.variables() and o[0] != '"':
                    o = '"%s"' % o
                newtriples.append((s, p, o))

            if adddoclevel:
                # not adding (yet) coreferences
                newvars = {} # concept: var
                #print(sent.docgraph.docgraph)
                if sent.docgraph:
                    if "temp" in adddoclevel and "temporal" in sent.docgraph.docgraph:
                        for s, p, o in sent.docgraph.docgraph["temporal"]:
                            #print("ttt", s,p,o)
                            if not self.is_from_here(s, sent.num):
                                #print("# S NOT HERE temp", sent.id, s,p,o, file=ofp)
                                continue
                            if not self.is_from_here(o, sent.num):
                                continue

                            if o in pg.variables():
                                # no references to other sentences (yet)
                                if s not in pg.variables():
                                    if s not in newvars:
                                        newvars[s] = "vv%d" % (len(newvars) + 1)
                                        newtriples.append((newvars[s], ":instance", s))
                                    newtriples.append((newvars[s], p, o))
                                    #print("AAA", newtriples[-1])
                                else:
                                    newtriples.append((s, p, o))
                                    #print("aaa", newtriples[-1])
                    if "modal" in adddoclevel and "modal" in sent.docgraph.docgraph:
                        # does not always work, without variables of preceding sentences, the graph becomes disconnected
                        # e.g. sanapana_umr-0001.umr:snt68
                        for s, p, o in sent.docgraph.docgraph["modal"]:
                            #print("mmmm", s,p,o, file=ofp)
                            if s in ["author", "author3", "author2", "have-condition", "null-conceiver", "purpose", "root", "have-condition-91", "have-purpose-91", "have-concession-91", "have-concessive-condition-91"]:
                                if s not in newvars:
                                    newvars[s] = "vv%d" % (len(newvars) + 1)
                                    newtriples.append((newvars[s], ":instance", s))
                                #newtriples.append((newvars[s], p, o))

                                if s == "root":
                                    newtriples.append((pg.top, ":root", newvars[s]))
                                s = newvars[s]
                            elif not self.is_from_here(s, sent.num):
                                print("# S NOT HERE temp", sent.id, s,p,o, file=ofp)
                                continue

                            if o in ["author", "have-condition", "null-conceiver", "purpose", "have-condition-91", "have-purpose-91"]:
                                if o not in newvars:
                                    newvars[o] = "vv%d" % (len(newvars) + 1)
                                    newtriples.append((newvars[o], ":instance", o))
                                o = newvars[o]
                            elif not self.is_from_here(o, sent.num):
                                print("# O NOT HERE temp", sent.id, s,p,o, file=ofp)
                                continue

                            newtriples.append((s, p, o))
                            #if o in pg.variables():
                            #    if s not in pg.variables():
                            #        if s not in newvars:
                            #            newvars[s] = "vv%d" % (len(newvars) + 1)
                            #            newtriples.append((newvars[s], ":instance", s))
                            #        print("ZZZZ", newtriples[-1], file=ofp)
                            #        newtriples.append((newvars[s], p, o))
            #for x in newtriples: print("TR", x, file=ofp)
            pg = penman.Graph(newtriples)

            if doalignments:
                alignment_indices = {} # triple: [index]
                ralignment_indices = {} # triple: [index]
                for tr in pg.instances():
                    if tr[0] in sent.alignments:
                        for al in sorted(sent.alignments[tr[0]]):
                            if al[0] > 0:
                                #pg.epidata[(tr[0], tr[1], tr[2])].append(penman.surface.Alignment((al[0],), "e."))
                                if (tr[0], tr[1], tr[2]) not in alignment_indices:
                                    alignment_indices[(tr[0], tr[1], tr[2])] = set()
                                alignment_indices[(tr[0], tr[1], tr[2])].add(al[0] - 1)
                                if al[1] != al[0]:
                                    alignment_indices[(tr[0], tr[1], tr[2])].add(al[1] - 1)
                for tr in pg.edges():
                    key = "%s#%s#%s#RA" % (tr[0], tr[2], tr[1][1:])
                    if key in sent.ralignments:
                        for al in sent.ralignments[key]:
                            if al[0] > 0:
                                #pg.epidata[(tr[0], tr[1], tr[2])].append(penman.surface.RoleAlignment((al[0],), "e."))
                                if (tr[0], tr[1], tr[2]) not in ralignment_indices:
                                    ralignment_indices[(tr[0], tr[1], tr[2])] = set()
                                ralignment_indices[(tr[0], tr[1], tr[2])].add(al[0] - 1)

                for tr in pg.attributes():
                    key = "%s#%s#%s#LA" % (tr[0], tr[1][1:], tr[2])
                    if key in sent.lalignments:
                        for al in sent.lalignments[key]:
                            if al[0] > 0:
                                #pg.epidata[(tr[0], tr[1], tr[2])].append(penman.surface.Alignment((al[0],), "e."))
                                if (tr[0], tr[1], tr[2]) not in alignment_indices:
                                    alignment_indices[(tr[0], tr[1], tr[2])] = set()
                                alignment_indices[(tr[0], tr[1], tr[2])].add(al[0] - 1)

                for k, v in alignment_indices.items():
                    if k not in pg.epidata:
                        pg.epidata[k] = []
                    pg.epidata[k].append(penman.surface.Alignment(sorted(v), "e."))
                for k, v in ralignment_indices.items():
                    if k not in pg.epidata:
                        pg.epidata[k] = []
                    pg.epidata[k].append(penman.surface.RoleAlignment(sorted(v), "e."))

            pt = penman.configure(pg)
            # {prefix} does not work with data from UMR since there are "concepts" with diacritics and digits in first pos
            # pt.reset_variables(fmt="{prefix}{j}") # TODO keeps accents on variable idconcepts starts with an accented letter

            pt.reset_variables(fmt="v{j}") # TODO keeps accents on variable idconcepts starts with an accented letter

            print(penman.format(pt, indent=4), file=ofp)
            ##print(penman.encode(pg, indent=4), file=ofp)
            print(file=ofp)

    def is_from_here(self, var, sentnum):
        mo = self.varname.match(var)
        if mo:
            # its a variable
            if mo.group(1) != sentnum:
                return False
        return True


if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser("UMR2AMR transformation")

    parser.add_argument("--infile", "-i", required=True, help="UMR input file to read")
    parser.add_argument("--outfile", "-o", help="AMR output file")
    parser.add_argument("--first", "-f", default=0, type=int, help="start at sentence number <first>")
    parser.add_argument("--last", "-l", default=0, type=int, help="stop after sentence number <last>")
    parser.add_argument("--filterid", "-F", help="ignore sentences which sentence id does not mutch given regex")
    parser.add_argument("--noalignments", action='store_false', default=True, help='do not output UMR aligments')
    #parser.add_argument("--adddoclevel", action='store_true', default=False, help='add document level annotation to graph (except references to other graphs)')
    parser.add_argument("--adddoclevel", nargs="*", default=None, help='current values: temp. modal. document level annotation to graph (except references to other graphs)')
    parser.add_argument("--id_prefix", help="prefix sentence ids with given string")

    if len(sys.argv) < 2:
        parser.print_help()
    else:
        args = parser.parse_args()
        a2u = UMR2AMR(args.infile, args.outfile, args.first, args.last, args.filterid, doalignments=args.noalignments, adddoclevel=args.adddoclevel, prefix=args.id_prefix)