platz/libs/ga.py

"""
Enthaelt alle Datenstrukturen um die einen genetischen Algorithmus darzustellen

"""
from random import *
from itertools import chain
from copy import deepcopy
from functools import total_ordering
import ast
import configparser, os


def _StringAusdruck(text):
	"""wertet einen einfachen String-Ausdruck sicher aus.

	   zyklus.cfg nutzt kompakte Schreibweisen wie
	     'e,'+'s,z,m,j,z,'*3+'s'
	   also String-Literale, mit '+' verkettet und mit '*' vervielfacht.
	   ast.literal_eval kann das nicht (nur reine Literale), eval() waere
	   unsicher. Dieser Mini-Auswerter erlaubt ausschliesslich str-Literale,
	   ganze Zahlen sowie die Operatoren + und * - kein Funktionsaufruf,
	   keine Namen, kein beliebiger Code.
	"""
	knoten = ast.parse(text, mode='eval').body

	def rechne(n):
		if isinstance(n, ast.Constant) and isinstance(n.value, (str, int)):
			return n.value
		if isinstance(n, ast.BinOp) and isinstance(n.op, (ast.Add, ast.Mult)):
			links, rechts = rechne(n.left), rechne(n.right)
			return links + rechts if isinstance(n.op, ast.Add) else links * rechts
		raise ValueError("unerlaubter Ausdruck in zyklus.cfg: %r" % text)

	ergebnis = rechne(knoten)
	if not isinstance(ergebnis, str):
		raise ValueError("zyklus.cfg-Ausdruck ergibt keinen String: %r" % text)
	return ergebnis


__author__ = "Michael Stangl"
__version__ = "$Revision: $"
__date__ = "$Date: $"
__copyright__ = "Copyright (c) 2005 Michael Stangl"
__license__ = "Python"

@total_ordering
class Loesung:
	"""ein einfaches Loesungsobjekt. Taugt um es als Superklasse zu ueberladen"""
	def __init__(self, *Argumente, **keywords):
		self.Id = -1 * randint(1,100)
		self.value = self.Id
		print(" initialisiere Loesung " + repr(self.Id))
	def mutieren(self):
		print(" mutiere Loesung " + repr(self.Id))
		self.Id = -1 * randint(1,100)
		self.value = self.Id
		return self
	def kreuzen(self,obj):
		print(" kreuze Loesung " + repr(self.Id))
	def laden(self):
		print(" lade Loesung " + repr(self.Id))
	def speichern(self):
		print(" speichere Loesung " + repr(self.Id))
	def __repr__(self):
		return ' L '+ repr(self.Id)
	def __eq__(self,other):
		return self.value == other.value
	def __lt__(self,other):
		return self.value < other.value

class Zyklus:
	"""Definiert wie ein Zyklus ablaufen soll:
	
		wann sollen Objekte generiert, gekreuzt, mutiert und selektiert werden
		wieviele sind davon jeweils betroffen
	"""
	def __init__(self, 	Name='default',
						Abfolge='e,m,s',
						Anzahl='10,10,10'):
		self.Name = Name
		self.Abfolge = Abfolge.split(',')
		self.Anzahl  = Anzahl.split(',')
	def __repr__(self):
		return( "Zyklus: %s\n"
				" Zyklusabfolge: %s\n"
				" Anzahl       : %s\n"
		% ( self.Name,
			self.Abfolge,
			self.Anzahl ) )
	def laden(self, FileName, ZyklusName):
		config = configparser.ConfigParser()
		ZyklusPath= os.path.split( FileName ) 
		ZyklusPath= os.path.join( FileName ) 
		config.read( ZyklusPath )
		self.Name = ZyklusName
		Abfolge = _StringAusdruck( config.get( ZyklusName, 'Abfolge' ))
		Anzahl  = _StringAusdruck( config.get( ZyklusName, 'Anzahl' ))
		self.Abfolge = Abfolge.split(',')
		self.Anzahl = Anzahl.split(',')
		pass

class Entwicklung:
	"""Beobachtungsobjekt: zeichnet den Verlauf der Loesungsfindung auf.

	   Nach jeder Zyklus-Aktion wird ein Messpunkt erfasst (bester,
	   durchschnittlicher und schlechtester Wert im aktuellen Pool).
	   alsSVG() zeichnet daraus eine Verlaufsgrafik: Zeitachse (Aktions-
	   index) nach rechts, Wertigkeit auf der Y-Achse. Domaenen-agnostisch;
	   funktioniert mit jeder Loesung, die ein Attribut .value hat.
	"""
	# Buchstaben der Zyklus-Abfolge -> lesbare Namen fuer die Beschriftung
	AktionsNamen = {
		'e': 'erzeugen', 'm': 'mutieren', 's': 'selektieren',
		'S': 'zufaellig', 'j': 'behalten', 'z': 'Generation' }

	def __init__(self):
		# je Messpunkt: (Schrittindex, Aktion, bester, schnitt, schlechtester, N)
		self.Punkte = []

	def erfassen(self, Aktion, Pool):
		"""einen Messpunkt aus dem aktuellen Pool aufnehmen"""
		if not Pool:
			return
		Werte = [ L.value for L in Pool ]
		self.Punkte.append( (
			len(self.Punkte),
			Aktion,
			max(Werte),
			sum(Werte) / len(Werte),
			min(Werte),
			len(Werte) ) )

	def __repr__(self):
		r = "-- Entwicklung (%d Messpunkte):\n" % len(self.Punkte)
		for (i, Aktion, best, schnitt, schlecht, n) in self.Punkte:
			r = r + ("  %3d %-12s best=%s schnitt=%.1f schlecht=%s (N=%d)\n"
			         % (i, self.AktionsNamen.get(Aktion, Aktion),
			            best, schnitt, schlecht, n))
		return r

	def alsSVG(self, OutputSVG, Breite=900, Hoehe=500, Rand=60):
		"""zeichnet den aufgezeichneten Verlauf als SVG-Liniendiagramm"""
		if not self.Punkte:
			raise ValueError("keine Messpunkte aufgezeichnet")

		AlleWerte = []
		for (_i, _a, best, schnitt, schlecht, _n) in self.Punkte:
			AlleWerte.extend( (best, schnitt, schlecht) )
		WMin, WMax = min(AlleWerte), max(AlleWerte)
		if WMax == WMin:
			WMax = WMin + 1		# flache Linie nicht auf den Rand kleben

		N = len(self.Punkte)
		PlotBreite = Breite - 2 * Rand
		PlotHoehe = Hoehe - 2 * Rand

		def X(i):
			if N == 1:
				return Rand + PlotBreite / 2
			return Rand + PlotBreite * i / (N - 1)

		def Y(wert):
			# hoher (besserer) Wert oben, niedriger unten
			return Rand + PlotHoehe * (WMax - wert) / (WMax - WMin)

		def Polyline(Index, Farbe, Beschriftung):
			punkte = ' '.join( '%.1f,%.1f' % (X(i), Y(p[Index]))
			                   for i, p in enumerate(self.Punkte) )
			return ('<polyline points="%s" fill="none" stroke="%s" '
			        'stroke-width="2"/>\n' % (punkte, Farbe), Beschriftung, Farbe)

		Teile = []
		Teile.append( '<?xml version="1.0" encoding="UTF-8"?>\n' )
		Teile.append(
			'<svg xmlns="http://www.w3.org/2000/svg" width="%d" height="%d" '
			'viewBox="0 0 %d %d" font-family="sans-serif">\n'
			% (Breite, Hoehe, Breite, Hoehe) )
		Teile.append( '<rect width="100%" height="100%" fill="white"/>\n' )
		Teile.append(
			'<text x="%d" y="30" text-anchor="middle" font-size="16" '
			'fill="#333">Entwicklung der Loesungsfindung</text>\n'
			% (Breite // 2) )

		# Achsen
		Teile.append(
			'<line x1="%d" y1="%d" x2="%d" y2="%d" stroke="#333" '
			'stroke-width="1.5"/>\n' % (Rand, Rand, Rand, Hoehe - Rand) )
		Teile.append(
			'<line x1="%d" y1="%d" x2="%d" y2="%d" stroke="#333" '
			'stroke-width="1.5"/>\n'
			% (Rand, Hoehe - Rand, Breite - Rand, Hoehe - Rand) )

		# Y-Gitter + Beschriftung (5 Linien)
		for k in range(6):
			wert = WMin + (WMax - WMin) * k / 5
			y = Y(wert)
			Teile.append(
				'<line x1="%d" y1="%.1f" x2="%d" y2="%.1f" stroke="#eee" '
				'stroke-width="1"/>\n' % (Rand, y, Breite - Rand, y) )
			Teile.append(
				'<text x="%d" y="%.1f" text-anchor="end" font-size="11" '
				'fill="#666">%.0f</text>\n' % (Rand - 8, y + 4, wert) )
		Teile.append(
			'<text x="18" y="%d" text-anchor="middle" font-size="12" '
			'fill="#333" transform="rotate(-90 18 %d)">Wertigkeit '
			'(hoeher = besser)</text>\n' % (Hoehe // 2, Hoehe // 2) )
		Teile.append(
			'<text x="%d" y="%d" text-anchor="middle" font-size="12" '
			'fill="#333">Zyklus-Aktion (Zeit)</text>\n'
			% (Breite // 2, Hoehe - 18) )

		# X-Beschriftung: Aktionsbuchstaben (bei wenigen Punkten alle,
		# sonst ausgeduennt, damit die Achse lesbar bleibt)
		Schritt = max(1, N // 30)
		for i, p in enumerate(self.Punkte):
			if i % Schritt == 0 or i == N - 1:
				Teile.append(
					'<text x="%.1f" y="%d" text-anchor="middle" font-size="10" '
					'fill="#999">%s</text>\n' % (X(i), Hoehe - Rand + 16, p[1]) )

		# die drei Kurven
		Kurven = [
			Polyline(2, '#2f6f4f', 'beste Loesung'),
			Polyline(3, '#c08a2d', 'Durchschnitt'),
			Polyline(4, '#a03030', 'schlechteste Loesung') ]
		for (svg, _besch, _farbe) in Kurven:
			Teile.append( svg )

		# Legende oben rechts
		LegX = Breite - Rand - 160
		LegY = Rand + 10
		for k, (_svg, besch, farbe) in enumerate(Kurven):
			y = LegY + k * 18
			Teile.append(
				'<line x1="%d" y1="%d" x2="%d" y2="%d" stroke="%s" '
				'stroke-width="2"/>\n' % (LegX, y, LegX + 24, y, farbe) )
			Teile.append(
				'<text x="%d" y="%d" font-size="11" fill="#333">%s</text>\n'
				% (LegX + 30, y + 4, besch) )

		Teile.append( '</svg>\n' )

		fout = open( OutputSVG, 'w' )
		fout.writelines( Teile )
		fout.close()


class Farm:
	"""enthaelt und pflegt Loesungsobjekte, kreuzt, mutiert, selektiert sie"""
	def __init__(self, Zyklus, Klasse, *Arguments, **keywords):
		"""initialisiere die Farm

		mit einem definerten Zyklus fuer die Zucht und dem gewunschten zu Zuchtobjekt

		Beobachtung: self.Entwicklung zeichnet nach jeder Zyklus-Aktion
		den Pool-Zustand auf und laesst sich per alsSVG() visualisieren.
		"""
		self.Position = 0
		self.Pool = []
		self.PoolNeu = []
		self.Zyklus = Zyklus
		self.Klasse = Klasse
		self.Entwicklung = Entwicklung()
		for Aktion, Zahl in zip(Zyklus.Abfolge,Zyklus.Anzahl):
			#print 'Aktion:',Aktion
			#print 'Zahl:',Zahl
			self.Position = self.Position + 1
			if   Aktion == 'e':
				self.LoesungenErzeugen(int(Zahl), Klasse, *Arguments, **keywords)
			elif Aktion == 'm':
				self.LoesungenMutieren(int(Zahl))
			elif Aktion == 's':
				self.LoesungenAuswaehlen(int(Zahl))
			elif Aktion == 'S':
				self.LoesungenZufaelligAuswaehlen(int(Zahl))
			elif Aktion == 'j':
				self.LoesungenBehalten()
			elif Aktion == 'z':
				self.NeueGeneration()
			else:
				pass
			# nach jeder Aktion den aktuellen Pool-Zustand aufzeichnen.
			# Der beobachtete Pool ist der, in dem gerade die Loesungen
			# liegen: PoolNeu, sobald dort etwas gesammelt wurde, sonst Pool.
			self.Entwicklung.erfassen(
				Aktion, self.PoolNeu if self.PoolNeu else self.Pool )
	def LoesungenErzeugen(self, N, Klasse, *Arguments, **keywords):
		""" Erzeuge Loesungen"""
		for i in range(N):
			L = Klasse( *Arguments, **keywords )
			self.Pool.append(L)
		#print(repr(len(self.Pool)) + ' Loesungsobjekte erzeugt')
	def LoesungenKreuzen(self,N):
		"""mache aus zwei Loesungen eine neue"""
		#print("+ Loesungen kreuzen")
		for i in range(N):
			self.Pool.append(self.Pool[i].kreuzen())
	def LoesungenMutieren(self, N):
		"""veraendere die Loesung, so dass sie hoffentlich besser wird"""
		print("+ Loesungen mutieren: ", N)
		for n in range(N):
			#print(" -"+repr(n+1)+" mal")
			for L in chain(self.Pool):
				O = deepcopy(L)
				self.PoolNeu.append(O.mutieren())
		#print(self)
	def LoesungenBehalten(self):
		"""Behalte alle Eltern aus dem alten Pool"""
		print("+ alten Loesungspool behalten")
		self.PoolNeu = self.PoolNeu + self.Pool
		#print(self)
	def NeueGeneration(self):
		"""Loesche alle bisherigen Eltern und mache einen neuen Zyklus"""
		print("+ Neue Generation erzeugen")
		self.Pool = self.PoolNeu
		self.PoolNeu = []
		#print(self)
		#print("---------------")
	def LoesungenAuswaehlen(self, N):
		"""Selektiere die besten Loesungen"""
		print("+ die besten -", repr(N) +" - Loesungen auswaehlen -")
		for i in range(N):
			L = max(self.Pool)
			self.PoolNeu.append(L)
			self.Pool.remove(L)
		#print(self)
	def LoesungenZufaelligAuswaehlen(self, N):
		"""Selektiere zufaellig Loesungen"""
		print("+ zufaellig auswaehlen -"+ repr(N) + "-")
		for i in range(N):
			self.PoolNeu.append(choice(self.Pool))
		#print(self)
	def __repr__(self):
		"""Drucke die Farm am Bildschirm aus"""
		return( " gewaehlte Objekte (%d): %s\n"
				" Elternobjekte (%d)    : %s\n"
		% ( len(self.PoolNeu),  self.PoolNeu,
			len(self.Pool), self.Pool))
	def laden(self):
		print("+ laden")
	def speichern(self):
		print("+ speichern")
	def Bester(self):
		return  max(self.Pool)


if __name__ == '__main__':
	print("\n-- Erzeuge Zyklus:")
	#Z = Zyklus( Name='Easy',
				#Abfolge='e,s,+,m,j,S,+',
				#Anzahl='10,5,x,2,x,5,x'
				#)
	Z = Zyklus( Name='Easy',
				Abfolge='e,'+'s,z,m,j,z,'*3+'s',
				Anzahl='10,'+'5,x,2,x,x,'*3+'1'
				)
	print(Z)

	print("\n-- Erzeuge Loesungen:")
	L1 = Loesung()
	L2 = Loesung()
	L3 = Loesung()
	print(L1, L2, L3)
	print('L1 > L2 =', L1 > L2)
	print('L2 > L3 =', L2 > L3)
	print('L1 > L3 =', L1 > L3)

	A=2
	B=3

	print("\n-- Erzeuge Farm:")
	F1 = Farm( Z, Loesung, Arg1=A, Arg2=B, Arg3=L1 )
	print(F1)

else:
	pass