Utils filehandling
utils_filehandling ¶
Module contains filesystem related utilities.
file_locked(filepath: str, waittime: float = 0.1) -> bool ¶
file_locked - check if file is locked
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
filepath | str | file to be checked | required |
waittime | float | waiting time for check, defaults to 0.1. | 0.1 |
Returns:
| Name | Type | Description |
|---|---|---|
bool | bool | lock indicator |
Source code in src/utils_mystuff/utils_filehandling.py
def file_locked(filepath: str, waittime: float = 0.1) -> bool:
"""
file_locked - check if file is locked
Args:
filepath (str): file to be checked
waittime (float, optional): waiting time for check, defaults to 0.1.
Returns:
bool: lock indicator
"""
if not (os.path.exists(filepath)):
return False
try:
f = open(filepath, 'r')
f.close()
except OSError:
return True
lockfile = filepath + ".lckchk"
if os.path.exists(lockfile):
os.remove(lockfile)
try:
os.rename(filepath, lockfile)
time.sleep(waittime)
os.rename(lockfile, filepath)
return False
except OSError:
return True
find_last_changed_file(searchpath: str, searchpattern: str, searchstart: datetime.datetime | None = None, timeout_sec: int = 10) -> str ¶
find_last_changed_file - find latest changed file in searchpath matching filename pattern provided as searchpattern
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
searchpath | str | path to search in | required |
searchpattern | str | file pattern | required |
searchstart | datetime | search for files changed after 'searchstart' only. Defaults to None. | None |
timeout_sec | int | timeout period. Defaults to 10. | 10 |
Returns:
| Name | Type | Description |
|---|---|---|
str | str | name of file found (if any) |
Source code in src/utils_mystuff/utils_filehandling.py
def find_last_changed_file(
searchpath: str,
searchpattern: str,
searchstart: datetime.datetime | None = None,
timeout_sec: int = 10
) -> str:
"""
find_last_changed_file - find latest changed file in searchpath matching filename pattern provided as searchpattern
Args:
searchpath (str): path to search in
searchpattern (str): file pattern
searchstart (datetime.datetime, optional): search for files changed after 'searchstart' only. Defaults to None.
timeout_sec (int, optional): timeout period. Defaults to 10.
Returns:
str: name of file found (if any)
"""
timenow = datetime.datetime.now(dateutil.tz.tz.tzlocal())
if searchstart is None:
searchstart = timenow - datetime.timedelta(seconds=5)
if searchstart.tzinfo is None:
searchstart = pytz.utc.localize(searchstart)
lastchangedfile = ""
# lastchangedtimestamp = searchstart - datetime.timedelta(seconds=1)
lastchangedtimestamp = searchstart.timestamp() - 1
while (lastchangedfile == "") and (datetime.datetime.now(dateutil.tz.tz.tzlocal()) <= timenow + datetime.timedelta(seconds=timeout_sec)):
with os.scandir(searchpath) as searchpathscan:
scanentry: os.DirEntry # overcome PyCharm IDE error - workaround derived from https://youtrack.jetbrains.com/issue/PY-46041
for scanentry in searchpathscan:
if fnmatch.fnmatch(scanentry.name, searchpattern):
filemodified = scanentry.stat().st_mtime_ns / 1000000000
if filemodified >= searchstart.timestamp():
pathfilename = os.path.join(searchpath, scanentry.name)
while os.path.isfile(pathfilename) and is_file_locked(pathfilename):
time.sleep(0.5)
if os.path.isfile(pathfilename) and not is_file_locked(pathfilename):
if filemodified > lastchangedtimestamp and scanentry.stat().st_size > 0:
lastchangedfile = scanentry.name
lastchangedtimestamp = filemodified
searchpathscan.close()
return lastchangedfile
is_file_locked(filepath: str, waittime: float = 0.1) -> bool ¶
is_file_locked - check if file is locked
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
filepath | str | file to be checked | required |
waittime | float | waiting time for check. Defaults to 0.1. | 0.1 |
Returns:
| Name | Type | Description |
|---|---|---|
bool | bool | lock indicator |
Source code in src/utils_mystuff/utils_filehandling.py
load_textfile(txt_filename: str, mode: str = 'str') -> list[str] | str ¶
load_textfile - load textfile to list of strings or str
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
txt_filename | str | source text file | required |
mode | str | return mode - 'list', 'str' or 'strcleaned'. Defaults to 'str'. | 'str' |
Returns:
| Type | Description |
|---|---|
list[str] | str | Union[list[str], str]: content of textfile as list of string or string |
Source code in src/utils_mystuff/utils_filehandling.py
def load_textfile(txt_filename: str, mode: str = "str") -> list[str] | str:
"""
load_textfile - load textfile to list of strings or str
Args:
txt_filename (str): source text file
mode (str, optional): return mode - 'list', 'str' or 'strcleaned'. Defaults to 'str'.
Returns:
Union[list[str], str]: content of textfile as list of string or string
"""
txt: list[str] | str = ""
fileexists = os.path.isfile(txt_filename)
# read to list of lines
if mode == "list":
if fileexists:
# via pathlib
# txt = pathlib.Path(txt_filename).read_text()
# via normal open file
with open(txt_filename, "r") as txt_file:
txt = txt_file.readlines()
txt_file.close()
else:
txt = []
# read to single string incl. CR
elif mode == "str":
if fileexists:
with open(txt_filename, "r") as txt_file:
txt = txt_file.read()
txt_file.close()
else:
txt = ""
# read to single string without CR
elif mode == "strcleaned":
if fileexists:
with open(txt_filename, "r") as txt_file:
txt = txt_file.read().replace("\n", "")
txt_file.close()
else:
txt = ""
else:
err_msg = "Invalid mode."
raise Exception(err_msg)
return txt
wait_for_file(filepath: str, timeout: int = 5, waittime: float = 0.1) -> bool ¶
wait_for_file - wait until file is available or timeout
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
filepath | str | filepath to be checked | required |
timeout | int | timeout period | 5 |
waittime | float | waittime between two checks | 0.1 |
Returns:
| Name | Type | Description |
|---|---|---|
bool | bool | True if file exists |
Source code in src/utils_mystuff/utils_filehandling.py
def wait_for_file(filepath: str, timeout: int = 5, waittime: float = 0.1) -> bool:
"""
wait_for_file - wait until file is available or timeout
Args:
filepath (str): filepath to be checked
timeout (int): timeout period
waittime (float): waittime between two checks
Returns:
bool: True if file exists
"""
start_time = time.time()
file_exists = os.path.exists(filepath)
while not file_exists:
time.sleep(waittime)
file_exists = os.path.exists(filepath)
elapsed_time = time.time() - start_time
if timeout > 0 and elapsed_time >= timeout:
break
return file_exists