lib
Functions
Identify the current Python execution environment. |
|
Recursively convert a dictionary and its nested structures into a SimpleNamespace. |
|
Download a file from a URL to the local file system (Universal Async). |
|
Calculate the absolute execution path and the workspace root directory. |
|
Send an HTTP request and return the JSON response. |
|
Initialize a standardized Loguru logger with custom naming and storage. |
|
Decorator that calculates and prints the execution time of a function. |
|
Truncate a number to a specified number of decimal places. |
Classes
A professional utility class for managing file archives and compression. |
|
A centralized utility for command-line argument parsing. |
|
Advanced utilities for manipulating Python Data Structures. |
|
Standardized SMTP client for persistent email configurations. |
|
Send emails using Microsoft Graph API (OAuth 2.0/MSAL) with persistent authentication. |
|
A high-performance utility class for robust file and directory management. |
|
A collection of static methods for standardized datetime formatting and time calculations. |
|
A high-performance, lazy-evaluating dictionary wrapper. |
|
|
A list-like proxy that lazily wraps elements upon access. |
A high-performance, truly lazy dictionary and list wrapper. |
- class Archive
Bases:
objectA professional utility class for managing file archives and compression.
The
Archiveclass provides robust, production tools for creating and extracting ZIP archives. It features advanced filtering by file age, size, and extension, and includes secure password support for encrypted archives.Summary of Methods
Method
Description
Unified archive tool with size/age filtering and source cleanup.
Extracts ZIP archives with password support and auto-folder creation.
Advanced purge tool for ZIPs based on age and size thresholds.
USAGE EXAMPLE: from ntsm.lib import Archive # 1. Archive a directory (only .csv files) and delete source after success Archive.make_zip( source='C:/Data/Logs', archive_name='Log_Backup', file_exts=['csv'], delete_source=True ) # 2. Archive a single file only if it exceeds 5MB Archive.make_zip( source='C:/Temp/report.pdf', max_size_mb=5.0, add_timestamp=True ) # 3. Unzip a password-protected archive Archive.un_zip( zip_path='C:/Backups/SecureData.zip', password='MySecretPassword' ) # 4. Clean up: Delete ZIPs older than 30 days OR larger than 1GB Archive.rem_zip('C:/Backups', max_days=30, min_size_mb=1024.0)
- static make_zip(source, dest_path=None, archive_name=None, add_timestamp=False, password=None, max_days=0, max_size_mb=0, file_exts=None, delete_source=False, debug=False)
Create a compressed ZIP archive with advanced size, age, and type filtering.
Handles archiving for both directories and individual files. If the source is a directory and
delete_sourceisTrue, the entire directory is removed after a successful ZIP operation.- Parameters:
source (
str) – The file or directory path to archive.dest_path (
str) – Directory where the ZIP will be saved. Defaults to the source’s parent location.archive_name (
str) – ZIP filename (no extension). Defaults to the source name.add_timestamp (
bool) – IfTrue, appendsYYYYMMDD_HHMMSSto the name.password (
str) – Applies a password to the ZIP archive.max_days (
int) – Only archive files modified X+ days ago.max_size_mb (
float) – Only archive if file (or total) exceeds X MB.file_exts (
Optional[List[str]]) – Filter by extensions (e.g.,['log', 'txt']).delete_source (
bool) – Deletes the source (file or directory) after compression.debug (
bool) – IfTrue, prints processing status.
USAGE EXAMPLES: from ntsm.lib import Archive # 1. Archive a directory of old logs and delete the source folder Archive.make_zip( source="C:/Project/Logs", dest_path="D:/Backups", max_days=30, file_exts=['log'], delete_source=True ) # 2. Archive a single large file only if it exceeds 100MB Archive.make_zip( source="C:/Data/heavy_output.csv", archive_name="HeavyData_Backup", add_timestamp=True, max_size_mb=100.0, delete_source=True )
- Return type:
- Returns:
The absolute path to the created ZIP archive, or an empty string if skipped.
- static rem_zip(zip_path, max_days=0, min_size_mb=0, recursive=False, debug=False)
Erase ZIP archives from a directory based on age OR size criteria.
Priority logic:
If both
max_daysandmin_size_mbare0, every ZIP is deleted.If only
max_days> 0, deletes ZIPs older than X days.If only
min_size_mb> 0, deletes ZIPs larger than X MB.If both are > 0, deletes if either condition is met.
- Parameters:
zip_path (
str) – Directory path containing ZIP archives.max_days (
int) – Delete ZIPs older than X days. Defaults to0.min_size_mb (
float) – Delete ZIPs larger than X MB. Defaults to0.recursive (
bool) – IfTrue, searches subdirectories for ZIPs.debug (
bool) – IfTrue, prints the paths of deleted files.
USAGE EXAMPLES: from ntsm.lib import Archive # Delete EVERY zip in the folder Archive.rem_zip("C:/Temp/Zips") # Delete if older than 7 days OR larger than 100MB Archive.rem_zip("C:/Backups", max_days=7, min_size_mb=100.0)
- static un_zip(zip_path, unzip_path=None, password=None, debug=False)
Extract a ZIP archive to a specified directory with optional password support.
- Parameters:
USAGE EXAMPLE: from ntsm.lib import Archive # Extract a protected archive to a specific folder Archive.un_zip( zip_path="C:/Backups/Project.zip", unzip_path="C:/Restored", password="SecurePassword123" )
- Return type:
- Returns:
The absolute path to the directory where files were extracted.
- Raises:
ValueError – If the file is not a valid ZIP archive.
- class Args
Bases:
objectA centralized utility for command-line argument parsing.
The
Argsclass provides a pre-configuredargparse.ArgumentParserinstance, optimized for GIS and automation scripts. It usesRawTextHelpFormatterto ensure that multi-line help descriptions and examples are rendered correctly in the console.Summary of Methods
Property / Method
Description
The core
argparse.ArgumentParserinstance.USAGE EXAMPLE: from ntsm.lib import Args # 1. Access the pre-configured parser parser = Args.parser # 2. Add custom arguments parser.add_argument("--t", type=str, choices=['one', 'two', 'three'], default='one', required=True, help="Deployment Type") parser.add_argument("--path", type=str, required=True, help="Input directory path") parser.add_argument("--force", action='store_true', help="Ignore safety locks") # 3. Parse and use args = parser.parse_args() # 4. Read parsed arguments values choice = args.t path = args.path force = args.force print(f"Running in {choice} mode at {path}")
- argparse = <module 'argparse' from '/home/docs/.asdf/installs/python/3.12.12/lib/python3.12/argparse.py'>
- parser = ArgumentParser(prog='__main__.py', usage=None, description='Argument Parser', formatter_class=<class 'argparse.RawTextHelpFormatter'>, conflict_handler='error', add_help=True)
- class DS
Bases:
objectAdvanced utilities for manipulating Python Data Structures.
Currently focused on high-performance dictionary filtering and sorting, especially for complex dictionaries containing nested lists or tuples.
Summary of Methods
Method
Description
Filters a dictionary based on equality/inequality of its values.
Filters a dictionary of lists based on a specific index and value.
Returns a new dictionary sorted by its keys (Asc/Desc).
Returns a new dictionary sorted by its values (Asc/Desc).
Returns a new dictionary sorted by an element within list values (Asc/Desc).
USAGE EXAMPLE: from ntsm.lib import DS # Input: { 'Key': [Val0, Val1, Val2] } in_dict = {'ab': ['1', '2', '3'], 'ac': ['4', '5', '6'], 'ad': ['1', '7', '8']} # 1. Filter dictionary by value at list index 0 # Get items where index 0 is equal to '1' out_dict = DS.filter_by_list_index(in_dict, 0, '1', equal=True) # Result: {'ab': [`'1'`, '2', '3'], 'ad': [`'1'`, '7', '8']} # 2. Filter for inequality # Get items where index 0 is NOT '1' out_dict = DS.filter_by_list_index(in_dict, 0, '1', equal=False) # Result: {'ac': ['4', '5', '6']} # 3. Sort dictionary by list index 0 (Ascending) sorted_dict = DS.sort_by_list_index(in_dict, 0, reverse=False) # Result: {'ab': ['1', '2', '3'], 'ad': ['1', '7', '8'], 'ac': ['4', '5', '6']}
- static filter_by_list_index(dict_in, list_idx, list_val, equal=True)
Filter a dictionary where values are lists, based on an item at a specific index.
- Parameters:
USAGE EXAMPLE: from ntsm.lib import DS # Data: { 'ID': [Name, Status, Priority] } data = { 101: ['Task_A', 'Active', 1], 102: ['Task_B', 'Pending', 2], 103: ['Task_C', 'Active', 3] } # Filter where Status (index 1) is 'Active' active_tasks = DS.filter_by_list_index(data, 1, 'Active') # Result: {101: ['Task_A', 'Active', 1], 103: ['Task_C', 'Active', 3]}
- Return type:
- Returns:
A new dictionary containing the filtered items.
- static filter_by_value(dict_in, dict_value, equal=True)
Filter a dictionary based on a specific value.
- Parameters:
USAGE EXAMPLE: from ntsm.lib import DS data = {'status': 'active', 'user': 'admin', 'flag': 'active'} # Filter for equality active_items = DS.filter_by_value(data, 'active') # Result: {'status': 'active', 'flag': 'active'} # Filter for inequality non_active = DS.filter_by_value(data, 'active', equal=False) # Result: {'user': 'admin'}
- Return type:
- Returns:
A new dictionary containing the filtered items.
- static sort_by_key(dict_in, reverse=False)
Sort a dictionary alphabetically or numerically based on its keys.
- Parameters:
USAGE EXAMPLE: from ntsm.lib import DS data = {'z': 100, 'a': 5, 'm': 50} # Sort Ascending sorted_data = DS.sort_by_key(data) # Result: {'a': 5, 'm': 50, 'z': 100} # Sort Descending desc_data = DS.sort_by_key(data, reverse=True) # Result: {'z': 100, 'm': 50, 'a': 5}
- Return type:
- Returns:
A new dictionary with keys in the specified order.
- static sort_by_list_index(dict_in, list_idx, reverse=False)
Sort a dictionary of lists based on an element at a specific index.
- Parameters:
USAGE EXAMPLE: from ntsm.lib import DS # Data: { 'ID': [Name, Score] } data = { 1: ['User_A', 85], 2: ['User_B', 92], 3: ['User_C', 78] } # Sort by Score (index 1) descending leaderboard = DS.sort_by_list_index(data, 1, reverse=True) # Result: {2: ['User_B', 92], 1: ['User_A', 85], 3: ['User_C', 78]}
- Return type:
- Returns:
A new dictionary sorted by the chosen list element.
- Raises:
ValueError – If values aren’t lists or the index is out of range.
- static sort_by_value(dict_in, reverse=False)
Sort a dictionary based on its values rather than its keys.
- Parameters:
USAGE EXAMPLE: from ntsm.lib import DS data = {'Item_A': 50, 'Item_B': 10, 'Item_C': 30} # Sort by values (Smallest to Largest) sorted_data = DS.sort_by_value(data) # Result: {'Item_B': 10, 'Item_C': 30, 'Item_A': 50}
- Return type:
- Returns:
A new dictionary with items sorted by value.
- class Email(srv_name, srv_port, acc_sender, acc_password)
Bases:
objectStandardized SMTP client for persistent email configurations.
Stores server credentials and connection settings, allowing for multiple email transmissions without re-passing sensitive configuration.
Summary of Methods
Method
Description
The primary interface for sending text/HTML emails with CC, BCC, and files.
- Parameters:
USAGE EXAMPLE: from ntsm.lib import Email # 1. Initialize once with server credentials mailer = Email('smtp.office365.com', 587, 'bot@nt.org.uk', 'app_password_123') # 2. Send complex email with multiple recipients, CC, and BCC mailer.send( recipient=['manager_01@nt.org.uk', 'manager_02@nt.org.uk'], # List of primary recipients subject='Weekly GIS Audit - All Regions', body='<h1>Audit Complete</h1><p>Reports for all regions are attached.</p>', cc=['supervisor@nt.org.uk'], # Visible Carbon Copy bcc=['audit_logs@nt.org.uk', 'admin_backup@nt.org.uk'], # Hidden Blind Carbon Copy attachments=['C:/Data/Region_A.csv', 'C:/Data/Region_B.csv'], is_html=True )
Initialize the SMTP server settings and credentials.
- Parameters:
- send(recipient, subject, body, cc=None, bcc=None, attachments=None, is_html=True)
Execute a comprehensive email transmission with lazy-loaded dependencies.
Handles
To,Cc, andBccaddressing logic, automatic MIME creation for attachments, and secure delivery via SSL or STARTTLS based on port.- Parameters:
subject (
str) – The subject line of the email.body (
str) – The content/body of the email.cc (
Union[str,list]) – Carbon Copy recipients (visible to all).bcc (
Union[str,list]) – Blind Carbon Copy recipients (hidden).attachments (
Union[str,list]) – Absolute file path(s) to attach.is_html (
bool) – Renders body as HTML ifTrue. Defaults toTrue.
- Return type:
- Returns:
Trueif the email was sent successfully,Falseotherwise.
- class EmailMsal(tenant_id, client_id, client_secret, client_email)
Bases:
objectSend emails using Microsoft Graph API (OAuth 2.0/MSAL) with persistent authentication.
Uses the Microsoft Authentication Library (MSAL) to acquire tokens from Azure AD and sends emails via the Graph API REST endpoint. Ideal for applications registered in Azure that require modern authentication (OAuth 2.0).
Summary of Methods
Method
Description
The primary interface for sending Graph API emails with CC, BCC, and files.
- Parameters:
USAGE EXAMPLE: from ntsm.lib import EmailMsal # 1. Initialize with Azure App Registration details ms_mailer = EmailMsal( tenant_id='your-tenant-id', client_id='your-client-id', client_secret='your-secret', client_email="automation@nt.org.uk" ) # 2. Send complex email with multiple recipients and attachments ms_mailer.send( recipient=['manager@nt.org.uk', 'lead@nt.org.uk'], subject="Automated Azure Report", body="<h1>Task Complete</h1><p>The Graph API transmission was successful.</p>", cc=['supervisor@nt.org.uk'], bcc='audit_archive@nt.org.uk', attachments=['C:/Reports/Audit.pdf'], is_html=True )
Initialize Azure AD credentials and set the authority endpoint.
- Parameters:
- send(recipient, subject, body, cc=None, bcc=None, attachments=None, is_html=True)
Execute an email transmission via Microsoft Graph API (Lazy Load).
Handles OAuth 2.0 token acquisition, MIME-to-JSON conversion for attachments, and RESTful delivery through the Graph API endpoint.
- Parameters:
subject (
str) – The subject line of the email.body (
str) – The content/body of the email.cc (
Union[str,list]) – Carbon Copy recipients (visible to all).bcc (
Union[str,list]) – Blind Carbon Copy recipients (hidden).attachments (
Union[str,list]) – Absolute file path(s) to attach.is_html (
bool) – Renders body as HTML ifTrue. Defaults toTrue.
- Return type:
- Returns:
Trueif the Graph API accepted the message (HTTP 202),Falseotherwise.
- class Files
Bases:
objectA high-performance utility class for robust file and directory management.
The
Filesclass provides a suite of static methods designed for production-grade automation, specifically optimized for GIS (Esri FGDB) workflows and large-scale data synchronization. It handles common failure points such as file locks, interrupted transfers, and directory-level verification.Summary of Methods
Method
Description
Syncs directories with resume logic and size-based verification.
Purges files based on age, extension, and recursion.
Generates filtered lists of absolute file paths.
Iterates and deletes a provided list of file paths.
Saves a string to a file, ensuring parent directories exist.
Appends text content to the end of an existing file.
Parses files into dictionaries, JSON, or dot-notation objects.
High-speed byte-size calculation for directories.
Appends data to CSV with automated header/encoding management.
Removal of files or folders with retry logic.
High-speed specific validation of file existence.
USAGE EXAMPLE: from ntsm.lib import Files # 1. Sync a 30GB Geodatabase (Silent, Resume-capable) dest = Files.copy_dir("C:/Data/Project.gdb", "D:/Backups") # 2. Cleanup old logs (Older than 30 days, specific extensions) Files.rem_files("C:/Logs", max_days=30, file_exts=['log', 'tmp'], recursive=True) # 3. Safe removal of a scratch directory Files.erase_if_exists("C:/Temp/scratch_workspace", retries=5)
- static append_dict_to_csv(csv_path, data_dict, field_names)
Append a dictionary as a new row to a CSV file, adding a header if needed.
Uses
csv.DictWriterto ensure data is mapped correctly to columns. If newly created, automatically writes field names as the first row (header).- Parameters:
USAGE EXAMPLE: from ntsm.lib import Files fields = ['timestamp', 'status', 'user'] entry = {'timestamp': '2026-02-17', 'status': 'Success', 'user': 'Admin'} Files.append_dict_to_csv("log.csv", entry, fields)
- static append_to_file(content, file_path)
Append new text content to the end of an existing file.
Adds a string to the current EOF without modifying existing data. Requires the file to already exist to prevent accidental file creation.
- Parameters:
USAGE EXAMPLE: from ntsm.lib import Files # Add a new entry to an existing log Files.append_to_file("New entry: Processing step 2 completed.", "./logs/audit.log")
- Return type:
- Raises:
FileNotFoundError – If the specified
file_pathdoes not exist.OSError – If the file cannot be written to due to permissions.
- static copy_dir(source, destination, overwrite=True, recursive=True, retries=3, verify=True, debug=False)
Copy a directory with resume capability, fail-over, and size-based verification.
Optimized for large structures like Esri FGDBs. Allows resuming interrupted transfers by comparing file sizes and timestamps, ensuring data integrity without unnecessary re-copying of existing data.
- Parameters:
source (
str) – The full path of the source folder to copy.destination (
str) – The parent directory where the copy will be saved.overwrite (
bool) – IfTrue, wipes the destination before starting. Defaults toTrue.recursive (
bool) – IfTrue, copies all subdirectories. Defaults toTrue.retries (
int) – Number of attempts per file if locks are found. Defaults to3.verify (
bool) – Performs a final byte-size comparison. Defaults toTrue.debug (
bool) – IfTrue, prints progress and status updates. Defaults toFalse.
USAGE EXAMPLE: from ntsm.lib import Files # Copy directory with all subfolders, without overwriting # existing data, and verify the copy new_path = Files.copy_dir( source="C:/Data/Project.gdb", destination="D:/Backups", overwrite=False )
- Return type:
- Returns:
The absolute path to the synced destination folder.
- Raises:
FileNotFoundError – If the source path does not exist.
OSError – If verification fails or files remain locked after retries.
- static erase_if_exists(path, retries=3, debug=False)
Remove a file or an entire directory tree if it exists on the system.
Identifies whether the target is a file or directory and applies the appropriate removal command. Includes a retry mechanism to handle temporary file locks common in GIS workflows.
- Parameters:
USAGE EXAMPLE: from ntsm.lib import Files # Silently and safely remove a temp folder or file Files.erase_if_exists("C:/Temp/scratch.gdb")
- Return type:
- Returns:
Trueif the path was removed or did not exist,Falseif removal failed.
- static file_exists(file_path)
Verify if a specific file exists and is accessible on the system.
Unlike a general path check, this specifically validates that the target is a file (not a directory). Designed for high-speed validation within complex production pipelines.
- Parameters:
file_path (
str) – The absolute or relative path to the file.
USAGE EXAMPLE: from ntsm.lib import Files if Files.file_exists("C:/Config/settings.json"): # Proceed with processing pass
- Return type:
- Returns:
Trueif the file exists and is a file;Falseotherwise.
- static file_to_type(file_path, output_type='dict')
Read a file and convert its content to a specific Python data structure.
Supports parsing standard Python literals, JSON strings, and converting the result into a dot-accessible object (
SimpleNamespace).- Parameters:
file_path (
str) – The full path to the file to be read.output_type (
Literal['dict','json','obj']) –Defines how the content is parsed:
"dict": Parses Python literals usingast.literal_eval."json": Parses content as standard JSON."obj": Parses and converts to a recursiveSimpleNamespace.
Defaults to
"dict".
USAGE EXAMPLE: from ntsm.lib import Files # Read a Python dictionary file data = Files.file_to_type("config.txt", output_type="dict") # Read a JSON file as an object for dot-notation access config = Files.file_to_type("settings.json", output_type="obj") print(config.database.port)
- Return type:
- Returns:
The parsed content in the requested format.
- Raises:
FileNotFoundError – If the file does not exist.
ValueError – If the file content is not valid for the requested type.
- static files_path_as_list(dir_path, file_exts=None, recursive=False)
Generate a list of full file paths from a directory with advanced filtering.
Scans a directory and returns a list of paths, optionally filtering by file extensions and searching through subdirectories.
- Parameters:
USAGE EXAMPLE: from ntsm.lib import Files # Get all PDF and DOCX files in a folder my_files = Files.files_path_as_list("C:/Docs", file_exts=['pdf', 'docx'])
- Return type:
- Returns:
A list of absolute file paths as strings.
- Raises:
NotADirectoryError – If the provided
dir_pathis invalid.
- static get_dir_size(path, recursive=True)
Calculate the total size of a directory in bytes.
High-speed scan primarily used for pre-copy estimation and post-copy verification to ensure data integrity in large-scale file transfers.
- Parameters:
USAGE EXAMPLE: from ntsm.lib import Files from pathlib import Path size_bytes = Files.get_dir_size("C:/Data/Production.gdb") size_gb = size_bytes / (1024**3) print(f"Total size: {size_gb:.2f} GB")
- Return type:
- Returns:
Total size in bytes. Returns
0if the path does not exist.
- static rem_files(dir_path, max_days=0, file_exts=None, recursive=False)
Remove files from a directory with advanced filtering.
Manages storage by purging old files. Can target specific file extensions and search recursively through subdirectories.
- Parameters:
dir_path (
str) – The full path to the directory to clean.max_days (
int) – Only delete files older than this many days. Defaults to0(deletes all matching files).file_exts (
Optional[List[str]]) – Filter by extensions (e.g.,['log', 'tmp']). IfNone, all file types are considered.recursive (
bool) – IfTrue, searches subdirectories. Defaults toFalse.
USAGE EXAMPLE: from ntsm.lib import Files # Clean all .log and .txt files older than 7 days Files.rem_files( dir_path="C:/Project/logs", max_days=7, file_exts=['log', 'txt'], recursive=True )
- Return type:
- Raises:
NotADirectoryError – If the provided
dir_pathdoes not exist.
- static rem_files_from_list(file_paths)
Iterate through a list of file paths and delete each file.
Designed to work seamlessly with the output of
files_path_as_list_adv. Includes safety checks to ensure the file exists before attempting deletion.USAGE EXAMPLE: from ntsm.lib import Files # Get and then delete a list of temp files tmp_files = Files.files_path_as_list("./temp", file_exts=['tmp']) Files.rem_files_from_list(file_paths=tmp_files)
- static write_to_file(content, file_path)
Save a string to a text file, overwriting any existing content.
Ensures the target directory exists before writing and uses UTF-8 encoding. If the file does not exist, it will be created.
- Parameters:
- Raises:
OSError – If the file cannot be written due to disk or permission issues.
USAGE EXAMPLE: Files.write_to_file(report, "./output/report.txt")
- Return type:
- class FormatTime
Bases:
objectA collection of static methods for standardized datetime formatting and time calculations.
This class provides a centralized suite of utilities to convert Python
datetimeobjects into specific string formats required for various environments such as OpenStreetMap (OSM), Esri Geodatabases (GDB), logging systems, and file naming conventions.Summary of Methods
Method
Description
Calculates the delta between two time points as
hh:mm:ss.Calculates elapsed time from a previous point until the current moment.
Returns a standard timestamp string without millisecond fractions.
The base formatting method supporting custom
strftimepatterns.Cleans OSM ISO-8601 strings into SQL-friendly timestamps.
Formats datetime objects for OSM Overpass API queries.
Formats datetime for standard logs (YYYY-MM-DD HH:MM:SS).
Formats datetime specifically for Esri GDB (DD/MM/YYYY HH:MM:SS).
Formats datetime for safe and sortable filenames (YYYYMMDD_HHMMSS).
USAGE EXAMPLE: from ntsm.lib import FormatTime as ft from datetime import datetime now = datetime.now() # Quick formatting for a filename fname = f"backup_{ft.for_files(now)}.zip" # Calculate process duration start = datetime.now() # ... some code ... duration = ft.timer_now(start)
- static for_files(time_in)
Format a datetime object as a filename-safe string (
YYYYMMDD_HHMMSS).Produces a sortable timestamp that avoids characters prohibited by operating systems (like colons or slashes), ideal for log files or exports.
Note
This is an underlying custom formatting logic derived from
to_str.- Parameters:
time_in (
datetime) – The datetime object to be formatted.
USAGE EXAMPLE: # import module from ntsm.lib import FormatTime # format for filename now = datetime.now() file_suffix = FormatTime.for_files(now) filename = f"report_{file_suffix}_data.csv" # result: "report_20240520_143005_data.csv"
- Return type:
- Returns:
Date and time formatted as a filename-safe string.
- static for_gdb(time_in)
Format a datetime object for Esri Geodatabase (GDB) inserts (
DD/MM/YYYY HH:MM:SS).Note
This is an underlying custom formatting logic derived from
to_str.- Parameters:
time_in (
datetime) – The datetime object to be formatted.
USAGE EXAMPLE: # import module from ntsm.lib import FormatTime # format for GDB now = datetime.now() gdb_timestamp = FormatTime.for_gdb(now) # result: "20/05/2024 14:30:05"
- Return type:
- Returns:
Date and time formatted as a GDB-compatible string.
- static for_log(time_in)
Format a datetime object as a standard log/DB timestamp (
YYYY-MM-DD HH:MM:SS).Note
This is an underlying custom formatting logic derived from
to_str.- Parameters:
time_in (
datetime) – The datetime object to be formatted.
USAGE EXAMPLE: # import module from ntsm.lib import FormatTime # format for log now = datetime.now() log_time = FormatTime.for_log(now) # result: "2024-05-20 14:30:05"
- Return type:
- Returns:
The formatted time string.
- static for_osm(time_in)
Format a datetime object for OpenStreetMap (OSM) Overpass API queries.
Produces the specific ISO-8601 string format
"YYYY-MM-DDTHH:MM:SSZ".- Parameters:
time_in (
datetime) – The datetime object to be formatted.
USAGE EXAMPLE: # import module from ntsm.lib import FormatTime # format for OSM now = datetime.now() osm_timestamp = FormatTime.for_osm(now) # result: '"2024-05-20T14:30:05Z"'
- Return type:
- Returns:
Date and time formatted as an OSM-compatible string.
- static from_osm(time_in)
Convert an OSM ISO-8601 timestamp string to a SQL-friendly text format.
Converts (e.g.,
"2023-10-27T10:30:00Z") into"2023-10-27 10:30:00".- Parameters:
time_in (
datetime) – The OSM-formatted timestamp string (ISO-8601).
USAGE EXAMPLE: # import module from ntsm.lib import FormatTime # format from OSM string osm_val = "2024-05-20T14:30:05Z" clean_val = FormatTime.from_osm(osm_val) # result: "2024-05-20 14:30:05"
- Return type:
- Returns:
A cleaned date-time string.
- static time_difference(time_start, time_end)
Calculate the time difference between two datetime objects.
- Parameters:
USAGE EXAMPLE: # import module from ntsm.lib import FormatTime as ft # calculate difference start = datetime.now() end = start + timedelta(hours=1, minutes=30, seconds=15) processing_time = ft.time_difference(start, end) # result: "1:30:15"
- Return type:
- Returns:
Time difference formatted as a string (
hh:mm:ss).
- static timer_now(time_in)
Calculate the time difference between a provided timestamp and the current moment.
Useful for tracking interim or partial processing times during long-running tasks by comparing the current system time against a start point.
- Parameters:
time_in (
datetime) – The previous timestamp to compare against now.
USAGE EXAMPLE: # import module from datetime import datetime import time from ntsm.lib import FormatTime # mark start start_point = datetime.now() # simulate work time.sleep(2) # get interim time elapsed = FormatTime.timer_now(start_point) # result: "0:00:02"
- Return type:
- Returns:
Time difference formatted as a string (
hh:mm:ss).
- static to_str(time_in, format='%Y-%m-%d')
Format a datetime object as a text string using a custom
strftimepattern.- Parameters:
USAGE EXAMPLE: # import module from ntsm.lib import FormatTime # format date now = datetime.now() date_str = FormatTime.to_str(now) # result: "2024-05-20"
- Returns:
YYYY-MM-DD).- Return type:
- static without_ms(time_in)
Convert a datetime object to a string while removing millisecond precision.
Note
This is an underlying custom formatting logic derived from
to_str.- Parameters:
time_in (
datetime) – The datetime object to be converted.
USAGE EXAMPLE: # import module from datetime import datetime from ntsm.lib import FormatTime # format without milliseconds now = datetime.now() # e.g., 2024-05-20 14:30:05.123456 clean_time = FormatTime.without_ms(now) # result: "2024-05-20 14:30:05"
- Return type:
- Returns:
Date and time string without milliseconds.
- class LazyAttrDict(data=None)
Bases:
objectA high-performance, lazy-evaluating dictionary wrapper.
Provides attribute-style access (dot notation) to dictionary keys, with nested dictionaries and lists recursively wrapped on-demand. Results are cached to ensure maximum performance on subsequent lookups.
Summary of Methods
Method
Description
Retrieves a value with an optional default if the key is missing.
Returns an iterable of the underlying dictionary keys.
Generates lazily-wrapped values for all keys.
Generates key-value pairs where values are lazily wrapped.
Returns the raw underlying dictionary.
Creates a deep copy of the wrapper and its data.
- Parameters:
data (
Optional[Dict[str,Any]]) – The initial dictionary to wrap. Defaults to an empty dict.
USAGE EXAMPLE: from ntsm.lib import LazyAttrDict config = { "api": {"key": "secret", "timeout": 30}, "servers": ["prod", "dev"] } lazy = LazyAttrDict(config) # Attribute access print(lazy.api.key) # Output: 'secret' # Indexed access print(lazy["api"]["timeout"]) # Output: 30 # Nested wrapping in lists print(lazy.servers[0]) # Output: 'prod'
Initialize the lazy dictionary and its internal cache.
- copy()
Create a deep copy of the LazyAttrDict.
- Return type:
- Returns:
A new LazyAttrDict instance with copied data.
- get(key, default=None)
Safely retrieve a value by key with a fallback default.
- items()
Generate (key, wrapped_value) pairs for the dictionary.
- class LazyNodes(data=None)
Bases:
objectA high-performance, truly lazy dictionary and list wrapper.
Provides attribute-style access to dictionary keys and lazy wrapping for both nested dictionaries and lists. Unlike standard wrappers, list elements are only transformed when specifically accessed, minimizing overhead for large datasets.
Summary of Methods
Method
Description
Retrieves a value with an optional default if the key is missing.
Returns an iterable of the underlying dictionary keys.
Generates lazily-wrapped values for all keys.
Generates key-value pairs where values are lazily wrapped.
Returns the raw underlying dictionary.
- Parameters:
data (
Optional[Dict[str,Any]]) – The initial dictionary to wrap. Defaults to an empty dict.
USAGE EXAMPLE: from ntsm.lib import LazyNodes data = { "metadata": {"version": 1.0}, "records": [{"id": 1, "val": "A"}, {"id": 2, "val": "B"}] } nodes = LazyNodes(data) # Attribute access print(nodes.metadata.version) # Output: 1.0 # Lazy list access (records[1] is only wrapped when accessed) print(nodes.records[1].val) # Output: 'B'
Initialize the lazy nodes wrapper.
- get(key, default=None)
Safely retrieve a value by key with a fallback default.
- check_py_env()
Identify the current Python execution environment.
Determines if the script is running within an interactive Jupyter Notebook (or similar IPython-based environment) or a standard Python interpreter.
USAGE EXAMPLE: from ntsm.lib import check_py_env # Get environment type env_type = check_py_env() if env_type == "Jupyter": print("Running in a Notebook - Enabling Rich Outputs") else: print("Running in standard Python - Console mode")
- Return type:
- Returns:
'Jupyter'if in a notebook environment, otherwise'Python'.
- dict_to_ns(in_dict)
Recursively convert a dictionary and its nested structures into a SimpleNamespace.
Transforms a standard dictionary into a
SimpleNamespace, enabling dot-notation access (e.g.,obj.key) at all nested levels.- Parameters:
in_dict (
Any) – The dictionary or list to convert. Non-collection types are returned unchanged.
USAGE EXAMPLE: from ntsm.lib import dict_to_ns data = { "project": "OSM_Query", "settings": {"threads": 4, "timeout": 30}, "workers": [{"id": 1}, {"id": 2}] } ns = dict_to_ns(data) # Access nested data with dots print(ns.settings.threads) # Output: 4 print(ns.workers[0].id) # Output: 1
- Return type:
- Returns:
The converted structure as a
SimpleNamespace, a list of namespaces, or the original value.
- async download(url, filename)
Download a file from a URL to the local file system (Universal Async).
Automatically detects the environment:
In Browser (Pyodide): Uses
pyfetch.In Desktop/Server: Uses
urllibwrapped in a thread to prevent blocking.
- Parameters:
USAGE EXAMPLE: import asyncio from ntsm.lib import download url = 'https://example.com/data.zip' path = './data.zip' # In a notebook, just use: await download(url, path)
- Return type:
- Returns:
None
- get_project_paths(file_ref)
Calculate the absolute execution path and the workspace root directory.
This utility standardizes path resolution regardless of whether the code is running as a standard Python script or within an interactive Jupyter notebook environment.
- Parameters:
file_ref (
str) – Typically__file__when called from a script. Used to determine the directory of the executing file.
USAGE EXAMPLE: from ntsm.lib import get_project_paths # When called from 'C:/Project/src/worker.py': path, wks = get_project_paths(__file__) # path -> 'C:/Project/src' # wks -> 'C:/Project'
- send_url_request(request)
Send an HTTP request and return the JSON response.
- Parameters:
request (
Any) – Aurllib.request.Requestobject or a URL string.
USAGE EXAMPLE: # import module from ntsm.lib import send_url_request # create request url = 'https://api.example.com/data' # get json data data = send_url_request(url)
- Return type:
- Returns:
The parsed JSON data from the server response.
- setup_logging(log_name, log_dir, level='INFO')
Initialize a standardized Loguru logger with custom naming and storage.
Configures a file-based logger with automatic rotation and compression. Removes the default system handler so logs are directed only to the file.
Level Behaviors:
DEBUG: Maximum detail. Enables
backtraceanddiagnose.INFO: Standard operation without diagnostic overhead.
WARNING/ERROR/CRITICAL: Enables
catchto prevent log failures from crashing the app.
- Parameters:
USAGE EXAMPLE: from ntsm.lib import setup_logging from pathlib import Path # Setup a specific log for a worker log_name = Path(__file__).stem log_path = Path(__wks__) / 'log' Log = setup_logging(log_name, log_path, level="DEBUG") Log.info("Logger initialized and ready.") Log.debug("Deep diagnostic info is captured here.") try: 1/0 except Exception: Log.exception("This captures the stack trace automatically.")
- Return type:
- Returns:
A pre-configured Loguru logger instance.
- timer_wrap(func)
Decorator that calculates and prints the execution time of a function.
- Parameters:
func (
Callable) – The function to be wrapped for timing.
USAGE EXAMPLE: # import module from ntsm.lib import timer_wrap @timer_wrap def my_process(): time.sleep(1) return "Done" # execution will print: - function 'my_process' took: 1.0000 sec result = my_process()
- Return type:
- Returns:
The wrapper function that executes the timing logic.
- truncate_decimals(n, d=0)
Truncate a number to a specified number of decimal places.
- Parameters:
USAGE EXAMPLE: # import module from ntsm.lib import truncate_decimals # truncate to 4 decimals val = 123.456789 result = truncate_decimals(val, 4) # result: 123.4567