utils Namespace Reference

Functions
	setup_logging (str level="INFO")
	find_oksincludes (list[str] includes, list[str] extra_dirs=[])
	find_free_port (int min_port_num=0, int max_port_num=65535)

Variables
list	log_levels = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]

Function Documentation

find_free_port()

utils.find_free_port	(	int	min_port_num = 0,
		int	max_port_num = 65535 )

Definition at line 93 of file utils.py.

def find_free_port(min_port_num:int=0, max_port_num:int=65535):
    # If the user didn't specify a minimum port number (or deliberately specified
    # zero), we can simply ask the system for an available port.
    if min_port_num == 0:
        with socket.socket() as s:
            s.bind(("", 0))
            s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            port = s.getsockname()[1]
            s.close()
            return port
    # If the user specified a minimum port number, use the specified range.
    else:
        if min_port_num < 1024:
            min_port_num = 1024
        while True:
            port = random.randint(min_port_num, max_port_num)
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
                try:
                    s.bind(("0.0.0.0", port))
                    return port
                except OSError:
                    continue

find_oksincludes()

utils.find_oksincludes	(	list[str]	includes,
		list[str]	extra_dirs = [] )

Definition at line 38 of file utils.py.

def find_oksincludes(includes:list[str], extra_dirs:list[str] = []):
    includefiles = []
 
    searchdirs = [path for path in os.environ["DUNEDAQ_DB_PATH"].split(":")]
    for dir in extra_dirs:
        searchdirs.append(dir)
 
    for inc in includes:
        # print (f"Searching for {inc}")
        match = False
        inc = inc.removesuffix(".xml")
        if inc.endswith(".data"):
            sub_dirs = ["config", "data"]
        elif inc.endswith(".schema"):
            sub_dirs = ["schema"]
        else:
            sub_dirs = ["*"]
            inc = inc + "*"
        for path in searchdirs:
            # print (f"   {path}/{inc}.xml")
            matches = glob.glob(f"{inc}.xml", root_dir=path)
            if len(matches) == 0:
                for search_dir in sub_dirs:
                    # print (f"   {path}/{search_dir}/{inc}.xml")
                    matches = glob.glob(f"{search_dir}/{inc}.xml", root_dir=path)
                    for filename in matches:
                        if filename not in includefiles:
                            print(f"Adding {filename} to include list")
                            includefiles.append(filename)
                        #else:
                        #    print(f"{filename} already in include list")
                        match = True
                        break
                    if match:
                        break
                if match:
                    break
            else:
                for filename in matches:
                    if filename not in includefiles:
                        print(f"Adding {filename} to include list")
                        includefiles.append(filename)
                    #else:
                    #    print(f"{filename} already in include list")
                    match = True
                    break
 
        if not match:
            print(f"Error could not find include file for {inc}")
            return [False, []]
 
    return [True, includefiles]
 
# This function returns a random available network port.  Users can optionally
# specify a range that should be used.

setup_logging()

utils.setup_logging

(

str

level = "INFO"

)

Definition at line 12 of file utils.py.

def setup_logging(level:str="INFO"):
    level = level.upper()
 
    loglevel = logging.INFO
 
    match level:
        case "DEBUG":
            loglevel = logging.DEBUG
        case "INFO":
            loglevel = logging.INFO
        case "WARNING":
            loglevel = logging.WARNING
        case "ERROR":
            loglevel = logging.ERROR
        case "CRITICAL":
            loglevel = logging.CRITICAL
        case _:
            loglevel = logging.INFO
 
    FORMAT = "%(message)s"
    logging.basicConfig(
        level="NOTSET", format=FORMAT, datefmt="[%X]", handlers=[RichHandler()]
    )
    logging.getLogger().setLevel(loglevel)
 
 

Variable Documentation

log_levels

list utils.log_levels = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]

Definition at line 9 of file utils.py.