qufe.base module

class qufe.base.ProgressBar

Bases: object

Progress bar utility for visual feedback during long operations.

Supports both console and Jupyter notebook environments.

static estimate_remaining(current: int, total: int, elapsed_seconds: float) → float | None

Estimate remaining time based on current progress.

Parameters:

• current – Current progress value

• total – Total value for completion

• elapsed_seconds – Elapsed time in seconds

Returns:

Estimated remaining time in seconds or None

Example

>>> remaining = ProgressBar.estimate_remaining(30, 100, 15.0)
>>> print(f"Remaining: {ProgressBar.format_time(remaining)}")

static format_time(seconds: float) → str

Format seconds into human-readable time string.

Parameters:

seconds – Time in seconds

Returns:

Formatted time string

Example

>>> ProgressBar.format_time(125.5)
'2m 5s'
>>> ProgressBar.format_time(3665)
'1h 1m 5s'

static show_bar(current: int, total: int, prefix: str = 'Progress', suffix: str = '', bar_length: int = 30, filled_char: str = '█', empty_char: str = '░') → str

Generate a text-based progress bar.

Parameters:

• current – Current progress value

• total – Total value for completion

• prefix – Text to display before the bar

• suffix – Text to display after the bar

• bar_length – Length of the progress bar in characters

• filled_char – Character for filled portion

• empty_char – Character for empty portion

Returns:

Formatted progress bar string

Example

>>> bar = ProgressBar.show_bar(30, 100, prefix="Processing")
>>> print(bar)
Processing: [█████████░░░░░░░░░░░░░░░░░░░░░] 30.0% (30/100)

static show_spinner(current: int, prefix: str = 'Processing', suffix: str = '') → str

Generate a simple spinner animation.

Parameters:

• current – Current iteration count

• prefix – Text before spinner

• suffix – Text after spinner

Returns:

Formatted spinner string

Example

>>> for i in range(10):
...     spinner = ProgressBar.show_spinner(i, "Loading")
...     print(f"

{spinner}”, end=””)

static update_display(content: str, jupyter_mode: bool = False, clear: bool = True) → None

Update display with content, supporting both console and Jupyter.

Parameters:

• content – Content to display

• jupyter_mode – Whether running in Jupyter environment

• clear – Whether to clear previous output

Example

>>> for i in range(100):
...     bar = ProgressBar.show_bar(i, 100)
...     ProgressBar.update_display(bar, jupyter_mode=True)

class qufe.base.ProgressTracker(total: int, prefix: str = 'Progress', jupyter_mode: bool = False, bar_length: int = 30, show_status: bool = True)

Bases: object

Advanced progress tracking with statistics and time estimation.

Supports status messages for richer feedback during processing.

__init__(total: int, prefix: str = 'Progress', jupyter_mode: bool = False, bar_length: int = 30, show_status: bool = True)

Initialize progress tracker.

Parameters:

• total – Total number of items to process

• prefix – Prefix text for progress bar

• jupyter_mode – Whether in Jupyter environment

• bar_length – Length of progress bar

• show_status – Whether to display status messages

finish(show_summary: bool = True) → Dict[str, Any]

Finish tracking and return summary.

Parameters:

show_summary – Whether to display summary

Returns:

Dictionary with tracking statistics including status summary

get_status_summary() → Dict[str, Any]

Get summary of all status messages.

Returns:

Dictionary with status statistics

update(increment: int = 1, item: Any = None, error: str | None = None, status: str | None = None) → None

Update progress with optional item tracking and status message.

Parameters:

• increment – Amount to increment progress

• item – Item that was processed (optional)

• error – Error message if processing failed (optional)

• status – Status message to display (optional)

class qufe.base.TS(time_zone: str | None = None, utc_offset: float | None = None)

Bases: object

Timestamp handling utility class with automatic timezone detection.

__init__(time_zone: str | None = None, utc_offset: float | None = None)

Initialize timestamp handler with automatic local timezone detection.

Parameters:

• time_zone – Timezone name (optional, auto-detects if None)

• utc_offset – UTC offset in hours (optional, auto-detects if None)

get_timezone_info() → Dict[str, Any]

Get information about current timezone settings.

Returns:

Dictionary with timezone information

get_ts_formatted(timestamp) → str

Get formatted timestamp string.

Parameters:

timestamp – Unix timestamp or datetime object

Returns:

Formatted timestamp string or None if invalid

Example

>>> ts = TS()  # Auto-detects local timezone
>>> formatted = ts.get_ts_formatted(1640995200)

timestamp_to_datetime(timestamp) → datetime

Convert timestamp to datetime object with timezone.

Parameters:

timestamp – Unix timestamp (int/float) or datetime object

Returns:

datetime object with timezone or None if invalid input

Example

>>> ts = TS()  # Auto-detects local timezone
>>> dt = ts.timestamp_to_datetime(1640995200)

class qufe.base.WOL(verbose: bool = True)

Bases: object

Wake-on-LAN utility for network device control.

__init__(verbose: bool = True)

Initialize Wake-on-LAN handler.

Parameters:

verbose – Enable detailed output messages (default: True)

create_magic_packet(mac: str) → bytes

Create WOL magic packet.

Magic packet structure: - 6 bytes of 0xFF (synchronization stream) - Target MAC address repeated 16 times

Parameters:

mac – Target device MAC address

Returns:

Magic packet as bytes

Example

>>> wol = WOL()
>>> packet = wol.create_magic_packet("AA:BB:CC:DD:EE:FF")
>>> len(packet)
102

format_mac(mac: str) → str

Format MAC address to standard colon notation.

Parameters:

mac – MAC address string in any supported format

Returns:

XX:XX:XX:XX:XX format

Return type:

MAC address in XX

Raises:

ValueError – If MAC address format is invalid

Example

>>> wol = WOL()
>>> wol.format_mac("aabbccddeeff")
'AA:BB:CC:DD:EE:FF'

send_packet(mac: str, broadcast_ip: str = '255.255.255.255', port: int = 9, attempts: int = 3, delay: float = 0.5) → bool

Send WOL magic packet to network.

Parameters:

• mac – Target device MAC address

• broadcast_ip – Broadcast IP address (default: 255.255.255.255)

• port – WOL port (default: 9, alternative: 7)

• attempts – Number of send attempts (default: 3)

• delay – Delay between attempts in seconds (default: 0.5)

Returns:

True if packet sent successfully

Example

>>> wol = WOL()
>>> success = wol.send_packet("AA:BB:CC:DD:EE:FF")

validate_mac(mac: str) → bool

Validate MAC address format.

Parameters:

mac – MAC address string

Returns:

True if valid MAC address format

Example

>>> wol = WOL()
>>> wol.validate_mac("AA:BB:CC:DD:EE:FF")
True
>>> wol.validate_mac("AA-BB-CC-DD-EE-FF")
True

wake(mac: str, device_name: str | None = None, subnet_broadcast: str | None = None) → bool

Wake network device using Wake-on-LAN.

Parameters:

• mac – Target device MAC address

• device_name – Device name for display (optional)

• subnet_broadcast – Subnet broadcast address (e.g., 192.168.1.255)

Returns:

True if wake signals sent successfully

Example

>>> wol = WOL()
>>> wol.wake("AA:BB:CC:DD:EE:FF", device_name="Development Server")

>>> # With subnet broadcast
>>> wol.wake("AA:BB:CC:DD:EE:FF",
...          device_name="Office PC",
...          subnet_broadcast="192.168.1.255")

qufe.base.diff_codes(left: str, right: str, mode: int = 0)

Compare two code strings with different diff formats.

Parameters:

• left – Left code string to compare

• right – Right code string to compare

• mode – Comparison mode (0=simple, 1=unified, 2=ndiff)

Example

>>> diff_codes("line1\nline2", "line1\nmodified", mode=1)

qufe.base.flatten(lst, max_depth=1, current_depth=0)

Flatten nested lists up to a specified depth.

Parameters:

• lst – The list to flatten

• max_depth – Maximum depth to flatten (default: 1)

• current_depth – Current recursion depth (internal use)

Returns:

Flattened list

Example

>>> flatten([1, [2, [3, 4], 5], [6, 7], 8])
[1, 2, 3, 4, 5, 6, 7, 8]

qufe.base.flatten_any(nested, max_depth=1, current_depth=0)

Flatten nested collections (list, tuple, set) up to specified depth.

Parameters:

• nested – The nested collection to flatten

• max_depth – Maximum depth to flatten (default: 1)

• current_depth – Current recursion depth (internal use)

Yields:

Flattened items one by one

Example

>>> list(flatten_any([1, (2, [3, {4, 5}])]))
[1, 2, 3, 4, 5]  # Order may vary for set items

qufe.base.flatten_gen(lst, max_depth=1, current_depth=0)

Flatten nested lists using generator (memory efficient).

Parameters:

• lst – The list to flatten

• max_depth – Maximum depth to flatten (default: 1)

• current_depth – Current recursion depth (internal use)

Yields:

Flattened items one by one

Example

>>> list(flatten_gen([1, [2, [3, [4]], 5]]))
[1, 2, 3, 4, 5]

qufe.base.flatten_three_levels_with_suffix(nested_dict: dict) → dict

Flatten 3-level nested dictionary by merging level2 into level1 with suffix notation for original parent keys.

Parameters:

nested_dict – 3-level nested dictionary

Returns:

Flattened dictionary with suffix notation

Example

>>> data = {'A': {'x': 1, 'y': {'p': 10, 'q': 20}, 'z': 3}}
>>> flatten_three_levels_with_suffix(data)
{'A': {'x': 1, 'p (y)': 10, 'q (y)': 20, 'z': 3}}

qufe.base.import_script(script_name: str, script_path: str)

Dynamically import a Python module from file path.

Parameters:

• script_name – Name for the imported module

• script_path – Path to the Python file to import

Returns:

Imported module object

Example

>>> module = import_script("my_module", "/path/to/script.py")
>>> module.some_function()

qufe.base.wake_device(mac: str, device_name: str | None = None, subnet_broadcast: str | None = None, verbose: bool = True) → bool

Wake network device using Wake-on-LAN (convenience function).

Parameters:

• mac – Target device MAC address

• device_name – Device name for display (optional)

• subnet_broadcast – Subnet broadcast address (optional)

• verbose – Enable detailed output (default: True)

Returns:

True if wake signals sent successfully

Example

>>> # Simple usage
>>> wake_device("AA:BB:CC:DD:EE:FF")

>>> # With device name and subnet
>>> wake_device("AA:BB:CC:DD:EE:FF",
...            device_name="Development Server",
...            subnet_broadcast="192.168.1.255")

>>> # Silent mode
>>> success = wake_device("AA:BB:CC:DD:EE:FF", verbose=False)

qufe.base.wake_multiple_devices(devices: Dict[str, str], subnet_broadcast: str | None = None, verbose: bool = True, delay: float = 1.0) → Dict[str, bool]

Wake multiple network devices sequentially.

Parameters:

• devices – Dictionary of {device_name: mac_address}

• subnet_broadcast – Subnet broadcast address (optional)

• verbose – Enable detailed output (default: True)

• delay – Delay between devices in seconds (default: 1.0)

Returns:

success_status}

Return type:

Dictionary of {device_name

Example

>>> devices = {
...     "Development Server": "AA:BB:CC:DD:EE:FF",
...     "Testing Machine": "11:22:33:44:55:66",
...     "Database Server": "99:88:77:66:55:44"
... }
>>> results = wake_multiple_devices(devices, subnet_broadcast="192.168.1.255")
>>> for device, success in results.items():
...     status = "OK" if success else "FAILED"
...     print(f"{device}: {status}")