Created
April 16, 2010 02:16
-
-
Save lamberta/367919 to your computer and use it in GitHub Desktop.
RemoteTCP - FontLab server
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#=============================================================================== | |
# RemoteTCP.py, v0.1, released 01.26.08. | |
# Installation notes, usage and commentary: | |
# http://lamberta.posterous.com/remotetcp-for-robofab-and-fontlab | |
# | |
# Copyright (c) William Lamberta, <www.lamberta.org> | |
# Released under the BSD License, | |
# see www.opensource.org/licenses/bsd-license.php for details. | |
# Includes code from RoboFab <www.robofab.org>, | |
# Copyright (c) The RoboFab Developers, Just van Rossum, Tal Leming, Erik van Bloklandwhich. | |
# Released under the RoboFab License Agreement (BSD License), | |
# see http://www.robofab.org/download/license.html for details. | |
#=============================================================================== | |
class FontLabServer: | |
_allowRemoteAccess = True | |
def __init__( self, remote=True ): | |
if remote is False: | |
FontLabServer._allowRemoteAccess = False | |
def run(self, port=1455): | |
print "Turning on the FontLab network remote." | |
startFontLabServer( port ) | |
#Add you own user/password combos for access (minus the comment of course) | |
users = { | |
#'adrian': 'Adrian Frutiger', | |
#'paul': 'Paul Renner' | |
} | |
passwords = { | |
#'adrian': 'univers', | |
#'paul': 'futura' | |
} | |
def addUser( self, newUser, password, fullName='' ): | |
if FontLabServer.users.has_key( newUser ): | |
FontLabServer.users[newUser].append( fullName ) | |
FontLabServer.passwords[newUser].append( password ) | |
else: | |
FontLabServer.users[newUser] = fullName | |
FontLabServer.passwords[newUser] = password | |
def getInternal_ip(self): | |
import socket | |
try: | |
return socket.gethostbyname( socket.gethostname() ) | |
except socket.error, e: | |
print "Unable to determine the internal network ip address: %s" %e | |
def getExternal_ip(self): | |
import urllib2 | |
try: | |
#it's quick and cute, but dependent on a remote service | |
return urllib2.urlopen('http://whatismyip.org/').read() | |
except urllib2.error, e: | |
print "Unable to determine the external network ip address: %s" %e | |
#=============================================================================== | |
# Server code. Using the Twisted framework. | |
#=============================================================================== | |
def startFontLabServer( port ): | |
from twisted.protocols import basic | |
from twisted.internet import reactor, protocol, defer | |
from twisted.cred import portal, checkers, credentials, error as credError | |
from zope.interface import Interface, implements | |
#authentication classes | |
class PasswordDictChecker( object ): | |
implements( checkers.ICredentialsChecker ) | |
credentialInterfaces = (credentials.IUsernamePassword,) | |
def __init__( self, passwords ): | |
"passwords: a dict-like object mapping usernames to passwords" | |
self.passwords = FontLabServer.passwords | |
def requestAvatarId( self, credentials ): | |
username = credentials.username | |
if self.passwords.has_key( username ): | |
if credentials.password == self.passwords[username]: | |
return defer.succeed( username ) | |
else: | |
return defer.fail( | |
credError.UnauthorizedLogin( "Bad password." )) | |
else: | |
return defer.fail( | |
credError.UnauthorizedLogin( "No such user." )) | |
class INamedUserAvatar(Interface): | |
"should have attributes username and fullname" | |
class NamedUserAvatar: | |
implements( INamedUserAvatar ) | |
def __init__( self, username, fullname ): | |
self.username = username | |
self.fullname = fullname | |
class FontLabRemoteRealm: | |
implements( portal.IRealm ) | |
def __init__( self, users ): | |
self.users = FontLabServer.users | |
def requestAvatar( self, avatarId, mind, *interfaces ): | |
if INamedUserAvatar in interfaces: | |
fullname = self.users[avatarId] | |
logout = lambda: None | |
return (INamedUserAvatar,NamedUserAvatar(avatarId, fullname),logout) | |
else: | |
raise KeyError("None of the requested interfaces is supported") | |
#Start with LineReceiver to handle commands, when authenticated drops to RawMode | |
class FontlabProtocol( basic.LineReceiver ): | |
def lineReceived(self, line): | |
#default LineReceiver line delimiter is \r\n, not \n | |
cmd = getattr( self, 'handle_' + self.currentCommand ) | |
cmd( line.strip() ) | |
def connectionMade( self ): | |
#If connection is from localhost than bypass authentication | |
if self.transport.getPeer().host == "127.0.0.1": | |
#the client uses this banner so don't chsnge | |
self.transport.write("Hello localhost!") | |
self.setRawMode() | |
elif FontLabServer._allowRemoteAccess is False: | |
#if the no remote flag is set, drop outside connections | |
self.transport.write("Sorry, remote access not allowed.") | |
self.transport.loseConnection() | |
else: | |
self.transport.write("Username: ") | |
self.currentCommand = 'user' | |
def handle_user( self, username ): | |
self.username = username | |
self.transport.write("Password: ") | |
self.currentCommand = 'pass' | |
def handle_pass( self, password ): | |
creds = credentials.UsernamePassword( self.username, password ) | |
self.factory.portal.login(creds, None, INamedUserAvatar).addCallback( | |
self._loginSucceeded).addErrback(self._loginFailed) | |
def _loginSucceeded( self, avatarInfo ): | |
avatarInterface, avatar, logout = avatarInfo | |
self.setRawMode() | |
def _logoutFinished( self, result ): | |
self.transport.loseConnection() | |
def _loginFailed( self, failure ): | |
self.transport.write("Denied: %s.\r\n" % failure.getErrorMessage() ) | |
self.transport.loseConnection() | |
#dropped into raw mode to handle python commands | |
def rawDataReceived( self, data ): | |
import sys, cStringIO, traceback | |
clientCommand = data | |
# create file-like string to capture output | |
codeOut = cStringIO.StringIO() | |
codeErr = cStringIO.StringIO() | |
# capture output and errors | |
sys.stdout = codeOut | |
sys.stderr = codeErr | |
namespace = {} | |
try: | |
try: | |
exec clientCommand in namespace | |
except: | |
traceback.print_exc() | |
finally: | |
# restore stdout and stderr | |
sys.stdout = sys.__stdout__ | |
sys.stderr = sys.__stderr__ | |
# see what we got and send a response to client | |
output = codeErr.getvalue() | |
if output: | |
print "< Failure executing remote command. >", self.transport.client | |
self.transport.write( output ) | |
output = codeOut.getvalue() | |
if output: | |
print "< Executing remote command. >", self.transport.client | |
self.transport.write( output ) | |
#reset and close connection. One request per connection | |
codeOut.close() | |
codeErr.close() | |
self.transport.loseConnection() | |
def connectionLost( self, reason ): | |
print "Server closed connection!", reason | |
def clientConnectionLost(self, reason): | |
print "Client closed connection!", reason | |
class FontlabServerFactory( protocol.ServerFactory ): | |
protocol = FontlabProtocol | |
def __init__(self, portal): | |
self.portal = portal | |
#a portal for authentication | |
p = portal.Portal( FontLabRemoteRealm( FontLabServer.users )) | |
p.registerChecker( PasswordDictChecker( FontLabServer.passwords )) | |
#and a factory for the server | |
factory = FontlabServerFactory( p ) | |
reactor.listenTCP( port, factory ) | |
print "Starting server on port %d." %port | |
reactor.run() | |
#=============================================================================== | |
# runFontLabRemote is a client function that sends Python code to a remote | |
# server to be evaluated and returned. If it is run on localhost the server will | |
# detect this and send a welcome banner that allows it to bypass authentication. | |
# Otherwise it is run from the network and the client must supply a username and | |
# password. It's a fairly straight forward socket connection with plenty of error | |
# checking. I would have liked to implement this using a twisted client but | |
# multiple calls to the reactor caused it to hang after one remote execution. | |
#=============================================================================== | |
def runFontLabRemote( remoteCode, host="localhost", port=1455, user='', pw='' ): | |
"""Runs code on a server (within FontLab or not) and returns evaluation.""" | |
import socket, sys | |
try: | |
mySocket = socket.socket( socket.AF_INET, socket.SOCK_STREAM ) | |
except socket.error, e: | |
print "Error creating socket: %s" %e | |
sys.exit(1) | |
try: | |
mySocket.connect(( host, port )) | |
except socket.gaierror, e: | |
print "Address-related error connecting to server: %s" %e | |
sys.exit(1) | |
except socket.error, e: | |
print "Connection error: %s" %e | |
sys.exit(1) | |
try: | |
serverResponse = mySocket.recv(512) #get welcome | |
except socket.error, e: | |
print "Error receiving data: %s" %e | |
sys.exit(1) | |
if serverResponse == "Hello localhost!": #bypass auth | |
try: | |
codeLen, received = mySocket.send( remoteCode ), 0 | |
except socket.error, e: | |
print "Error sending data: %s" %e | |
sys.exit(1) | |
if codeLen != len( remoteCode ): | |
print "Failed to send complete message" | |
try: | |
serverResponse = mySocket.recv( 1024 )#get what we came for | |
except socket.error, e: | |
print "Error receiving data: %s" %e | |
sys.exit(1) | |
elif serverResponse == "Username: ": | |
try: | |
mySocket.send( user + "\r\n" )#basic.LineReceiver needs \r\n | |
except socket.error, e: | |
print "Error sending data: %s" %e | |
sys.exit(1) | |
try: | |
serverResponse = mySocket.recv(512) | |
except socket.error, e: | |
print "Error receiving data: %s" %e | |
sys.exit(1) | |
if serverResponse == "Password: ": | |
try: | |
mySocket.send( pw + "\r\n" ) | |
except socket.error, e: | |
print "Error sending data: %s" %e | |
sys.exit(1) | |
try: | |
codeLen, received = mySocket.send( remoteCode ), 0 | |
except socket.error, e: | |
print "Error sending data: %s" %e | |
sys.exit(1) | |
if codeLen != len( remoteCode ): | |
print "Failed to send complete message" | |
try: | |
serverResponse = mySocket.recv( 1024 )#what we came for | |
except socket.error, e: | |
print "Error receiving data: %s" %e | |
sys.exit(1) | |
else: | |
print "Error: Received response from server that I didn't forsee:\n", serverResponse | |
sys.exit(1) | |
mySocket.close() | |
return serverResponse | |
#=============================================================================== | |
# The glyph transmit and interface functions from robofab.tools.remote, | |
# modified slightly by taking out the Mac only stuff where needed. | |
#=============================================================================== | |
def Glyph2String( glyph ): | |
from robofab.pens.digestPen import DigestPointPen | |
import pickle | |
p = DigestPointPen( glyph ) | |
glyph.drawPoints( p ) | |
info = {} | |
info['name'] = glyph.name | |
info['width'] = glyph.width | |
info['points'] = p.getDigest() | |
return str( pickle.dumps( info ) ) | |
def String2Glyph( gString, penClass, font ): | |
import pickle | |
if gString is None: | |
return None | |
info = pickle.loads( gString ) | |
name = info['name'] | |
if not name in font.keys(): | |
glyph = font.newGlyph( name ) | |
else: | |
glyph = font[name] | |
pen = penClass( glyph ) | |
for p in info['points']: | |
if p == "beginPath": | |
pen.beginPath() | |
elif p == "endPath": | |
pen.endPath() | |
else: | |
pt, type = p | |
pen.addPoint(pt, type) | |
glyph.width = info['width'] | |
glyph.update() | |
return glyph | |
_makeFLGlyph = """ | |
from robofab.world import CurrentFont | |
from robofab.tools.remoteTCP import receiveGlyph | |
receiveGlyph( '''%s''', CurrentFont() )""" | |
def transmitGlyph( glyph, host="localhost", port=1455, user='', pw='' ): | |
from robofab.world import world | |
if world.inFontLab is True: | |
print Glyph2String( glyph ) | |
else: | |
# takes our new glyph string and inserts it in the above code | |
remoteCode = _makeFLGlyph % Glyph2String( glyph ) | |
#all packed up, now pass it to our client function | |
return runFontLabRemote( remoteCode, host, port, user, pw ) | |
def receiveGlyph( glyphString, font=None ): | |
from robofab.world import world | |
if world.inFontLab is True: | |
# for use inside fontlab, writes to CurrentFont() | |
from robofab.pens.flPen import FLPointPen | |
print String2Glyph( glyphString, FLPointPen, font ) | |
else: | |
# catch output for use elsewhere (in our scripts) | |
from robofab.pens.rfUFOPen import RFUFOPointPen | |
print String2Glyph( glyphString, RFUFOPointPen, font ) | |
#=============================================================================== | |
# Functions for FontLab to open a UFO and save as a VFB. | |
# Taken verbatim from robofab.tools.remote, included for convenience. | |
#=============================================================================== | |
def os9PathConvert(path): | |
"""Attempt to convert a unix style path to a Mac OS9 style path. | |
No support for relative paths! | |
""" | |
if path.find("/Volumes") == 0: | |
# it's on the volumes list, some sort of external volume | |
path = path[len("/Volumes")+1:] | |
elif path[0] == "/": | |
# a dir on the root volume | |
path = path[1:] | |
new = path.replace("/", ":") | |
return new | |
_remoteUFOImportProgram = """ | |
from robofab.objects.objectsFL import NewFont | |
import os.path | |
destinationPathVFB = "%(destinationPathVFB)s" | |
font = NewFont() | |
font.readUFO("%(sourcePathUFO)s", doProgress=True) | |
font.update() | |
font.save(destinationPathVFB) | |
print font, "done" | |
font.close() | |
""" | |
def makeVFB(sourcePathUFO, destinationPathVFB=None): | |
"""FontLab convenience function to import a UFO and save it as a VFB""" | |
import os | |
fl = FontLab("FLab", start=1) | |
if destinationPathVFB is None: | |
destinationPathVFB = os.path.splitext(sourcePathUFO)[0]+".vfb" | |
src9 = os9PathConvert(sourcePathUFO) | |
dst9 = os9PathConvert(destinationPathVFB) | |
code = _remoteUFOImportProgram%{'sourcePathUFO': src9, 'destinationPathVFB':dst9} | |
ae, parms, attrs = fl.send("Rfab", "exec", {"----": code}) | |
output = parms.get("----") | |
return output |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment