We show here some examples of how to run Python on a Pynq platform. Python 3.6 is running exclusively on the ARM processor.
In the first example, which is based on calculating the factors and primes of integer numbers, give us a sense of the performance available when running on an ARM processor running Linux.
In the second set of examples, we leverage Python's numpy package and asyncio
module to demonstrate how Python can communicate
with programmable logic.
Code is provided in the cell below for a function to calculate factors and
primes. It contains some sample functions to calculate the factors and primes
of integers. We will use three functions from the factors_and_primes module
to demonstrate Python programming.
In [1]:
"""Factors-and-primes functions.
Find factors or primes of integers, int ranges and int lists
and sets of integers with most factors in a given integer interval
"""
def factorize(n):
"""Calculate all factors of integer n.
"""
factors = []
if isinstance(n, int) and n > 0:
if n == 1:
factors.append(n)
return factors
else:
for x in range(1, int(n**0.5)+1):
if n % x == 0:
factors.append(x)
factors.append(n//x)
return sorted(set(factors))
else:
print('factorize ONLY computes with one integer argument > 0')
def primes_between(interval_min, interval_max):
"""Find all primes in the interval.
"""
primes = []
if (isinstance(interval_min, int) and interval_min > 0 and
isinstance(interval_max, int) and interval_max > interval_min):
if interval_min == 1:
primes = [1]
for i in range(interval_min, interval_max):
if len(factorize(i)) == 2:
primes.append(i)
return sorted(primes)
else:
print('primes_between ONLY computes over the specified range.')
def primes_in(integer_list):
"""Calculate all unique prime numbers.
"""
primes = []
try:
for i in (integer_list):
if len(factorize(i)) == 2:
primes.append(i)
return sorted(set(primes))
except TypeError:
print('primes_in ONLY computes over lists of integers.')
def get_ints_with_most_factors(interval_min, interval_max):
"""Finds the integers with the most factors.
"""
max_no_of_factors = 1
all_ints_with_most_factors = []
# Find the lowest number with most factors between i_min and i_max
if interval_check(interval_min, interval_max):
for i in range(interval_min, interval_max):
factors_of_i = factorize(i)
no_of_factors = len(factors_of_i)
if no_of_factors > max_no_of_factors:
max_no_of_factors = no_of_factors
results = (i, max_no_of_factors, factors_of_i,\
primes_in(factors_of_i))
all_ints_with_most_factors.append(results)
# Find any larger numbers with an equal number of factors
for i in range(all_ints_with_most_factors[0][0]+1, interval_max):
factors_of_i = factorize(i)
no_of_factors = len(factors_of_i)
if no_of_factors == max_no_of_factors:
results = (i, max_no_of_factors, factors_of_i, \
primes_in(factors_of_i))
all_ints_with_most_factors.append(results)
return all_ints_with_most_factors
else:
print_error_msg()
def interval_check(interval_min, interval_max):
"""Check type and range of integer interval.
"""
if (isinstance(interval_min, int) and interval_min > 0 and
isinstance(interval_max, int) and interval_max > interval_min):
return True
else:
return False
def print_error_msg():
"""Print invalid integer interval error message.
"""
print('ints_with_most_factors ONLY computes over integer intervals where'
' interval_min <= int_with_most_factors < interval_max and'
' interval_min >= 1')
Next we will call the factorize() function to calculate the factors of an integer.
In [2]:
factorize(1066)
Out[2]:
The primes_between() function can tell us how many prime numbers there are in an integer range. Let’s try it for the interval 1 through 1066. We can also use one of Python’s built-in methods len() to count them all.
In [3]:
len(primes_between(1, 1066))
Out[3]:
Additionally, we can combine len() with another built-in method, sum(), to calculate the average of the 180 prime numbers.
In [4]:
primes_1066 = primes_between(1, 1066)
primes_1066_average = sum(primes_1066) / len(primes_1066)
primes_1066_average
Out[4]:
This result makes sense intuitively because prime numbers are known to become less frequent for larger number intervals. These examples demonstrate how Python treats functions as first-class objects so that functions may be passed as parameters to other functions. This is a key property of functional programming and demonstrates the power of Python.
In the next code snippet, we can use list comprehensions (a ‘Pythonic’ form of the map-filter-reduce template) to ‘mine’ the factors of 1066 to find those factors that end in the digit ‘3’.
In [5]:
primes_1066_ends3 = [x for x in primes_between(1, 1066)
if str(x).endswith('3')]
print('{}'.format(primes_1066_ends3))
This code tells Python to first convert each prime between 1 and 1066 to a string and then to return those numbers whose string representation end with the number ‘3’. It uses the built-in str() and endswith() methods to test each prime for inclusion in the list.
And because we really want to know what fraction of the 180 primes of 1066 end in a ‘3’, we can calculate ...
In [6]:
len(primes_1066_ends3) / len(primes_1066)
Out[6]:
These examples demonstrate how Python is a modern, multi-paradigmatic language. More simply, it continually integrates the best features of other leading languages, including functional programming constructs. Consider how many lines of code you would need to implement the list comprehension above in C and you get an appreciation of the power of productivity-layer languages. Higher levels of programming abstraction really do result in higher programmer productivity!
Code in the cells below show a very simple data movement code snippet that can be used to share data with programmable logic. We leverage the Python numpy package to manipulate the buffer on the ARM processors and can then send a buffer pointer to programmable logic for sharing data.
We do not assume what programmable logic design is loaded, so here we only allocate the needed memory space and show that it can manipulated as a numpy array and contains a buffer pointer attribute. That pointer can then can be passed to programmable logic hardware.
In [7]:
import numpy as np
import pynq
def get_pynq_buffer(shape, dtype):
""" Simple function to call PYNQ's memory allocator with numpy attributes
"""
return pynq.allocate(shape, dtype)
With the simple wrapper above, we can get access to memory that can be shared by both numpy methods and programmable logic.
In [8]:
buffer = get_pynq_buffer(shape=(4,4), dtype=np.uint32)
buffer
Out[8]:
To double-check we show that the buffer is indeed a numpy array.
In [9]:
isinstance(buffer,np.ndarray)
Out[9]:
To send the buffer pointer to programmable logic, we use its physical address which is what programmable logic would need to communicate using this shared buffer.
In [10]:
pl_buffer_address = hex(buffer.physical_address)
pl_buffer_address
Out[10]:
In this short example, we showed a simple allocation of a numpy array that is now ready to be shared with programmable logic devices. With numpy arrays that are accessible to programmable logic, we can quickly manipulate and move data across software and hardware.
PYNQ also leverages the Python asyncio module for communicating with programmable logic devices through events (namely interrupts).
A Python program running on PYNQ can use the asyncio library to manage multiple IO-bound tasks asynchronously, thereby avoiding any blocking caused by waiting for responses from slower IO subsystems. Instead, the program can continue to execute other tasks that are ready to run. When the previously-busy tasks are ready to resume, they will be executed in turn, and the cycle is repeated.
Again, since we won't assume what interrupt enabled devices are loaded on programmable logic, we will show an example here a software-only asyncio example that uses asyncio's sleep method.
In [11]:
import asyncio
import random
import time
# Coroutine
async def wake_up(delay):
'''A function that will yield to asyncio.sleep() for a few seconds
and then resume, having preserved its state while suspended
'''
start_time = time.time()
print(f'The time is: {time.strftime("%I:%M:%S")}')
print(f"Suspending coroutine 'wake_up' at 'await` statement\n")
await asyncio.sleep(delay)
print(f"Resuming coroutine 'wake_up' from 'await` statement")
end_time = time.time()
sleep_time = end_time - start_time
print(f"'wake-up' was suspended for precisely: {sleep_time} seconds")
With the wake_up function defined, we then can add a new task to the event loop.
In [12]:
delay = random.randint(1,5)
my_event_loop = asyncio.get_event_loop()
try:
print("Creating task for coroutine 'wake_up'\n")
wake_up_task = my_event_loop.create_task(wake_up(delay))
my_event_loop.run_until_complete(wake_up_task)
except RuntimeError as err:
print (f'{err}' +
' - restart the Jupyter kernel to re-run the event loop')
finally:
my_event_loop.close()
All the above examples show standard Python 3.6 running on the PYNQ platform. This entire notebook can be run on the PYNQ board - see the getting_started folder on the Jupyter landing page to rerun this notebook.