core, docutils: Revamp error reporting in the SerAPI module

* alectryon/core.py (SerAPI): Add `observer` field.
(SerAPI._warn_on_exn): Call `observer.notify()`.
(SerAPI.annotate): New function (uses `self.observer`).
* alectryon/docutils.py (AlectryonTransform.annotate_cached): Call SerAPI with
`observer` set to instance of new `DocutilsObserver` class.

Author: cpitclaudel

alectryon/cli.py

@@ -62,11 +62,12 @@ def rst_to_coq(coq, fpath, point, marker):
     return _catch_parsing_errors(fpath, rst2coq_marked, coq, point, marker)
 
 def annotate_chunks(chunks, fpath, cache_directory, cache_compression, sertop_args):
-    from .core import SerAPI, annotate
+    from .core import SerAPI
     from .json import Cache
+    api = SerAPI(sertop_args)
     metadata = {"sertop_args": sertop_args}
     cache = Cache(cache_directory, fpath, metadata, cache_compression)
-    return cache.update(chunks, lambda c: annotate(c, sertop_args), SerAPI.version_info())
+    return cache.update(chunks, api.annotate, SerAPI.version_info())
 
 def register_docutils(v, ctx):
     from . import docutils

alectryon/core.py

@@ -18,7 +18,7 @@
 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 # SOFTWARE.
 
-from typing import Any, Iterator, List, Tuple, Union
+from typing import Any, Iterator, List, Tuple, Union, NamedTuple, TypeVar, Generic
 
 from collections import namedtuple, defaultdict
 from shlex import quote
@@ -144,16 +144,24 @@ def __init__(self, _s, pos, col_offset):
         super().__init__()
         self.pos, self.col_offset = pos, col_offset
 
-class PosView(bytes):
+class View(bytes):
+    def __getitem__(self, key):
+        return memoryview(self).__getitem__(key)
+
+    def __init__(self, s):
+        super().__init__()
+        self.s = s
+
+class PosView(View):
     NL = b"\n"
 
     def __new__(cls, s):
         bs = s.encode("utf-8")
         # https://stackoverflow.com/questions/20221858/
-        return super().__new__(cls, bs) if isinstance(s, PosStr) else memoryview(bs)
+        return super().__new__(cls, bs) if isinstance(s, PosStr) else View(bs)
 
     def __init__(self, s):
-        super().__init__()
+        super().__init__(s)
         self.pos, self.col_offset = s.pos, s.col_offset
 
     def __getitem__(self, key):
@@ -183,6 +191,27 @@ def translate_span(self, beg, end):
         return Range(self.translate_offset(beg),
                      self.translate_offset(end))
 
+_T = TypeVar("_T")
+class Notification(NamedTuple, Generic[_T]): # type: ignore
+    obj: _T
+    message: str
+    location: Range
+    level: int
+
+class Observer:
+    def _notify(self, n: Notification):
+        raise NotImplementedError()
+
+    def notify(self, obj, message, location, level):
+        self._notify(Notification(obj, message, location, level))
+
+class StderrObserver(Observer):
+    def _notify(self, n: Notification):
+        header = n.location.as_header() if n.location else "!!"
+        message = n.message.rstrip().replace("\n", "\n   ")
+        level_name = {2: "WARNING", 3: "ERROR"}.get(n.level, "??")
+        stderr.write("{} ({}/{}) {}\n".format(header, level_name, n.level, message))
+
 PrettyPrinted = namedtuple("PrettyPrinted", "sid pp")
 
 def sexp_hd(sexp):
@@ -221,12 +250,13 @@ def version_info(sertop_bin=SERTOP_BIN):
     def __init__(self, args=(), # pylint: disable=dangerous-default-value
                  sertop_bin=SERTOP_BIN,
                  pp_args=DEFAULT_PP_ARGS):
-        """Configure and start a ``sertop`` instance."""
+        """Configure a ``sertop`` instance."""
         self.args, self.sertop_bin = [*args, *SerAPI.DEFAULT_ARGS], sertop_bin
         self.sertop = None
         self.next_qid = 0
         self.pp_args = {**SerAPI.DEFAULT_PP_ARGS, **pp_args}
         self.last_response = None
+        self.observer : Observer = StderrObserver()
 
     def __enter__(self):
         self.reset()
@@ -375,15 +405,7 @@ def _clip_span(loc, chunk):
     def _range_of_span(span, chunk):
         return chunk.translate_span(*span) if isinstance(chunk, PosView) else None
 
-    @staticmethod
-    def _print_warning(msg, loc):
-        header = loc.as_header() if loc else "!!"
-        msg = msg.rstrip().replace("\n", "\n   ")
-        stderr.write("{} (WARNING/2) {}\n".format(header, msg))
-        # FIXME report this error to calling context
-
-    @staticmethod
-    def _warn_on_exn(response, chunk):
+    def _warn_on_exn(self, response, chunk):
         QUOTE = '  > '
         ERR_FMT = "Coq raised an exception:\n{}"
         msg = sx.tostr(response.exn)
@@ -392,7 +414,7 @@ def _warn_on_exn(response, chunk):
         if chunk:
             err += "\n" + SerAPI._highlight_exn(span, chunk, prefix=QUOTE)
         err += "\n" + "Results past this point may be unreliable."
-        SerAPI._print_warning(err, SerAPI._range_of_span(span, chunk))
+        self.observer.notify(chunk.s, err, SerAPI._range_of_span(span, chunk), level=3)
 
     def _collect_messages(self, typs: Tuple[type, ...], chunk, sid) -> Iterator[Any]:
         warn_on_exn = ApiExn not in typs
@@ -404,7 +426,7 @@ def _collect_messages(self, typs: Tuple[type, ...], chunk, sid) -> Iterator[Any]
                     return
                 if warn_on_exn and isinstance(response, ApiExn):
                     if sid is None or response.sids is None or sid in response.sids:
-                        SerAPI._warn_on_exn(response, chunk)
+                        self._warn_on_exn(response, chunk)
                 if (not typs) or isinstance(response, typs): # type: ignore
                     yield response
 
@@ -498,11 +520,16 @@ def run(self, chunk):
             if fragment is None:
                 err = "Orphaned message for sid {}:".format(message.sid)
                 err += "\n" + indent(message.pp, " >  ")
-                SerAPI._print_warning(err, SerAPI._range_of_span((0, len(chunk)), chunk))
+                err_range = SerAPI._range_of_span((0, len(chunk)), chunk)
+                self.observer.notify(chunk.s, err, err_range, level=2)
             else:
                 fragment.messages.append(Message(message.pp))
         return fragments
 
+    def annotate(self, chunks):
+        with self as api:
+            return [api.run(chunk) for chunk in chunks]
+
 def annotate(chunks, sertop_args=()):
     r"""Annotate multiple `chunks` of Coq code.
 
@@ -514,5 +541,4 @@ def annotate(chunks, sertop_args=()):
     >>> annotate(["Check 1."])
     [[Sentence(contents='Check 1.', messages=[Message(contents='1\n     : nat')], goals=[])]]
     """
-    with SerAPI(args=sertop_args) as api:
-        return [api.run(chunk) for chunk in chunks]
+    return SerAPI(args=sertop_args).annotate(chunks)

alectryon/docutils.py

@@ -87,7 +87,7 @@
 from docutils.writers import html4css1, html5_polyglot, latex2e, xetex
 
 from . import core, transforms, html, latex, markers
-from .core import Gensym, annotate, SerAPI, Position, PosStr
+from .core import Gensym, SerAPI, Position, PosStr
 from .pygments import highlight_html, highlight_latex, added_tokens, \
     resolve_token, replace_builtin_coq_lexer
 
@@ -288,6 +288,16 @@ def apply(self, **kwargs):
         for node in self.document.traverse(self.is_math):
             node.attributes.setdefault("classes", []).append("mathjax_process")
 
+class DocutilsObserver(core.Observer):
+    def __init__(self, document):
+        self.document = document
+
+    def _notify(self, n: core.Notification):
+        beg, end = n.location.beg, n.location.end
+        self.document.reporter.system_message(
+            n.level, n.message, line=beg.line, column=beg.col,
+            end_line=end.line, end_column=end.col)
+
 class AlectryonTransform(OneTimeTransform):
     default_priority = 800
     auto_toggle = True
@@ -316,9 +326,11 @@ def check_for_long_lines(self, node, fragments):
 
     def annotate_cached(self, chunks, sertop_args):
         from .json import Cache
+        prover = SerAPI(sertop_args)
+        prover.observer = DocutilsObserver(self.document)
         metadata = {"sertop_args": sertop_args}
         cache = Cache(CACHE_DIRECTORY, self.document['source'], metadata, CACHE_COMPRESSION)
-        annotated = cache.update(chunks, lambda c: annotate(c, sertop_args), SerAPI.version_info())
+        annotated = cache.update(chunks, prover.annotate, SerAPI.version_info())
         return cache.generator, annotated
 
     def annotate(self, pending_nodes):

recipes/_output/tests/errors.txt

@@ -45,17 +45,17 @@ Missing search pattern for key ``.s`` in expression ``.s.g``. (maybe an invalid
    Check nat.
 
 tests/errors.rst:5: (ERROR/3) In `:alectryon/pygments/xyz:`: Unknown token kind: xyz
-tests/errors.rst:(109:62)-(109:64): (WARNING/2) Coq raised an exception:
-     > Syntax error: '.' expected after [vernac:syntax_command] (in [vernac_aux]).
-   The offending chunk is delimited by >>>…<<< below:
-     > Notation "'🆄🄽🅘ⓒ⒪𝖉∈' α ⊕ ⤋ β" := α >>>++<<< β. (* Bytes vs. str *)
-   Results past this point may be unreliable.
-tests/errors.rst:(114:8)-(114:12): (WARNING/2) Coq raised an exception:
-     > The term "true" has type "bool" which should be Set, Prop or Type.
-   The offending chunk is delimited by >>>…<<< below:
-     > Definition a
-     >   : >>>true<<< := 1. (* Next line *)
-   Results past this point may be unreliable.
+tests/errors.rst:109: (ERROR/3) Coq raised an exception:
+  > Syntax error: '.' expected after [vernac:syntax_command] (in [vernac_aux]).
+The offending chunk is delimited by >>>…<<< below:
+  > Notation "'🆄🄽🅘ⓒ⒪𝖉∈' α ⊕ ⤋ β" := α >>>++<<< β. (* Bytes vs. str *)
+Results past this point may be unreliable.
+tests/errors.rst:114: (ERROR/3) Coq raised an exception:
+  > The term "true" has type "bool" which should be Set, Prop or Type.
+The offending chunk is delimited by >>>…<<< below:
+  > Definition a
+  >   : >>>true<<< := 1. (* Next line *)
+Results past this point may be unreliable.
 tests/errors.rst:15: (WARNING/2) Long line (107 characters)
      exact I. (* A very long line *) (* whose contents are split *) (* across *) (* multiple *) (* comments *)
 tests/errors.rst:87: (ERROR/3) In `.. coq::`: