1"""ICE40 ASC-file circuit implementation.
2
3Provides :class:`FileBasedCircuit`, which reads and writes bitstreams in the
4IceStorm ASCII (``.asc``) format, compiles them via ``icepack``, and uploads
5to hardware via ``iceprog``. Uses memory-mapped file I/O for performance.
6
7.. note::
8 Magic tile coordinates and routing types are specific to the ICE40 HX1K.
9 Different FPGA models will require different constants.
10"""
11
12from mmap import mmap
13from pathlib import Path
14from shutil import copyfile
15from subprocess import run
16import os
17from time import sleep
18from BitstreamEvolutionProtocols import FPGA_Compilation_Data
19from Circuit.Circuit import Circuit
20from Directories import Directories
21from Logger import Logger
22from returns.result import Result, Success, Failure # type: ignore
23
24COMPILE_CMD = "icepack"
25RUN_CMD = "iceprog"
26
[docs]
27class FileBasedCircuit(Circuit):
28 """
29 Represents a Circuit that is based on an ASC file, in a format
30 ready to be compiled by IceStorm tools.
31 Provides useful methods for working with hardware files
32 """
33
34 def __init__(self, *, index: int, filename: str, template: Path, logger: Logger, directories: Directories,
35 routing_type: str, accessed_columns: list[int]):
36 Circuit.__init__(self, index, filename, logger)
37 self.__routing_type = routing_type
38 self.__accessed_columns = accessed_columns
39
40 asc_dir = directories.asc_dir
41 bin_dir = directories.bin_dir
42 data_dir = directories.data_dir
43 self.__hardware_filepath = asc_dir.joinpath(filename + ".asc")
44 self.__bitstream_filepath = bin_dir.joinpath(filename + ".bin")
45
46 # Create directories if they don't exist
47 os.makedirs(asc_dir, exist_ok=True)
48 os.makedirs(bin_dir, exist_ok=True)
49 os.makedirs(data_dir, exist_ok=True)
50
51 # NOTE Using log files instead of a data buffer in the event of premature termination
52 self._data_filepath = data_dir.joinpath(filename + ".log")
53 # Create the data file if it doesn't exist
54 open(self._data_filepath, "w+").close()
55
56 if template:
57 copyfile(template, self.__hardware_filepath)
58
59 # Since the hardware file is written to and read from a lot, we
60 # mmap it to improve preformance.
61 hardware_file = open(self.__hardware_filepath, "r+")
62 self._hardware_file = mmap(hardware_file.fileno(), 0)
63 hardware_file.close()
64
[docs]
65 def copy_from(self, other: 'FileBasedCircuit'):
66 # Close our mmap so the file can be overwritten (required on Windows)
67 self._hardware_file.close()
68 copyfile(other.__hardware_filepath, self.__hardware_filepath)
69 # Re-open and re-mmap
70 hardware_file = open(self.__hardware_filepath, "r+")
71 self._hardware_file = mmap(hardware_file.fileno(), 0)
72 hardware_file.close()
73
[docs]
74 def compile(self, fpga: FPGA_Compilation_Data) -> Result[None,Exception]:
75 """
76 Compiles and uploads the compiled circuit and runs it on the FPGA
77 """
78
79 self.__compile()
80
81 cmd_str = [
82 RUN_CMD,
83 self.__bitstream_filepath,
84 "-d",
85 fpga.id
86 ]
87 print(cmd_str)
88 run(cmd_str)
89 sleep(1)
90
91 return Success(None)
92
93 # if switching fpgas every sample, need to upload to the second fpga also
94 # if self._config.get_transfer_sample():
95 # cmd_str = [
96 # RUN_CMD,
97 # self._bitstream_filepath,
98 # "-d",
99 # self._config.get_fpga2()
100 # ]
101 # print(cmd_str)
102 # run(cmd_str)
103 # sleep(1)
104
105 def __run_at_each_modifiable(self, lambda_func, hardware_file = None, accessible_columns = None,
106 routing_type=None):
107 """
108 Runs the lambda_func at every modifiable position
109 Args passed to lambda_func: value of bit (as a byte), row, col
110 If lambda_func returns a value (byte), then the bit is set to that number
111 If lambda_func returns None, then the bit is left unmodified
112 Keep in mind the bytes are the ASCII codes, so for example 49 = 1
113
114 .. warning::
115 Go over this with someone who can clarify what all of the data types are.
116
117 Parameters
118 ----------
119 lambda_func : Callable
120 This function is called for each modifiable bit. The bit value is passed into lambda_func.
121 If lambda_func returns None, the bit is left unmodified. If lambda_func returns another value,
122 then the bit at that position is replaced with the return value of lambda_func.
123 hardware_file : str | None
124 The path to the hardware file to read from/write to. If no value provided, uses this Circuit's hardware file.
125 accessible_columns : list[str] | None
126 The accessible columns. If no value provided, uses the current configuration value.
127 routing_type: str | None
128 The routing type (MOORE or NEWSE). If no value provided, uses the current configuration value.
129 """
130
131 if hardware_file is None:
132 hardware_file = self._hardware_file
133
134 if accessible_columns is None:
135 accessible_columns = self.__accessed_columns
136
137 if routing_type is None:
138 routing_type = self.__routing_type
139
140 # Set tile to the first location of the substring ".logic_tile"
141 # The b prefix makes the string an instance of the "bytes" type
142 # The .logic_tile header indicates that there is a tile, so the "tile" variable stores the starting point of the current tile
143 tile = hardware_file.find(b".logic_tile")
144
145 while tile > 0:
146 # Set pos to the position of this tile, but with the length of ".logic_tile" added so it is in front of where we have the x/y coords
147 pos = tile + len(".logic_tile")
148
149 # Check if the position is legal to modify
150 if self.__tile_is_included(hardware_file, pos):
151 # Find the start and end of the line; the positions of the \n newline just before and at the end of this line
152 # The start is the newline position + 1, so the first valid bit character
153 # line_size is self-explanatory
154 # This finds the length of a standard line of bits (so the width of each data-containing line in this tile)
155 line_start = hardware_file.find(b"\n", tile) + 1
156 line_end = hardware_file.find(b"\n", line_start + 1)
157 line_size = line_end - line_start + 1
158
159 # Determine which rows we can modify
160 # TODO ALIFE2021 The routing protocol here is dated and needs to mimic that of the Tone Discriminator
161 rows = []
162 if routing_type == "MOORE":
163 rows = [1, 2, 13]
164 elif routing_type == "NEWSE":
165 rows = [1, 2]
166 # Iterate over each row and the columns that we can access within each row
167 for row in rows:
168 for col in accessible_columns:
169 # This will get us to individual bits. If the mutation probability passes
170 # Our position is now going to be the start of the first line, plus the line size multiplied to get to our desired row,
171 # and finally added to the column (with the int cast to sanitize user input)
172 pos = line_start + line_size * (row - 1) + int(col)
173 bit_value = hardware_file[pos]
174 lambda_return = lambda_func(bit_value, row, col)
175 if lambda_return is not None:
176 # need to re-assign the bit
177 hardware_file[pos] = lambda_return
178
179 # Find the next logic tile, and start again
180 # Will return -1 if .logic_tile isn't found, and the while loop will exit
181 tile = hardware_file.find(b".logic_tile", tile + 1)
182
183 def __compile(self):
184 """
185 Compile circuit ASC file to a BIN file for hardware upload.
186 """
187 self.__log_event(2, "Compiling", self, "with icepack...")
188
189 # Ensure the file backing the mmap is up to date with the latest
190 # changes to the mmap.
191 self._hardware_file.flush()
192
193 compile_command = [
194 COMPILE_CMD,
195 self.__hardware_filepath,
196 self.__bitstream_filepath
197 ]
198 run(compile_command)
199
200 self.__log_event(2, "Finished compiling", self)
201
202 def __tile_is_included(self, hardware_file, pos):
203 """
204 Determines whether a given tile is available for modificiation.
205 NOTE: Tile = the .logic_tile in the asc file.
206
207 .. warning::
208 Preexisting todo: Replace magic values with a more generalized solution.
209 These magic values are indicative of the underlying hardware (ice40kh1k)
210
211 Parameters
212 ----------
213 hardware_file : mmap
214 Memory Mapped hardware file
215 pos : int
216 Index of the first byte in the .asc file for the hardware
217
218 Returns
219 -------
220 bool
221 True if the tile at that position is valid (The Tiles we can modigy)
222 """
223 # Replace these magic values with a more generalized solution
224 # Magic values are indicative of the underlying hardware (ice40hx1k)
225 # A different model will require different magic values (i.e. ice40hx8k)
226 VALID_TILE_X = range(4, 10)
227 VALID_TILE_Y = range(1, 17)
228
229 # NOTE x and y are stored as ints to aid the loops that search and identify
230 # tiles while scraping the asc files
231 # This is in the actual asc file; this is why we can simply pull from "pos"
232 # i.e. you'll see the header ".logic_file 1 1" - x=1, y=1
233
234 # This is where we had a fundamental issue before: The value in the hardware at this position is going to be an ASCII char value, not the actual
235 # number. Here, we parse the byte as an integer, then to a char, then back to an integer
236
237 # However, we have a great problem now: what about multi-digit numbers?
238 # Find the space that separates the x and y, and find the end of the line
239 # Then, grab the bytes for x, grab the bytes for y, convert to strings, and parse those strings
240 space_pos = hardware_file.find(b" ", pos + 1)
241 eol_pos = hardware_file.find(b"\n", pos)
242 x_bytes = hardware_file[pos:space_pos]
243 y_bytes = hardware_file[space_pos:eol_pos]
244 x_str = x_bytes.decode("utf-8").strip()
245 y_str = y_bytes.decode("utf-8").strip()
246 x = int(x_str)
247 y = int(y_str)
248 is_x_valid = x in VALID_TILE_X
249 is_y_valid = y in VALID_TILE_Y
250
251 return is_x_valid and is_y_valid
252
[docs]
253 def get_bitstream(self) -> list[bool]:
254 bitstream: list[bool] = []
255 def add_bit(bit: int, *rest):
256 if bit == 48:
257 bitstream.append(False)
258 elif bit == 49:
259 bitstream.append(True)
260 self.__run_at_each_modifiable(add_bit)
261 return bitstream
262
[docs]
263 def set_bitstream(self, bitstream: list[bool]):
264 index = 0
265 def get_bit(*rest):
266 nonlocal index
267 bit = bitstream[index]
268 index = index + 1
269 if bit:
270 return 49
271 else:
272 return 48
273 self.__run_at_each_modifiable(get_bit)
274
[docs]
275 @staticmethod
276 def get_file_attribute_st(mmapped_file, attribute):
277 '''
278 Returns the value of the stored attribute from the hardware file.
279 Circuits are capable of storing string name-value pairs in their hardware file, for purposes such as
280 tracking most recently-evaluated fitness of a Circuit
281 Static version of get_file_attribute that requires the memory-mapped file to be provided
282
283 Parameters
284 ----------
285 mmapped_file : mmap
286 The memory-mapped hardware file of the circuit
287 attribute : str
288 Attribute name to lookup
289
290 Returns
291 -------
292 str
293 File attribute value
294 '''
295 index = mmapped_file.find(b".comment FILE_ATTRIBUTES")
296 if index < 0:
297 return '0'
298 else:
299 newline_index = mmapped_file.find(b'\n', index)
300 searchable_area = mmapped_file[index:newline_index]
301 attr_index = searchable_area.find(bytes(attribute + '={', 'utf-8'))
302 if attr_index < 0: # Value doesn't exist yet
303 return '0'
304 attr_index = attr_index + len(attribute + "={")
305 end_index = searchable_area.find(b'}', attr_index)
306 value_bytes = searchable_area[attr_index:end_index]
307 return str(value_bytes, 'utf-8')
308
[docs]
309 @staticmethod
310 def set_file_attribute_st(hardware_file, attribute, value):
311 '''
312 Sets a Circuit's file attribute to the specified value
313 Circuits are capable of storing string name-value pairs in their hardware file, for purposes such as
314 tracking most recently-evaluated fitness of a Circuit
315 Static version of set_file_attribute that requires the memory-mapped file to be provided
316
317 Parameters
318 ----------
319 hardware_file : mmap
320 The memory-mapped hardware file of the Circuit
321 attribute : str
322 The name of the attribute to modify
323 value : str
324 The value to assign to the attribute
325 '''
326 # Check if the comment exists
327 #hardware_file = open(file_path, "r+")
328 mmapped_file = mmap(hardware_file.fileno(), 0)
329 index = mmapped_file.find(b".comment FILE_ATTRIBUTES")
330 if index < 0:
331 # Create the comment
332 comment_line = ".comment FILE_ATTRIBUTES " + attribute + "={" + value + "}\n"
333 # This requires re-mapping the self._hardware_file
334 #hardware_file = open(file_path, "r+")
335 content = hardware_file.read()
336 hardware_file.seek(0, 0)
337 hardware_file.write(comment_line + content)
338 else:
339 # Check if the attribute exists
340 end_index = mmapped_file.find(b'\n', index)
341 line = str(mmapped_file[index:end_index], 'utf-8')
342 attr_index = line.find(attribute + "={")
343 lines = hardware_file.readlines()
344 line_index = 0 # Index of the line that contains the attribute comment
345 for l in lines:
346 if l.find(".comment FILE_ATTRIBUTES") >= 0:
347 break
348 line_index = line_index + 1
349
350 if attr_index < 0:
351 # Attribute doesn't exist yet
352 line = line + " " + attribute + "={" + value + "}\n"
353 else:
354 attr_end_index = line.find('}', attr_index) + 2
355 before_attr = line[:attr_index]
356 after_attr = line[attr_end_index:]
357 line = before_attr + attribute + "={" + value + "} " + after_attr + '\n'
358 lines[line_index] = line
359 hardware_file.truncate(0)
360 hardware_file.seek(0)
361 hardware_file.writelines(lines)
362
[docs]
363 def get_file_attribute(self, attribute) -> str | None:
364 '''
365 Returns the value of the stored attribute for this Circuit
366 Circuits are capable of storing string name-value pairs in their hardware file, for purposes such as
367 tracking most recently-evaluated fitness of a Circuit
368
369 Parameters
370 ----------
371 attrbute : str
372 The name of the attribute of this circuit you want
373
374 Returns
375 -------
376 str
377 The value of the attribute
378 '''
379 return FileBasedCircuit.get_file_attribute_st(self._hardware_file, attribute)
380
[docs]
381 def set_file_attribute(self, attribute, value):
382 '''
383 Sets this Circuit's file attribute to the specified value
384 Circuits are capable of storing string name-value pairs in their hardware file, for purposes such as
385 tracking most recently-evaluated fitness of a Circuit
386
387 Parameters
388 ----------
389 attribute : str
390 The name of the attribute to modify
391 value : str
392 The value to assign to the attribute
393 '''
394 hardware_file = open(self.__hardware_filepath, "r+")
395 FileBasedCircuit.set_file_attribute_st(hardware_file, attribute, value)
396 # Re-map our hardware file
397 self._hardware_file = mmap(hardware_file.fileno(), 0)
398 hardware_file.close()
399
400 def __log_event(self, level, *event):
401 """
402 Emit an event-level log. This function is fulfilled through
403 the logger.
404 """
405 self._logger.log_event(level, *event)
