From dd60a9843f088225497238e559828842c2e80c5d Mon Sep 17 00:00:00 2001
From: ooonea <35407790+ooonea@users.noreply.github.com>
Date: Sun, 19 Apr 2026 21:56:34 +0200
Subject: [PATCH 1/8] Fix correctness bugs in throttled.py and mmio.py

- get_msr_list(): sort numerically and skip non-digit entries (like
  /dev/cpu/microcode). os.listdir() ordering is arbitrary, so the
  previous code returned readmsr() results indexed by enumeration
  order rather than CPU number, breaking the cpu= and flatten=True
  paths on systems where listdir didn't happen to return CPUs in
  order; it would also crash with ValueError when /dev/cpu contained
  any non-CPU entry.
- readmsr/writemsr: close MSR file descriptors via try/finally so a
  failed read/write doesn't leak the fd.
- is_on_battery(): replace bare `raise` (no active exception) and
  catch-all `except:` with specific exceptions; restore the intended
  fall-through to the upower fallback.
- test_msr_rw_capabilities(): declare TESTMSR global. The previous
  `TESTMSR = True` only created a local, so the in-flight detection
  could not actually catch MSR errors and would call fatal() instead.
- power_thread(): always sleep at end of loop. Previously the
  exit_event.wait was only reached in the else-branch of the HWP
  check, so when HWP fired the thread spun without sleeping.
- set_disable_bdprochot(): use `read_value == 0` for the debug
  match check; `~read_value` was always truthy.
- main(): initialize cpuid = None so --force does not raise
  UnboundLocalError when starting the power thread.
- mmio.py: map /dev/mem with PROT_READ | PROT_WRITE so read32 is
  defined and portable; close fd if mmap.mmap raises; fix __str__
  which referenced nonexistent self.base/self.size; drop dead
  self._fd assignment.

Signed-off-by: ooonea <35407790+ooonea@users.noreply.github.com>
---
 mmio.py      | 12 ++++++----
 throttled.py | 65 +++++++++++++++++++++++++++++-----------------------
 2 files changed, 44 insertions(+), 33 deletions(-)

diff --git a/mmio.py b/mmio.py
index 4568af9..27c343b 100644
--- a/mmio.py
+++ b/mmio.py
@@ -61,8 +61,14 @@ class MMIO(object):
 
         try:
             self.mapping = mmap.mmap(
-                fd, self._aligned_size, flags=mmap.MAP_SHARED, prot=mmap.PROT_WRITE, offset=self._aligned_physaddr)
+                fd,
+                self._aligned_size,
+                flags=mmap.MAP_SHARED,
+                prot=mmap.PROT_READ | mmap.PROT_WRITE,
+                offset=self._aligned_physaddr,
+            )
         except OSError as e:
+            os.close(fd)
             raise MMIOError(e.errno, "Mapping /dev/mem: " + e.strerror)
 
         try:
@@ -126,9 +132,7 @@ class MMIO(object):
         self.mapping.close()
         self.mapping = None
 
-        self._fd = None
-
     # String representation
 
     def __str__(self):
-        return "MMIO 0x%08x (size=%d)" % (self.base, self.size)
+        return "MMIO 0x%08x (size=%d)" % (self._physaddr, self._size)
diff --git a/throttled.py b/throttled.py
index dcd474d..9396d5a 100755
--- a/throttled.py
+++ b/throttled.py
@@ -223,7 +223,8 @@ def warning(msg, oneshot=True, end='\n'):
 
 
 def get_msr_list():
-    return ['/dev/cpu/{:d}/msr'.format(int(x)) for x in os.listdir("/dev/cpu")]
+    cpus = sorted(int(x) for x in os.listdir('/dev/cpu') if x.isdigit())
+    return ['/dev/cpu/{:d}/msr'.format(cpu) for cpu in cpus]
 
 def writemsr(msr, val):
     msr_list = get_msr_list()
@@ -235,9 +236,11 @@ def writemsr(msr, val):
     try:
         for addr in msr_list:
             f = os.open(addr, os.O_WRONLY)
-            os.lseek(f, MSR_DICT[msr], os.SEEK_SET)
-            os.write(f, struct.pack('Q', val))
-            os.close(f)
+            try:
+                os.lseek(f, MSR_DICT[msr], os.SEEK_SET)
+                os.write(f, struct.pack('Q', val))
+            finally:
+                os.close(f)
     except (IOError, OSError) as e:
         if TESTMSR:
             raise e
@@ -267,9 +270,11 @@ def readmsr(msr, from_bit=0, to_bit=63, cpu=None, flatten=False):
         output = []
         for addr in msr_list:
             f = os.open(addr, os.O_RDONLY)
-            os.lseek(f, MSR_DICT[msr], os.SEEK_SET)
-            val = struct.unpack('Q', os.read(f, 8))[0]
-            os.close(f)
+            try:
+                os.lseek(f, MSR_DICT[msr], os.SEEK_SET)
+                val = struct.unpack('Q', os.read(f, 8))[0]
+            finally:
+                os.close(f)
             output.append(get_value_for_bits(val, from_bit, to_bit))
         if flatten:
             if len(set(output)) > 1:
@@ -312,15 +317,16 @@ def is_on_battery(config):
         for path in glob.glob(config.get('GENERAL', 'Sysfs_Power_Path', fallback=DEFAULT_SYSFS_POWER_PATH)):
             with open(path) as f:
                 return not bool(int(f.read()))
-        raise
-    except:
+    except (IOError, OSError, ValueError) as e:
+        warning('Sysfs_Power_Path read failed ({}). Trying upower method.'.format(e))
+    else:
         warning('No valid Sysfs_Power_Path found! Trying upower method')
     try:
         bus = dbus.SystemBus()
         proxy = bus.get_object('org.freedesktop.UPower', '/org/freedesktop/UPower')
         iface = dbus.Interface(proxy, 'org.freedesktop.DBus.Properties')
-        return iface.Get('org.freedesktop.UPower', 'OnBattery')
-    except:
+        return bool(iface.Get('org.freedesktop.UPower', 'OnBattery'))
+    except dbus.DBusException:
         pass
 
     warning('No valid power detection methods found. Assuming that the system is running on battery power.')
@@ -660,7 +666,7 @@ def set_disable_bdprochot():
     writemsr('MSR_POWER_CTL', new_val)
     if args.debug:
         read_value = readmsr('MSR_POWER_CTL', from_bit=0, to_bit=0)[0]
-        match = OK if ~read_value else ERR
+        match = OK if read_value == 0 else ERR
         log('[D] BDPROCHOT - write "{:#02x}" - read "{:#02x}" - match {}'.format(0, read_value, match))
 
 
@@ -790,8 +796,7 @@ def power_thread(config, regs, exit_event, cpuid):
             set_hwp(enable_hwp_mode)
             next_hwp_write = time() + HWP_INTERVAL
 
-        else:
-            exit_event.wait(wait_t)
+        exit_event.wait(wait_t)
 
 
 def check_kernel():
@@ -859,24 +864,25 @@ def check_cpu():
 
 
 def test_msr_rw_capabilities():
+    global TESTMSR
     TESTMSR = True
-
     try:
-        log('[I] Testing if undervolt is supported...')
-        get_undervolt()
-    except:
-        warning('Undervolt seems not to be supported by your system, disabling.')
-        UNSUPPORTED_FEATURES.append('UNDERVOLT')
+        try:
+            log('[I] Testing if undervolt is supported...')
+            get_undervolt()
+        except (IOError, OSError):
+            warning('Undervolt seems not to be supported by your system, disabling.')
+            UNSUPPORTED_FEATURES.append('UNDERVOLT')
 
-    try:
-        log('[I] Testing if HWP is supported...')
-        cur_val = readmsr('IA32_HWP_REQUEST', cpu=0)
-        writemsr('IA32_HWP_REQUEST', cur_val)
-    except:
-        warning('HWP seems not to be supported by your system, disabling.')
-        UNSUPPORTED_FEATURES.append('HWP')
-
-    TESTMSR = False
+        try:
+            log('[I] Testing if HWP is supported...')
+            cur_val = readmsr('IA32_HWP_REQUEST', cpu=0)
+            writemsr('IA32_HWP_REQUEST', cur_val)
+        except (IOError, OSError):
+            warning('HWP seems not to be supported by your system, disabling.')
+            UNSUPPORTED_FEATURES.append('HWP')
+    finally:
+        TESTMSR = False
 
 
 def monitor(exit_event, wait):
@@ -958,6 +964,7 @@ def main():
             args.log = None
             fatal('Unable to write to the log file!')
 
+    cpuid = None
     if not args.force:
         check_kernel()
         cpuid = check_cpu()

From a6a95bdc1c6a4622eaf53b22e6d19dd68f5a0a2f Mon Sep 17 00:00:00 2001
From: ooonea <35407790+ooonea@users.noreply.github.com>
Date: Sun, 19 Apr 2026 21:56:34 +0200
Subject: [PATCH 2/8] Improve robustness: tighten exception handling and skip
 missing MSRs

- Replace bare `except:` clauses in set_msr_allow_writes, check_cpu,
  and the --log file open with specific exception types so we no
  longer swallow KeyboardInterrupt/SystemExit and we surface the
  actual error in the log.
- check_cpu now re-raises SystemExit instead of calling sys.exit
  again, and includes the underlying error in the fatal message.
- power_thread skips the PKG_POWER_LIMIT (and matching MCHBAR) write
  if no package power limit was computed, instead of KeyError'ing
  when all PL1/PL2 entries are commented out in the config.
- Strip ANSI color escapes when writing to the --log file. The
  OK/ERR/LIM constants embed ANSI codes intended for terminals and
  were leaking into log files.

Signed-off-by: ooonea <35407790+ooonea@users.noreply.github.com>
---
 throttled.py | 66 ++++++++++++++++++++++++++++------------------------
 1 file changed, 35 insertions(+), 31 deletions(-)

diff --git a/throttled.py b/throttled.py
index 9396d5a..12bc08b 100755
--- a/throttled.py
+++ b/throttled.py
@@ -195,30 +195,33 @@ LIM = bcolors.YELLOW + bcolors.BOLD + 'LIM' + bcolors.RESET
 
 log_history = set()
 
+ANSI_ESCAPE_RE = re.compile(r'\x1b\[[0-9;]*m')
+
+
+def _format(prefix, msg):
+    tstamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]
+    if args.log:
+        return '{:s}: {:s}{:s}'.format(tstamp, prefix, ANSI_ESCAPE_RE.sub('', msg))
+    return '{:s}{:s}'.format(prefix, msg)
+
 
 def log(msg, oneshot=False, end='\n'):
     outfile = args.log if args.log else sys.stdout
     if msg.strip() not in log_history or oneshot is False:
-        tstamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]
-        full_msg = '{:s}: {:s}'.format(tstamp, msg) if args.log else msg
-        print(full_msg, file=outfile, end=end)
+        print(_format('', msg), file=outfile, end=end)
         log_history.add(msg.strip())
 
 
 def fatal(msg, code=1, end='\n'):
     outfile = args.log if args.log else sys.stderr
-    tstamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]
-    full_msg = '{:s}: [E] {:s}'.format(tstamp, msg) if args.log else '[E] {:s}'.format(msg)
-    print(full_msg, file=outfile, end=end)
+    print(_format('[E] ', msg), file=outfile, end=end)
     sys.exit(code)
 
 
 def warning(msg, oneshot=True, end='\n'):
     outfile = args.log if args.log else sys.stderr
     if msg.strip() not in log_history or oneshot is False:
-        tstamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]
-        full_msg = '{:s}: [W] {:s}'.format(tstamp, msg) if args.log else '[W] {:s}'.format(msg)
-        print(full_msg, file=outfile, end=end)
+        print(_format('[W] ', msg), file=outfile, end=end)
         log_history.add(msg.strip())
 
 
@@ -308,7 +311,7 @@ def set_msr_allow_writes():
         try:
             with open('/sys/module/msr/parameters/allow_writes', 'w') as f:
                 f.write('on')
-        except:
+        except OSError:
             warning('Unable to set MSR allow_writes to on. You might experience warnings in kernel logs.')
 
 
@@ -754,28 +757,29 @@ def power_thread(config, regs, exit_event, cpuid):
                 )
 
         # set PL1/2 on MSR
-        write_value = regs[power['source']]['MSR_PKG_POWER_LIMIT']
-        writemsr('MSR_PKG_POWER_LIMIT', write_value)
-        if args.debug:
-            read_value = readmsr('MSR_PKG_POWER_LIMIT', 0, 55, flatten=True)
-            match = OK if write_value == read_value else ERR
-            log(
-                '[D] MSR PACKAGE_POWER_LIMIT - write {:#x} - read {:#x} - match {}'.format(
-                    write_value, read_value, match
-                )
-            )
-        if mchbar_mmio is not None:
-            # set MCHBAR register to the same PL1/2 values
-            mchbar_mmio.write32(0, write_value & 0xFFFFFFFF)
-            mchbar_mmio.write32(4, write_value >> 32)
+        if 'MSR_PKG_POWER_LIMIT' in regs[power['source']]:
+            write_value = regs[power['source']]['MSR_PKG_POWER_LIMIT']
+            writemsr('MSR_PKG_POWER_LIMIT', write_value)
             if args.debug:
-                read_value = mchbar_mmio.read32(0) | (mchbar_mmio.read32(4) << 32)
+                read_value = readmsr('MSR_PKG_POWER_LIMIT', 0, 55, flatten=True)
                 match = OK if write_value == read_value else ERR
                 log(
-                    '[D] MCHBAR PACKAGE_POWER_LIMIT - write {:#x} - read {:#x} - match {}'.format(
+                    '[D] MSR PACKAGE_POWER_LIMIT - write {:#x} - read {:#x} - match {}'.format(
                         write_value, read_value, match
                     )
                 )
+            if mchbar_mmio is not None:
+                # set MCHBAR register to the same PL1/2 values
+                mchbar_mmio.write32(0, write_value & 0xFFFFFFFF)
+                mchbar_mmio.write32(4, write_value >> 32)
+                if args.debug:
+                    read_value = mchbar_mmio.read32(0) | (mchbar_mmio.read32(4) << 32)
+                    match = OK if write_value == read_value else ERR
+                    log(
+                        '[D] MCHBAR PACKAGE_POWER_LIMIT - write {:#x} - read {:#x} - match {}'.format(
+                            write_value, read_value, match
+                        )
+                    )
 
         # Disable BDPROCHOT
         disable_bdprochot = config.getboolean(power['source'], 'Disable_BDPROCHOT', fallback=None)
@@ -858,9 +862,9 @@ def check_cpu():
         log('[I] Detected CPU architecture: Intel {:s}'.format(supported_cpus[cpuid]))
         return cpuid
     except SystemExit:
-        sys.exit(1)
-    except:
-        fatal('Unable to identify CPU model.')
+        raise
+    except (OSError, KeyError, ValueError) as e:
+        fatal('Unable to identify CPU model: {}'.format(e))
 
 
 def test_msr_rw_capabilities():
@@ -960,9 +964,9 @@ def main():
     if args.log:
         try:
             args.log = open(args.log, 'w')
-        except:
+        except OSError as e:
             args.log = None
-            fatal('Unable to write to the log file!')
+            fatal('Unable to write to the log file: {}'.format(e))
 
     cpuid = None
     if not args.force:

From 2fd89b747bad97bdbd00ca98c9fd5216c90ae95b Mon Sep 17 00:00:00 2001
From: ooonea <35407790+ooonea@users.noreply.github.com>
Date: Sun, 19 Apr 2026 21:56:35 +0200
Subject: [PATCH 3/8] Clarity: drop dead code, add docstrings, prune
 dev-commentary

- Remove `from __future__ import print_function` (Python 3 only).
- mmio.py: drop the Python 2 `long = int` shim and `(int, long)`
  isinstance checks; make __enter__ return self so `with MMIO(...)`
  works as expected.
- Drop the redundant `value = config.set(...)` assignment in
  load_config (config.set returns None).
- Replace the personal "for my CPU" comments and the
  "ugly code, just testing..." comment with brief docstrings that
  describe what each helper computes.
- Add concise one-line docstrings to the public-ish functions
  (readmsr/writemsr, get_value_for_bits, is_on_battery, the various
  MSR getters/setters, load_config, calc_reg_values, power_thread,
  monitor, main, etc.) so a reader can navigate without reading the
  body.

Signed-off-by: ooonea <35407790+ooonea@users.noreply.github.com>
---
 mmio.py      | 32 +++++++++++++-------------------
 throttled.py | 52 +++++++++++++++++++++++++++++++++++++++-------------
 2 files changed, 52 insertions(+), 32 deletions(-)

diff --git a/mmio.py b/mmio.py
index 27c343b..cfd610f 100644
--- a/mmio.py
+++ b/mmio.py
@@ -1,19 +1,13 @@
-'''
+"""
 Stripped down version from https://github.com/vsergeev/python-periphery/blob/master/periphery/mmio.py
-'''
+"""
 import mmap
 import os
 import struct
-import sys
-
-# Alias long to int on Python 3
-if sys.version_info[0] >= 3:
-    long = int
 
 
 class MMIOError(IOError):
     """Base class for MMIO errors."""
-    pass
 
 
 class MMIO(object):
@@ -21,8 +15,8 @@ class MMIO(object):
         """Instantiate an MMIO object and map the region of physical memory
         specified by the address base `physaddr` and size `size` in bytes.
         Args:
-            physaddr (int, long): base physical address of memory region.
-            size (int, long): size of memory region.
+            physaddr (int): base physical address of memory region.
+            size (int): size of memory region.
         Returns:
             MMIO: MMIO object.
         Raises:
@@ -36,15 +30,15 @@ class MMIO(object):
         self.close()
 
     def __enter__(self):
-        pass
+        return self
 
     def __exit__(self, t, value, traceback):
         self.close()
 
     def _open(self, physaddr, size):
-        if not isinstance(physaddr, (int, long)):
+        if not isinstance(physaddr, int):
             raise TypeError("Invalid physaddr type, should be integer.")
-        if not isinstance(size, (int, long)):
+        if not isinstance(size, int):
             raise TypeError("Invalid size type, should be integer.")
 
         pagesize = os.sysconf(os.sysconf_names['SC_PAGESIZE'])
@@ -89,14 +83,14 @@ class MMIO(object):
         """Read 32-bits from the specified `offset` in bytes, relative to the
         base physical address of the MMIO region.
         Args:
-            offset (int, long): offset from base physical address, in bytes.
+            offset (int): offset from base physical address, in bytes.
         Returns:
             int: 32-bit value read.
         Raises:
             TypeError: if `offset` type is invalid.
             ValueError: if `offset` is out of bounds.
         """
-        if not isinstance(offset, (int, long)):
+        if not isinstance(offset, int):
             raise TypeError("Invalid offset type, should be integer.")
 
         offset = self._adjust_offset(offset)
@@ -107,15 +101,15 @@ class MMIO(object):
         """Write 32-bits to the specified `offset` in bytes, relative to the
         base physical address of the MMIO region.
         Args:
-            offset (int, long): offset from base physical address, in bytes.
-            value (int, long): 32-bit value to write.
+            offset (int): offset from base physical address, in bytes.
+            value (int): 32-bit value to write.
         Raises:
             TypeError: if `offset` or `value` type are invalid.
             ValueError: if `offset` or `value` are out of bounds.
         """
-        if not isinstance(offset, (int, long)):
+        if not isinstance(offset, int):
             raise TypeError("Invalid offset type, should be integer.")
-        if not isinstance(value, (int, long)):
+        if not isinstance(value, int):
             raise TypeError("Invalid value type, should be integer.")
         if value < 0 or value > 0xffffffff:
             raise ValueError("Value out of bounds.")
diff --git a/throttled.py b/throttled.py
index 12bc08b..96e634e 100755
--- a/throttled.py
+++ b/throttled.py
@@ -1,6 +1,4 @@
 #!/usr/bin/env python3
-from __future__ import print_function
-
 import argparse
 import configparser
 import glob
@@ -226,10 +224,13 @@ def warning(msg, oneshot=True, end='\n'):
 
 
 def get_msr_list():
+    """Return the per-CPU MSR device paths in CPU-index order."""
     cpus = sorted(int(x) for x in os.listdir('/dev/cpu') if x.isdigit())
     return ['/dev/cpu/{:d}/msr'.format(cpu) for cpu in cpus]
 
+
 def writemsr(msr, val):
+    """Write a 64-bit value to the named MSR on every online CPU."""
     msr_list = get_msr_list()
     if not os.path.exists(msr_list[0]):
         try:
@@ -258,8 +259,12 @@ def writemsr(msr, val):
             raise e
 
 
-# returns the value between from_bit and to_bit as unsigned long
 def readmsr(msr, from_bit=0, to_bit=63, cpu=None, flatten=False):
+    """Read the named MSR and return the [from_bit, to_bit] field as
+    an unsigned integer. By default returns one value per CPU; with
+    cpu=N returns just CPU N, with flatten=True returns the shared value
+    (warning if CPUs disagree).
+    """
     assert cpu is None or cpu in range(cpu_count())
     if from_bit > to_bit:
         fatal('Wrong readmsr bit params')
@@ -296,11 +301,13 @@ def readmsr(msr, from_bit=0, to_bit=63, cpu=None, flatten=False):
 
 
 def get_value_for_bits(val, from_bit=0, to_bit=63):
+    """Extract bits [from_bit, to_bit] (inclusive) from val."""
     mask = sum(2 ** x for x in range(from_bit, to_bit + 1))
     return (val & mask) >> from_bit
 
 
 def set_msr_allow_writes():
+    """Try to enable msr.allow_writes; tolerate kernels that don't expose it."""
     log('[I] Trying to unlock MSR allow_writes.')
     if not os.path.exists('/sys/module/msr'):
         try:
@@ -316,6 +323,9 @@ def set_msr_allow_writes():
 
 
 def is_on_battery(config):
+    """Return True if the system is on battery power; falls back to UPower
+    over D-Bus if the configured sysfs path is unreadable.
+    """
     try:
         for path in glob.glob(config.get('GENERAL', 'Sysfs_Power_Path', fallback=DEFAULT_SYSFS_POWER_PATH)):
             with open(path) as f:
@@ -337,6 +347,7 @@ def is_on_battery(config):
 
 
 def get_cpu_platform_info():
+    """Decode MSR_PLATFORM_INFO into a dict of named feature bits."""
     features_msr_value = readmsr('MSR_PLATFORM_INFO', cpu=0)
     cpu_platform_info = {}
     for key, value in platform_info_bits.items():
@@ -345,7 +356,7 @@ def get_cpu_platform_info():
 
 
 def get_reset_thermal_status():
-    # read thermal status
+    """Read IA32_THERM_STATUS for every CPU, then clear the sticky log bits."""
     thermal_status_msr_value = readmsr('IA32_THERM_STATUS')
     thermal_status = []
     for core in range(cpu_count()):
@@ -359,23 +370,23 @@ def get_reset_thermal_status():
 
 
 def get_time_unit():
-    # 0.000977 is the time unit of my CPU
-    # TODO formula might be different for other CPUs
+    """Return the RAPL time unit in seconds (Intel SDM Vol. 4, MSR 0x606)."""
     return 1.0 / 2 ** readmsr('MSR_RAPL_POWER_UNIT', 16, 19, cpu=0)
 
 
 def get_power_unit():
-    # 0.125 is the power unit of my CPU
-    # TODO formula might be different for other CPUs
+    """Return the RAPL power unit in watts (Intel SDM Vol. 4, MSR 0x606)."""
     return 1.0 / 2 ** readmsr('MSR_RAPL_POWER_UNIT', 0, 3, cpu=0)
 
 
 def get_critical_temp():
-    # the critical temperature for my CPU is 100 'C
+    """Return the package critical temperature offset in degrees Celsius."""
     return readmsr('MSR_TEMPERATURE_TARGET', 16, 23, cpu=0)
 
 
 def get_cur_pkg_power_limits():
+    """Return the current PL1/PL2 power and time-window fields from
+    MSR_PKG_POWER_LIMIT."""
     value = readmsr('MSR_PKG_POWER_LIMIT', 0, 55, flatten=True)
     return {
         'PL1': get_value_for_bits(value, 0, 14),
@@ -386,6 +397,8 @@ def get_cur_pkg_power_limits():
 
 
 def calc_time_window_vars(t):
+    """Encode a time-window duration (s) as the (Y, Z) pair used by
+    MSR_PKG_POWER_LIMIT."""
     time_unit = get_time_unit()
     for Y in range(2 ** 5):
         for Z in range(2 ** 2):
@@ -413,6 +426,7 @@ def calc_undervolt_mv(msr_value):
 
 
 def get_undervolt(plane=None, convert=False):
+    """Read the current undervolt offset from one or all voltage planes."""
     if 'UNDERVOLT' in UNSUPPORTED_FEATURES:
         return 0
     planes = [plane] if plane in VOLTAGE_PLANES else VOLTAGE_PLANES
@@ -426,6 +440,7 @@ def get_undervolt(plane=None, convert=False):
 
 
 def undervolt(config):
+    """Apply the undervolt offsets from the config to all voltage planes."""
     if ('UNDERVOLT.{:s}'.format(power['source']) not in config and 'UNDERVOLT' not in config) or (
         'UNDERVOLT' in UNSUPPORTED_FEATURES
     ):
@@ -464,6 +479,7 @@ def calc_icc_max_amp(msr_value):
 
 
 def get_icc_max(plane=None, convert=False):
+    """Read the IccMax setting from one or all current planes."""
     planes = [plane] if plane in CURRENT_PLANES else CURRENT_PLANES
     out = {}
     for plane in planes:
@@ -475,6 +491,7 @@ def get_icc_max(plane=None, convert=False):
 
 
 def set_icc_max(config):
+    """Apply the IccMax limits from the config to all current planes."""
     for plane in CURRENT_PLANES:
         try:
             write_current_amp = config.getfloat(
@@ -498,6 +515,7 @@ def set_icc_max(config):
 
 
 def load_config():
+    """Parse the config file, validating and clamping out-of-range values."""
     config = configparser.ConfigParser()
     config.read(args.config)
 
@@ -506,7 +524,7 @@ def load_config():
         for option in ('Update_Rate_s', 'PL1_Tdp_W', 'PL1_Duration_s', 'PL2_Tdp_W', 'PL2_Duration_S'):
             value = config.getfloat(power_source, option, fallback=None)
             if value is not None:
-                value = config.set(power_source, option, str(max(0.001, value)))
+                config.set(power_source, option, str(max(0.001, value)))
             elif option == 'Update_Rate_s':
                 fatal('The mandatory "Update_Rate_s" parameter is missing.')
 
@@ -574,12 +592,12 @@ def load_config():
 
 
 def calc_reg_values(platform_info, config):
+    """Compute the MSR values to apply for each power source from the config."""
     regs = defaultdict(dict)
     for power_source in ('AC', 'BATTERY'):
         if platform_info['feature_programmable_temperature_target'] != 1:
             warning("Setting temperature target is not supported by this CPU")
         else:
-            # the critical temperature for my CPU is 100 'C
             critical_temp = get_critical_temp()
             # update the allowed temp range to keep at least 3 'C from the CPU critical temperature
             global TRIP_TEMP_RANGE
@@ -647,6 +665,7 @@ def calc_reg_values(platform_info, config):
 
 
 def set_hwp(performance_mode):
+    """Set the IA32_HWP_REQUEST energy/performance preference field."""
     if performance_mode not in (True, False) or 'HWP' in UNSUPPORTED_FEATURES:
         return
     # set HWP energy performance preference
@@ -662,7 +681,7 @@ def set_hwp(performance_mode):
 
 
 def set_disable_bdprochot():
-    # Disable BDPROCHOT
+    """Clear bit 0 of MSR_POWER_CTL to disable BDPROCHOT."""
     cur_val = readmsr('MSR_POWER_CTL', flatten=True)
     new_val = cur_val & 0xFFFFFFFFFFFFFFFE
 
@@ -674,6 +693,7 @@ def set_disable_bdprochot():
 
 
 def get_config_write_time():
+    """Return the config file's mtime, or None if it doesn't exist."""
     try:
         return os.stat(args.config).st_mtime
     except FileNotFoundError:
@@ -681,6 +701,7 @@ def get_config_write_time():
 
 
 def reload_config():
+    """Re-read the config and re-apply undervolt, IccMax and HWP settings."""
     config = load_config()
     regs = calc_reg_values(get_cpu_platform_info(), config)
     undervolt(config)
@@ -691,6 +712,7 @@ def reload_config():
 
 
 def power_thread(config, regs, exit_event, cpuid):
+    """Daemon main loop: periodically (re-)apply throttling MSRs."""
     try:
         MCHBAR_BASE = int(check_output(('setpci', '-s', '0:0.0', '48.l')), 16)
     except CalledProcessError:
@@ -804,6 +826,7 @@ def power_thread(config, regs, exit_event, cpuid):
 
 
 def check_kernel():
+    """Verify we run as root and that the kernel exposes MSR/devmem."""
     if os.geteuid() != 0:
         fatal('No root no party. Try again with sudo.')
 
@@ -830,6 +853,7 @@ def check_kernel():
 
 
 def check_cpu():
+    """Identify the CPU from /proc/cpuinfo and refuse to run on unsupported models."""
     try:
         with open('/proc/cpuinfo') as f:
             cpuinfo = {}
@@ -868,6 +892,7 @@ def check_cpu():
 
 
 def test_msr_rw_capabilities():
+    """Probe undervolt and HWP support; mark unavailable features as such."""
     global TESTMSR
     TESTMSR = True
     try:
@@ -890,6 +915,7 @@ def test_msr_rw_capabilities():
 
 
 def monitor(exit_event, wait):
+    """Live-display throttling causes and per-domain power until exit_event is set."""
     wait = max(0.1, wait)
     rapl_power_unit = 0.5 ** readmsr('MSR_RAPL_POWER_UNIT', from_bit=8, to_bit=12, cpu=0)
     power_plane_msr = {
@@ -917,7 +943,6 @@ def monitor(exit_event, wait):
         offsets = {'Thermal': 0, 'Power': 10, 'Current': 12, 'Cross-domain (e.g. GPU)': 14}
         output = ('{:s}: {:s}'.format(cause, LIM if bool((value >> offsets[cause]) & 1) else OK) for cause in offsets)
 
-        # ugly code, just testing...
         vcore = readmsr('IA32_PERF_STATUS', from_bit=32, to_bit=47, cpu=0) / (2.0 ** 13) * 1000
         stats2 = {'VCore': '{:.0f} mV'.format(vcore)}
         total = 0.0
@@ -943,6 +968,7 @@ def monitor(exit_event, wait):
 
 
 def main():
+    """Daemon entrypoint: parse args, validate platform, start power thread."""
     global args
 
     parser = argparse.ArgumentParser()

From 1b815ea3fe28bf3beac40e9b883cb1280786b2fe Mon Sep 17 00:00:00 2001
From: ooonea <35407790+ooonea@users.noreply.github.com>
Date: Sun, 19 Apr 2026 21:56:35 +0200
Subject: [PATCH 4/8] Style: convert to f-strings, add type hints to mmio, run
 Black

- Replace remaining .format() calls with f-strings (ruff UP032 plus
  manual cleanups for the formats ruff couldn't rewrite); replace
  the lone %-format in mmio.MMIO.__str__ as well.
- Hoist `UNDERVOLT.<source>` and `ICCMAX.<source>` section names to
  local variables in undervolt() / set_icc_max() so the f-string
  appears once.
- Add type hints to the public mmio.MMIO methods (the only real
  module boundary in the project).
- Run Black with the existing pyproject.toml config (line-length 120,
  skip-string-normalization). No semantic changes.

Signed-off-by: ooonea <35407790+ooonea@users.noreply.github.com>
---
 mmio.py      |  21 ++++----
 throttled.py | 134 +++++++++++++++++++++------------------------------
 2 files changed, 66 insertions(+), 89 deletions(-)

diff --git a/mmio.py b/mmio.py
index cfd610f..e07411a 100644
--- a/mmio.py
+++ b/mmio.py
@@ -1,6 +1,7 @@
 """
 Stripped down version from https://github.com/vsergeev/python-periphery/blob/master/periphery/mmio.py
 """
+
 import mmap
 import os
 import struct
@@ -11,7 +12,7 @@ class MMIOError(IOError):
 
 
 class MMIO(object):
-    def __init__(self, physaddr, size):
+    def __init__(self, physaddr: int, size: int) -> None:
         """Instantiate an MMIO object and map the region of physical memory
         specified by the address base `physaddr` and size `size` in bytes.
         Args:
@@ -35,7 +36,7 @@ class MMIO(object):
     def __exit__(self, t, value, traceback):
         self.close()
 
-    def _open(self, physaddr, size):
+    def _open(self, physaddr: int, size: int) -> None:
         if not isinstance(physaddr, int):
             raise TypeError("Invalid physaddr type, should be integer.")
         if not isinstance(size, int):
@@ -79,7 +80,7 @@ class MMIO(object):
         if (offset + length) > self._aligned_size:
             raise ValueError("Offset out of bounds.")
 
-    def read32(self, offset):
+    def read32(self, offset: int) -> int:
         """Read 32-bits from the specified `offset` in bytes, relative to the
         base physical address of the MMIO region.
         Args:
@@ -95,9 +96,9 @@ class MMIO(object):
 
         offset = self._adjust_offset(offset)
         self._validate_offset(offset, 4)
-        return struct.unpack("=L", self.mapping[offset:offset + 4])[0]
+        return struct.unpack("=L", self.mapping[offset : offset + 4])[0]
 
-    def write32(self, offset, value):
+    def write32(self, offset: int, value: int) -> None:
         """Write 32-bits to the specified `offset` in bytes, relative to the
         base physical address of the MMIO region.
         Args:
@@ -111,14 +112,14 @@ class MMIO(object):
             raise TypeError("Invalid offset type, should be integer.")
         if not isinstance(value, int):
             raise TypeError("Invalid value type, should be integer.")
-        if value < 0 or value > 0xffffffff:
+        if value < 0 or value > 0xFFFFFFFF:
             raise ValueError("Value out of bounds.")
 
         offset = self._adjust_offset(offset)
         self._validate_offset(offset, 4)
-        self.mapping[offset:offset + 4] = struct.pack("=L", value)
+        self.mapping[offset : offset + 4] = struct.pack("=L", value)
 
-    def close(self):
+    def close(self) -> None:
         """Unmap the MMIO object's mapped physical memory."""
         if self.mapping is None:
             return
@@ -128,5 +129,5 @@ class MMIO(object):
 
     # String representation
 
-    def __str__(self):
-        return "MMIO 0x%08x (size=%d)" % (self._physaddr, self._size)
+    def __str__(self) -> str:
+        return f"MMIO 0x{self._physaddr:08x} (size={self._size:d})"
diff --git a/throttled.py b/throttled.py
index 96e634e..574a228 100755
--- a/throttled.py
+++ b/throttled.py
@@ -197,10 +197,10 @@ ANSI_ESCAPE_RE = re.compile(r'\x1b\[[0-9;]*m')
 
 
 def _format(prefix, msg):
-    tstamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]
     if args.log:
-        return '{:s}: {:s}{:s}'.format(tstamp, prefix, ANSI_ESCAPE_RE.sub('', msg))
-    return '{:s}{:s}'.format(prefix, msg)
+        tstamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]
+        return f'{tstamp}: {prefix}{ANSI_ESCAPE_RE.sub("", msg)}'
+    return f'{prefix}{msg}'
 
 
 def log(msg, oneshot=False, end='\n'):
@@ -226,7 +226,7 @@ def warning(msg, oneshot=True, end='\n'):
 def get_msr_list():
     """Return the per-CPU MSR device paths in CPU-index order."""
     cpus = sorted(int(x) for x in os.listdir('/dev/cpu') if x.isdigit())
-    return ['/dev/cpu/{:d}/msr'.format(cpu) for cpu in cpus]
+    return [f'/dev/cpu/{cpu:d}/msr' for cpu in cpus]
 
 
 def writemsr(msr, val):
@@ -250,11 +250,11 @@ def writemsr(msr, val):
             raise e
         if e.errno == EPERM or e.errno == EACCES:
             fatal(
-                'Unable to write to MSR {} ({:x}). Try to disable Secure Boot '
-                'and check if your kernel does not restrict access to MSR.'.format(msr, MSR_DICT[msr])
+                f'Unable to write to MSR {msr} ({MSR_DICT[msr]:x}). Try to disable Secure Boot '
+                'and check if your kernel does not restrict access to MSR.'
             )
         elif e.errno == EIO:
-            fatal('Unable to write to MSR {} ({:x}). Unknown error.'.format(msr, MSR_DICT[msr]))
+            fatal(f'Unable to write to MSR {msr} ({MSR_DICT[msr]:x}). Unknown error.')
         else:
             raise e
 
@@ -286,23 +286,23 @@ def readmsr(msr, from_bit=0, to_bit=63, cpu=None, flatten=False):
             output.append(get_value_for_bits(val, from_bit, to_bit))
         if flatten:
             if len(set(output)) > 1:
-                warning('Found multiple values for {:s} ({:x}). This should never happen.'.format(msr, MSR_DICT[msr]))
+                warning(f'Found multiple values for {msr:s} ({MSR_DICT[msr]:x}). This should never happen.')
             return output[0]
         return output[cpu] if cpu is not None else output
     except (IOError, OSError) as e:
         if TESTMSR:
             raise e
         if e.errno == EPERM or e.errno == EACCES:
-            fatal('Unable to read from MSR {} ({:x}). Try to disable Secure Boot.'.format(msr, MSR_DICT[msr]))
+            fatal(f'Unable to read from MSR {msr} ({MSR_DICT[msr]:x}). Try to disable Secure Boot.')
         elif e.errno == EIO:
-            fatal('Unable to read to MSR {} ({:x}). Unknown error.'.format(msr, MSR_DICT[msr]))
+            fatal(f'Unable to read to MSR {msr} ({MSR_DICT[msr]:x}). Unknown error.')
         else:
             raise e
 
 
 def get_value_for_bits(val, from_bit=0, to_bit=63):
     """Extract bits [from_bit, to_bit] (inclusive) from val."""
-    mask = sum(2 ** x for x in range(from_bit, to_bit + 1))
+    mask = sum(2**x for x in range(from_bit, to_bit + 1))
     return (val & mask) >> from_bit
 
 
@@ -331,7 +331,7 @@ def is_on_battery(config):
             with open(path) as f:
                 return not bool(int(f.read()))
     except (IOError, OSError, ValueError) as e:
-        warning('Sysfs_Power_Path read failed ({}). Trying upower method.'.format(e))
+        warning(f'Sysfs_Power_Path read failed ({e}). Trying upower method.')
     else:
         warning('No valid Sysfs_Power_Path found! Trying upower method')
     try:
@@ -400,9 +400,9 @@ def calc_time_window_vars(t):
     """Encode a time-window duration (s) as the (Y, Z) pair used by
     MSR_PKG_POWER_LIMIT."""
     time_unit = get_time_unit()
-    for Y in range(2 ** 5):
-        for Z in range(2 ** 2):
-            if t <= (2 ** Y) * (1.0 + Z / 4.0) * time_unit:
+    for Y in range(2**5):
+        for Z in range(2**2):
+            if t <= (2**Y) * (1.0 + Z / 4.0) * time_unit:
                 return (Y, Z)
     raise ValueError('Unable to find a good combination!')
 
@@ -441,14 +441,11 @@ def get_undervolt(plane=None, convert=False):
 
 def undervolt(config):
     """Apply the undervolt offsets from the config to all voltage planes."""
-    if ('UNDERVOLT.{:s}'.format(power['source']) not in config and 'UNDERVOLT' not in config) or (
-        'UNDERVOLT' in UNSUPPORTED_FEATURES
-    ):
+    section = f"UNDERVOLT.{power['source']}"
+    if (section not in config and 'UNDERVOLT' not in config) or 'UNDERVOLT' in UNSUPPORTED_FEATURES:
         return
     for plane in VOLTAGE_PLANES:
-        write_offset_mv = config.getfloat(
-            'UNDERVOLT.{:s}'.format(power['source']), plane, fallback=config.getfloat('UNDERVOLT', plane, fallback=0.0)
-        )
+        write_offset_mv = config.getfloat(section, plane, fallback=config.getfloat('UNDERVOLT', plane, fallback=0.0))
         write_value = calc_undervolt_msr(plane, write_offset_mv)
         writemsr('MSR_OC_MAILBOX', write_value)
         if args.debug:
@@ -457,9 +454,7 @@ def undervolt(config):
             read_offset_mv = calc_undervolt_mv(read_value)
             match = OK if write_value == read_value else ERR
             log(
-                '[D] Undervolt plane {:s} - write {:.0f} mV ({:#x}) - read {:.0f} mV ({:#x}) - match {}'.format(
-                    plane, write_offset_mv, write_value, read_offset_mv, read_value, match
-                )
+                f'[D] Undervolt plane {plane:s} - write {write_offset_mv:.0f} mV ({write_value:#x}) - read {read_offset_mv:.0f} mV ({read_value:#x}) - match {match}'
             )
 
 
@@ -492,10 +487,11 @@ def get_icc_max(plane=None, convert=False):
 
 def set_icc_max(config):
     """Apply the IccMax limits from the config to all current planes."""
+    section = f"ICCMAX.{power['source']}"
     for plane in CURRENT_PLANES:
         try:
             write_current_amp = config.getfloat(
-                'ICCMAX.{:s}'.format(power['source']), plane, fallback=config.getfloat('ICCMAX', plane, fallback=-1.0)
+                section, plane, fallback=config.getfloat('ICCMAX', plane, fallback=-1.0)
             )
             if write_current_amp > 0:
                 write_value = calc_icc_max_msr(plane, write_current_amp)
@@ -506,9 +502,7 @@ def set_icc_max(config):
                     read_current_A = calc_icc_max_amp(read_value)
                     match = OK if write_value == read_value else ERR
                     log(
-                        '[D] IccMax plane {:s} - write {:.2f} A ({:#x}) - read {:.2f} A ({:#x}) - match {}'.format(
-                            plane, write_current_amp, write_value, read_current_A, read_value, match
-                        )
+                        f'[D] IccMax plane {plane:s} - write {write_current_amp:.2f} A ({write_value:#x}) - read {read_current_A:.2f} A ({read_value:#x}) - match {match}'
                     )
         except (configparser.NoSectionError, configparser.NoOptionError):
             pass
@@ -534,9 +528,7 @@ def load_config():
             if trip_temp != valid_trip_temp:
                 config.set(power_source, 'Trip_Temp_C', str(valid_trip_temp))
                 log(
-                    '[!] Overriding invalid "Trip_Temp_C" value in "{:s}": {:.1f} -> {:.1f}'.format(
-                        power_source, trip_temp, valid_trip_temp
-                    )
+                    f'[!] Overriding invalid "Trip_Temp_C" value in "{power_source:s}": {trip_temp:.1f} -> {valid_trip_temp:.1f}'
                 )
 
     # fix any invalid value (ie. > 0) in the undervolt settings
@@ -548,9 +540,7 @@ def load_config():
                 if value != valid_value:
                     config.set(key, plane, str(valid_value))
                     log(
-                        '[!] Overriding invalid "{:s}" value in "{:s}" voltage plane: {:.0f} -> {:.0f}'.format(
-                            key, plane, value, valid_value
-                        )
+                        f'[!] Overriding invalid "{key:s}" value in "{plane:s}" voltage plane: {value:.0f} -> {valid_value:.0f}'
                     )
 
     # handle the case where only one of UNDERVOLT.AC, UNDERVOLT.BATTERY keys exists
@@ -581,7 +571,7 @@ def load_config():
                         raise ValueError
                     iccmax_enabled = True
                 except ValueError:
-                    warning('Invalid value for {:s} in {:s}'.format(plane, key), oneshot=False)
+                    warning(f'Invalid value for {plane:s} in {key:s}', oneshot=False)
                     config.remove_option(key, plane)
                 except configparser.NoOptionError:
                     pass
@@ -608,7 +598,7 @@ def calc_reg_values(platform_info, config):
                 trip_offset = int(round(critical_temp - Trip_Temp_C))
                 regs[power_source]['MSR_TEMPERATURE_TARGET'] = trip_offset << 24
             else:
-                log('[I] {:s} trip temperature is disabled in config.'.format(power_source))
+                log(f'[I] {power_source:s} trip temperature is disabled in config.')
 
         power_unit = get_power_unit()
 
@@ -621,26 +611,26 @@ def calc_reg_values(platform_info, config):
             cur_pkg_power_limits = get_cur_pkg_power_limits()
             if PL1_Tdp_W is None:
                 PL1 = cur_pkg_power_limits['PL1']
-                log('[I] {:s} PL1_Tdp_W disabled in config.'.format(power_source))
+                log(f'[I] {power_source:s} PL1_Tdp_W disabled in config.')
             else:
                 PL1 = int(round(PL1_Tdp_W / power_unit))
 
             if PL1_Duration_s is None:
                 TW1 = cur_pkg_power_limits['TW1']
-                log('[I] {:s} PL1_Duration_s disabled in config.'.format(power_source))
+                log(f'[I] {power_source:s} PL1_Duration_s disabled in config.')
             else:
                 Y, Z = calc_time_window_vars(PL1_Duration_s)
                 TW1 = Y | (Z << 5)
 
             if PL2_Tdp_W is None:
                 PL2 = cur_pkg_power_limits['PL2']
-                log('[I] {:s} PL2_Tdp_W disabled in config.'.format(power_source))
+                log(f'[I] {power_source:s} PL2_Tdp_W disabled in config.')
             else:
                 PL2 = int(round(PL2_Tdp_W / power_unit))
 
             if PL2_Duration_s is None:
                 TW2 = cur_pkg_power_limits['TW2']
-                log('[I] {:s} PL2_Duration_s disabled in config.'.format(power_source))
+                log(f'[I] {power_source:s} PL2_Duration_s disabled in config.')
             else:
                 Y, Z = calc_time_window_vars(PL2_Duration_s)
                 TW2 = Y | (Z << 5)
@@ -649,7 +639,7 @@ def calc_reg_values(platform_info, config):
                 PL1 | (1 << 15) | (1 << 16) | (TW1 << 17) | (PL2 << 32) | (1 << 47) | (TW2 << 49)
             )
         else:
-            log('[I] {:s} package power limits are disabled in config.'.format(power_source))
+            log(f'[I] {power_source:s} package power limits are disabled in config.')
 
         # cTDP
         c_tdp_target_value = config.getint(power_source, 'cTDP', fallback=None)
@@ -677,7 +667,7 @@ def set_hwp(performance_mode):
     if args.debug:
         read_value = readmsr('IA32_HWP_REQUEST', from_bit=24, to_bit=31)[0]
         match = OK if hwp_mode == read_value else ERR
-        log('[D] HWP - write "{:#02x}" - read "{:#02x}" - match {}'.format(hwp_mode, read_value, match))
+        log(f'[D] HWP - write "{hwp_mode:#02x}" - read "{read_value:#02x}" - match {match}')
 
 
 def set_disable_bdprochot():
@@ -689,7 +679,7 @@ def set_disable_bdprochot():
     if args.debug:
         read_value = readmsr('MSR_POWER_CTL', from_bit=0, to_bit=0)[0]
         match = OK if read_value == 0 else ERR
-        log('[D] BDPROCHOT - write "{:#02x}" - read "{:#02x}" - match {}'.format(0, read_value, match))
+        log(f'[D] BDPROCHOT - write "{0:#02x}" - read "{read_value:#02x}" - match {match}')
 
 
 def get_config_write_time():
@@ -718,7 +708,7 @@ def power_thread(config, regs, exit_event, cpuid):
     except CalledProcessError:
         warning('Please ensure that "setpci" is in path. This is typically provided by the "pciutils" package.')
         warning('Trying to guess the MCHBAR address from the CPUID. This MIGHT NOT WORK!')
-        if cpuid in ((6, 140, 1),(6, 140, 2),(6, 141, 1),(6, 151, 2),(6, 151, 5), (6, 154, 3),(6, 154, 4)):
+        if cpuid in ((6, 140, 1), (6, 140, 2), (6, 141, 1), (6, 151, 2), (6, 151, 5), (6, 154, 3), (6, 154, 4)):
             MCHBAR_BASE = 0xFEDC0001
         else:
             MCHBAR_BASE = 0xFED10001
@@ -739,7 +729,7 @@ def power_thread(config, regs, exit_event, cpuid):
             thermal_status = get_reset_thermal_status()
             for index, core_thermal_status in enumerate(thermal_status):
                 for key, value in core_thermal_status.items():
-                    log('[D] core {} thermal status: {} = {}'.format(index, key.replace("_", " "), value))
+                    log(f'[D] core {index} thermal status: {key.replace("_", " ")} = {value}')
 
         # Reload config on changes (unless it's deleted)
         if config.getboolean('GENERAL', 'Autoreload', fallback=False):
@@ -759,11 +749,7 @@ def power_thread(config, regs, exit_event, cpuid):
             if args.debug:
                 read_value = readmsr('MSR_TEMPERATURE_TARGET', 24, 29, flatten=True)
                 match = OK if write_value >> 24 == read_value else ERR
-                log(
-                    '[D] TEMPERATURE_TARGET - write {:#x} - read {:#x} - match {}'.format(
-                        write_value >> 24, read_value, match
-                    )
-                )
+                log(f'[D] TEMPERATURE_TARGET - write {write_value >> 24:#x} - read {read_value:#x} - match {match}')
 
         # set cTDP
         if 'MSR_CONFIG_TDP_CONTROL' in regs[power['source']]:
@@ -772,11 +758,7 @@ def power_thread(config, regs, exit_event, cpuid):
             if args.debug:
                 read_value = readmsr('MSR_CONFIG_TDP_CONTROL', 0, 1, flatten=True)
                 match = OK if write_value == read_value else ERR
-                log(
-                    '[D] CONFIG_TDP_CONTROL - write {:#x} - read {:#x} - match {}'.format(
-                        write_value, read_value, match
-                    )
-                )
+                log(f'[D] CONFIG_TDP_CONTROL - write {write_value:#x} - read {read_value:#x} - match {match}')
 
         # set PL1/2 on MSR
         if 'MSR_PKG_POWER_LIMIT' in regs[power['source']]:
@@ -785,11 +767,7 @@ def power_thread(config, regs, exit_event, cpuid):
             if args.debug:
                 read_value = readmsr('MSR_PKG_POWER_LIMIT', 0, 55, flatten=True)
                 match = OK if write_value == read_value else ERR
-                log(
-                    '[D] MSR PACKAGE_POWER_LIMIT - write {:#x} - read {:#x} - match {}'.format(
-                        write_value, read_value, match
-                    )
-                )
+                log(f'[D] MSR PACKAGE_POWER_LIMIT - write {write_value:#x} - read {read_value:#x} - match {match}')
             if mchbar_mmio is not None:
                 # set MCHBAR register to the same PL1/2 values
                 mchbar_mmio.write32(0, write_value & 0xFFFFFFFF)
@@ -798,9 +776,7 @@ def power_thread(config, regs, exit_event, cpuid):
                     read_value = mchbar_mmio.read32(0) | (mchbar_mmio.read32(4) << 32)
                     match = OK if write_value == read_value else ERR
                     log(
-                        '[D] MCHBAR PACKAGE_POWER_LIMIT - write {:#x} - read {:#x} - match {}'.format(
-                            write_value, read_value, match
-                        )
+                        f'[D] MCHBAR PACKAGE_POWER_LIMIT - write {write_value:#x} - read {read_value:#x} - match {match}'
                     )
 
         # Disable BDPROCHOT
@@ -832,7 +808,7 @@ def check_kernel():
 
     kernel_config = None
     try:
-        with open(os.path.join('/boot', 'config-{:s}'.format(uname()[2]))) as f:
+        with open(os.path.join('/boot', f'config-{uname()[2]:s}')) as f:
             kernel_config = f.read()
     except IOError:
         config_gz_path = os.path.join('/proc', 'config.gz')
@@ -883,12 +859,12 @@ def check_cpu():
                 'from /proc/cpuinfo.'
             )
 
-        log('[I] Detected CPU architecture: Intel {:s}'.format(supported_cpus[cpuid]))
+        log(f'[I] Detected CPU architecture: Intel {supported_cpus[cpuid]:s}')
         return cpuid
     except SystemExit:
         raise
     except (OSError, KeyError, ValueError) as e:
-        fatal('Unable to identify CPU model: {}'.format(e))
+        fatal(f'Unable to identify CPU model: {e}')
 
 
 def test_msr_rw_capabilities():
@@ -930,21 +906,21 @@ def monitor(exit_event, wait):
     }
 
     undervolt_values = get_undervolt(convert=True)
-    undervolt_output = ' | '.join('{:s}: {:.2f} mV'.format(plane, undervolt_values[plane]) for plane in VOLTAGE_PLANES)
-    log('[D] Undervolt offsets: {:s}'.format(undervolt_output))
+    undervolt_output = ' | '.join(f'{plane:s}: {undervolt_values[plane]:.2f} mV' for plane in VOLTAGE_PLANES)
+    log(f'[D] Undervolt offsets: {undervolt_output:s}')
 
     iccmax_values = get_icc_max(convert=True)
-    iccmax_output = ' | '.join('{:s}: {:.2f} A'.format(plane, iccmax_values[plane]) for plane in CURRENT_PLANES)
-    log('[D] IccMax: {:s}'.format(iccmax_output))
+    iccmax_output = ' | '.join(f'{plane:s}: {iccmax_values[plane]:.2f} A' for plane in CURRENT_PLANES)
+    log(f'[D] IccMax: {iccmax_output:s}')
 
     log('[D] Realtime monitoring of throttling causes:\n')
     while not exit_event.is_set():
         value = readmsr('IA32_THERM_STATUS', from_bit=0, to_bit=15, cpu=0)
         offsets = {'Thermal': 0, 'Power': 10, 'Current': 12, 'Cross-domain (e.g. GPU)': 14}
-        output = ('{:s}: {:s}'.format(cause, LIM if bool((value >> offsets[cause]) & 1) else OK) for cause in offsets)
+        output = (f'{cause:s}: {LIM if bool((value >> offsets[cause]) & 1) else OK:s}' for cause in offsets)
 
-        vcore = readmsr('IA32_PERF_STATUS', from_bit=32, to_bit=47, cpu=0) / (2.0 ** 13) * 1000
-        stats2 = {'VCore': '{:.0f} mV'.format(vcore)}
+        vcore = readmsr('IA32_PERF_STATUS', from_bit=32, to_bit=47, cpu=0) / (2.0**13) * 1000
+        stats2 = {'VCore': f'{vcore:.0f} mV'}
         total = 0.0
         for power_plane in ('Package', 'Graphics', 'DRAM'):
             energy_j = readmsr(power_plane_msr[power_plane], cpu=0) * rapl_power_unit
@@ -953,15 +929,15 @@ def monitor(exit_event, wait):
                 (energy_j, now),
                 (energy_j - prev_energy[power_plane][0]) / (now - prev_energy[power_plane][1]),
             )
-            stats2[power_plane] = '{:.1f} W'.format(energy_w)
+            stats2[power_plane] = f'{energy_w:.1f} W'
             total += energy_w
 
-        stats2['Total'] = '{:.1f} W'.format(total)
+        stats2['Total'] = f'{total:.1f} W'
 
-        output2 = ('{:s}: {:s}'.format(label, stats2[label]) for label in stats2)
+        output2 = (f'{label}: {stats2[label]}' for label in stats2)
         terminator = '\n' if args.log else '\r'
         log(
-            '[{}] {}  ||  {}{}'.format(power['source'], ' - '.join(output), ' - '.join(output2), ' ' * 10),
+            f"[{power['source']}] {' - '.join(output)}  ||  {' - '.join(output2)}{' ' * 10}",
             end=terminator,
         )
         exit_event.wait(wait)
@@ -992,7 +968,7 @@ def main():
             args.log = open(args.log, 'w')
         except OSError as e:
             args.log = None
-            fatal('Unable to write to the log file: {}'.format(e))
+            fatal(f'Unable to write to the log file: {e}')
 
     cpuid = None
     if not args.force:
@@ -1013,7 +989,7 @@ def main():
     platform_info = get_cpu_platform_info()
     if args.debug:
         for key, value in platform_info.items():
-            log('[D] cpu platform info: {} = {}'.format(key.replace("_", " "), value))
+            log(f'[D] cpu platform info: {key.replace("_", " ")} = {value}')
     regs = calc_reg_values(platform_info, config)
 
     if not config.getboolean('GENERAL', 'Enabled'):

From 8432e1dcdcc704ff2d6bf2f3b42f680a553b929d Mon Sep 17 00:00:00 2001
From: ooonea <35407790+ooonea@users.noreply.github.com>
Date: Sun, 19 Apr 2026 21:56:36 +0200
Subject: [PATCH 5/8] Efficiency: drop the duplicate is_on_battery call in
 power_thread

In polling mode, power['source'] is already set from is_on_battery
at the top of each iteration, so the HWP check was calling
is_on_battery a second time for the same value. Replace the whole
dbus-vs-polling branch with `power['source'] == 'AC'`, which is
correct for both methods and avoids the extra glob/file read (and a
fallback DBus round-trip).

Signed-off-by: ooonea <35407790+ooonea@users.noreply.github.com>
---
 throttled.py | 9 +--------
 1 file changed, 1 insertion(+), 8 deletions(-)

diff --git a/throttled.py b/throttled.py
index 574a228..31e0d18 100755
--- a/throttled.py
+++ b/throttled.py
@@ -787,14 +787,7 @@ def power_thread(config, regs, exit_event, cpuid):
         wait_t = config.getfloat(power['source'], 'Update_Rate_s')
         enable_hwp_mode = config.getboolean('AC', 'HWP_Mode', fallback=None)
         # set HWP less frequently. Just to be safe since (e.g.) TLP might reset this value
-        if (
-            enable_hwp_mode
-            and next_hwp_write <= time()
-            and (
-                (power['method'] == 'dbus' and power['source'] == 'AC')
-                or (power['method'] == 'polling' and not is_on_battery(config))
-            )
-        ):
+        if enable_hwp_mode and next_hwp_write <= time() and power['source'] == 'AC':
             set_hwp(enable_hwp_mode)
             next_hwp_write = time() + HWP_INTERVAL
 

From 2b8f94b23ee2aeaa6f29e17c05807de41f985851 Mon Sep 17 00:00:00 2001
From: ooonea <35407790+ooonea@users.noreply.github.com>
Date: Sun, 19 Apr 2026 21:56:37 +0200
Subject: [PATCH 6/8] mmio: add read64/write64 helpers Mirror the existing
 read32/write32 pattern for 64-bit accesses. The MCHBAR PKG_POWER_LIMIT
 register is naturally a single 64-bit field, and having a symmetric helper
 avoids the open-coded split into two 32-bit writes at the call site.

Signed-off-by: ooonea <35407790+ooonea@users.noreply.github.com>
---
 mmio.py | 20 ++++++++++++++++++++
 1 file changed, 20 insertions(+)

diff --git a/mmio.py b/mmio.py
index e07411a..aa9ecb7 100644
--- a/mmio.py
+++ b/mmio.py
@@ -119,6 +119,26 @@ class MMIO(object):
         self._validate_offset(offset, 4)
         self.mapping[offset : offset + 4] = struct.pack("=L", value)
 
+    def read64(self, offset: int) -> int:
+        if not isinstance(offset, int):
+            raise TypeError("Invalid offset type, should be integer.")
+
+        offset = self._adjust_offset(offset)
+        self._validate_offset(offset, 8)
+        return struct.unpack("=Q", self.mapping[offset : offset + 8])[0]
+
+    def write64(self, offset: int, value: int) -> None:
+        if not isinstance(offset, int):
+            raise TypeError("Invalid offset type, should be integer.")
+        if not isinstance(value, int):
+            raise TypeError("Invalid value type, should be integer.")
+        if value < 0 or value > 0xFFFFFFFFFFFFFFFF:
+            raise ValueError("Value out of bounds.")
+
+        offset = self._adjust_offset(offset)
+        self._validate_offset(offset, 8)
+        self.mapping[offset : offset + 8] = struct.pack("=Q", value)
+
     def close(self) -> None:
         """Unmap the MMIO object's mapped physical memory."""
         if self.mapping is None:

From b9a0d8634de4e0ac8e830bb0a26a2c2d4479c7ac Mon Sep 17 00:00:00 2001
From: ooonea <35407790+ooonea@users.noreply.github.com>
Date: Sun, 19 Apr 2026 21:56:37 +0200
Subject: [PATCH 7/8] throttled: use mmio.write64/read64 for MCHBAR
 PKG_POWER_LIMIT Replaces the open-coded write32(0, low) + write32(4, high)
 pair (and the matching read32 merge in the debug path) with a single 64-bit
 access. Besides being less code, this also closes a small window between the
 two 32-bit writes in which the register briefly held (new_low, old_high).

Signed-off-by: ooonea <35407790+ooonea@users.noreply.github.com>
---
 throttled.py | 5 ++---
 1 file changed, 2 insertions(+), 3 deletions(-)

diff --git a/throttled.py b/throttled.py
index 31e0d18..f9030f7 100755
--- a/throttled.py
+++ b/throttled.py
@@ -770,10 +770,9 @@ def power_thread(config, regs, exit_event, cpuid):
                 log(f'[D] MSR PACKAGE_POWER_LIMIT - write {write_value:#x} - read {read_value:#x} - match {match}')
             if mchbar_mmio is not None:
                 # set MCHBAR register to the same PL1/2 values
-                mchbar_mmio.write32(0, write_value & 0xFFFFFFFF)
-                mchbar_mmio.write32(4, write_value >> 32)
+                mchbar_mmio.write64(0, write_value)
                 if args.debug:
-                    read_value = mchbar_mmio.read32(0) | (mchbar_mmio.read32(4) << 32)
+                    read_value = mchbar_mmio.read64(0)
                     match = OK if write_value == read_value else ERR
                     log(
                         f'[D] MCHBAR PACKAGE_POWER_LIMIT - write {write_value:#x} - read {read_value:#x} - match {match}'

From 4bf1faa900c33a1aad3c88baef41c53f1d96b4dd Mon Sep 17 00:00:00 2001
From: ooonea <35407790+ooonea@users.noreply.github.com>
Date: Sun, 19 Apr 2026 21:56:38 +0200
Subject: [PATCH 8/8] mmio: document read64/write64 helpers Short one-line
 docstrings mirroring the style of the existing read32/write32 methods.

Signed-off-by: ooonea <35407790+ooonea@users.noreply.github.com>
---
 mmio.py | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/mmio.py b/mmio.py
index aa9ecb7..a188bb7 100644
--- a/mmio.py
+++ b/mmio.py
@@ -120,6 +120,7 @@ class MMIO(object):
         self.mapping[offset : offset + 4] = struct.pack("=L", value)
 
     def read64(self, offset: int) -> int:
+        """Read 64-bits from the specified `offset` in bytes, relative to the base physical address of the MMIO region."""
         if not isinstance(offset, int):
             raise TypeError("Invalid offset type, should be integer.")
 
@@ -128,6 +129,7 @@ class MMIO(object):
         return struct.unpack("=Q", self.mapping[offset : offset + 8])[0]
 
     def write64(self, offset: int, value: int) -> None:
+        """Write 64-bits to the specified `offset` in bytes, relative to the base physical address of the MMIO region."""
         if not isinstance(offset, int):
             raise TypeError("Invalid offset type, should be integer.")
         if not isinstance(value, int):