Merge pull request #1019 from googlefonts/protocols

Add backend protocols

src/fontra/__main__.py

@@ -4,12 +4,13 @@
 from importlib.metadata import entry_points
 
 from . import __version__ as fontraVersion
+from .core.protocols import ProjectManager, ProjectManagerFactory
 from .core.server import FontraServer, findFreeTCPPort
 
 DEFAULT_PORT = 8000
 
 
-def main():
+def main() -> None:
     logging.basicConfig(
         format="%(asctime)s %(name)-17s %(levelname)-8s %(message)s",
         level=logging.INFO,
@@ -41,15 +42,15 @@ def main():
             # See https://github.com/googlefonts/fontra/issues/141
             continue
         subParser = subParsers.add_parser(entryPoint.name)
-        pmFactory = entryPoint.load()
+        pmFactory: ProjectManagerFactory = entryPoint.load()
         pmFactory.addArguments(subParser)
         subParser.set_defaults(getProjectManager=pmFactory.getProjectManager)
 
     args = parser.parse_args()
 
     host = args.host
     httpPort = args.http_port
-    manager = args.getProjectManager(args)
+    manager: ProjectManager = args.getProjectManager(args)
     server = FontraServer(
         host=host,
         httpPort=httpPort if httpPort is not None else findFreeTCPPort(DEFAULT_PORT),

src/fontra/backends/__init__.py

@@ -1,19 +1,22 @@
 import logging
 import pathlib
 from importlib.metadata import entry_points
+from os import PathLike
+
+from ..core.protocols import ReadableFontBackend, WritableFontBackend
 
 logger = logging.getLogger(__name__)
 
 
-def getFileSystemBackend(path):
+def getFileSystemBackend(path: PathLike) -> ReadableFontBackend:
     return _getFileSystemBackend(path, False)
 
 
-def newFileSystemBackend(path):
+def newFileSystemBackend(path: PathLike) -> WritableFontBackend:
     return _getFileSystemBackend(path, True)
 
 
-def _getFileSystemBackend(path, create):
+def _getFileSystemBackend(path: PathLike, create: bool) -> WritableFontBackend:
     logVerb = "creating" if create else "loading"
 
     path = pathlib.Path(path)
@@ -34,5 +37,10 @@ def _getFileSystemBackend(path, create):
     else:
         backend = backendClass.fromPath(path)
 
+    if create:
+        assert isinstance(backend, WritableFontBackend)
+    else:
+        assert isinstance(backend, ReadableFontBackend)
+
     logger.info(f"done {logVerb} {path.name}")
     return backend

src/fontra/backends/copy.py

@@ -5,22 +5,28 @@
 import shutil
 from contextlib import closing
 
+from ..core.protocols import ReadableFontBackend, WritableFontBackend
 from . import getFileSystemBackend, newFileSystemBackend
 
 logger = logging.getLogger(__name__)
 
 
 async def copyFont(
-    sourceBackend, destBackend, *, glyphNames=None, numTasks=1, progressInterval=0
-):
+    sourceBackend: ReadableFontBackend,
+    destBackend: WritableFontBackend,
+    *,
+    glyphNames=None,
+    numTasks=1,
+    progressInterval=0,
+) -> None:
     await destBackend.putGlobalAxes(await sourceBackend.getGlobalAxes())
     await destBackend.putCustomData(await sourceBackend.getCustomData())
     glyphMap = await sourceBackend.getGlyphMap()
     glyphNamesInFont = set(glyphMap)
     glyphNamesToCopy = sorted(
         glyphNamesInFont if not glyphNames else set(glyphNames) & set(glyphMap)
     )
-    glyphNamesCopied = set()
+    glyphNamesCopied: set[str] = set()
 
     tasks = [
         asyncio.create_task(
@@ -38,21 +44,25 @@ async def copyFont(
     done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_EXCEPTION)
     for task in pending:
         task.cancel()
-    exceptions = [task.exception() for task in done if task.exception()]
+    exceptions: list[BaseException | None] = [
+        task.exception() for task in done if task.exception()
+    ]
     if exceptions:
         if len(exceptions) > 1:
             logger.error(f"Multiple exceptions were raised: {exceptions}")
-        raise exceptions[0]
+        e = exceptions[0]
+        assert e is not None
+        raise e
 
 
 async def copyGlyphs(
-    sourceBackend,
-    destBackend,
-    glyphMap,
-    glyphNamesToCopy,
-    glyphNamesCopied,
-    progressInterval,
-):
+    sourceBackend: ReadableFontBackend,
+    destBackend: WritableFontBackend,
+    glyphMap: dict[str, list[int]],
+    glyphNamesToCopy: list[str],
+    glyphNamesCopied: set[str],
+    progressInterval: int,
+) -> None:
     while glyphNamesToCopy:
         if progressInterval and not (len(glyphNamesToCopy) % progressInterval):
             logger.info(f"{len(glyphNamesToCopy)} glyphs left to copy")
@@ -73,14 +83,10 @@ async def copyGlyphs(
         }
         glyphNamesToCopy.extend(sorted(componentNames - glyphNamesCopied))
 
-        error = await destBackend.putGlyph(glyphName, glyph, glyphMap[glyphName])
-        if error:
-            # FIXME: putGlyph should always raise, and not return some error string
-            # This may be unique to the rcjk backend, though.
-            raise ValueError(error)
+        await destBackend.putGlyph(glyphName, glyph, glyphMap[glyphName])
 
 
-async def mainAsync():
+async def mainAsync() -> None:
     logging.basicConfig(
         format="%(asctime)s %(name)-17s %(levelname)-8s %(message)s",
         level=logging.INFO,
@@ -118,7 +124,7 @@ async def mainAsync():
         elif destPath.exists():
             destPath.unlink()
     elif destPath.exists():
-        raise argparse.ArgumentError("the destination file already exists")
+        raise argparse.ArgumentError(None, "the destination file already exists")
 
     sourceBackend = getFileSystemBackend(sourcePath)
     destBackend = newFileSystemBackend(destPath)

src/fontra/backends/designspace.py

@@ -10,7 +10,9 @@
 from dataclasses import asdict, dataclass
 from datetime import datetime
 from functools import cache, cached_property, singledispatch
+from os import PathLike
 from types import SimpleNamespace
+from typing import Any, AsyncGenerator, Callable
 
 import watchfiles
 from fontTools.designspaceLib import (
@@ -19,6 +21,7 @@
     DiscreteAxisDescriptor,
 )
 from fontTools.misc.transform import DecomposedTransform
+from fontTools.pens.pointPen import AbstractPointPen
 from fontTools.pens.recordingPen import RecordingPointPen
 from fontTools.ufoLib import UFOReaderWriter
 from fontTools.ufoLib.glifLib import GlyphSet
@@ -35,6 +38,7 @@
     VariableGlyph,
 )
 from ..core.path import PackedPathPointPen
+from ..core.protocols import WritableFontBackend
 from .ufo_utils import extractGlyphNameAndUnicodes
 
 logger = logging.getLogger(__name__)
@@ -60,11 +64,11 @@
 
 class DesignspaceBackend:
     @classmethod
-    def fromPath(cls, path):
+    def fromPath(cls, path: PathLike) -> WritableFontBackend:
         return cls(DesignSpaceDocument.fromfile(path))
 
     @classmethod
-    def createFromPath(cls, path):
+    def createFromPath(cls, path: PathLike) -> WritableFontBackend:
         path = pathlib.Path(path)
         ufoDir = path.parent
 
@@ -174,10 +178,13 @@ def updateGlyphSetContents(self, glyphSet):
         for glyphName, fileName in glyphSet.contents.items():
             glifFileNames[fileName] = glyphName
 
-    async def getGlyphMap(self):
+    async def getGlyphMap(self) -> dict[str, list[int]]:
         return dict(self.glyphMap)
 
-    async def getGlyph(self, glyphName):
+    async def putGlyphMap(self, value: dict[str, list[int]]) -> None:
+        pass
+
+    async def getGlyph(self, glyphName: str) -> VariableGlyph | None:
         if glyphName not in self.glyphMap:
             return None
 
@@ -278,7 +285,9 @@ def _unpackLocalDesignSpace(self, dsDict, defaultLayerName):
             )
         return axes, sources
 
-    async def putGlyph(self, glyphName, glyph, unicodes):
+    async def putGlyph(
+        self, glyphName: str, glyph: VariableGlyph, unicodes: list[int]
+    ) -> None:
         assert isinstance(unicodes, list)
         assert all(isinstance(cp, int) for cp in unicodes)
         self.glyphMap[glyphName] = unicodes
@@ -534,7 +543,7 @@ async def deleteGlyph(self, glyphName):
         del self.glyphMap[glyphName]
         self.savedGlyphModificationTimes[glyphName] = None
 
-    async def getGlobalAxes(self):
+    async def getGlobalAxes(self) -> list[GlobalAxis | GlobalDiscreteAxis]:
         return self.axes
 
     async def putGlobalAxes(self, axes):
@@ -557,17 +566,27 @@ async def putGlobalAxes(self, axes):
         self.updateAxisInfo()
         self.loadUFOLayers()
 
-    async def getUnitsPerEm(self):
+    async def getUnitsPerEm(self) -> int:
         return self.defaultFontInfo.unitsPerEm
 
-    async def getCustomData(self):
+    async def putUnitsPerEm(self, value: int) -> None:
+        del self.defaultFontInfo
+        ufoPaths = sorted(set(self.ufoLayers.iterAttrs("path")))
+        for ufoPath in ufoPaths:
+            reader = self.ufoManager.getReader(ufoPath)
+            info = UFOFontInfo()
+            reader.readInfo(info)
+            info.unitsPerEm = value
+            reader.writeInfo(info)
+
+    async def getCustomData(self) -> dict[str, Any]:
         return deepcopy(self.dsDoc.lib)
 
     async def putCustomData(self, lib):
         self.dsDoc.lib = deepcopy(lib)
         self.dsDoc.write(self.dsDoc.path)
 
-    async def watchExternalChanges(self):
+    async def watchExternalChanges(self) -> AsyncGenerator[tuple[Any, Any], None]:
         ufoPaths = sorted(set(self.ufoLayers.iterAttrs("path")))
         async for changes in watchfiles.awatch(*ufoPaths):
             changes = cleanupWatchFilesChanges(changes)
@@ -910,7 +929,7 @@ def readGlyphOrCreate(
     glyphSet: GlyphSet,
     glyphName: str,
     unicodes: list[int],
-):
+) -> UFOGlyph:
     layerGlyph = UFOGlyph()
     layerGlyph.lib = {}
     if glyphName in glyphSet:
@@ -925,7 +944,7 @@ def populateUFOLayerGlyph(
     layerGlyph: UFOGlyph,
     staticGlyph: StaticGlyph,
     forceVariableComponents: bool = False,
-) -> None:
+) -> Callable[[AbstractPointPen], None]:
     pen = RecordingPointPen()
     layerGlyph.width = staticGlyph.xAdvance
     layerGlyph.height = staticGlyph.yAdvance

src/fontra/backends/fontra.py

@@ -6,9 +6,18 @@
 import shutil
 from copy import deepcopy
 from dataclasses import dataclass, field
-from typing import Callable
-
-from fontra.core.classes import Font, VariableGlyph, structure, unstructure
+from os import PathLike
+from typing import Any, Callable
+
+from fontra.core.classes import (
+    Font,
+    GlobalAxis,
+    GlobalDiscreteAxis,
+    VariableGlyph,
+    structure,
+    unstructure,
+)
+from fontra.core.protocols import WritableFontBackend
 
 from .filenames import stringToFileName
 
@@ -21,23 +30,23 @@ class FontraBackend:
     glyphsDirName = "glyphs"
 
     @classmethod
-    def fromPath(cls, path):
+    def fromPath(cls, path) -> WritableFontBackend:
         return cls(path=path)
 
     @classmethod
-    def createFromPath(cls, path):
+    def createFromPath(cls, path) -> WritableFontBackend:
         return cls(path=path, create=True)
 
-    def __init__(self, *, path=None, create=False):
-        self.path = pathlib.Path(path).resolve() if path is not None else None
+    def __init__(self, *, path: PathLike, create: bool = False):
+        self.path = pathlib.Path(path).resolve()
         if create:
             if self.path.is_dir():
                 shutil.rmtree(self.path)
             elif self.path.exists():
                 self.path.unlink()
             self.path.mkdir()
         self.glyphsDir.mkdir(exist_ok=True)
-        self.glyphMap = {}
+        self.glyphMap: dict[str, list[int]] = {}
         if not create:
             self._readGlyphInfo()
             self._readFontData()
@@ -64,21 +73,29 @@ def close(self):
     def flush(self):
         self._scheduler.flush()
 
-    async def getUnitsPerEm(self):
+    async def getUnitsPerEm(self) -> int:
         return self.fontData.unitsPerEm
 
     async def putUnitsPerEm(self, unitsPerEm):
         self.fontData.unitsPerEm = unitsPerEm
         self._scheduler.schedule(self._writeFontData)
 
-    async def getGlyphMap(self):
+    async def getGlyphMap(self) -> dict[str, list[int]]:
         return dict(self.glyphMap)
 
-    async def getGlyph(self, glyphName):
-        jsonSource = self.getGlyphData(glyphName)
+    async def putGlyphMap(self, value: dict[str, list[int]]) -> None:
+        pass
+
+    async def getGlyph(self, glyphName: str) -> VariableGlyph | None:
+        try:
+            jsonSource = self.getGlyphData(glyphName)
+        except KeyError:
+            return None
         return deserializeGlyph(jsonSource, glyphName)
 
-    async def putGlyph(self, glyphName, glyph, codePoints):
+    async def putGlyph(
+        self, glyphName: str, glyph: VariableGlyph, codePoints: list[int]
+    ) -> None:
         jsonSource = serializeGlyph(glyph, glyphName)
         filePath = self.getGlyphFilePath(glyphName)
         filePath.write_text(jsonSource, encoding="utf=8")
@@ -88,14 +105,14 @@ async def putGlyph(self, glyphName, glyph, codePoints):
     async def deleteGlyph(self, glyphName):
         self.glyphMap.pop(glyphName, None)
 
-    async def getGlobalAxes(self):
+    async def getGlobalAxes(self) -> list[GlobalAxis | GlobalDiscreteAxis]:
         return deepcopy(self.fontData.axes)
 
     async def putGlobalAxes(self, axes):
         self.fontData.axes = deepcopy(axes)
         self._scheduler.schedule(self._writeFontData)
 
-    async def getCustomData(self):
+    async def getCustomData(self) -> dict[str, Any]:
         return deepcopy(self.fontData.customData)
 
     async def putCustomData(self, customData):