Module tf.convert.makewatm

Classes

class MakeWATM (fileLoc)

Base class for running conversions to WATM.

This class has methods to convert corpora from TEI or PageXML to TF and then to WATM.

But if the corpus needs additional preparation, you can make a sub class based on this with additional tasks defined an implemented.

Any class M in m.py based on this class can be called from the command line as follows:

python m.py flags tasks

If you base a superclass on this, you can register the additional tasks as follows:

For each extra task xxx, write an method

doTask_xxx(self)

Then provide for each task a simple doc line and register them all by:

self.setOptions(
    taskSpecs=(
        (task1, docLine1),
        (task2, docLine2),
        ...
    ),
    flagSpecs=(
        (flag1, docLine1),
        (flag2, docLine2),
        ...
    ),
)

Localize upon creation.

When an object of this class is initialized, we assume that the script doing it is localized in the programs directory in a corpus repo.

Parameters

fileLoc : string

The full path of the file that creates an instance of this class.

Expand source code Browse git

class MakeWATM:
    """Base class for running conversions to WATM.

    This class has methods to convert corpora from TEI or PageXML to TF and then
    to WATM.

    But if the corpus needs additional preparation, you can make a sub class
    based on this with additional tasks defined an implemented.

    Any class M in m.py based on this class can be called from the command line
    as follows:

        python m.py flags tasks

    If you base a superclass on this, you can register the additional tasks as follows:

    For each extra task xxx, write an method

        doTask_xxx(self)

    Then provide for each task a simple doc line and register them all by:

        self.setOptions(
            taskSpecs=(
                (task1, docLine1),
                (task2, docLine2),
                ...
            ),
            flagSpecs=(
                (flag1, docLine1),
                (flag2, docLine2),
                ...
            ),
        )
    """

    def __init__(self, fileLoc):
        """Localize upon creation.

        When an object of this class is initialized, we assume that the script
        doing it is localized in the `programs` directory in a corpus repo.

        Parameters
        ----------
        fileLoc: string
            The full path of the file that creates an instance of this class.
        """

        self.BASE_HELP = dedent(
            """
        Convert corpus data to WATM.

        USAGE

        python make.py tasks

        Converts data from TEI files

        ARGS

        tfversion
            Any arg that contains a . is considered to be the tf version number.
            If no version is passed, we resort to the hard coded default.

        all
            run all (enabled) tasks

        """
        )
        self.BASE_DOCS = dict(
            tei2tf="TEI => TF",
            page2tf="PageXML => TF",
            watm="TF => WATM",
            watms="multi TF => WATMS",
        )
        self.BASE_FLAGS = dict(
            silent="To run a bit more silent",
            relaxed="Accept XML validation errors",
            usenlp="Will run an NLP pipeline to mark tokens, sentences and entities",
            prod="Delivers WATM in production location",
            force="Force execution even if (uptodate) data is present",
        )
        self.BASE_TASKS = ("tei2tf", "page2tf", "watm", "watms")
        self.TF_VERSION = "0.0.0test"

        self.fileLoc = fileLoc

        self.good = True

        repoBase = dirNm(dirNm(abspath(fileLoc)))
        (self.backend, self.org, self.repo, self.relative) = getLocation(
            targetDir=repoBase
        )
        self.repoBase = repoBase
        self.setOptions()

    def setOptions(
        self,
        taskSpecs=(
            ("tei2tf", None),
            ("watm", None),
        ),
        flagSpecs=(
            ("silent", None),
            ("relaxed", None),
            ("usenlp", None),
            ("prod", None),
            ("force", None),
        ),
        intro=None,
    ):
        self.TASKS = []
        self.DOCS = {}
        self.FLAGS = {}

        good = True

        for name, doc in taskSpecs:
            if name not in self.BASE_TASKS:
                if doc is None:
                    console(f"task {name}: no help text given", error=True)
                    good = False

                method = f"doTask_{name}"

                if not hasattr(self, method):
                    console(
                        f"task {name}: no method {method} defined in subclass",
                        error=True,
                    )
                    good = False

            if not good:
                continue

            self.TASKS.append(name)
            self.DOCS[name] = self.BASE_DOCS[name] if doc is None else doc

        for name, doc in flagSpecs:
            if name not in self.BASE_FLAGS:
                if doc is None:
                    console(f"flag --{name}: no help text given", error=True)
                    good = False

            if not good:
                continue

            self.FLAGS[name] = doc

        self.good = good

        self.HELP = (
            (f"{intro}\n\n" if intro else "")
            + self.BASE_HELP
            + "\nFLAGS\n\n"
            + "".join(f"{name}\n\t{doc}" for (name, doc) in self.FLAGS.items())
            + "\nTASKS\n\n"
            + "".join(f"{name}\n\t{doc}" for (name, doc) in self.DOCS.items())
            + "\n\nall\n\trun all (enabled) tasks\n\n"
        )

    def main(self, cmdLine=None, cargs=sys.argv[1:]):
        FLAGS = self.FLAGS
        TASKS = self.TASKS

        if cmdLine is not None:
            cargs = cmdLine.split()

        if "--help" in cargs:
            console(self.HELP)
            return 0

        unrecognized = set()
        tasks = set()
        version = None

        for flag in FLAGS:
            setattr(self, f"flag_{flag}", False)

        for carg in cargs:
            if carg.startswith("--"):
                flag = carg[2:]

                if flag not in FLAGS:
                    unrecognized.add(carg)
                else:
                    setattr(self, f"flag_{flag}", True)
            elif carg == "all":
                for task in TASKS:
                    tasks.add(task)
            elif carg in TASKS:
                tasks.add(carg)
            elif "." in carg:
                version = carg
            else:
                unrecognized.add(carg)

        silent = self.flag_silent

        if not silent:
            console(f"Enabled tasks: {' '.join(self.TASKS)}")
        if version is None:
            console(
                f"No version for the TF data given. Using default: {self.TF_VERSION}"
            )
            version = self.TF_VERSION
        else:
            if not silent:
                console(f"Using TF version: {version}")

        self.version = version

        if len(unrecognized):
            console(self.HELP)
            console(
                f"Unrecognized arguments: {', '.join(sorted(unrecognized))}", error=True
            )
            return -1

        if len(tasks) == 0:
            console("Nothing to do")
            return 0

        self.prepareRun(tasks)

        return self.run(tasks)

    def prepareRun(self, tasks):
        return

    def run(self, tasks):
        TASKS = self.TASKS
        DOCS = self.DOCS

        for task in TASKS:
            if task not in tasks:
                continue

            console(f"{DOCS[task]} ...")

            method = getattr(self, f"doTask_{task}")
            method()

        return 0 if self.good else 1

    def doTask_tei2tf(self):
        good = self.good
        silent = self.flag_silent
        relaxed = self.flag_relaxed
        usenlp = self.flag_usenlp

        if not good:
            if not silent:
                console("Skipping 'produce TF' because of an error condition")
            return

        tfVersion = self.version
        verbose = -1 if silent else 0
        loadVerbose = DEEP if silent else TERSE

        Tei = TEI(verbose=verbose, tei=0, tf=f"{tfVersion}pre" if usenlp else tfVersion)
        self.teiVersion = Tei.teiVersion

        console("\tValidating TEI ...")

        if not Tei.task(check=True, verbose=verbose, validate=True):
            if relaxed:
                Tei.good = True
            else:
                self.good = False
                return

        console("\tConverting TEI ...")

        if not Tei.task(convert=True, verbose=verbose):
            self.good = False
            return

        console("\tLoading TF ...")

        if not Tei.task(load=True, verbose=verbose):
            self.good = False
            return

        if not silent:
            console("Set up TF-app ...")

        if not Tei.task(app=True, verbose=verbose):
            self.good = False
            return

        if usenlp:
            console("\tAdding NLP data ...")

            org = self.org
            repo = self.repo
            backend = self.backend

            Apre = use(
                f"{org}/{repo}:clone",
                backend=backend,
                checkout="clone",
                silent=loadVerbose,
            )
            NLP = NLPipeline(
                lang="it", ner=True, parser=True, verbose=verbose, write=True
            )
            NLP.loadApp(Apre)
            NLP.task(plaintext=True, lingo=True, ingest=True)

            if not NLP.good:
                self.good = False
                return

            if not silent:
                console("Set up TF-app ...")

            if not Tei.task(apptoken=True, verbose=-1):
                self.good = False
                return

            if not Tei.task(load=True, verbose=-1):
                self.good = False
                return

    def doTask_page2tf(self):
        good = self.good
        silent = self.flag_silent

        if not good:
            if not silent:
                console("Skipping 'produce TF' because of an error condition")
            return

        tfVersion = self.version
        repoBase = self.repoBase
        sourceDir = f"{repoBase}/organized/source"

        verbose = -1 if silent else 0

        P = PageXML(sourceDir, repoBase, verbose=verbose, source=0, tf=tfVersion)

        if not P.task(convert=True, verbose=verbose):
            self.good = False
            return

        console("\tLoading TF")

        if not P.task(load=True, verbose=verbose):
            self.good = False
            return

        if not silent:
            console("Set up TF-app ...")

        if not P.task(app=True, verbose=verbose):
            self.good = False
            return

        if not P.good:
            self.good = False
            return

    def doTask_watm(self):
        good = self.good
        silent = self.flag_silent
        prod = self.flag_prod

        if not good:
            if not silent:
                console("Skipping 'produce WATM' because of an error condition")
            return

        backend = self.backend
        org = self.org
        repo = self.repo

        console("\tLoading TF ...")

        loadVerbose = DEEP if silent else TERSE

        A = use(
            f"{org}/{repo}:clone", backend=backend, checkout="clone", silent=loadVerbose
        )

        console(f"\tMaking WATM for version {A.version}")

        WA = WATM(A, "tei", skipMeta=False, silent=silent, prod=prod)
        WA.makeText()
        WA.makeAnno()
        WA.writeAll()
        WA.testAll()

    def doTask_watms(self):
        good = self.good
        silent = self.flag_silent
        prod = self.flag_prod

        if not good:
            if not silent:
                console("Skipping 'produce WATM' because of an error condition")
            return

        backend = self.backend
        org = self.org
        repo = self.repo

        console("\tMaking WATMs")

        W = WATMS(org, repo, backend, "pagexml", silent=silent)
        W.produce(prod=prod)

Methods

def doTask_page2tf(self)

def doTask_tei2tf(self)

def doTask_watm(self)

def doTask_watms(self)

def main(self, cmdLine=None, cargs=['--force', '--html', '--output-dir', 'site', '--template-dir', 'tf/docs/templates', 'tf'])

def prepareRun(self, tasks)

def run(self, tasks)

def setOptions(self, taskSpecs=(('tei2tf', None), ('watm', None)), flagSpecs=(('silent', None), ('relaxed', None), ('usenlp', None), ('prod', None), ('force', None)), intro=None)