Expressions, Functions & Time

regular expressions

manipulating strings

`findall()`	returning list with all matches
`search()`	returns matching object, if existing
`split()`	returns list of fragments
`replace()`	replacing one or many matches

`[ ]`	set of characters
`\`	special sequence (escape character)
`.`	as single character (except newline)
`^`	starts with
`$`	ends with
`*`	zero or more occurrences
`+`	one or more occurrences
`?`	zero or one occurrences
`{ }`	exactly the specified number of occurrences
`\|`	either or
`( )`	capture and grouping

#!/usr/bin/env python3

import re   # regular expressions

txt = "The rain in Spain falls mostly in ..."  
# ... the plain.

xFind = re.findall("in", txt)
print("xFind   : ", xFind) 

xSplit = re.split(" ", txt)
print("xSplit  : ", xSplit) 

# splitting at first position  
xSplit_1 = re.split(" ", txt, 1)
print("xSplit_1: ", xSplit_1) 

# replace 3 times
xSub = re.sub(" ", "_", txt, 3)
print("xSub    : ", xSub) 

# search for an upper case "S" character 
# in the beginning of a word 
xSpan = re.search(r"\bS\w+", txt)
if xSpan:
    print("match object :", xSpan)
    print("matched text :", xSpan.group())
    print("start index  :", xSpan.start())
    print("end index    :", xSpan.end())
    print("span         :", xSpan.span())

lambda expressions

lambda expressions are inline functions
:: no name (anonymous), no access elsewhere
only a single expression (command)
several arguments (bound variables)
() at the end for immediate call
line continuation with \

ternary expressions

x%2 and 'odd' or 'even'
:: returns either 'odd' or 'even'
'odd' if x%2 else 'even'
:: returns either 'odd' or 'even'

sorting (key,value) pairs

sorted(): for any iterable (values only)
list.sort(): idem for lists (in place)
enumerate(): : returns iterator over (key,value) tuples

#!/usr/bin/env python3

print("# *************************")
print("# simple lambda expressions")
print("# *************************")
x = lambda a : a + 10
print("lambda    a : a + 10     :",x(5)) 

y = lambda a, b : a * b
print("lambda a, b : a * b      :",y(5, 6)) 

# line continuation
# no empty space after  '\'
z = lambda x: \
    'odd' if x%2 else 'even'   # inline 'if then else" 
#   x % 2 and 'odd' or 'even'   # identical

print("\'odd\' if x%2 else \'even\' :",z(5)) 

print()
print("# **************************************")
print("# function returning a lambda expression")
print("# **************************************")

def myfunc(n):
  return lambda a : a * n

mytripler = myfunc(3)
# equivalent to
# mytripler = lambda a : a*3

print("return lambda a : a * n  :",mytripler(11))
   
print()
print("# ***********************")
print("# top-k (indices, values)")
print("# ***********************")
lst = [10, 3, 45, 6, 23, 89, 5]
print("          lst   ", lst)
print("enumerate(lst)  ", enumerate(lst))
print("list(enum(lst)) ", list(enumerate(lst))[:4])
print()
k = 3
sortedPairs = sorted(enumerate(lst),        # tuples (index,value)
                     key=lambda x: x[1],    # sort by values
                     reverse=True)
print("top-k entries   ", sortedPairs[:k])  # slicing

print()
print("# **************")
print("# direct calling")
print("# **************")
print("(lambda x: x + 1)(2)     :",(lambda x: x + 1)(2))

#!/usr/bin/env python3

print("# *************************")
print("# simple lambda expressions")
print("# *************************")
x = lambda a : a + 10
print("lambda    a : a + 10     :",x(5))

y = lambda a, b : a * b
print("lambda a, b : a * b      :",y(5, 6))

# line continuation
# no empty space after  '\'
z = lambda x: \
    'odd' if x%2 else 'even'   # inline 'if then else" 
#   x % 2 and 'odd' or 'even'   # identical

print("\'odd\' if x%2 else \'even\' :",z(5))

print()
print("# **************************************")
print("# function returning a lambda expression")
print("# **************************************")

def myfunc(n):
  return lambda a : a * n

mytripler = myfunc(3)
# equivalent to
# mytripler = lambda a : a*3

print("return lambda a : a * n  :",mytripler(11))
   
print()
print("# ***********************")
print("# top-k (indices, values)")
print("# ***********************")
lst = [10, 3, 45, 6, 23, 89, 5]
print("          lst   ", lst)
print("enumerate(lst)  ", enumerate(lst))
print("list(enum(lst)) ", list(enumerate(lst))[:4])
print()
k = 3
sortedPairs = sorted(enumerate(lst),        # tuples (index,value)
                     key=lambda x: x[1],    # sort by values
                     reverse=True)
print("top-k entries   ", sortedPairs[:k])  # slicing

print()
print("# **************")
print("# direct calling")
print("# **************")
print("(lambda x: x + 1)(2)     :",(lambda x: x + 1)(2))

magic functions

numbers are objects
:: member functions used for arithmetics, etc.
var.__add__(5)
this is what happens when evaluating var+5
dir(int): directory listing of magic methods (for int class)
:: listing for all objects possible
:: actually int.__dir__

#!/usr/bin/env python3

# as simple function
def myFunction(a):
  return a**2

# intVar+5                is just a short for
# intVar.__add__(5)
intVar = 10
print()
print("using __add__() : ", intVar+5, intVar.__add__(5))
print()

print("*******************")
print("content of dir(int)")
print("*******************")
print()
#for mF in int.__dir__:    # idem
for mF in dir(int):        # dir: directory of magic methods
  print(mF)

print()
print("**************************")
print("content of dir(myFunction)")
print("**************************")
print()
for mF in dir(myFunction):  
  print(mF)

built-in functions

A
abs()
aiter()
all()
any()
anext()
ascii()

B
bin()
bool()
breakpoint()
bytearray()
bytes()

C
callable()
chr()
classmethod()
compile()
complex()

D
delattr()
dict()
dir()
divmod()

E
enumerate()
eval()
exec()

F
filter()
float()
format()
frozenset()

G
getattr()
globals()

H
hasattr()
hash()
help()
hex()

I
id()
input()
int()
isinstance()
issubclass()
iter()
__import__()

L
len()
list()
locals()

M
map()
max()
memoryview()
min()

N
next()

O
object()
oct()
open()
ord()

P
pow()
print()
property()

R
range()
repr()
reversed()
round()

S
set()
setattr()
slice()
sorted()
staticmethod()
str()
sum()
super()

T
tuple()
type()

V
vars()

Z
zip()

threading

threading.Thread()
:: execute another part of the code asynchronously
:: either by time sharing, or on another CPU
threading.Timer()
:: a thread starting with a delay
if __name__ == "__main__":
:: execute only when in main thread
:: e.g., not when loaded as a module

#!/usr/bin/env python3

import threading, time, math

def fetch_data(source_id):
    """e.g., reading from file, doing math"""
    print(f"starting  fetch_data({source_id})")
#
    if (1==2): 
      time.sleep(1)   
    else:
      for _ in range(int(0.5e8)):
        math.sqrt(3.3 * 3.3)
#
    print(f"finishing fetch_data({source_id})\n")
    return

# ***
# *** main start
# ***
start_time = time.time()
nJobs = 3

print("# --- ------------------ ---")
print("# --- sequential reading ---")
print("# --- ------------------ ---")
for i in range(nJobs):
    fetch_data(i)

print(f"total time: {time.time() - start_time:.2f} seconds")

print()
print("# --- ------------------ ---")
print("# --- threaded execution ---")
print("# --- ------------------ ---")

start_time = time.time()
threads = []

# creating, storing, and starting threads
for i in range(nJobs):
    t = threading.Thread(target=fetch_data, args=(i,))
    threads.append(t)
    t.start()

# waiting for all threads to complete, before moving on
for t in threads:
    t.join()

print(f"total time: {time.time() - start_time:.2f} seconds")

#!/usr/bin/env python3

import threading, time, math

def fetch_data(source_id):
    """e.g., reading from file, doing math"""
    print(f"starting  fetch_data({source_id})")
#
    if (1==2): 
      time.sleep(1)   
    else:
      for _ in range(int(0.5e8)):
        math.sqrt(3.3 * 3.3)
#
    print(f"finishing fetch_data({source_id})\n")
    return

# ***
# *** main start
# ***
start_time = time.time()
nJobs = 3

print("# --- ------------------ ---")
print("# --- sequential reading ---")
print("# --- ------------------ ---")
for i in range(nJobs):
    fetch_data(i)

print(f"total time: {time.time() - start_time:.2f} seconds")

print()
print("# --- ------------------ ---")
print("# --- threaded execution ---")
print("# --- ------------------ ---")

start_time = time.time()
threads = []

# creating, storing, and starting threads
for i in range(nJobs):
    t = threading.Thread(target=fetch_data, args=(i,))
    threads.append(t)
    t.start()

# waiting for all threads to complete, before moving on
for t in threads:
    t.join()

print(f"total time: {time.time() - start_time:.2f} seconds")

global interpreter lock (GIL)

threads should not modifiy a given object
when another thread is trying to do so
→ may lead to uncontrolled interference
strictly enforced by Python GIL
:: waiting is ok
threading essential for communication
:: client ↔ server
:: agent ↔ LLM

thread pools

concurrent.futures
high-level interface for asynchronous execution
ThreadPoolExecutor(max_workers=5)
for I/O-bound tasks (network, disk, database)
ProcessPoolExecutor(max_workers=5)
for CPU-bound tasks (calculations, processing)
:: threads are actually not best for GPU heavy tasks
executor.map()
maps functions (with arguments) to the thread pool

#!/usr/bin/env python3

import threading, time
from concurrent.futures import ThreadPoolExecutor

def fetch_data(source_id):
    """e.g., reading from file"""
    print(f"starting  fetch_data({source_id})")
    time.sleep(2)    # simulated working time
    print(f"finishing fetch_data({source_id})\n")
    return "result " + str(source_id)

start_time = time.time()
print("# --- ------------------ ---")
print("# --- sequential reading ---")
print("# --- ------------------ ---")

sources = [1, 2, 3]

# ThreadPoolExecutor handles starting and joining threads automatically
with ThreadPoolExecutor(max_workers=3) as executor:
    results = executor.map(fetch_data, sources)

for result in results:
    print(result)

multi-processing

creating separate operating-system processes
same API as for threads
:: start(), join(), ...
threads: one process, shared memory
processes: specfic/isolated memory spaces
exchanging data with inter-process communication (IPC)
:: beyond this tutorial

#!/usr/bin/env python3

import threading, time, math, multiprocessing

def fetch_data(source_id):
    """e.g., reading from file, doing math"""
    print(f"starting  fetch_data({source_id})")
#
    if (1==2): 
      time.sleep(1)   
    else:
      for _ in range(int(0.5e8)):
        math.sqrt(3.3 * 3.3)
#
    print(f"finishing fetch_data({source_id})\n")
    return

# ***
# *** main start
# ***
start_time = time.time()
nJobs = 3

print("# --- -------------------- ---")
print("# --- sequential execution ---")
print("# --- -------------------- ---")
for i in range(nJobs):
    fetch_data(i)

print(f"total time: {time.time() - start_time:.2f} seconds")

print()
print("# --- ------------------ ---")
print("# --- threaded execution ---")
print("# --- ------------------ ---")

start_time = time.time()
threads = []

# creating, storing, and starting threads
for i in range(nJobs):
    t = threading.Thread(target=fetch_data, args=(i,))
    threads.append(t)
    t.start()

# waiting for all threads to complete, before moving on
for t in threads:
    t.join()

print(f"total time: {time.time() - start_time:.2f} seconds")

print()
print("# --- ---------------- ---")
print("# --- multi-processing ---")
print("# --- ---------------- ---")

start_time = time.time()
processes = []

# create separate OS processes (bypassing the GIL)
for i in range(nJobs):
    p = multiprocessing.Process(target=fetch_data, args=(i,))
    processes.append(p)
    p.start()

# waiting for termination
for p in processes:
    p.join()

print(f"total time: {time.time() - start_time:.2f} seconds")

#!/usr/bin/env python3

import threading, time, math, multiprocessing

def fetch_data(source_id):
    """e.g., reading from file, doing math"""
    print(f"starting  fetch_data({source_id})")
#
    if (1==2): 
      time.sleep(1)   
    else:
      for _ in range(int(0.5e8)):
        math.sqrt(3.3 * 3.3)
#
    print(f"finishing fetch_data({source_id})\n")
    return

# ***
# *** main start
# ***
start_time = time.time()
nJobs = 3

print("# --- -------------------- ---")
print("# --- sequential execution ---")
print("# --- -------------------- ---")
for i in range(nJobs):
    fetch_data(i)

print(f"total time: {time.time() - start_time:.2f} seconds")

print()
print("# --- ------------------ ---")
print("# --- threaded execution ---")
print("# --- ------------------ ---")

start_time = time.time()
threads = []

# creating, storing, and starting threads
for i in range(nJobs):
    t = threading.Thread(target=fetch_data, args=(i,))
    threads.append(t)
    t.start()

# waiting for all threads to complete, before moving on
for t in threads:
    t.join()

print(f"total time: {time.time() - start_time:.2f} seconds")

print()
print("# --- ---------------- ---")
print("# --- multi-processing ---")
print("# --- ---------------- ---")

start_time = time.time()
processes = []

# create separate OS processes (bypassing the GIL)
for i in range(nJobs):
    p = multiprocessing.Process(target=fetch_data, args=(i,))
    processes.append(p)
    p.start()

# waiting for termination
for p in processes:
    p.join()

print(f"total time: {time.time() - start_time:.2f} seconds")