"""
.. deprecated:: 1.10.0
The p3d packaging system has been replaced with the new setuptools-based
system. See the :ref:`distribution` manual section.
"""
__all__ = ["PatchMaker"]
from direct.p3d.FileSpec import FileSpec
from direct.p3d.SeqValue import SeqValue
from panda3d.core import *
import copy
[docs]class PatchMaker:
""" This class will operate on an existing package install
directory, as generated by the Packager, and create patchfiles
between versions as needed. It is also used at runtime, to apply
the downloaded patches. """
[docs] class PackageVersion:
""" A specific patch version of a package. This is not just
the package's "version" string; it also corresponds to the
particular patch version, which increments independently of
the "version". """
[docs] def __init__(self, packageName, platform, version, hostUrl, file):
self.packageName = packageName
self.platform = platform
self.version = version
self.hostUrl = hostUrl
self.file = file
self.printName = None
# The Package object that produces this version, if this
# is the current, base, or top form, respectively.
self.packageCurrent = None
self.packageBase = None
self.packageTop = None
# A list of patchfiles that can produce this version.
self.fromPatches = []
# A list of patchfiles that can start from this version.
self.toPatches = []
# A temporary file for re-creating the archive file for
# this version.
self.tempFile = None
[docs] def cleanup(self):
if self.tempFile:
self.tempFile.unlink()
[docs] def getPatchChain(self, startPv, alreadyVisited = []):
""" Returns a list of patches that, when applied in
sequence to the indicated PackageVersion object, will
produce this PackageVersion object. Returns None if no
chain can be found. """
if self is startPv:
# We're already here. A zero-length patch chain is
# therefore the answer.
return []
if self in alreadyVisited:
# We've already been here; this is a loop. Avoid
# infinite recursion.
return None
alreadyVisited = alreadyVisited[:]
alreadyVisited.append(self)
bestPatchChain = None
for patchfile in self.fromPatches:
fromPv = patchfile.fromPv
patchChain = fromPv.getPatchChain(startPv, alreadyVisited)
if patchChain is not None:
# There's a path through this patchfile.
patchChain = patchChain + [patchfile]
if bestPatchChain is None or len(patchChain) < len(bestPatchChain):
bestPatchChain = patchChain
# Return the shortest path found, or None if there were no
# paths found.
return bestPatchChain
[docs] def getRecreateFilePlan(self, alreadyVisited = []):
""" Returns the tuple (startFile, startPv, plan),
describing how to recreate the archive file for this
version. startFile and startPv is the Filename and
packageVersion of the file to start with, and plan is a
list of tuples (patchfile, pv), listing the patches to
apply in sequence, and the packageVersion object
associated with each patch. Returns (None, None, None) if
there is no way to recreate this archive file. """
if self.tempFile:
return (self.tempFile, self, [])
if self in alreadyVisited:
# We've already been here; this is a loop. Avoid
# infinite recursion.
return (None, None, None)
alreadyVisited = alreadyVisited[:]
alreadyVisited.append(self)
if self.packageCurrent:
# This PackageVersion instance represents the current
# version of some package.
package = self.packageCurrent
return (Filename(package.packageDir, package.compressedFilename), self, [])
if self.packageBase:
# This PackageVersion instance represents the base
# (oldest) version of some package.
package = self.packageBase
return (Filename(package.packageDir, package.baseFile.filename + '.pz'), self, [])
# We'll need to re-create the file.
bestPlan = None
bestStartFile = None
bestStartPv = None
for patchfile in self.fromPatches:
fromPv = patchfile.fromPv
startFile, startPv, plan = fromPv.getRecreateFilePlan(alreadyVisited)
if plan is not None:
# There's a path through this patchfile.
plan = plan + [(patchfile, self)]
if bestPlan is None or len(plan) < len(bestPlan):
bestPlan = plan
bestStartFile = startFile
bestStartPv = startPv
# Return the shortest path found, or None if there were no
# paths found.
return (bestStartFile, bestStartPv, bestPlan)
[docs] def getFile(self):
""" Returns the Filename of the archive file associated
with this version. If the file doesn't actually exist on
disk, a temporary file will be created. Returns None if
the file can't be recreated. """
startFile, startPv, plan = self.getRecreateFilePlan()
if startFile.getExtension() in ('pz', 'gz'):
# If the starting file is compressed, we have to
# decompress it first.
assert startPv.tempFile is None
startPv.tempFile = Filename.temporary('', 'patch_')
if not decompressFile(startFile, startPv.tempFile):
# Failure trying to decompress the file.
return None
startFile = startPv.tempFile
if not plan:
# If plan is a zero-length list, we're already
# here--return startFile. If plan is None, there's no
# solution, and startFile is None. In either case, we
# can return startFile.
return startFile
# If plan is a non-empty list, we have to walk the list to
# apply the patch plan.
prevFile = startFile
for patchfile, pv in plan:
fromPv = patchfile.fromPv
patchFilename = Filename(patchfile.package.packageDir, patchfile.file.filename)
result = self.applyPatch(prevFile, patchFilename)
if not result:
# Failure trying to re-create the file.
return None
pv.tempFile = result
prevFile = result
# Successfully patched.
assert pv is self and prevFile is self.tempFile
return prevFile
[docs] def applyPatch(self, origFile, patchFilename):
""" Applies the named patch to the indicated original
file, storing the results in a temporary file, and returns
that temporary Filename. Returns None on failure. """
result = Filename.temporary('', 'patch_')
p = Patchfile()
if not p.apply(patchFilename, origFile, result):
print("Internal patching failed: %s" % (patchFilename))
return None
return result
[docs] def getNext(self, package):
""" Gets the next patch in the chain towards this
package. """
for patch in self.toPatches:
if patch.packageName == package.packageName and \
patch.platform == package.platform and \
patch.version == package.version and \
patch.hostUrl == package.hostUrl:
return patch.toPv
return None
[docs] class Patchfile:
""" A single patchfile for a package. """
[docs] def __init__(self, package):
self.package = package
self.packageName = package.packageName
self.platform = package.platform
self.version = package.version
self.hostUrl = None
# FileSpec for the patchfile itself
self.file = None
# FileSpec for the package file that the patch is applied to
self.sourceFile = None
# FileSpec for the package file that the patch generates
self.targetFile = None
# The PackageVersion corresponding to our sourceFile
self.fromPv = None
# The PackageVersion corresponding to our targetFile
self.toPv = None
[docs] def getSourceKey(self):
""" Returns the key for locating the package that this
patchfile can be applied to. """
return (self.packageName, self.platform, self.version, self.hostUrl, self.sourceFile)
[docs] def getTargetKey(self):
""" Returns the key for locating the package that this
patchfile will generate. """
return (self.packageName, self.platform, self.version, self.hostUrl, self.targetFile)
[docs] def fromFile(self, packageDir, patchFilename, sourceFile, targetFile):
""" Creates the data structures from an existing patchfile
on disk. """
self.file = FileSpec()
self.file.fromFile(packageDir, patchFilename)
self.sourceFile = sourceFile
self.targetFile = targetFile
[docs] def loadXml(self, xpatch):
""" Reads the data structures from an xml file. """
self.packageName = xpatch.Attribute('name') or self.packageName
self.platform = xpatch.Attribute('platform') or self.platform
self.version = xpatch.Attribute('version') or self.version
self.hostUrl = xpatch.Attribute('host') or self.hostUrl
self.file = FileSpec()
self.file.loadXml(xpatch)
xsource = xpatch.FirstChildElement('source')
if xsource:
self.sourceFile = FileSpec()
self.sourceFile.loadXml(xsource)
xtarget = xpatch.FirstChildElement('target')
if xtarget:
self.targetFile = FileSpec()
self.targetFile.loadXml(xtarget)
[docs] def makeXml(self, package):
xpatch = TiXmlElement('patch')
if self.packageName != package.packageName:
xpatch.SetAttribute('name', self.packageName)
if self.platform != package.platform:
xpatch.SetAttribute('platform', self.platform)
if self.version != package.version:
xpatch.SetAttribute('version', self.version)
if self.hostUrl != package.hostUrl:
xpatch.SetAttribute('host', self.hostUrl)
self.file.storeXml(xpatch)
xsource = TiXmlElement('source')
self.sourceFile.storeMiniXml(xsource)
xpatch.InsertEndChild(xsource)
xtarget = TiXmlElement('target')
self.targetFile.storeMiniXml(xtarget)
xpatch.InsertEndChild(xtarget)
return xpatch
[docs] class Package:
""" This is a particular package. This contains all of the
information needed to reconstruct the package's desc file. """
[docs] def __init__(self, packageDesc, patchMaker, xpackage = None):
self.packageDir = Filename(patchMaker.installDir, packageDesc.getDirname())
self.packageDesc = packageDesc
self.patchMaker = patchMaker
self.contentsDocPackage = xpackage
self.patchVersion = 1
self.currentPv = None
self.basePv = None
self.topPv = None
self.packageName = None
self.platform = None
self.version = None
self.hostUrl = None
self.currentFile = None
self.baseFile = None
self.doc = None
self.anyChanges = False
self.patches = []
[docs] def getCurrentKey(self):
""" Returns the key to locate the current version of this
package. """
return (self.packageName, self.platform, self.version, self.hostUrl, self.currentFile)
[docs] def getBaseKey(self):
""" Returns the key to locate the "base" or oldest version
of this package. """
return (self.packageName, self.platform, self.version, self.hostUrl, self.baseFile)
[docs] def getTopKey(self):
""" Returns the key to locate the "top" or newest version
of this package. """
return (self.packageName, self.platform, self.version, self.hostUrl, self.topFile)
[docs] def getGenericKey(self, fileSpec):
""" Returns the key that has the indicated hash. """
return (self.packageName, self.platform, self.version, self.hostUrl, fileSpec)
[docs] def readDescFile(self, doProcessing = False):
""" Reads the existing package.xml file and stores it in
this class for later rewriting. if doProcessing is true,
it may massage the file and the directory contents in
preparation for building patches. Returns true on
success, false on failure. """
self.anyChanges = False
packageDescFullpath = Filename(self.patchMaker.installDir, self.packageDesc)
self.doc = TiXmlDocument(packageDescFullpath.toOsSpecific())
if not self.doc.LoadFile():
print("Couldn't read %s" % (packageDescFullpath))
return False
xpackage = self.doc.FirstChildElement('package')
if not xpackage:
return False
self.packageName = xpackage.Attribute('name')
self.platform = xpackage.Attribute('platform')
self.version = xpackage.Attribute('version')
# All packages we defined in-line are assigned to the
# "none" host. TODO: support patching from packages on
# other hosts, which means we'll need to fill in a value
# here for those hosts.
self.hostUrl = None
self.currentFile = None
self.baseFile = None
self.topFile = None
self.compressedFilename = None
compressedFile = None
# Assume there are changes for this version, until we
# discover that there aren't.
isNewVersion = True
# Get the actual current version.
xarchive = xpackage.FirstChildElement('uncompressed_archive')
if xarchive:
self.currentFile = FileSpec()
self.currentFile.loadXml(xarchive)
# Get the top_version--the top (newest) of the patch
# chain.
xarchive = xpackage.FirstChildElement('top_version')
if xarchive:
self.topFile = FileSpec()
self.topFile.loadXml(xarchive)
if self.topFile.hash == self.currentFile.hash:
# No new version this pass.
isNewVersion = False
else:
# There's a new version this pass. Update it.
self.anyChanges = True
else:
# If there isn't a top_version yet, we have to make
# one, by duplicating the currentFile.
self.topFile = copy.copy(self.currentFile)
self.anyChanges = True
# Get the current patch version. If we have a
# patch_version attribute, it refers to this particular
# instance of the file, and that is the current patch
# version number. If we only have a last_patch_version
# attribute, it means a patch has not yet been built for
# this particular instance, and that number is the
# previous version's patch version number.
patchVersion = xpackage.Attribute('patch_version')
if patchVersion:
self.patchVersion = int(patchVersion)
else:
patchVersion = xpackage.Attribute('last_patch_version')
if patchVersion:
self.patchVersion = int(patchVersion)
if isNewVersion:
self.patchVersion += 1
self.anyChanges = True
# Put the patchVersion in the compressed filename, for
# cache-busting. This means when the version changes, its
# URL will also change, guaranteeing that users will
# download the latest version, and not some stale cache
# file.
xcompressed = xpackage.FirstChildElement('compressed_archive')
if xcompressed:
compressedFile = FileSpec()
compressedFile.loadXml(xcompressed)
oldCompressedFilename = compressedFile.filename
self.compressedFilename = oldCompressedFilename
if doProcessing:
newCompressedFilename = '%s.%s.pz' % (self.currentFile.filename, self.patchVersion)
if newCompressedFilename != oldCompressedFilename:
oldCompressedPathname = Filename(self.packageDir, oldCompressedFilename)
newCompressedPathname = Filename(self.packageDir, newCompressedFilename)
if oldCompressedPathname.renameTo(newCompressedPathname):
compressedFile.fromFile(self.packageDir, newCompressedFilename)
compressedFile.storeXml(xcompressed)
self.compressedFilename = newCompressedFilename
self.anyChanges = True
# Get the base_version--the bottom (oldest) of the patch
# chain.
xarchive = xpackage.FirstChildElement('base_version')
if xarchive:
self.baseFile = FileSpec()
self.baseFile.loadXml(xarchive)
else:
# If there isn't a base_version yet, we have to make
# one, by duplicating the currentFile.
self.baseFile = copy.copy(self.currentFile)
# Note that the we only store the compressed version
# of base_filename on disk, but we store the md5 of
# the uncompressed version in the xml file. To
# emphasize this, we name it without the .pz extension
# in the xml file, even though the compressed file on
# disk actually has a .pz extension.
self.baseFile.filename += '.base'
# Also duplicate the (compressed) file itself.
if doProcessing and self.compressedFilename:
fromPathname = Filename(self.packageDir, self.compressedFilename)
toPathname = Filename(self.packageDir, self.baseFile.filename + '.pz')
fromPathname.copyTo(toPathname)
self.anyChanges = True
self.patches = []
xpatch = xpackage.FirstChildElement('patch')
while xpatch:
patchfile = PatchMaker.Patchfile(self)
patchfile.loadXml(xpatch)
self.patches.append(patchfile)
xpatch = xpatch.NextSiblingElement('patch')
return True
[docs] def writeDescFile(self):
""" Rewrites the desc file with the new patch
information. """
if not self.anyChanges:
# No need to rewrite.
return
xpackage = self.doc.FirstChildElement('package')
if not xpackage:
return
packageSeq = SeqValue()
packageSeq.loadXml(xpackage, 'seq')
packageSeq += 1
packageSeq.storeXml(xpackage, 'seq')
# Remove all of the old patch entries from the desc file
# we read earlier.
xremove = []
for value in ['base_version', 'top_version', 'patch']:
xpatch = xpackage.FirstChildElement(value)
while xpatch:
xremove.append(xpatch)
xpatch = xpatch.NextSiblingElement(value)
for xelement in xremove:
xpackage.RemoveChild(xelement)
xpackage.RemoveAttribute('last_patch_version')
# Now replace them with the current patch information.
xpackage.SetAttribute('patch_version', str(self.patchVersion))
xarchive = TiXmlElement('base_version')
self.baseFile.storeXml(xarchive)
xpackage.InsertEndChild(xarchive)
# The current version is now the top version.
xarchive = TiXmlElement('top_version')
self.currentFile.storeXml(xarchive)
xpackage.InsertEndChild(xarchive)
for patchfile in self.patches:
xpatch = patchfile.makeXml(self)
xpackage.InsertEndChild(xpatch)
self.doc.SaveFile()
# Also copy the seq to the import desc file, for
# documentation purposes.
importDescFilename = str(self.packageDesc)[:-3] + 'import.xml'
importDescFullpath = Filename(self.patchMaker.installDir, importDescFilename)
doc = TiXmlDocument(importDescFullpath.toOsSpecific())
if doc.LoadFile():
xpackage = doc.FirstChildElement('package')
if xpackage:
packageSeq.storeXml(xpackage, 'seq')
doc.SaveFile()
else:
print("Couldn't read %s" % (importDescFullpath))
if self.contentsDocPackage:
# Now that we've rewritten the xml file, we have to
# change the contents.xml file that references it to
# indicate the new file hash.
fileSpec = FileSpec()
fileSpec.fromFile(self.patchMaker.installDir, self.packageDesc)
fileSpec.storeXml(self.contentsDocPackage)
# Also important to update the import.xml hash.
ximport = self.contentsDocPackage.FirstChildElement('import')
if ximport:
fileSpec = FileSpec()
fileSpec.fromFile(self.patchMaker.installDir, importDescFilename)
fileSpec.storeXml(ximport)
# Also copy the package seq value into the
# contents.xml file, mainly for documentation purposes
# (the authoritative seq value is within the desc
# file).
packageSeq.storeXml(self.contentsDocPackage, 'seq')
# PatchMaker constructor.
[docs] def __init__(self, installDir):
self.installDir = installDir
self.packageVersions = {}
self.packages = []
[docs] def buildPatches(self, packageNames = None):
""" Makes the patches required in a particular directory
structure on disk. If packageNames is None, this makes
patches for all packages; otherwise, it should be a list of
package name strings, limiting the set of packages that are
processed. """
if not self.readContentsFile():
return False
self.buildPatchChains()
if packageNames is None:
self.processAllPackages()
else:
self.processSomePackages(packageNames)
self.writeContentsFile()
self.cleanup()
return True
[docs] def cleanup(self):
""" Should be called on exit to remove temporary files and
such created during processing. """
for pv in self.packageVersions.values():
pv.cleanup()
[docs] def getPatchChainToCurrent(self, descFilename, fileSpec):
""" Reads the package defined in the indicated desc file, and
constructs a patch chain from the version represented by
fileSpec to the current version of this package, if possible.
Returns the patch chain if successful, or None otherwise. """
package = self.readPackageDescFile(descFilename)
if not package:
return None
self.buildPatchChains()
fromPv = self.getPackageVersion(package.getGenericKey(fileSpec))
toPv = package.currentPv
patchChain = None
if toPv and fromPv:
patchChain = toPv.getPatchChain(fromPv)
return patchChain
[docs] def readPackageDescFile(self, descFilename):
""" Reads a desc file associated with a particular package,
and adds the package to self.packages. Returns the Package
object, or None on failure. """
package = self.Package(Filename(descFilename), self)
if not package.readDescFile(doProcessing = False):
return None
self.packages.append(package)
return package
[docs] def readContentsFile(self):
""" Reads the contents.xml file at the beginning of
processing. """
contentsFilename = Filename(self.installDir, 'contents.xml')
doc = TiXmlDocument(contentsFilename.toOsSpecific())
if not doc.LoadFile():
# Couldn't read file.
print("couldn't read %s" % (contentsFilename))
return False
xcontents = doc.FirstChildElement('contents')
if xcontents:
contentsSeq = SeqValue()
contentsSeq.loadXml(xcontents)
contentsSeq += 1
contentsSeq.storeXml(xcontents)
xpackage = xcontents.FirstChildElement('package')
while xpackage:
solo = xpackage.Attribute('solo')
solo = int(solo or '0')
filename = xpackage.Attribute('filename')
if filename and not solo:
filename = Filename(filename)
package = self.Package(filename, self, xpackage)
package.readDescFile(doProcessing = True)
self.packages.append(package)
xpackage = xpackage.NextSiblingElement('package')
self.contentsDoc = doc
return True
[docs] def writeContentsFile(self):
""" Writes the contents.xml file at the end of processing. """
# We also have to write the desc file for all packages that
# might need it, because we might have changed some of them on
# read.
for package in self.packages:
package.writeDescFile()
# The above writeDescFile() call should also update each
# package's element within the contents.xml document, so all
# we have to do now is write out the document.
self.contentsDoc.SaveFile()
[docs] def getPackageVersion(self, key):
""" Returns a shared PackageVersion object for the indicated
key. """
packageName, platform, version, hostUrl, file = key
# We actually key on the hash, not the FileSpec itself.
k = (packageName, platform, version, hostUrl, file.hash)
pv = self.packageVersions.get(k, None)
if not pv:
pv = self.PackageVersion(*key)
self.packageVersions[k] = pv
return pv
[docs] def buildPatchChains(self):
""" Builds up the chains of PackageVersions and the patchfiles
that connect them. """
self.patchFilenames = {}
for package in self.packages:
if not package.baseFile:
# This package doesn't have any versions yet.
continue
currentPv = self.getPackageVersion(package.getCurrentKey())
package.currentPv = currentPv
currentPv.packageCurrent = package
currentPv.printName = package.currentFile.filename
basePv = self.getPackageVersion(package.getBaseKey())
package.basePv = basePv
basePv.packageBase = package
basePv.printName = package.baseFile.filename
topPv = self.getPackageVersion(package.getTopKey())
package.topPv = topPv
topPv.packageTop = package
for patchfile in package.patches:
self.recordPatchfile(patchfile)
[docs] def recordPatchfile(self, patchfile):
""" Adds the indicated patchfile to the patch chains. """
self.patchFilenames[patchfile.file.filename] = patchfile
fromPv = self.getPackageVersion(patchfile.getSourceKey())
patchfile.fromPv = fromPv
fromPv.toPatches.append(patchfile)
toPv = self.getPackageVersion(patchfile.getTargetKey())
patchfile.toPv = toPv
toPv.fromPatches.append(patchfile)
toPv.printName = patchfile.file.filename
[docs] def processSomePackages(self, packageNames):
""" Builds missing patches only for the named packages. """
remainingNames = packageNames[:]
for package in self.packages:
if package.packageName in packageNames:
self.processPackage(package)
if package.packageName in remainingNames:
remainingNames.remove(package.packageName)
if remainingNames:
print("Unknown packages: %s" % (remainingNames,))
[docs] def processAllPackages(self):
""" Walks through the list of packages, and builds missing
patches for each one. """
for package in self.packages:
self.processPackage(package)
[docs] def processPackage(self, package):
""" Builds missing patches for the indicated package. """
if not package.baseFile:
# No versions.
return
# What's the current version on the top of the tree?
topPv = package.topPv
currentPv = package.currentPv
if topPv != currentPv:
# They're different, so build a new patch.
filename = Filename(package.currentFile.filename + '.%s.patch' % (package.patchVersion))
assert filename not in self.patchFilenames
if not self.buildPatch(topPv, currentPv, package, filename):
raise Exception("Couldn't build patch.")
[docs] def buildPatch(self, v1, v2, package, patchFilename):
""" Builds a patch from PackageVersion v1 to PackageVersion
v2, and stores it in patchFilename.pz. Returns true on
success, false on failure."""
pathname = Filename(package.packageDir, patchFilename)
if not self.buildPatchFile(v1.getFile(), v2.getFile(), pathname,
v1.printName, v2.printName):
return False
compressedPathname = Filename(pathname + '.pz')
compressedPathname.unlink()
if not compressFile(pathname, compressedPathname, 9):
raise Exception("Couldn't compress patch.")
pathname.unlink()
patchfile = self.Patchfile(package)
patchfile.fromFile(package.packageDir, patchFilename + '.pz',
v1.file, v2.file)
package.patches.append(patchfile)
package.anyChanges = True
self.recordPatchfile(patchfile)
return True
[docs] def buildPatchFile(self, origFilename, newFilename, patchFilename,
printOrigName, printNewName):
""" Creates a patch file from origFilename to newFilename,
storing the result in patchFilename. Returns true on success,
false on failure. """
if not origFilename.exists():
# No original version to patch from.
return False
print("Building patch from %s to %s" % (printOrigName, printNewName))
patchFilename.unlink()
p = Patchfile() # The C++ class
if p.build(origFilename, newFilename, patchFilename):
return True
# Unable to build a patch for some reason.
patchFilename.unlink()
return False
