617b8c6396
varStore.py. Commit the correct function references. psCharstrings.py. Add decompiler class specifically for flattening subroutinized charstrings. mutator.py. Flatten the subroutinized charstrings, delete the Subrs. TestCFF2VF.otf, Updated to version with subroutinization. Updated expected ttx output.
582 lines
16 KiB
Python
582 lines
16 KiB
Python
from __future__ import print_function, division, absolute_import
|
|
from fontTools.misc.py23 import *
|
|
from fontTools.misc.fixedTools import otRound
|
|
from fontTools.ttLib.tables import otTables as ot
|
|
from fontTools.varLib.models import supportScalar
|
|
from fontTools.varLib.builder import (buildVarRegionList, buildVarStore,
|
|
buildVarRegion, buildVarData)
|
|
from functools import partial
|
|
from collections import defaultdict
|
|
from array import array
|
|
|
|
|
|
def _getLocationKey(loc):
|
|
return tuple(sorted(loc.items(), key=lambda kv: kv[0]))
|
|
|
|
|
|
class OnlineVarStoreBuilder(object):
|
|
|
|
def __init__(self, axisTags):
|
|
self._axisTags = axisTags
|
|
self._regionMap = {}
|
|
self._regionList = buildVarRegionList([], axisTags)
|
|
self._store = buildVarStore(self._regionList, [])
|
|
self._data = None
|
|
self._model = None
|
|
self._supports = None
|
|
self._varDataIndices = {}
|
|
self._varDataCaches = {}
|
|
self._cache = {}
|
|
|
|
def setModel(self, model):
|
|
self.setSupports(model.supports)
|
|
self._model = model
|
|
|
|
def setSupports(self, supports):
|
|
self._model = None
|
|
self._supports = list(supports)
|
|
if not self._supports[0]:
|
|
del self._supports[0] # Drop base master support
|
|
self._cache = {}
|
|
self._data = None
|
|
|
|
def finish(self, optimize=True):
|
|
self._regionList.RegionCount = len(self._regionList.Region)
|
|
self._store.VarDataCount = len(self._store.VarData)
|
|
for data in self._store.VarData:
|
|
data.ItemCount = len(data.Item)
|
|
data.calculateNumShorts(optimize=optimize)
|
|
return self._store
|
|
|
|
def _add_VarData(self):
|
|
regionMap = self._regionMap
|
|
regionList = self._regionList
|
|
|
|
regions = self._supports
|
|
regionIndices = []
|
|
for region in regions:
|
|
key = _getLocationKey(region)
|
|
idx = regionMap.get(key)
|
|
if idx is None:
|
|
varRegion = buildVarRegion(region, self._axisTags)
|
|
idx = regionMap[key] = len(regionList.Region)
|
|
regionList.Region.append(varRegion)
|
|
regionIndices.append(idx)
|
|
|
|
# Check if we have one already...
|
|
key = tuple(regionIndices)
|
|
varDataIdx = self._varDataIndices.get(key)
|
|
if varDataIdx is not None:
|
|
self._outer = varDataIdx
|
|
self._data = self._store.VarData[varDataIdx]
|
|
self._cache = self._varDataCaches[key]
|
|
if len(self._data.Item) == 0xFFF:
|
|
# This is full. Need new one.
|
|
varDataIdx = None
|
|
|
|
if varDataIdx is None:
|
|
self._data = buildVarData(regionIndices, [], optimize=False)
|
|
self._outer = len(self._store.VarData)
|
|
self._store.VarData.append(self._data)
|
|
self._varDataIndices[key] = self._outer
|
|
if key not in self._varDataCaches:
|
|
self._varDataCaches[key] = {}
|
|
self._cache = self._varDataCaches[key]
|
|
|
|
|
|
def storeMasters(self, master_values):
|
|
deltas = self._model.getDeltas(master_values)
|
|
base = otRound(deltas.pop(0))
|
|
return base, self.storeDeltas(deltas)
|
|
|
|
def storeDeltas(self, deltas):
|
|
# Pity that this exists here, since VarData_addItem
|
|
# does the same. But to look into our cache, it's
|
|
# good to adjust deltas here as well...
|
|
deltas = [otRound(d) for d in deltas]
|
|
if len(deltas) == len(self._supports) + 1:
|
|
deltas = tuple(deltas[1:])
|
|
else:
|
|
assert len(deltas) == len(self._supports)
|
|
deltas = tuple(deltas)
|
|
|
|
varIdx = self._cache.get(deltas)
|
|
if varIdx is not None:
|
|
return varIdx
|
|
|
|
if not self._data:
|
|
self._add_VarData()
|
|
inner = len(self._data.Item)
|
|
if inner == 0xFFFF:
|
|
# Full array. Start new one.
|
|
self._add_VarData()
|
|
return self.storeDeltas(deltas)
|
|
self._data.addItem(deltas)
|
|
|
|
varIdx = (self._outer << 16) + inner
|
|
self._cache[deltas] = varIdx
|
|
return varIdx
|
|
|
|
def VarData_addItem(self, deltas):
|
|
deltas = [otRound(d) for d in deltas]
|
|
|
|
countUs = self.VarRegionCount
|
|
countThem = len(deltas)
|
|
if countUs + 1 == countThem:
|
|
deltas = tuple(deltas[1:])
|
|
else:
|
|
assert countUs == countThem, (countUs, countThem)
|
|
deltas = tuple(deltas)
|
|
self.Item.append(list(deltas))
|
|
self.ItemCount = len(self.Item)
|
|
|
|
ot.VarData.addItem = VarData_addItem
|
|
|
|
def VarRegion_get_support(self, fvar_axes):
|
|
return {fvar_axes[i].axisTag: (reg.StartCoord,reg.PeakCoord,reg.EndCoord)
|
|
for i,reg in enumerate(self.VarRegionAxis)}
|
|
|
|
ot.VarRegion.get_support = VarRegion_get_support
|
|
|
|
class VarStoreInstancer(object):
|
|
|
|
def __init__(self, varstore, fvar_axes, location={}):
|
|
self.fvar_axes = fvar_axes
|
|
assert varstore is None or varstore.Format == 1
|
|
self._varData = varstore.VarData if varstore else []
|
|
self._regions = varstore.VarRegionList.Region if varstore else []
|
|
self.setLocation(location)
|
|
|
|
def setLocation(self, location):
|
|
self.location = dict(location)
|
|
self._clearCaches()
|
|
|
|
def _clearCaches(self):
|
|
self._scalars = {}
|
|
|
|
def _getScalar(self, regionIdx):
|
|
scalar = self._scalars.get(regionIdx)
|
|
if scalar is None:
|
|
support = self._regions[regionIdx].get_support(self.fvar_axes)
|
|
scalar = supportScalar(self.location, support)
|
|
self._scalars[regionIdx] = scalar
|
|
return scalar
|
|
|
|
@staticmethod
|
|
def interpolateFromDeltasAndScalars(deltas, scalars):
|
|
delta = 0.
|
|
for d,s in zip(deltas, scalars):
|
|
if not s: continue
|
|
delta += d * s
|
|
return delta
|
|
|
|
def __getitem__(self, varidx):
|
|
major, minor = varidx >> 16, varidx & 0xFFFF
|
|
varData = self._varData
|
|
scalars = [self._getScalar(ri) for ri in varData[major].VarRegionIndex]
|
|
deltas = varData[major].Item[minor]
|
|
return self.interpolateFromDeltasAndScalars(deltas, scalars)
|
|
|
|
def interpolateFromDeltas(self, varDataIndex, deltas):
|
|
varData = self._varData
|
|
scalars = [self._getScalar(ri) for ri in
|
|
varData[varDataIndex].VarRegionIndex]
|
|
return self.interpolateFromDeltasAndScalars(deltas, scalars)
|
|
|
|
|
|
#
|
|
# Optimizations
|
|
#
|
|
|
|
def VarStore_subset_varidxes(self, varIdxes, optimize=True):
|
|
|
|
# Sort out used varIdxes by major/minor.
|
|
used = {}
|
|
for varIdx in varIdxes:
|
|
major = varIdx >> 16
|
|
minor = varIdx & 0xFFFF
|
|
d = used.get(major)
|
|
if d is None:
|
|
d = used[major] = set()
|
|
d.add(minor)
|
|
del varIdxes
|
|
|
|
#
|
|
# Subset VarData
|
|
#
|
|
|
|
varData = self.VarData
|
|
newVarData = []
|
|
varDataMap = {}
|
|
for major,data in enumerate(varData):
|
|
usedMinors = used.get(major)
|
|
if usedMinors is None:
|
|
continue
|
|
newMajor = varDataMap[major] = len(newVarData)
|
|
newVarData.append(data)
|
|
|
|
items = data.Item
|
|
newItems = []
|
|
for minor in sorted(usedMinors):
|
|
newMinor = len(newItems)
|
|
newItems.append(items[minor])
|
|
varDataMap[(major<<16)+minor] = (newMajor<<16)+newMinor
|
|
|
|
data.Item = newItems
|
|
data.ItemCount = len(data.Item)
|
|
|
|
data.calculateNumShorts(optimize=optimize)
|
|
|
|
self.VarData = newVarData
|
|
self.VarDataCount = len(self.VarData)
|
|
|
|
self.prune_regions()
|
|
|
|
return varDataMap
|
|
|
|
ot.VarStore.subset_varidxes = VarStore_subset_varidxes
|
|
|
|
def VarStore_prune_regions(self):
|
|
"""Remove unused VarRegions."""
|
|
#
|
|
# Subset VarRegionList
|
|
#
|
|
|
|
# Collect.
|
|
usedRegions = set()
|
|
for data in self.VarData:
|
|
usedRegions.update(data.VarRegionIndex)
|
|
# Subset.
|
|
regionList = self.VarRegionList
|
|
regions = regionList.Region
|
|
newRegions = []
|
|
regionMap = {}
|
|
for i in sorted(usedRegions):
|
|
regionMap[i] = len(newRegions)
|
|
newRegions.append(regions[i])
|
|
regionList.Region = newRegions
|
|
regionList.RegionCount = len(regionList.Region)
|
|
# Map.
|
|
for data in self.VarData:
|
|
data.VarRegionIndex = [regionMap[i] for i in data.VarRegionIndex]
|
|
|
|
ot.VarStore.prune_regions = VarStore_prune_regions
|
|
|
|
|
|
def _visit(self, func):
|
|
"""Recurse down from self, if type of an object is ot.Device,
|
|
call func() on it. Works on otData-style classes."""
|
|
|
|
if type(self) == ot.Device:
|
|
func(self)
|
|
|
|
elif isinstance(self, list):
|
|
for that in self:
|
|
_visit(that, func)
|
|
|
|
elif hasattr(self, 'getConverters') and not hasattr(self, 'postRead'):
|
|
for conv in self.getConverters():
|
|
that = getattr(self, conv.name, None)
|
|
if that is not None:
|
|
_visit(that, func)
|
|
|
|
elif isinstance(self, ot.ValueRecord):
|
|
for that in self.__dict__.values():
|
|
_visit(that, func)
|
|
|
|
def _Device_recordVarIdx(self, s):
|
|
"""Add VarIdx in this Device table (if any) to the set s."""
|
|
if self.DeltaFormat == 0x8000:
|
|
s.add((self.StartSize<<16)+self.EndSize)
|
|
|
|
def Object_collect_device_varidxes(self, varidxes):
|
|
adder = partial(_Device_recordVarIdx, s=varidxes)
|
|
_visit(self, adder)
|
|
|
|
ot.GDEF.collect_device_varidxes = Object_collect_device_varidxes
|
|
ot.GPOS.collect_device_varidxes = Object_collect_device_varidxes
|
|
|
|
def _Device_mapVarIdx(self, mapping, done):
|
|
"""Map VarIdx in this Device table (if any) through mapping."""
|
|
if id(self) in done:
|
|
return
|
|
done.add(id(self))
|
|
if self.DeltaFormat == 0x8000:
|
|
varIdx = mapping[(self.StartSize<<16)+self.EndSize]
|
|
self.StartSize = varIdx >> 16
|
|
self.EndSize = varIdx & 0xFFFF
|
|
|
|
def Object_remap_device_varidxes(self, varidxes_map):
|
|
mapper = partial(_Device_mapVarIdx, mapping=varidxes_map, done=set())
|
|
_visit(self, mapper)
|
|
|
|
ot.GDEF.remap_device_varidxes = Object_remap_device_varidxes
|
|
ot.GPOS.remap_device_varidxes = Object_remap_device_varidxes
|
|
|
|
|
|
class _Encoding(object):
|
|
|
|
def __init__(self, chars):
|
|
self.chars = chars
|
|
self.width = self._popcount(chars)
|
|
self.overhead = self._characteristic_overhead(chars)
|
|
self.items = set()
|
|
|
|
def append(self, row):
|
|
self.items.add(row)
|
|
|
|
def extend(self, lst):
|
|
self.items.update(lst)
|
|
|
|
def get_room(self):
|
|
"""Maximum number of bytes that can be added to characteristic
|
|
while still being beneficial to merge it into another one."""
|
|
count = len(self.items)
|
|
return max(0, (self.overhead - 1) // count - self.width)
|
|
room = property(get_room)
|
|
|
|
@property
|
|
def gain(self):
|
|
"""Maximum possible byte gain from merging this into another
|
|
characteristic."""
|
|
count = len(self.items)
|
|
return max(0, self.overhead - count * (self.width + 1))
|
|
|
|
def sort_key(self):
|
|
return self.width, self.chars
|
|
|
|
def __len__(self):
|
|
return len(self.items)
|
|
|
|
def can_encode(self, chars):
|
|
return not (chars & ~self.chars)
|
|
|
|
def __sub__(self, other):
|
|
return self._popcount(self.chars & ~other.chars)
|
|
|
|
@staticmethod
|
|
def _popcount(n):
|
|
# Apparently this is the fastest native way to do it...
|
|
# https://stackoverflow.com/a/9831671
|
|
return bin(n).count('1')
|
|
|
|
@staticmethod
|
|
def _characteristic_overhead(chars):
|
|
"""Returns overhead in bytes of encoding this characteristic
|
|
as a VarData."""
|
|
c = 6
|
|
while chars:
|
|
if chars & 3:
|
|
c += 2
|
|
chars >>= 2
|
|
return c
|
|
|
|
|
|
def _find_yourself_best_new_encoding(self, done_by_width):
|
|
self.best_new_encoding = None
|
|
for new_width in range(self.width+1, self.width+self.room+1):
|
|
for new_encoding in done_by_width[new_width]:
|
|
if new_encoding.can_encode(self.chars):
|
|
break
|
|
else:
|
|
new_encoding = None
|
|
self.best_new_encoding = new_encoding
|
|
|
|
|
|
class _EncodingDict(dict):
|
|
|
|
def __missing__(self, chars):
|
|
r = self[chars] = _Encoding(chars)
|
|
return r
|
|
|
|
def add_row(self, row):
|
|
chars = self._row_characteristics(row)
|
|
self[chars].append(row)
|
|
|
|
@staticmethod
|
|
def _row_characteristics(row):
|
|
"""Returns encoding characteristics for a row."""
|
|
chars = 0
|
|
i = 1
|
|
for v in row:
|
|
if v:
|
|
chars += i
|
|
if not (-128 <= v <= 127):
|
|
chars += i * 2
|
|
i <<= 2
|
|
return chars
|
|
|
|
|
|
def VarStore_optimize(self):
|
|
"""Optimize storage. Returns mapping from old VarIdxes to new ones."""
|
|
|
|
# TODO
|
|
# Check that no two VarRegions are the same; if they are, fold them.
|
|
|
|
n = len(self.VarRegionList.Region) # Number of columns
|
|
zeroes = array('h', [0]*n)
|
|
|
|
front_mapping = {} # Map from old VarIdxes to full row tuples
|
|
|
|
encodings = _EncodingDict()
|
|
|
|
# Collect all items into a set of full rows (with lots of zeroes.)
|
|
for major,data in enumerate(self.VarData):
|
|
regionIndices = data.VarRegionIndex
|
|
|
|
for minor,item in enumerate(data.Item):
|
|
|
|
row = array('h', zeroes)
|
|
for regionIdx,v in zip(regionIndices, item):
|
|
row[regionIdx] += v
|
|
row = tuple(row)
|
|
|
|
encodings.add_row(row)
|
|
front_mapping[(major<<16)+minor] = row
|
|
|
|
# Separate encodings that have no gain (are decided) and those having
|
|
# possible gain (possibly to be merged into others.)
|
|
encodings = sorted(encodings.values(), key=_Encoding.__len__, reverse=True)
|
|
done_by_width = defaultdict(list)
|
|
todo = []
|
|
for encoding in encodings:
|
|
if not encoding.gain:
|
|
done_by_width[encoding.width].append(encoding)
|
|
else:
|
|
todo.append(encoding)
|
|
|
|
# For each encoding that is possibly to be merged, find the best match
|
|
# in the decided encodings, and record that.
|
|
todo.sort(key=_Encoding.get_room)
|
|
for encoding in todo:
|
|
encoding._find_yourself_best_new_encoding(done_by_width)
|
|
|
|
# Walk through todo encodings, for each, see if merging it with
|
|
# another todo encoding gains more than each of them merging with
|
|
# their best decided encoding. If yes, merge them and add resulting
|
|
# encoding back to todo queue. If not, move the enconding to decided
|
|
# list. Repeat till done.
|
|
while todo:
|
|
encoding = todo.pop()
|
|
best_idx = None
|
|
best_gain = 0
|
|
for i,other_encoding in enumerate(todo):
|
|
combined_chars = other_encoding.chars | encoding.chars
|
|
combined_width = _Encoding._popcount(combined_chars)
|
|
combined_overhead = _Encoding._characteristic_overhead(combined_chars)
|
|
combined_gain = (
|
|
+ encoding.overhead
|
|
+ other_encoding.overhead
|
|
- combined_overhead
|
|
- (combined_width - encoding.width) * len(encoding)
|
|
- (combined_width - other_encoding.width) * len(other_encoding)
|
|
)
|
|
this_gain = 0 if encoding.best_new_encoding is None else (
|
|
+ encoding.overhead
|
|
- (encoding.best_new_encoding.width - encoding.width) * len(encoding)
|
|
)
|
|
other_gain = 0 if other_encoding.best_new_encoding is None else (
|
|
+ other_encoding.overhead
|
|
- (other_encoding.best_new_encoding.width - other_encoding.width) * len(other_encoding)
|
|
)
|
|
separate_gain = this_gain + other_gain
|
|
|
|
if combined_gain > separate_gain:
|
|
best_idx = i
|
|
best_gain = combined_gain - separate_gain
|
|
|
|
if best_idx is None:
|
|
# Encoding is decided as is
|
|
done_by_width[encoding.width].append(encoding)
|
|
else:
|
|
other_encoding = todo[best_idx]
|
|
combined_chars = other_encoding.chars | encoding.chars
|
|
combined_encoding = _Encoding(combined_chars)
|
|
combined_encoding.extend(encoding.items)
|
|
combined_encoding.extend(other_encoding.items)
|
|
combined_encoding._find_yourself_best_new_encoding(done_by_width)
|
|
del todo[best_idx]
|
|
todo.append(combined_encoding)
|
|
|
|
# Assemble final store.
|
|
back_mapping = {} # Mapping from full rows to new VarIdxes
|
|
encodings = sum(done_by_width.values(), [])
|
|
encodings.sort(key=_Encoding.sort_key)
|
|
self.VarData = []
|
|
for major,encoding in enumerate(encodings):
|
|
data = ot.VarData()
|
|
self.VarData.append(data)
|
|
data.VarRegionIndex = range(n)
|
|
data.VarRegionCount = len(data.VarRegionIndex)
|
|
data.Item = sorted(encoding.items)
|
|
for minor,item in enumerate(data.Item):
|
|
back_mapping[item] = (major<<16)+minor
|
|
|
|
# Compile final mapping.
|
|
varidx_map = {}
|
|
for k,v in front_mapping.items():
|
|
varidx_map[k] = back_mapping[v]
|
|
|
|
# Remove unused regions.
|
|
self.prune_regions()
|
|
|
|
# Recalculate things and go home.
|
|
self.VarRegionList.RegionCount = len(self.VarRegionList.Region)
|
|
self.VarDataCount = len(self.VarData)
|
|
for data in self.VarData:
|
|
data.ItemCount = len(data.Item)
|
|
data.optimize()
|
|
|
|
return varidx_map
|
|
|
|
ot.VarStore.optimize = VarStore_optimize
|
|
|
|
|
|
def main(args=None):
|
|
from argparse import ArgumentParser
|
|
from fontTools import configLogger
|
|
from fontTools.ttLib import TTFont
|
|
from fontTools.ttLib.tables.otBase import OTTableWriter
|
|
|
|
parser = ArgumentParser(prog='varLib.varStore')
|
|
parser.add_argument('fontfile')
|
|
parser.add_argument('outfile', nargs='?')
|
|
options = parser.parse_args(args)
|
|
|
|
# TODO: allow user to configure logging via command-line options
|
|
configLogger(level="INFO")
|
|
|
|
fontfile = options.fontfile
|
|
outfile = options.outfile
|
|
|
|
font = TTFont(fontfile)
|
|
gdef = font['GDEF']
|
|
store = gdef.table.VarStore
|
|
|
|
writer = OTTableWriter()
|
|
store.compile(writer, font)
|
|
size = len(writer.getAllData())
|
|
print("Before: %7d bytes" % size)
|
|
|
|
varidx_map = store.optimize()
|
|
|
|
gdef.table.remap_device_varidxes(varidx_map)
|
|
if 'GPOS' in font:
|
|
font['GPOS'].table.remap_device_varidxes(varidx_map)
|
|
|
|
writer = OTTableWriter()
|
|
store.compile(writer, font)
|
|
size = len(writer.getAllData())
|
|
print("After: %7d bytes" % size)
|
|
|
|
if outfile is not None:
|
|
font.save(outfile)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import sys
|
|
if len(sys.argv) > 1:
|
|
sys.exit(main())
|
|
import doctest
|
|
sys.exit(doctest.testmod().failed)
|