1 | from subprocess import Popen, PIPE, check_output
|
---|
2 | from enum import Enum
|
---|
3 | from typing import List
|
---|
4 | import json
|
---|
5 | import sys, os
|
---|
6 | import argparse
|
---|
7 | import numpy as np
|
---|
8 |
|
---|
9 |
|
---|
10 | class FramsticksCLI:
|
---|
11 | """Runs Framsticks CLI (command-line) executable and communicates with it using standard input and output.
|
---|
12 | You can perform basic operations like mutation, crossover, and evaluation of genotypes.
|
---|
13 | This way you can perform evolution controlled by python, access and manipulate genotypes.
|
---|
14 | You can even design and use in evolution your own genetic representation implemented entirely in python.
|
---|
15 |
|
---|
16 | You need to provide one or two parameters when you run this class: the path to Framsticks CLI
|
---|
17 | and the name of the Framsticks CLI executable (if it is non-standard). See::
|
---|
18 | FramsticksCLI.py -h"""
|
---|
19 |
|
---|
20 | PRINT_FRAMSTICKS_OUTPUT: bool = False # set to True for debugging
|
---|
21 | DETERMINISTIC: bool = False # set to True to have the same results on each run
|
---|
22 |
|
---|
23 | GENO_SAVE_FILE_FORMAT = Enum('GENO_SAVE_FILE_FORMAT', 'NATIVEFRAMS RAWGENO') # how to save genotypes
|
---|
24 | OUTPUT_DIR = "scripts_output"
|
---|
25 | STDOUT_ENDOPER_MARKER = "FileObject.write" # we look for this message on Framsticks CLI stdout to detect when Framsticks created a file with the result we expect
|
---|
26 |
|
---|
27 | FILE_PREFIX = 'framspy_'
|
---|
28 |
|
---|
29 | RANDOMIZE_CMD = "rnd" + "\n"
|
---|
30 | SETEXPEDEF_CMD = "expdef standard-eval" + "\n"
|
---|
31 | GETSIMPLEST_CMD = "getsimplest"
|
---|
32 | GETSIMPLEST_FILE = FILE_PREFIX + "simplest.gen"
|
---|
33 | EVALUATE_CMD = "evaluate eval-allcriteria.sim "
|
---|
34 | EVALUATE_FILE = "genos_eval.json"
|
---|
35 | CROSSOVER_CMD = "crossover"
|
---|
36 | CROSSOVER_FILE = FILE_PREFIX + "child.gen"
|
---|
37 | DISSIMIL_CMD = "dissimil"
|
---|
38 | DISSIMIL_FILE = FILE_PREFIX + "dissimilarity_matrix.gen"
|
---|
39 | ISVALID_CMD = "isvalid"
|
---|
40 | ISVALID_FILE = FILE_PREFIX + "validity.gen"
|
---|
41 | MUTATE_CMD = "mutate"
|
---|
42 | MUTATE_FILE = FILE_PREFIX + "mutant.gen"
|
---|
43 |
|
---|
44 | CLI_INPUT_FILE = FILE_PREFIX + "genotypes.gen"
|
---|
45 |
|
---|
46 |
|
---|
47 | def __init__(self, framspath, framsexe):
|
---|
48 | self.frams_path = framspath
|
---|
49 | self.frams_exe = framsexe if framsexe is not None else 'frams.exe' if os.name == "nt" else 'frams.linux'
|
---|
50 | self.writing_path = None
|
---|
51 | mainpath = os.path.join(self.frams_path, self.frams_exe)
|
---|
52 | exe_call = [mainpath, '-Q', '-s', '-c', '-icliutils.ini'] # -c will be ignored in Windows Framsticks (this option is meaningless because the Windows version does not support color console, so no need to deactivate this feature using -c)
|
---|
53 | exe_call_to_get_version = [mainpath, '-V']
|
---|
54 | exe_call_to_get_path = [mainpath, '-?']
|
---|
55 | try:
|
---|
56 | print("\n".join(self.__readAllOutput(exe_call_to_get_version)))
|
---|
57 | help = self.__readAllOutput(exe_call_to_get_path)
|
---|
58 | for helpline in help:
|
---|
59 | if 'dDIRECTORY' in helpline:
|
---|
60 | self.writing_path = helpline.split("'")[1]
|
---|
61 | except FileNotFoundError:
|
---|
62 | print("Could not find Framsticks executable ('%s') in the given location ('%s')." % (self.frams_exe, self.frams_path))
|
---|
63 | sys.exit(1)
|
---|
64 | print("Temporary files with results will be saved in detected writable working directory '%s'" % self.writing_path)
|
---|
65 | self.__spawnFramsticksCLI(exe_call)
|
---|
66 |
|
---|
67 |
|
---|
68 | def __readAllOutput(self, command):
|
---|
69 | frams_process = Popen(command, stdout=PIPE, stderr=PIPE, stdin=PIPE)
|
---|
70 | return [line.decode('utf-8').rstrip() for line in iter(frams_process.stdout.readlines())]
|
---|
71 |
|
---|
72 |
|
---|
73 | def __spawnFramsticksCLI(self, args):
|
---|
74 | # the child app (Framsticks CLI) should not buffer outputs and we need to immediately read its stdout, hence we use pexpect/wexpect
|
---|
75 | print('Spawning Framsticks CLI for continuous stdin/stdout communication... ', end='')
|
---|
76 | if os.name == "nt": # Windows:
|
---|
77 | import wexpect # https://pypi.org/project/wexpect/
|
---|
78 | # https://github.com/raczben/wexpect/tree/master/examples
|
---|
79 | self.child = wexpect.spawn(' '.join(args))
|
---|
80 | else:
|
---|
81 | import pexpect # https://pexpect.readthedocs.io/en/stable/
|
---|
82 | self.child = pexpect.spawn(' '.join(args))
|
---|
83 | self.child.setecho(False) # linux only
|
---|
84 | print('OK.')
|
---|
85 |
|
---|
86 | self.__readFromFramsCLIUntil("UserScripts.autoload")
|
---|
87 | print('Performing a basic test 1/3... ', end='')
|
---|
88 | assert self.getSimplest("1") == "X"
|
---|
89 | print('OK.')
|
---|
90 | print('Performing a basic test 2/3... ', end='')
|
---|
91 | assert self.isValid("X[0:0]") == True
|
---|
92 | print('OK.')
|
---|
93 | print('Performing a basic test 3/3... ', end='')
|
---|
94 | assert self.isValid("X[0:0],") == False
|
---|
95 | print('OK.')
|
---|
96 | if not self.DETERMINISTIC:
|
---|
97 | self.child.sendline(self.RANDOMIZE_CMD)
|
---|
98 | self.child.sendline(self.SETEXPEDEF_CMD)
|
---|
99 |
|
---|
100 |
|
---|
101 | def closeFramsticksCLI(self):
|
---|
102 | # End gracefully by sending end-of-file character: ^Z or ^D
|
---|
103 | # Without -Q argument ("quiet mode"), Framsticks CLI would print "Shell closed." for goodbye.
|
---|
104 | self.child.sendline(chr(26 if os.name == "nt" else 4))
|
---|
105 |
|
---|
106 |
|
---|
107 | def __saveGenotypeToFile(self, genotype, name, mode):
|
---|
108 | outpath = os.path.join(self.writing_path, name)
|
---|
109 | outfile = open(outpath, mode)
|
---|
110 | outfile.write("org:\n")
|
---|
111 | outfile.write("genotype:~\n")
|
---|
112 | outfile.write(genotype + "~\n\n") # TODO proper quoting of special characters in genotype...
|
---|
113 | outfile.close()
|
---|
114 | return name
|
---|
115 |
|
---|
116 |
|
---|
117 | def __saveToFile(self, genotype, name, mode):
|
---|
118 | outpath = os.path.join(self.writing_path, name)
|
---|
119 | outfile = open(outpath, mode)
|
---|
120 | outfile.write(genotype)
|
---|
121 | outfile.close()
|
---|
122 | return name
|
---|
123 |
|
---|
124 |
|
---|
125 | def __removeFile(self, path):
|
---|
126 | filepath = os.path.join(self.writing_path, path)
|
---|
127 | if os.path.exists(filepath):
|
---|
128 | os.remove(filepath)
|
---|
129 |
|
---|
130 |
|
---|
131 | def __readFromFramsCLIUntil(self, until_marker: str):
|
---|
132 | while True:
|
---|
133 | self.child.expect('\n')
|
---|
134 | msg = str(self.child.before)
|
---|
135 | if self.PRINT_FRAMSTICKS_OUTPUT or msg.startswith("[ERROR]"):
|
---|
136 | print(msg)
|
---|
137 | if until_marker in msg:
|
---|
138 | break
|
---|
139 |
|
---|
140 |
|
---|
141 | def __runCommand(self, command, genotypes, result_file_name, saveformat) -> List[str]:
|
---|
142 | filenames = [] # list of file names with input data for the command
|
---|
143 | if saveformat == self.GENO_SAVE_FILE_FORMAT["RAWGENO"]:
|
---|
144 | for i in range(len(genotypes)):
|
---|
145 | filenames.append(self.__saveToFile(genotypes[i], "genotype" + str(i) + ".gen", "w")) # plain text format = must have a separate file for each genotype
|
---|
146 | elif saveformat == self.GENO_SAVE_FILE_FORMAT["NATIVEFRAMS"]:
|
---|
147 | self.__removeFile(self.CLI_INPUT_FILE) # ensure there is nothing left from the last run of the program because we "a"ppend to file in the loop below
|
---|
148 | for i in range(len(genotypes)):
|
---|
149 | outfilename = self.__saveGenotypeToFile(genotypes[i], self.CLI_INPUT_FILE, "a")
|
---|
150 | filenames.append(outfilename) # since we use the same file in the loop above, add this file only once (i.e., outside of the loop)
|
---|
151 |
|
---|
152 | if result_file_name != self.EVALUATE_FILE: # all functions except for evaluate provide frams with the file name to write to
|
---|
153 | self.child.sendline(command + " " + " ".join(filenames) + " " + result_file_name + "\n")
|
---|
154 | else:
|
---|
155 | self.child.sendline(command + " " + " ".join(filenames) + "\n")
|
---|
156 | self.__readFromFramsCLIUntil(self.STDOUT_ENDOPER_MARKER)
|
---|
157 | filenames.append(os.path.join(self.writing_path, self.OUTPUT_DIR, result_file_name))
|
---|
158 | return filenames # last element is a path to the file containing results
|
---|
159 |
|
---|
160 |
|
---|
161 | def __cleanUpCommandResults(self, filepaths):
|
---|
162 | """Deletes files with results created by the command."""
|
---|
163 | for i in filepaths:
|
---|
164 | if i == filepaths[-1]:
|
---|
165 | os.remove(i) # the result is written with its full path and we have used it before so the file surely exists
|
---|
166 | else:
|
---|
167 | self.__removeFile(i)
|
---|
168 |
|
---|
169 |
|
---|
170 | def getSimplest(self, genetic_format) -> str:
|
---|
171 | assert len(genetic_format) == 1, "Genetic format should be a single character"
|
---|
172 | files = self.__runCommand(self.GETSIMPLEST_CMD + " " + genetic_format + " ", [], self.GETSIMPLEST_FILE, self.GENO_SAVE_FILE_FORMAT["RAWGENO"])
|
---|
173 | with open(files[-1]) as f:
|
---|
174 | genotype = "".join(f.readlines())
|
---|
175 | self.__cleanUpCommandResults(files)
|
---|
176 | return genotype
|
---|
177 |
|
---|
178 |
|
---|
179 | def evaluate(self, genotype: str):
|
---|
180 | """
|
---|
181 | Returns:
|
---|
182 | Dictionary -- genotype evaluated with self.EVALUATE_COMMAND. Note that for whatever reason (e.g. incorrect genotype),
|
---|
183 | the dictionary you will get may be empty or partially empty and may not have the fields you expected, so handle such cases properly.
|
---|
184 | """
|
---|
185 | files = self.__runCommand(self.EVALUATE_CMD, [genotype], self.EVALUATE_FILE, self.GENO_SAVE_FILE_FORMAT["NATIVEFRAMS"])
|
---|
186 | with open(files[-1]) as f:
|
---|
187 | data = json.load(f)
|
---|
188 | if len(data) > 0:
|
---|
189 | self.__cleanUpCommandResults(files)
|
---|
190 | return data
|
---|
191 | else:
|
---|
192 | print("Evaluating genotype: no performance data was returned in", self.EVALUATE_FILE) # we do not delete files here
|
---|
193 | return None
|
---|
194 |
|
---|
195 |
|
---|
196 | def mutate(self, genotype: str) -> str:
|
---|
197 | files = self.__runCommand(self.MUTATE_CMD, [genotype], self.MUTATE_FILE, self.GENO_SAVE_FILE_FORMAT["RAWGENO"])
|
---|
198 | with open(files[-1]) as f:
|
---|
199 | newgenotype = "".join(f.readlines())
|
---|
200 | self.__cleanUpCommandResults(files)
|
---|
201 | return newgenotype
|
---|
202 |
|
---|
203 |
|
---|
204 | def crossOver(self, genotype1: str, genotype2: str) -> str:
|
---|
205 | files = self.__runCommand(self.CROSSOVER_CMD, [genotype1, genotype2], self.CROSSOVER_FILE, self.GENO_SAVE_FILE_FORMAT["RAWGENO"])
|
---|
206 | with open(files[-1]) as f:
|
---|
207 | child_genotype = "".join(f.readlines())
|
---|
208 | self.__cleanUpCommandResults(files)
|
---|
209 | return child_genotype
|
---|
210 |
|
---|
211 |
|
---|
212 | def dissimilarity(self, genotype1: str, genotype2: str) -> float:
|
---|
213 | files = self.__runCommand(self.DISSIMIL_CMD, [genotype1, genotype2], self.DISSIMIL_FILE, self.GENO_SAVE_FILE_FORMAT["NATIVEFRAMS"])
|
---|
214 | with open(files[-1]) as f:
|
---|
215 | dissimilarity_matrix = np.genfromtxt(f, dtype=np.float64, comments='#', encoding=None, delimiter='\t')
|
---|
216 | # We would like to skip column #1 while reading and read everything else, but... https://stackoverflow.com/questions/36091686/exclude-columns-from-genfromtxt-with-numpy
|
---|
217 | # This would be too complicated, so strings (names) in column #1 become NaN as floats (unless they accidentally are valid numbers) - not great, not terrible
|
---|
218 | EXPECTED_SHAPE = (2, 4)
|
---|
219 | assert dissimilarity_matrix.shape == EXPECTED_SHAPE, f"Not a correct dissimilarity matrix, expected {EXPECTED_SHAPE} "
|
---|
220 | for i in range(len(dissimilarity_matrix)):
|
---|
221 | assert dissimilarity_matrix[i][i + 2] == 0, "Not a correct dissimilarity matrix, diagonal expected to be 0"
|
---|
222 | assert dissimilarity_matrix[0][3] == dissimilarity_matrix[1][2], "Probably not a correct dissimilarity matrix, expecting symmetry, verify this"
|
---|
223 | self.__cleanUpCommandResults(files)
|
---|
224 | return dissimilarity_matrix[0][3]
|
---|
225 |
|
---|
226 |
|
---|
227 | def isValid(self, genotype: str) -> bool:
|
---|
228 | files = self.__runCommand(self.ISVALID_CMD, [genotype], self.ISVALID_FILE, self.GENO_SAVE_FILE_FORMAT["RAWGENO"])
|
---|
229 | with open(files[-1]) as f:
|
---|
230 | valid = f.readline() == "1"
|
---|
231 | self.__cleanUpCommandResults(files)
|
---|
232 | return valid
|
---|
233 |
|
---|
234 |
|
---|
235 | def parseArguments():
|
---|
236 | parser = argparse.ArgumentParser(description='Run this program with "python -u %s" if you want to disable buffering of its output.' % sys.argv[0])
|
---|
237 | parser.add_argument('-path', type=ensureDir, required=True, help='Path to Framsticks CLI without trailing slash.')
|
---|
238 | parser.add_argument('-exe', required=False, help='Executable name. If not given, "frams.exe" or "frams.linux" is assumed.')
|
---|
239 | parser.add_argument('-genformat', required=False, help='Genetic format for the demo run, for example 4, 9, or B. If not given, f1 is assumed.')
|
---|
240 | return parser.parse_args()
|
---|
241 |
|
---|
242 |
|
---|
243 | def ensureDir(string):
|
---|
244 | if os.path.isdir(string):
|
---|
245 | return string
|
---|
246 | else:
|
---|
247 | raise NotADirectoryError(string)
|
---|
248 |
|
---|
249 |
|
---|
250 | if __name__ == "__main__":
|
---|
251 | # A demo run.
|
---|
252 |
|
---|
253 | # TODO ideas:
|
---|
254 | # - check_validity with three levels (invalid, corrected, valid)
|
---|
255 | # - "vectorize" some operations (isvalid, evaluate) so that a number of genotypes is handled in one call
|
---|
256 | # - use threads for non-blocking reading from frams' stdout and thus not relying on specific strings printed by frams
|
---|
257 | # - a pool of binaries run at the same time, balance load - in particular evaluation
|
---|
258 | # - if we read genotypes in "org:" format anywhere: import https://pypi.org/project/framsreader/0.1.2/ and use it if successful,
|
---|
259 | # if not then print a message "framsreader not available, using simple internal method to save a genotype" and proceed as it is now.
|
---|
260 | # So far we don't read, but we should use the proper writer to handle all special cases like quoting etc.
|
---|
261 |
|
---|
262 | parsed_args = parseArguments()
|
---|
263 | framsCLI = FramsticksCLI(parsed_args.path, parsed_args.exe)
|
---|
264 |
|
---|
265 | simplest = framsCLI.getSimplest('1' if parsed_args.genformat is None else parsed_args.genformat)
|
---|
266 | print("\tSimplest genotype:", simplest)
|
---|
267 | parent1 = framsCLI.mutate(simplest)
|
---|
268 | parent2 = parent1
|
---|
269 | MUTATE_COUNT = 10
|
---|
270 | for x in range(MUTATE_COUNT): # example of a chain of 20 mutations
|
---|
271 | parent2 = framsCLI.mutate(parent2)
|
---|
272 | print("\tParent1 (mutated simplest):", parent1)
|
---|
273 | print("\tParent2 (Parent1 mutated %d times):" % MUTATE_COUNT, parent2)
|
---|
274 | offspring = framsCLI.crossOver(parent1, parent2)
|
---|
275 | print("\tCrossover (Offspring):", offspring)
|
---|
276 | print('\tDissimilarity of Parent1 and Offspring:', framsCLI.dissimilarity(offspring, parent1))
|
---|
277 | print('\tPerformance of Offspring:', framsCLI.evaluate(offspring))
|
---|
278 | print('\tValidity of Offspring:', framsCLI.isValid(offspring))
|
---|
279 |
|
---|
280 | framsCLI.closeFramsticksCLI()
|
---|