PY_HLP.py

#!/usr/bin/env python3
"""
Helper file that has executable examples.
Ex:
  ~/hlp/PY_HLP.py map_simple_example
"""

def ex_args_kwargs( *args, **kwargs ):
    for i in kwargs.items():
        print(f'{i}: {kwargs[i]}')
    for i in args:
        print(f'{i}{type(i)}')

def ex_dbg_interactive( *args, **kwargs ):
    #import pdb: pdb.set_trace() # shorter but need to proceed command with 'p' to get completion
    import pdb, rlcompleter
    pdb.Pdb.complete=rlcompleter.Completer(locals()).complete
    this = { "that": "other" }
    pdb.set_trace()

def ex_map_simple( *args, **kwargs ):
    ''' The most simplified MAP example '''
    def double(x):
        return x * 2

    numbers = [1,2,3,4,5]

    # Map with function pointer
    doubles = map(double, numbers)
    print(doubles)
    print(numbers,list(doubles))
    # Same but using lanbda (unamed function)
    doubles = map(lambda x: x * 2, numbers)
    print(doubles)
    print(numbers,list(doubles))

def ex_dict_comp( *args, **kwargs ):
   # dict comprehension (programmatically build a dict)
    myMap = { i: 2*i for i in range(3) }
    print(myMap)

def ex_zip( *args, **kwargs ):
    nums1=[1,2,3,4]
    nums2=["a","b","c"]
    for n1, n2 in zip(nums1,nums2):
        print(n1,n2)

def ex_filter_simple( *args, **kwargs ):
    numbers = [1,2,3,4,5]
    evens = filter(lambda x: x % 2 == 1, numbers)
    odds  = filter(lambda x: x % 2 == 0, numbers)
    print(evens,odds)
    print(list(evens),list(odds))


def ex_heap( *args, **kwargs ):
    import heapq

    # under the hood are arrays
    minHeap=[]
    heapq.heappush(miniHeap, 3)
    heapq.heappush(miniHeap, 2)
    heapq.heappush(miniHeap, 4)

    # Min is always at index 0
    print(miniHeap[0]) # 2

    # Build heap from initial values
    arr=[1,2,3,4,7,84,3]
    heapq.heapify(arr)
    while arr:
        print(heapq.heappop(arr))

def ex_palindrome( *args, **kwargs ):
    # Ex of two pointers problem
    def is_palindrome(string):
        start=0
        end=len(string)-1

        while start < end:
            if string[start] != string[end]:
                return False
            start +=1
            end   -=1

        return True
    print( is_palindrome("racecar"))
    print( is_palindrome("abccba") )
    print( is_palindrome("123421") )

def ex_slidewindow( *args, **kwargs ):
  def max_subarry_sum_sliding_window( arr, k):
    n=len(arr)
    window_sum=sum(arr[:k])
    max_sum=window_sum
    max_start_index=0
    for i in range(n-k):
      window_sum=window_sum - arr[i] + arr[i+k]
      print( window_sum )
      if window_sum>max_sum:
        max_sum=window_sum
        max_start_index=i+1
    return arr[max_start_index:max_start_index+k], max_start_index, max_sum

  a=[3,2,7,5,9,6,2,1,1,8,7,8,0]
  print( max_subarry_sum_sliding_window( a, 4 ) )
  print( max_subarry_sum_sliding_window( a, 3 ) )
  print( max_subarry_sum_sliding_window( a, 2 ) )
  print( max_subarry_sum_sliding_window( a, 1 ) )


# python type hints (they are not enforced by python but third party tools can use) https://www.uninformativ.de/blog/postings/2022-04-21/0/POSTING-en.html

def ex_decorator_func_w_func( *args, **kwargs ):
    import functools, time

    def timer(func):
        """Print the runtime of the decorated function"""
        @functools.wraps(func)
        def wrapper_timer(*args, **kwargs):
            start_time = time.perf_counter()
            value = func(*args, **kwargs)
            end_time = time.perf_counter()
            run_time = end_time - start_time
            print(f"Finished {func.__name__}() in {run_time:.4f} secs")
            return value
        return wrapper_timer

    @timer
    def waste_some_time(num_times):
        val=0
        for _ in range(num_times):
            val+=sum([number**2 for number in range(10_000)])
        return val

    print( waste_some_time(1) )

def ex_decorator_func_w_class( *args, **kwargs ):
    class Power(object):
        def __init__(self, func):
            self._func = func

        def __call__(self, a, b):
            retval = self._func( a, b )
            return retval ** 2

    @Power
    def multiply_together(a, b):
         return a * b

    print(multiply_together)
    print(multiply_together(2,2))

def ex_decorator_w_param( *args, **kwargs ):
    class Power(object):
        def __init__(self, arg):
            self._arg = arg

        def __call__(self, *param_arg):
            """If there are decorator arguments, __call__() is only called once as part of the decoration process. You can only give it a single argument, which is the function object If there are no decorator arguments, the function to be decorated is passed to the constructor.  """
            if len(param_arg) == 1:
                def wrapper(a, b):
                    retval = param_arg[0](a, b)
                    return retval ** self._arg
                return wrapper
            else:
                expo = 2
                retval = self._arg(param_arg[0], param_arg[1])
                return retval ** expo

    @Power
    def multiply_together(a, b):
        return a * b
    print(multiply_together(2, 2))

    @Power(3)
    def multiply_together(a, b):
      return a * b
    print(multiply_together(2, 2))

def ex_constraint_solver( *args, **kwargs ):
    import constraint, random
    problem = constraint.Problem()
    problem.addVariable("a", [1,2,3])
    problem.addVariable("b", [4,5,6])
    print( problem.getSolutions() )
    #  [{'a': 3, 'b': 6}, {'a': 3, 'b': 5}, {'a': 3, 'b': 4},
    #   {'a': 2, 'b': 6}, {'a': 2, 'b': 5}, {'a': 2, 'b': 4},
    #   {'a': 1, 'b': 6}, {'a': 1, 'b': 5}, {'a': 1, 'b': 4}]

    problem.addConstraint(lambda a, b: a*2 == b, ("a", "b"))
    print( problem.getSolutions() )
    #  [{'a': 3, 'b': 6}, {'a': 2, 'b': 4}]

    problem = constraint.Problem()
    problem.addVariables(["a", "b"], [1, 2, 3])
    problem.addConstraint(constraint.AllDifferentConstraint())
    print( problem.getSolutions() )
    #  [{'a': 3, 'b': 2}, {'a': 3, 'b': 1}, {'a': 2, 'b': 3},
    #   {'a': 2, 'b': 1}, {'a': 1, 'b': 2}, {'a': 1, 'b': 3}]:

    problem = constraint.Problem()
    problem.addVariables(["a", "b"], [1, 2, 3, 4])
    problem.addConstraint(lambda a, b: b==1 if (a%2==0) else b==2, ("a", "b"))
    rslt = problem.getSolutions()
    print( rslt )
    random.seed(100)        # keep the results stable
    random.shuffle( rslt ) #in-place shuffle
    for i in range(len(rslt)):
        print(f"{rslt[i]}")

def ex_udp_msg( *args, **kwargs ):
    ''' Test out sending UDP packets. Run in seperate terminals: ~/HLP/PY_HLP.py ex_udp_msg; nc -u localhost 12345 '''
    # Creating a socket:
    #   TCP: socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    #   UDP: socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    # Establishing a connection (TCP only):
    #   socket.connect((host, port))
    # Sending data:
    #   TCP: socket.send(data)
    #   UDP: socket.sendto(data, (host, port))
    # Receiving data:
    #   TCP: socket.recv(buffer_size)
    #   UDP: socket.recvfrom(buffer_size)
    #UDPServerSocket = socket.socket(family=socket.AF_INET, type=socket.SOCK_DGRAM)
    #UDPServerSocket.bind((localIP, localPort))
    import socket

    # Server setup
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    server_address = ('localhost', 12345)
    server_socket.bind(server_address)

    print("Chatbot server listening on {}:{}".format(*server_address))

    while True:
        data, client_address = server_socket.recvfrom(1024)
        message = data.decode('utf-8')

        print("Received message from {}:{}: {}".format(*client_address, message))

        # Process message and generate a response
        response = "You said: " + message

        server_socket.sendto(response.encode('utf-8'), client_address)

def ex_bytearray_reverse( *args, **kwargs ): # reverse the bytes
    msg = bytearray(b"Hello World!")
    msg.reverse()
    print(msg)

def ex_bytearray_xor_lrc( *args, **kwargs ): # Longitudinal Redundancy Check (LRC)
    msg = bytearray(b"Hello World!")
    lrc = 0
    for b in msg:
        lrc ^= b
    msg.append(lrc)
    print(msg)
    # python2 version:
    msg = "Hello World!"
    lrc = 0
    for b in msg:
        lrc ^= ord(b)
    msg += chr(lrc)
    print(msg)

def ex_bytearray_xor_cypher( *args, **kwargs ):
    key = 37
    msg = bytearray(b"Hello World!")
    enc_msg = bytearray(x ^ key for x in msg)
    print(enc_msg)
    dec_msg=bytearray(x ^ key for x in enc_msg)
    print(dec_msg)

def ex_bytearray_data2bin_file( *args, **kwargs ):
    import struct
    points = [(1,2),(3,4),(5,6),(7,8),(9,0)]
    f = open("points.bin","wb")
    msg = bytearray()
    msg.extend(struct.pack("I",len(points)))
    for x,y in points:
        msg.extend(struct.pack("II",x,y))
    f.write(msg)
    f.close()

def ex_bytearray_endianness( *args, **kwargs ):
    '''
      Import struct: The struct module provides functions for packing and unpacking binary data.
      Create a bytearray: This is the data you want to convert.
      Convert using struct.unpack:
        The first argument to unpack is the format string, specifying the data type and endianness.
        "<I" means unpack as a 4-byte unsigned integer in little-endian format.
        ">I" means unpack as a 4-byte unsigned integer in big-endian format.
        unpack returns a tuple, so we take the first element to get the integer value.
    '''
    import struct
    data = bytearray([0x12, 0x34, 0x56, 0x78])

    little_endian = struct.unpack("<I", data)[0]
    print("little_endian: "+hex(little_endian))

    big_endian = struct.unpack(">I", data)[0]
    print("big_endian:    "+hex(big_endian))


def get_func_names():
    ''' Get the name of functions defined in this "module" '''
    module = sys.modules[__name__]
    is_function_in_module = lambda obj: inspect.isfunction(obj) and inspect.getmodule(obj) == module
    script_functions =  inspect.getmembers(module, is_function_in_module)
    func_names = []
    for func in script_functions:
        if func[0].startswith("ex_"):
            func_names.append( func[0] )
    return f"Callable functions:\n  {func_names}\n"



# Here’s the complete code for an agent and its components:
from pyuvm import *

class MyDriver(uvm_driver):
    def run_phase(self):
        while True:
            transaction = self.seq_item_port.get_next_item()
            self.drive(transaction)
            self.seq_item_port.item_done()

    def drive(self, transaction):
        pass

class MyMonitor(uvm_monitor):
    def run_phase(self):
        while True:
            transaction = self.collect()
            self.analysis_port.write(transaction)

    def collect(self):
        return None  # Replace with actual collection code

class MySequencer(uvm_sequencer):
    def run_phase(self):
        while True:
            transaction = MyTransaction()
            self.seq_item_export.put_next_item(transaction)

class MyAgent(uvm_agent):
    def build_phase(self):
        super().build_phase()
        self.driver = MyDriver("driver", self)
        self.monitor = MyMonitor("monitor", self)
        self.sequencer = MySequencer("sequencer", self)

    def connect_phase(self):
        super().connect_phase()
        self.driver.seq_item_port.connect(self.sequencer.seq_item_export)

class MyEnv(uvm_env):
    def build_phase(self):
        super().build_phase()
        self.agent = MyAgent("agent", self)

# TODO -
#  def ex_pyuvm( *args, **kwargs ):
#  def ex_zip( *args, **kwargs ):
#  def ex_graph( *args, **kwargs ): # simple example of creating graph with nodes and archs

if __name__ == '__main__':
    import inspect, sys
    from argparse import ArgumentParser, RawTextHelpFormatter
    parser = ArgumentParser(
        description     = __doc__+get_func_names(),
        formatter_class = RawTextHelpFormatter
    )
    # Add your arguments here
    parser.add_argument("name", help="Name of the function to run")
    parser.add_argument('unparsed_args', nargs='*')
    args = parser.parse_args()

    try:
        eval( f"{args.name}"+f"(*args.unparsed_args)" )
    except:
        raise
    print( "Done" )