fixing gitlab?

fetch.py

+# filename: fetch.py
+import itertools
+from pydigital.memory import readmemh
+from pydigital.register import Register
+ 
+# the PC register
+PC = Register()
+ 
+# construct a memory segment for instruction memory
+# load the contents from the 32-bit fetch_test hex file (big endian)
+imem = readmemh('riscv_isa/programs/fetch_test.hex', word_size = 4, byteorder = 'big')
+ 
+def display():
+   if pc_val == None:
+       return "PC: xxxxxxxx, IR: xxxxxxxx"
+   else:
+       return f"PC: {pc_val:08x}, IR: {instr_val:08x}"
+ 
+startup = True
+# generate system clocks until we reach a stopping condition
+# this is basically the run function from the last lab
+for t in itertools.count():
+   # sample inputs
+   pc_val = PC.out()
+ 
+   # RESET the PC register
+   if startup:
+       PC.reset(imem.begin_addr)       
+       startup = False
+       print(f"{t:20d}:", display())
+       continue
+ 
+   # access instruction memory
+   instr_val = imem[pc_val]
+ 
+   # print one line at the end of the clock cycle
+   print(f"{t:20d}:", display())
+ 
+   # clock logic blocks, PC is the only clocked module!
+   # here the next pc value is always +4
+   PC.clock(4 + pc_val)
+ 
+   # check stopping conditions
+   if pc_val > 0x1100:
+       print("STOP -- PC is large! Is something wrong?")
+       break
+   if instr_val == 0:
+       print("Done -- end of program.")
+       break

fetch_decode.py

+# filename: fetch_decode.py
+import itertools
+from pydigital.memory import readmemh
+from pydigital.register import Register
+from riscv_isa import Instruction
+ 
+# the PC register
+PC = Register()
+ 
+# construct a memory segment for instruction memory
+# load the contents from the 32-bit fetch_test hex file (big endian)
+imem = readmemh('riscv_isa/programs/fetch_test.hex',
+   word_size = 4, byteorder = 'big')
+ 
+def display():
+   if pc_val == None:
+       return "PC: xxxxxxxx, IR: xxxxxxxx"
+   else:
+       return f"PC: {pc_val:08x}, IR: {instr.val:08x}, {instr}"
+ 
+startup = True
+# generate system clocks until we reach a stopping condition
+# this is basically the run function from the last lab
+for t in itertools.count():
+   # sample inputs
+   pc_val = PC.out()
+ 
+   # RESET the PC register
+   if startup:
+       PC.reset(imem.begin_addr)       
+       startup = False
+       print(f"{t:20d}:", display())
+       continue
+  
+   # access instruction memory
+   instr = Instruction(imem[pc_val], pc_val)
+ 
+   # print one line at the end of the clock cycle
+   print(f"{t:20d}:", display())
+ 
+   # clock logic blocks, PC is the only clocked module!
+   # here the next pc value is always +4
+   PC.clock(4 + pc_val)
+  
+   # check stopping conditions on NEXT instruction
+   if PC.out() > 0x1100:
+       print("STOP -- PC is large! Is something wrong?")
+       break
+   if imem[PC.out()] == 0:
+       print("Done -- end of program.")
+       break

pydigital/README.md

+# pydigital
+
+For simulating digital systems in python
+

pydigital/__init__.py

+from .system import System
+

pydigital/elfloader.py

+"""
+elfloader.py
+------------
+A wrapper for pyelftools https://github.com/eliben/pyelftools
+"""
+
+from elftools.elf.elffile import ELFFile
+from elftools.elf.enums import *
+from elftools.elf.sections import SymbolTableSection
+import elftools.elf.constants as elfconst
+
+from .memory import ELFMemory, MemorySegment
+
+class Elf():
+    """
+    Simplified ELF wrapper class, use with a context manager as:
+    with Elf('elffile') as foo:
+        ...
+    """
+    def __init__(self, elffilename, quiet = False):
+        self.elffilename = elffilename
+        self.quiet = quiet
+    def __enter__(self):
+        self.f = open(self.elffilename, 'rb')
+        self.ef = ELFFile(self.f)
+        self.byteorder = "little" if self.ef.little_endian else "big"
+
+        if not self.quiet:
+            print(f"Loading ELF binary \"{self.elffilename}\".")
+            print(f'{self.ef.get_machine_arch()} {self.byteorder} endian {self.ef["e_ident"]["EI_CLASS"]} has {self.ef.num_segments()} segments')
+        
+        if self.ef['e_ident']['EI_CLASS'] == 'ELFCLASS64':
+            self.bytes_per_word = 8
+        elif self.ef['e_ident']['EI_CLASS'] == 'ELFCLASS32':
+            self.bytes_per_word = 4
+        else:
+            raise ValueError(f"Unsupported word size: {self.ef['e_ident']['EI_CLASS'] }")
+
+        self.symtab = self.ef.get_section_by_name('.symtab')
+        # build a symbol lookup map forward and reverse
+        self.symbol_map = {sym.entry["st_value"]:sym.name for sym in self.symtab.iter_symbols()}
+        self.symbol_map.update({sym.name:sym.entry["st_value"] for sym in self.symtab.iter_symbols()})
+        
+        return self
+    def entry_point(self):
+        return self.ef["e_entry"]
+
+    def segments(self):
+        if not self.quiet:
+            print( '  --- SEGMENTS ---')
+        for idx, segment in enumerate(self.ef.iter_segments()):
+            d = segment.data()
+            if not self.quiet:
+                print(f'    {idx}: {segment["p_type"].ljust(8)} '
+                    f'@{segment["p_vaddr"]:08x} '
+                    f'size = {segment["p_memsz"]:4x}, '
+                    f'data = {len(d):4x}')
+            yield segment["p_vaddr"], segment["p_memsz"], d
+
+    def sections(self):
+        # print( '  --- SECTIONS ---')
+        for idx, section in enumerate(self.ef.iter_sections()):
+            # print(f'    {idx}: {section.name} {section["sh_type"]}',
+            #         hex(section["sh_addr"]), 
+            #         hex(section["sh_offset"]),
+            #         hex(section["sh_size"]))
+            yield section["sh_addr"], section["sh_size"], section.data()
+            
+    def __exit__(self, *args):
+        self.f.close()
+
+def load_elf(elffile, stack_size = 64 * 2**10, quiet = False):
+    "this loads an elf file into memory segments for simulation"
+    # initialize memories as a unified memory (instruction + data)
+    sys_mem = ELFMemory()
+    # TODO this should read the word size from the file, now it assumes 32 bit.
+    with Elf(elffile, quiet=quiet) as e:
+        for addr, size, data in e.segments():
+            if len(data) == 0:
+                # non initalized segments need to be allocated
+                data = bytearray(size)
+            elif len(data) < size:
+                # bss segments are in size but not data
+                # need to zero initialize this memory.
+                data = bytearray(data) + bytearray(size-len(data))
+
+            ms = MemorySegment(
+                begin_addr = addr,
+                data = data,
+                byteorder = e.byteorder,
+                word_size = 4)
+
+            # add the segment to system memory
+            sys_mem += ms
+
+        symbols = e.symbol_map    
+    # allocate stack immediately at the end of the elf segments
+    # this is how the UCB linker script expects memory 
+    sys_mem += MemorySegment(
+        begin_addr = sys_mem.end_addr(),
+        count = stack_size,
+        byteorder = sys_mem.byteorder,
+        word_size = 4)
+    if not quiet:
+        print(f"Created system memory in range {sys_mem.begin_addr():08x}:{sys_mem.end_addr():08x}")                
+        print( "Segments:\n" + str(sys_mem))
+        print(f"Total allocated system memory is {len(sys_mem) / 1024:4.1f} kilobytes.")
+        print("-"*60)
+
+
+    return sys_mem, symbols
\ No newline at end of file

pydigital/memory.py

+"""
+memory.py
+=========
+Provides a byte-addressed memory.
+"""
+from pydigital.utils import sextend
+class Memory:
+    "Memory module which implements the risc-v sodor memory interface"
+    def __init__(self, segment = None):
+        "initialize with a memory segment"
+        self.mem = segment
+    def out(self, addr, byte_count = 4, signed = True):
+        "read access"
+        if addr == None:
+            return None
+        #TODO check of byte_count is larger than word size, that won't work!
+        if signed:
+            f = sextend
+        else:
+            f = lambda x, c: x
+        if byte_count == 1:
+            return f(self.mem[addr] & 0xff, 8)
+        elif byte_count == 2:
+            return f(self.mem[addr] & 0xffff, 16)
+        elif byte_count == 4:
+            return f(self.mem[addr] & 0xffffffff, 32)
+        elif byte_count == 8:
+            return self.mem[addr] & 0xffffffffffffffff
+        else:
+            raise ValueError("Mem can only access Bytes/Half Words/Words.")
+    def clock(self, addr, data, mem_rw = 0, byte_count = 4):
+        "synchronous write, mem_rw=1 for write"
+        if mem_rw == 1:
+            mask = (2**(byte_count*8))-1
+            #print(f"MEM write mask {mask:08x} masked val is {(mask&data):08x} of {byte_count} bytes")
+            # mask out any upper bits so to_bytes doesn't complain
+            val = (mask & data).to_bytes(length=byte_count,
+                byteorder = self.mem.byteorder, signed = False)
+            #print(f'MEM val is {val}')
+            self.mem[addr] = val
+class ELFMemory:
+    "ELFMemory is a collection of memory segments that supports get/set"
+    def __init__(self):        
+        self.mems = []
+        self.byteorder = None
+    def __getitem__(self, i):
+        if i == None: 
+            return None
+        for m in self.mems:
+            if i in m:
+                return m[i]
+        raise IndexError(f"Address {i:08x} not found in memory.")
+    def __setitem__(self, i, val):
+        if i == None:
+            return
+        for m in self.mems:
+            if i in m:                  
+                m[i] = val
+    def __iadd__(self, seg):
+        if self.byteorder == None:
+            self.byteorder = seg.byteorder      
+        elif self.byteorder != seg.byteorder:
+            raise ValueError("Byteorder does not match previous segments.")
+        self.mems.append(seg)
+        return self
+    def begin_addr(self):
+        "return the lowest begin address included"
+        return min([m.begin_addr for m in self.mems])
+    def end_addr(self):
+        "return the highest end address included"
+        return max([m.end_addr for m in self.mems])
+    def __str__(self):
+        "debug segment addresses"
+        s = []
+        for i, seg in enumerate(self.mems):
+            s += [f"[{i}] {seg.begin_addr:08x}:{seg.end_addr:08x} ({len(seg.data):4x} bytes)"]
+        return "\n".join(s)
+    def __len__(self):
+        return sum([len(m.data) for m in self.mems])
+class MemorySegment:
+    "A continuous segment of byte addressable memory"
+    def __init__(self, begin_addr = 0x1000, count = None, 
+        word_size = 4, data = None, byteorder = 'big'):
+        "create a new memory from begin_addr with count words (32 or 64 bits per word)"
+        self.word_size = word_size
+        self.byteorder = byteorder
+        if data == None:
+            if count == None:
+                raise ValueError("Count must be given without data.")
+            self.data = bytearray(self.word_size * count)           
+        else:
+            if count != None:
+                 raise ValueError("Count must NOT be given with data.")            
+            if type(data) is bytearray:
+                self.data = data
+            else:
+                # attempt to convert to bytearray
+                self.data = bytearray(data)
+        self.end_addr = begin_addr + len(self.data)
+        self.begin_addr = begin_addr
+    def __str__(self):
+        return f"Memory[{self.begin_addr:8x}:{self.end_addr:8x}] ({len(self.data)})"
+    def __getitem__(self, i):
+        "get a word from a given *byte* address"
+        if i == None:
+            return None
+        if isinstance(i, slice):
+            # if you ask for a slice, you get raw bytes
+            data = self.data[i.start - self.begin_addr: i.stop - self.begin_addr: i.step]            
+            return data
+            # this decodes, but that's less useful
+            #return [int.from_bytes(data[_i:_i+self.word_size], byteorder=self.byteorder, signed=False) \
+                    #for _i in data[::self.word_size]]
+        else:
+            try:
+                i -= self.begin_addr
+            except TypeError as err:
+                print("can't access", i)
+                raise err
+            # returns the given word size value as an unsigned int (preserving 2s comp)
+            return int.from_bytes(
+                self.data[i: i+self.word_size],
+                byteorder=self.byteorder, signed=False)
+    def __setitem__(self, i, val, signed=False):
+        "set a word at given *byte* address"
+        if type(val) == int:
+            # convert to a words/bytes
+            val = val.to_bytes(length=self.word_size, 
+                byteorder=self.byteorder, signed=signed)
+        if type(val) == bytes or type(val) == bytearray:
+            # print('setting bytes', val)
+            # byte by byte copy
+            i -= self.begin_addr
+            for v in val:
+                self.data[i] = v
+                i += 1 
+        else:
+            raise ValueError("Value must be bytes or int.")
+        # self.data[(i - self.begin_addr) // self.word_size] = self.fromTwosComp(val)
+    def __contains__(self, addr):
+        "is the given byte address in this memory segment?"
+        if isinstance(addr, slice):
+            return addr.start in self and addr.stop in self
+        else:
+            return addr >= self.begin_addr and addr < self.end_addr
+    def to_hex(self):
+        s = ["@" + format(int(self.begin_addr / self.word_size), "x")]
+        
+        fmt = f"0{2*self.word_size}x"
+        num = int(len(self.data) / self.word_size)
+        print(fmt)
+        for wordaddr in range(num):
+            byteaddr = wordaddr * self.word_size
+            d = int.from_bytes(self.data[byteaddr: byteaddr + self.word_size], 
+                byteorder=self.byteorder)    
+            s.append(format(d, fmt))
+
+        return " ".join(s)
+
+def readmemh(filename, begin_addr = 0, word_size = 4, byteorder = 'big'):
+    "reads a verilog hex file and returns a memory segment"
+    at = begin_addr
+    data = None
+    with open (filename, 'r') as f:
+        for statement in f.read().split():            
+            if statement[0] == '@':
+                if data != None:
+                    # output segment before creating a new one
+                    # multi segment hex files are not supported atm.
+                    raise NotImplementedError()
+                # the hex file is indexed by words
+                at = word_size * int (statement[1:], base=16)
+                data = []
+            else:
+                data.append(int(statement, 16).to_bytes(word_size, byteorder))
+
+    data = b"".join(data)
+    return MemorySegment(begin_addr = at, 
+        data = data, 
+        word_size = word_size,
+        byteorder = byteorder)
+    
\ No newline at end of file

pydigital/register.py

+"""
+register.py
+===========
+A simple clocked register.
+"""
+class Register:
+    """
+    A generic register, at the clock edge the input is evaluated 
+    and assigned to the output.
+    """
+    def __init__(self, *args):        
+        self._val = None
+        self._args = args
+    def out(self):
+        # return the current value stored in the reg
+        return self._val
+    def reset(self, value):    
+        # asynchronous (re)set
+        self._val = value
+    def clock (self, next_val):
+        # the system should evaluate all inputs and pass in the next value here
+        # we just need to assign (copy) them to the stored values
+        self._val = next_val
\ No newline at end of file

pydigital/system.py

+"""
+pydigital is a verlog-like simulation engine for digital systems.
+
+The system class provides a clock to various modules.
+
+Alan Marchiori 2021
+"""
+
+from .utils import verilog_fmt
+
+class System():
+    """
+    The system clock generates clock ticks and keeps track of time
+    Similar to verilog:
+        reg clk = 0;
+        always #1 clk = !clk;
+    """
+    def __init__(self, posedge=[], negedge=[]):
+        "Pass in componets to be clocked on positive and negative edges"
+        self.time = 0
+        self._val = False
+        self._pos = posedge
+        self._neg = negedge
+        self._mon_str = None
+        self._mon_vals = []
+        self._disp_str = None
+        self._disp_vals = []
+    def monitor(self, mon_str, *mon_vals):
+        "Attach a monitor, mon_vals must be functions that return the current value"
+        self._mon_str = mon_str
+        self._mon_vals = mon_vals
+        self._mon_last_vals = None
+        self.do_monitor()
+    def display(self, mon_str, *mon_vals):
+        "Attach a display, mon_vals must be functions that return the current value"
+        self._disp_str = mon_str
+        self._disp_vals = mon_vals
+        self.do_display()
+    def do_display(self):
+        "evaluates the expressions and prints if anything has changed"                       
+        if self._disp_str and self._disp_vals:
+            clk = '-'
+            if self._val:
+                clk = '+'
+            print (clk + verilog_fmt(self._disp_str, 
+                    *[x() for x in self._disp_vals], 
+                    timeval = self.time))        
+
+    def do_monitor(self):
+        "evaluates the monitored expressions and prints if anything has changed"
+        # evaluate all monitored values and store
+        current_vals = [x() for x in self._mon_vals]
+
+        # only print if there is a monitor and something changed
+        if self._mon_str and self._mon_vals and current_vals != self._mon_last_vals:
+            clk = '-'
+            if self._val:
+                clk = '+'
+            print (clk + verilog_fmt(self._mon_str, 
+                    *current_vals, 
+                    timeval = self.time))
+        self._mon_last_vals = current_vals
+
+    def __iter__(self):
+        return self
+    def __next__(self):        
+        # update monitor
+        self.do_monitor()
+
+        self._val ^= True # invert the clock level
+        self.time += 1    # increment time 
+
+        def evaluate(modules):
+            # sample inputs
+            vals = []
+            for x in modules:
+                vals.append(list(y() for y in x.inputs))
+            # clock passing in input vals
+            for x,y in zip(modules, vals):
+                x.clock(*y)
+
+        # execute the pos/negedge methods
+        if self._val:            
+            evaluate(self._pos)
+        else:
+            evaluate(self._neg)
+
+        # update monitor
+        self.do_monitor()
+        
+        # execute displays
+        self.do_display()
+        return self._val  # return new clock level
+
+    def run(self, ticks=2):
+        """run the system for the given number of clock ticks (clock half-cycles), 
+           like a #ticks; in verilog
+           The default (2) runs one full clock period.
+        """       
+        for _i in range(ticks):
+            next(self)
\ No newline at end of file

pydigital/utils.py

+"""
+utils.py
+========
+Misc utilities for working on digital systems including sign extension of python integers.
+"""
+
+import re
+from functools import partial
+def verilog_fmt(fstr, *args, timeval = -1):
+    """
+    Verilog % style formating
+    Supports %t for time and %d or %x for integers only!
+    It does support width specifiers on all arg types.
+
+    Example:
+    verilog_fmt("At time %3t, value = 0x%05x (%d)", 99, 99, timeval = 33)
+    At time  33, value = 0x00063 (99)
+    """    
+    argidx = 0
+    pos = 0
+    s = ""
+    for part in re.finditer(r"(\%\d*[stxd])", fstr):
+        s += fstr[pos:part.start(0)]
+        pos = part.end(0)
+        fmt = part[0][1:]   # strip leading % from format
+        if part[0][-1] == 't':
+            val = timeval   # use special global time val
+            fmt = fmt[:-1] + 'd' # replace trailing t with d for format
+            if fmt == 'd':  # check if no width specified because
+                fmt = "20d" # verilog defaults to 20 digits for time.
+        else:
+            val = args[argidx]
+            argidx += 1
+            if val == None: # if arg value is None, treat as Undefined (X)
+                widthstr = fmt[:-1]
+                if widthstr == "":
+                    val = 'x'
+                else:
+                    try:
+                        val = 'x'*int(widthstr)
+                    except ValueError:
+                        val = 'x'
+                fmt = ""
+        s += format(val, fmt)
+    s += fstr[pos:]
+    return s
+
+def sextend(val, c=32):
+    "sign extend a c bit val to 32 bits as a python integer"
+    # this converts from a twos-complement number to a python signed integer
+    sign = 0b1 & (val >> (c-1))
+    mask = (1 << c) - 1
+    
+
+    if sign == 0b1:        
+        uppermask = 0xffffffff ^ mask
+        #print(f"sign {sign:02x} mask = {mask:08x}, uppermask {uppermask:08x}, val = {val:08x}")
+        # invert plus one, after sign extension
+        return - (0xffffffff - (uppermask | val) + 1) 
+    else:
+        # positive values (should we check if > 32 bits and truncate?)
+        return val 
+
+def sextend12(val):
+    "12 bit sign extender, generates a 32 bit 2s comp value from a 12-bit 2s comp value."
+    return partial(sextend, c=12)
+