DEV Community: vast cow

How to Disable Suspend on Lid Close in Rocky Linux

vast cow — Sun, 21 Jun 2026 06:51:22 +0000

In Rocky Linux, lid-close handling is generally managed by systemd-logind, even when using multi-user.target. Set HandleLidSwitch=ignore. The Rocky Linux documentation also recommends setting HandleLidSwitch to ignore in /etc/systemd/logind.conf. (Rocky Linux Docs)

Recommended Configuration

sudo cp -a /etc/systemd/logind.conf /etc/systemd/logind.conf.bak.$(date +%F-%H%M%S)

sudo mkdir -p /etc/systemd/logind.conf.d

sudo tee /etc/systemd/logind.conf.d/99-ignore-lid.conf >/dev/null <<'EOF'
[Login]
HandleLidSwitch=ignore
HandleLidSwitchExternalPower=ignore
HandleLidSwitchDocked=ignore
EOF

sudo systemctl restart systemd-logind.service

This ignores lid-close events in all cases: normal operation, when connected to AC power, and when docked. The default value of HandleLidSwitch is suspend; changing it to ignore prevents logind from suspending the system when the lid is closed. (man7.org)

Verification

systemd-analyze cat-config systemd/logind.conf | grep -E 'HandleLidSwitch'

Expected output:

HandleLidSwitch=ignore
HandleLidSwitchExternalPower=ignore
HandleLidSwitchDocked=ignore

To check the current default target:

systemctl get-default

If you want to keep the system on multi-user.target:

sudo systemctl set-default multi-user.target

If Editing `/etc/systemd/logind.conf` Directly

If drop-in configuration files are not supported in your environment, you can edit the main configuration file directly:

sudo vi /etc/systemd/logind.conf

Add the following under the [Login] section:

[Login]
HandleLidSwitch=ignore
HandleLidSwitchExternalPower=ignore
HandleLidSwitchDocked=ignore

Apply the changes:

sudo systemctl restart systemd-logind.service

If the System Still Suspends

First, inspect the logs:

journalctl -u systemd-logind -b | grep -i -E 'lid|suspend|sleep'

Then check whether another process is managing power events:

systemd-inhibit --list

With multi-user.target, desktop environments typically do not interfere. However, if a graphical session such as GNOME is running, the desktop environment may take over suspend handling. The systemd documentation notes that Handle* settings may be bypassed when another application holds a low-level inhibitor lock. (man7.org)

As a last resort, you can disable suspend-related targets entirely:

sudo systemctl mask sleep.target suspend.target hibernate.target hybrid-sleep.target

To restore them later:

sudo systemctl unmask sleep.target suspend.target hibernate.target hybrid-sleep.target

In most cases, setting HandleLidSwitch=ignore in logind is sufficient. If you are using a laptop as a server and plan to operate it under heavy load with the lid closed, be aware that some models have reduced cooling performance when closed. The Red Hat documentation provides a similar caution. (Red Hat Documentation)

Use Unix Domain Sockets on Windows Python: Building an AF_UNIX Compatibility API

vast cow — Sun, 21 Jun 2026 06:43:11 +0000

Python provides socket.AF_UNIX, asyncio.open_unix_connection(), and asyncio.start_unix_server() for working with Unix Domain Sockets on Unix-like operating systems.

On Windows, however, support for Unix Domain Sockets tends to depend on the Python version and runtime environment. In particular, differences become apparent when trying to use the higher-level asyncio APIs in the same way as on Unix.

To address this, I created a compatibility layer that hides the differences between Unix and Windows and allows AF_UNIX sockets to be used through a largely identical API.

This article covers two types of APIs:

An asyncio-based AF_UNIX compatibility API
A synchronous socket-based AF_UNIX compatibility API

Goal

The objective is straightforward.

On Unix, use the standard library APIs as-is. On Windows, fill in the missing functionality so that application code can remain as unified as possible.

For example, on Unix you can write:

reader, writer = await asyncio.open_unix_connection(path)

And on the server side:

server = await asyncio.start_unix_server(handle_client, path)

The goal is to preserve this style of programming on Windows as much as possible.

What Was Built

The compatibility layer consists of two major components.

1. Asyncio Version

This is the asynchronous implementation designed to match the asyncio Unix Domain Socket APIs.

The main APIs are:

await open_unix_connection(path, *, limit=...)
await start_unix_server(callback, path, *, limit=..., backlog=...)
await create_unix_connection(protocol_factory, path, ...)
await create_unix_server(protocol_factory, path, ...)
install()

On Unix-like systems, these simply delegate to the standard asyncio implementation.

On Windows, they use Winsock AF_UNIX sockets and combine WSAEventSelect with event-loop handle waiting to implement asynchronous operations.

2. Synchronous Socket Version

This version provides a traditional blocking-socket-style API without using asyncio.

The main APIs are:

socket(AF_UNIX, SOCK_STREAM)
create_connection(path, timeout=None)
create_server(path, backlog=100, cleanup_socket=False)
unlink_socket(path)
install()

Like the asyncio version, it delegates to the standard socket implementation on Unix while supplying missing functionality on Windows.

Using the Asyncio Version

Example server:

import asyncio
from af_unix_asyncio_compat import start_unix_server

SOCK = "./example.sock"

async def handle_client(reader, writer):
    data = await reader.read(1024)
    writer.write(b"echo: " + data)
    await writer.drain()

    writer.close()
    await writer.wait_closed()

async def main():
    server = await start_unix_server(handle_client, SOCK)

    async with server:
        await server.serve_forever()

asyncio.run(main())

Client:

import asyncio
from af_unix_asyncio_compat import open_unix_connection

SOCK = "./example.sock"

async def main():
    reader, writer = await open_unix_connection(SOCK)

    writer.write(b"hello")
    await writer.drain()

    response = await reader.read(1024)
    print(response)

    writer.close()
    await writer.wait_closed()

asyncio.run(main())

On Unix, this feels almost identical to using the standard asyncio APIs.

On Windows, the application still interacts through the familiar Stream API (reader / writer).

Using the Synchronous Socket Version

The synchronous version is a lower-level and simpler API.

Server:

import af_unix_socket_compat as unix_socket

SOCK = "./example.sock"

server = unix_socket.create_server(SOCK, cleanup_socket=True)

try:
    conn, addr = server.accept()

    try:
        data = conn.recv(1024)
        conn.sendall(b"echo: " + data)
    finally:
        conn.close()

finally:
    server.close()
    unix_socket.unlink_socket(SOCK)

Client:

import af_unix_socket_compat as unix_socket

SOCK = "./example.sock"

client = unix_socket.create_connection(SOCK)

try:
    client.sendall(b"hello")
    response = client.recv(1024)
    print(response)

finally:
    client.close()

This version is easier to integrate into existing synchronous socket code or applications that do not use asyncio.

The Role of `install()`

Both compatibility layers provide an install() function.

install() serves as an entry point for adopting the compatibility API without requiring extensive modifications to existing code.

Conceptually, the asyncio version performs replacements such as:

asyncio.open_unix_connection = compat.open_unix_connection
asyncio.start_unix_server = compat.start_unix_server

As a result, existing code like:

reader, writer = await asyncio.open_unix_connection(path)

can transparently invoke the compatibility implementation on Windows.

However, install() is a monkey patch that globally modifies standard library behavior.

For new code, it is generally safer to explicitly import the compatibility APIs:

from af_unix_asyncio_compat import open_unix_connection

Or, for the synchronous version:

import af_unix_socket_compat as unix_socket

install() is best viewed as a migration aid for existing codebases that cannot be easily modified.

Implementation Strategy

The implementation follows separate paths for Unix and Windows.

On Unix-Like Systems

Python's standard library already provides robust AF_UNIX support.

Therefore, the compatibility layer avoids unnecessary intervention and simply delegates to the standard implementation.

await asyncio.open_unix_connection(path=path)

Or:

socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)

On Windows

Windows uses Winsock AF_UNIX sockets.

The synchronous implementation provides equivalents of:

connect()
bind()
listen()
accept()
recv()
send()

The asyncio implementation must additionally integrate with the asynchronous event loop.

To accomplish this, the Windows implementation:

Creates AF_UNIX sockets
Creates Winsock events
Monitors socket events with WSAEventSelect
Waits for read, write, connect, accept, and close events
Exposes the result through asyncio.StreamReader and StreamWriter

This allows application code to continue using the standard asyncio Stream API.

API Compatibility

Achieving complete standard-library compatibility requires matching many implementation details.

For example, the asyncio version raises questions such as:

ssl
server_hostname
ssl_handshake_timeout
ssl_shutdown_timeout
start_serving
cleanup_socket
StreamReader.limit
Server.close()
Server.wait_closed()
Server.serve_forever()

This compatibility layer focuses primarily on the higher-level APIs that are most commonly used.

At the same time, Windows AF_UNIX and Unix AF_UNIX differ at the operating-system level, so perfectly identical behavior is not always achievable.

Limitations

The primary limitations are as follows.

SSL Is Not Supported

SSL over Unix Domain Sockets is uncommon in many practical use cases.

For that reason, the Windows compatibility implementation does not support SSL.

If ssl is specified, it raises an explicit error rather than silently ignoring the option.

Abstract Namespace Sockets Are Not Supported

Linux supports abstract namespace Unix Domain Sockets that do not use filesystem paths.

Because the Windows AF_UNIX model differs significantly, these sockets are outside the scope of the compatibility layer.

Windows Path Length Restrictions

Windows AF_UNIX sockets still impose limits on socket path length.

Paths that exceed the allowed length must therefore be rejected.

Use `install()` Carefully

Although convenient, install() globally alters standard-library behavior.

It should not be called automatically by libraries. Instead, it is best invoked explicitly during application startup.

Choosing Between the Asyncio and Synchronous Versions

The choice is simple.

If your application already uses asyncio, use the asyncio version:

reader, writer = await open_unix_connection(path)

If a traditional blocking socket is sufficient, use the synchronous version:

client = create_connection(path)
client.sendall(b"hello")

The same principle applies to servers.

Use the asyncio version when handling multiple concurrent connections asynchronously, and use the synchronous version for simple IPC or testing scenarios.

Conclusion

Unix Domain Sockets are a convenient mechanism for local inter-process communication.

However, differences in API support between Unix and Windows often force application code to contain platform-specific branches.

By introducing an AF_UNIX compatibility layer, the architecture becomes cleaner:

Delegate to the standard library on Unix
Fill in missing functionality on Windows
Provide separate asyncio and synchronous implementations
Prefer explicit imports in new code
Offer install() for legacy codebases

This approach confines platform-specific behavior to the compatibility layer and keeps application code simple and portable.

# af_unix_socket_compat.py
"""
Synchronous AF_UNIX socket compatibility layer.

Use this module when you want ordinary/blocking socket-style Unix-domain socket
APIs, not asyncio stream APIs.

Typical usage:

    import af_unix_socket_compat as socket

    srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    srv.bind("./server.sock")
    srv.listen(5)
    conn, addr = srv.accept()

The module is intentionally shaped like the stdlib socket module for AF_UNIX
use-cases:

    socket(AF_UNIX, SOCK_STREAM)
    create_connection(path, timeout=None)
    create_server(path, backlog=100, cleanup_socket=False)
    install()

On POSIX platforms it delegates to the stdlib socket implementation.  On modern
Windows/Python where socket.AF_UNIX already works, it delegates to the stdlib
implementation too.  A ctypes/Winsock fallback is included for environments
where AF_UNIX is available in Winsock but not exposed by Python's socket module.

Windows limitations:
- Only SOCK_STREAM is supported by the fallback.
- Linux abstract namespace addresses are not supported on Windows.
- makefile() is available only when the native Python socket backend is used.
"""

from __future__ import annotations

import errno
import os
import select
import socket as _socket_module
import sys
import time
from typing import Any, Iterable

_ORIGINAL_SOCKET_FACTORY = _socket_module.socket
_IS_WINDOWS = sys.platform == "win32"
_NATIVE_AF_UNIX = getattr(_socket_module, "AF_UNIX", 1)
AF_UNIX = _NATIVE_AF_UNIX
AF_LOCAL = AF_UNIX
SOCK_STREAM = _socket_module.SOCK_STREAM
SOL_SOCKET = _socket_module.SOL_SOCKET
SO_ERROR = _socket_module.SO_ERROR
SHUT_RD = _socket_module.SHUT_RD
SHUT_WR = _socket_module.SHUT_WR
SHUT_RDWR = _socket_module.SHUT_RDWR
error = _socket_module.error
timeout = _socket_module.timeout

__all__ = [
    "AF_UNIX",
    "AF_LOCAL",
    "SOCK_STREAM",
    "SOL_SOCKET",
    "SO_ERROR",
    "SHUT_RD",
    "SHUT_WR",
    "SHUT_RDWR",
    "error",
    "timeout",
    "socket",
    "UnixSocket",
    "create_connection",
    "create_server",
    "unlink_socket",
    "install",
]


def __getattr__(name: str) -> Any:
    """Expose the rest of stdlib socket's constants/functions lazily."""
    return getattr(_socket_module, name)


def _path_bytes(path: os.PathLike[str] | str | bytes) -> bytes:
    if isinstance(path, bytes):
        raw = path
    else:
        raw = os.fsencode(os.fspath(path))
    if b"\x00" in raw:
        raise ValueError("AF_UNIX path must not contain NUL bytes on Windows")
    if len(raw) >= 108:
        raise OSError(errno.ENAMETOOLONG, f"AF_UNIX path is too long: {len(raw)} bytes")
    return raw


def unlink_socket(path: os.PathLike[str] | str | bytes) -> None:
    """Best-effort unlink helper matching common Unix-domain socket setup code."""
    try:
        os.unlink(path)
    except FileNotFoundError:
        return


def _native_af_unix_works() -> bool:
    if not hasattr(_socket_module, "AF_UNIX"):
        return False
    try:
        s = _ORIGINAL_SOCKET_FACTORY(_socket_module.AF_UNIX, _socket_module.SOCK_STREAM)
    except OSError:
        return False
    else:
        s.close()
        return True


_USE_NATIVE = (not _IS_WINDOWS) or _native_af_unix_works()


class _NativeUnixSocket:
    """Thin delegating wrapper around a real Python socket.socket object."""

    def __init__(self, sock: _socket_module.socket):
        self._sock = sock

    def __getattr__(self, name: str) -> Any:
        return getattr(self._sock, name)

    def __enter__(self) -> "_NativeUnixSocket":
        self._sock.__enter__()
        return self

    def __exit__(self, *args: Any) -> Any:
        return self._sock.__exit__(*args)

    def fileno(self) -> int:
        return self._sock.fileno()

    def bind(self, address: os.PathLike[str] | str | bytes) -> None:
        self._sock.bind(os.fspath(address) if not isinstance(address, bytes) else address)

    def connect(self, address: os.PathLike[str] | str | bytes) -> None:
        self._sock.connect(os.fspath(address) if not isinstance(address, bytes) else address)

    def accept(self) -> tuple["_NativeUnixSocket", Any]:
        conn, addr = self._sock.accept()
        return _NativeUnixSocket(conn), addr

    def close(self) -> None:
        self._sock.close()

    def detach(self) -> int:
        return self._sock.detach()

    @property
    def family(self) -> int:
        return self._sock.family

    @property
    def type(self) -> int:
        return self._sock.type

    @property
    def proto(self) -> int:
        return self._sock.proto


UnixSocket = _NativeUnixSocket


# ---------------------------------------------------------------------------
# Windows ctypes fallback.
# ---------------------------------------------------------------------------

if _IS_WINDOWS and not _USE_NATIVE:
    import ctypes
    import ctypes.wintypes as wintypes

    AF_UNIX = 1
    AF_LOCAL = AF_UNIX
    UNIX_PATH_MAX = 108

    WSAEWOULDBLOCK = 10035
    WSAEINPROGRESS = 10036
    WSAEALREADY = 10037
    WSAEINVAL = 10022
    WSAECONNRESET = 10054
    WSAENOTCONN = 10057

    FIONBIO = 0x8004667E
    INVALID_SOCKET = -1
    SOCKET_ERROR = -1

    SOCKET_T = ctypes.c_uint64 if ctypes.sizeof(ctypes.c_void_p) == 8 else ctypes.c_uint32
    INVALID_SOCKET_VALUE = (1 << (ctypes.sizeof(SOCKET_T) * 8)) - 1

    class SockaddrUn(ctypes.Structure):
        _fields_ = [
            ("sun_family", wintypes.USHORT),
            ("sun_path", ctypes.c_char * UNIX_PATH_MAX),
        ]

    class WSAData(ctypes.Structure):
        _fields_ = [
            ("wVersion", wintypes.WORD),
            ("wHighVersion", wintypes.WORD),
            ("szDescription", ctypes.c_char * 257),
            ("szSystemStatus", ctypes.c_char * 129),
            ("iMaxSockets", wintypes.USHORT),
            ("iMaxUdpDg", wintypes.USHORT),
            ("lpVendorInfo", ctypes.c_char_p),
        ]

    ws2_32 = ctypes.WinDLL("ws2_32", use_last_error=True)

    ws2_32.WSAStartup.restype = ctypes.c_int
    ws2_32.WSAStartup.argtypes = (wintypes.WORD, ctypes.POINTER(WSAData))

    ws2_32.WSAGetLastError.restype = ctypes.c_int
    ws2_32.WSAGetLastError.argtypes = ()

    ws2_32.socket.restype = SOCKET_T
    ws2_32.socket.argtypes = (ctypes.c_int, ctypes.c_int, ctypes.c_int)

    ws2_32.closesocket.restype = ctypes.c_int
    ws2_32.closesocket.argtypes = (SOCKET_T,)

    ws2_32.bind.restype = ctypes.c_int
    ws2_32.bind.argtypes = (SOCKET_T, ctypes.POINTER(SockaddrUn), ctypes.c_int)

    ws2_32.connect.restype = ctypes.c_int
    ws2_32.connect.argtypes = (SOCKET_T, ctypes.POINTER(SockaddrUn), ctypes.c_int)

    ws2_32.listen.restype = ctypes.c_int
    ws2_32.listen.argtypes = (SOCKET_T, ctypes.c_int)

    ws2_32.accept.restype = SOCKET_T
    ws2_32.accept.argtypes = (SOCKET_T, ctypes.POINTER(SockaddrUn), ctypes.POINTER(ctypes.c_int))

    ws2_32.recv.restype = ctypes.c_int
    ws2_32.recv.argtypes = (SOCKET_T, ctypes.c_void_p, ctypes.c_int, ctypes.c_int)

    ws2_32.send.restype = ctypes.c_int
    ws2_32.send.argtypes = (SOCKET_T, ctypes.c_void_p, ctypes.c_int, ctypes.c_int)

    ws2_32.shutdown.restype = ctypes.c_int
    ws2_32.shutdown.argtypes = (SOCKET_T, ctypes.c_int)

    ws2_32.ioctlsocket.restype = ctypes.c_int
    ws2_32.ioctlsocket.argtypes = (SOCKET_T, ctypes.c_long, ctypes.POINTER(ctypes.c_ulong))

    ws2_32.getsockopt.restype = ctypes.c_int
    ws2_32.getsockopt.argtypes = (
        SOCKET_T,
        ctypes.c_int,
        ctypes.c_int,
        ctypes.c_void_p,
        ctypes.POINTER(ctypes.c_int),
    )

    ws2_32.setsockopt.restype = ctypes.c_int
    ws2_32.setsockopt.argtypes = (
        SOCKET_T,
        ctypes.c_int,
        ctypes.c_int,
        ctypes.c_void_p,
        ctypes.c_int,
    )

    _wsa_started = False

    def _ensure_winsock() -> None:
        global _wsa_started
        if _wsa_started:
            return
        data = WSAData()
        rc = ws2_32.WSAStartup(0x0202, ctypes.byref(data))
        if rc:
            raise OSError(rc, "WSAStartup failed")
        _wsa_started = True

    def _last_error() -> int:
        return int(ws2_32.WSAGetLastError())

    def _raise_socket_error(err: int | None = None) -> None:
        if err is None:
            err = _last_error()
        # socket.error/OSError on Windows conventionally stores the WSA error.
        raise OSError(err, os.strerror(err) if err < 10000 else f"Winsock error {err}")

    def _invalid_socket(value: int) -> bool:
        return value in (-1, INVALID_SOCKET_VALUE)

    def _sockaddr(path: os.PathLike[str] | str | bytes) -> SockaddrUn:
        raw = _path_bytes(path)
        addr = SockaddrUn()
        addr.sun_family = AF_UNIX
        addr.sun_path = raw
        return addr

    def _deadline(timeout_value: float | None) -> float | None:
        if timeout_value is None:
            return None
        if timeout_value < 0:
            raise ValueError("Timeout value out of range")
        return time.monotonic() + timeout_value

    def _remaining(deadline: float | None) -> float | None:
        if deadline is None:
            return None
        rem = deadline - time.monotonic()
        if rem <= 0:
            raise timeout("timed out")
        return rem

    class _CtypesUnixSocket:
        def __init__(
            self,
            family: int = AF_UNIX,
            type: int = SOCK_STREAM,
            proto: int = 0,
            fileno: int | None = None,
        ) -> None:
            if family not in (AF_UNIX, AF_LOCAL):
                raise OSError(errno.EAFNOSUPPORT, "only AF_UNIX is supported by this fallback")
            if type != SOCK_STREAM:
                raise OSError(errno.EPROTONOSUPPORT, "only SOCK_STREAM is supported by this fallback")
            _ensure_winsock()
            if fileno is None:
                handle = int(ws2_32.socket(AF_UNIX, type, proto))
                if _invalid_socket(handle):
                    _raise_socket_error()
            else:
                handle = int(fileno)
            self._handle = handle
            self._family = family
            self._type = type
            self._proto = proto
            self._timeout: float | None = _socket_module.getdefaulttimeout()
            self._blocking = self._timeout != 0.0
            self._closed = False
            self._sockname: os.PathLike[str] | str | bytes | None = None
            self._peername: os.PathLike[str] | str | bytes | None = None
            if self._timeout == 0.0:
                self.setblocking(False)

        def __enter__(self) -> "_CtypesUnixSocket":
            return self

        def __exit__(self, *args: Any) -> None:
            self.close()

        @property
        def family(self) -> int:
            return self._family

        @property
        def type(self) -> int:
            return self._type

        @property
        def proto(self) -> int:
            return self._proto

        def fileno(self) -> int:
            return -1 if self._closed else self._handle

        def detach(self) -> int:
            if self._closed:
                raise OSError(errno.EBADF, "socket is closed")
            handle = self._handle
            self._closed = True
            self._handle = -1
            return handle

        def _check_open(self) -> None:
            if self._closed:
                raise OSError(errno.EBADF, "socket is closed")

        def setblocking(self, flag: bool) -> None:
            self._check_open()
            mode = ctypes.c_ulong(0 if flag else 1)
            if ws2_32.ioctlsocket(self._handle, FIONBIO, ctypes.byref(mode)) != 0:
                _raise_socket_error()
            self._blocking = bool(flag)
            self._timeout = None if flag else 0.0

        def getblocking(self) -> bool:
            return self._blocking

        def settimeout(self, value: float | None) -> None:
            if value is not None:
                value = float(value)
                if value < 0:
                    raise ValueError("Timeout value out of range")
            self._timeout = value
            # Python sockets use non-blocking mode internally for finite timeouts.
            mode = ctypes.c_ulong(1 if value is not None else 0)
            if ws2_32.ioctlsocket(self._handle, FIONBIO, ctypes.byref(mode)) != 0:
                _raise_socket_error()
            self._blocking = value is None

        def gettimeout(self) -> float | None:
            return self._timeout

        def bind(self, address: os.PathLike[str] | str | bytes) -> None:
            self._check_open()
            addr = _sockaddr(address)
            if ws2_32.bind(self._handle, ctypes.byref(addr), ctypes.sizeof(addr)) != 0:
                _raise_socket_error()
            self._sockname = address

        def listen(self, backlog: int = 128) -> None:
            self._check_open()
            if ws2_32.listen(self._handle, int(backlog)) != 0:
                _raise_socket_error()

        def connect(self, address: os.PathLike[str] | str | bytes) -> None:
            self._check_open()
            addr = _sockaddr(address)
            rc = ws2_32.connect(self._handle, ctypes.byref(addr), ctypes.sizeof(addr))
            if rc == 0:
                self._peername = address
                return
            err = _last_error()
            if self._timeout == 0.0:
                if err in (WSAEWOULDBLOCK, WSAEINPROGRESS, WSAEALREADY, WSAEINVAL):
                    raise BlockingIOError(err, "operation would block")
                _raise_socket_error(err)
            if err not in (WSAEWOULDBLOCK, WSAEINPROGRESS, WSAEALREADY, WSAEINVAL):
                _raise_socket_error(err)

            deadline = _deadline(self._timeout)
            self._wait_writable(deadline)
            so_error = self.getsockopt(SOL_SOCKET, SO_ERROR)
            if so_error:
                _raise_socket_error(int(so_error))
            self._peername = address

        def connect_ex(self, address: os.PathLike[str] | str | bytes) -> int:
            try:
                self.connect(address)
                return 0
            except BlockingIOError as exc:
                return int(exc.errno or WSAEWOULDBLOCK)
            except OSError as exc:
                return int(exc.errno or SOCKET_ERROR)

        def accept(self) -> tuple["_CtypesUnixSocket", Any]:
            self._check_open()
            deadline = _deadline(self._timeout)
            while True:
                addr = SockaddrUn()
                addr_len = ctypes.c_int(ctypes.sizeof(addr))
                client = int(ws2_32.accept(self._handle, ctypes.byref(addr), ctypes.byref(addr_len)))
                if not _invalid_socket(client):
                    sock = _CtypesUnixSocket(AF_UNIX, SOCK_STREAM, 0, fileno=client)
                    sock.settimeout(self._timeout)
                    return sock, b""
                err = _last_error()
                if err != WSAEWOULDBLOCK:
                    _raise_socket_error(err)
                if self._timeout == 0.0:
                    raise BlockingIOError(err, "operation would block")
                self._wait_readable(deadline)

        def recv(self, bufsize: int, flags: int = 0) -> bytes:
            self._check_open()
            if bufsize < 0:
                raise ValueError("negative buffersize in recv")
            deadline = _deadline(self._timeout)
            while True:
                buf = ctypes.create_string_buffer(bufsize)
                n = ws2_32.recv(self._handle, buf, int(bufsize), int(flags))
                if n >= 0:
                    return bytes(buf.raw[:n])
                err = _last_error()
                if err == WSAEWOULDBLOCK:
                    if self._timeout == 0.0:
                        raise BlockingIOError(err, "operation would block")
                    self._wait_readable(deadline)
                    continue
                if err in (WSAECONNRESET, WSAENOTCONN):
                    return b""
                _raise_socket_error(err)

        def recv_into(self, buffer: Any, nbytes: int | None = None, flags: int = 0) -> int:
            data = self.recv(len(buffer) if nbytes is None else nbytes, flags)
            buffer[: len(data)] = data
            return len(data)

        def send(self, data: bytes | bytearray | memoryview, flags: int = 0) -> int:
            self._check_open()
            view = memoryview(data)
            deadline = _deadline(self._timeout)
            while True:
                # Copy to a stable ctypes buffer for the duration of send().
                chunk = view[: min(len(view), 1024 * 1024)]
                cbuf = (ctypes.c_char * len(chunk)).from_buffer_copy(chunk)
                n = ws2_32.send(self._handle, cbuf, len(chunk), int(flags))
                if n >= 0:
                    return int(n)
                err = _last_error()
                if err == WSAEWOULDBLOCK:
                    if self._timeout == 0.0:
                        raise BlockingIOError(err, "operation would block")
                    self._wait_writable(deadline)
                    continue
                if err in (WSAECONNRESET, WSAENOTCONN):
                    raise ConnectionResetError(err, "connection reset")
                _raise_socket_error(err)

        def sendall(self, data: bytes | bytearray | memoryview, flags: int = 0) -> None:
            view = memoryview(data)
            total = 0
            while total < len(view):
                sent = self.send(view[total:], flags)
                if sent == 0:
                    raise RuntimeError("socket connection broken")
                total += sent

        def sendmsg(self, *args: Any, **kwargs: Any) -> None:
            raise NotImplementedError("sendmsg is not implemented by the Windows fallback")

        def recvmsg(self, *args: Any, **kwargs: Any) -> None:
            raise NotImplementedError("recvmsg is not implemented by the Windows fallback")

        def shutdown(self, how: int) -> None:
            self._check_open()
            if ws2_32.shutdown(self._handle, int(how)) != 0:
                err = _last_error()
                if err not in (WSAENOTCONN,):
                    _raise_socket_error(err)

        def close(self) -> None:
            if self._closed:
                return
            self._closed = True
            if self._handle != -1:
                ws2_32.closesocket(self._handle)
            self._handle = -1

        def getsockname(self) -> Any:
            return self._sockname if self._sockname is not None else b""

        def getpeername(self) -> Any:
            if self._peername is None:
                raise OSError(errno.ENOTCONN, "socket is not connected")
            return self._peername

        def getsockopt(self, level: int, optname: int, buflen: int | None = None) -> Any:
            self._check_open()
            if buflen is None:
                value = ctypes.c_int()
                value_len = ctypes.c_int(ctypes.sizeof(value))
                rc = ws2_32.getsockopt(
                    self._handle,
                    int(level),
                    int(optname),
                    ctypes.byref(value),
                    ctypes.byref(value_len),
                )
                if rc != 0:
                    _raise_socket_error()
                return int(value.value)
            buf = ctypes.create_string_buffer(int(buflen))
            value_len = ctypes.c_int(int(buflen))
            rc = ws2_32.getsockopt(
                self._handle,
                int(level),
                int(optname),
                buf,
                ctypes.byref(value_len),
            )
            if rc != 0:
                _raise_socket_error()
            return bytes(buf.raw[: value_len.value])

        def setsockopt(self, level: int, optname: int, value: int | bytes | bytearray) -> None:
            self._check_open()
            if isinstance(value, int):
                cval = ctypes.c_int(value)
                ptr = ctypes.byref(cval)
                size = ctypes.sizeof(cval)
            else:
                cbuf = (ctypes.c_char * len(value)).from_buffer_copy(value)
                ptr = cbuf
                size = len(value)
            if ws2_32.setsockopt(self._handle, int(level), int(optname), ptr, size) != 0:
                _raise_socket_error()

        def makefile(self, *args: Any, **kwargs: Any) -> Any:
            raise NotImplementedError(
                "makefile() is not implemented by the ctypes fallback; "
                "use recv/send or run on a Python build with native socket.AF_UNIX"
            )

        def _wait_readable(self, deadline: float | None) -> None:
            rem = _remaining(deadline)
            readable, _, _ = select.select([self], [], [], rem)
            if not readable:
                raise timeout("timed out")

        def _wait_writable(self, deadline: float | None) -> None:
            rem = _remaining(deadline)
            _, writable, _ = select.select([], [self], [], rem)
            if not writable:
                raise timeout("timed out")

    UnixSocket = _CtypesUnixSocket


# ---------------------------------------------------------------------------
# Public factory/convenience APIs.
# ---------------------------------------------------------------------------


def socket(
    family: int = AF_UNIX,
    type: int = SOCK_STREAM,
    proto: int = 0,
    fileno: int | None = None,
) -> Any:
    """
    socket.socket-compatible factory.

    For AF_UNIX sockets this returns a compatibility object.  For all other
    families it returns the stdlib socket.socket object unchanged.
    """
    if family not in (AF_UNIX, AF_LOCAL):
        if fileno is None:
            return _ORIGINAL_SOCKET_FACTORY(family, type, proto)
        return _ORIGINAL_SOCKET_FACTORY(family, type, proto, fileno=fileno)

    if _USE_NATIVE:
        native_family = getattr(_socket_module, "AF_UNIX", AF_UNIX)
        if fileno is None:
            sock = _ORIGINAL_SOCKET_FACTORY(native_family, type, proto)
        else:
            try:
                sock = _ORIGINAL_SOCKET_FACTORY(native_family, type, proto, fileno=fileno)
            except TypeError:
                sock = _ORIGINAL_SOCKET_FACTORY(fileno=fileno)
        return _NativeUnixSocket(sock)

    return UnixSocket(family, type, proto, fileno=fileno)


def create_connection(
    address: os.PathLike[str] | str | bytes,
    timeout: float | None | object = _socket_module._GLOBAL_DEFAULT_TIMEOUT,
    source_address: Any = None,
) -> Any:
    """AF_UNIX equivalent of socket.create_connection(address, timeout=...)."""
    if source_address is not None:
        raise ValueError("source_address is not meaningful for AF_UNIX sockets")
    sock = socket(AF_UNIX, SOCK_STREAM)
    try:
        if timeout is not _socket_module._GLOBAL_DEFAULT_TIMEOUT:
            sock.settimeout(timeout)  # type: ignore[arg-type]
        sock.connect(address)
        return sock
    except BaseException:
        sock.close()
        raise


def create_server(
    address: os.PathLike[str] | str | bytes,
    *,
    backlog: int = 100,
    cleanup_socket: bool = False,
) -> Any:
    """
    Create, bind, and listen on an AF_UNIX SOCK_STREAM server socket.

    cleanup_socket=True unlinks an existing path before bind, matching the
    common Unix-domain server pattern.  It does not unlink on close; call
    unlink_socket(path) explicitly when you want post-close cleanup.
    """
    if cleanup_socket:
        unlink_socket(address)
    srv = socket(AF_UNIX, SOCK_STREAM)
    try:
        srv.bind(address)
        srv.listen(backlog)
        return srv
    except BaseException:
        srv.close()
        raise


def install() -> None:
    """
    Install this compatibility factory into the stdlib socket module.

    This is optional and intentionally narrow: existing code can instead do
    `import af_unix_socket_compat as socket`.
    """
    if not hasattr(_socket_module, "AF_UNIX"):
        setattr(_socket_module, "AF_UNIX", AF_UNIX)
    if not hasattr(_socket_module, "AF_LOCAL"):
        setattr(_socket_module, "AF_LOCAL", AF_LOCAL)
    _socket_module.socket = socket  # type: ignore[assignment]

# af_unix_asyncio_compat.py
"""
asyncio AF_UNIX compatibility helpers.

This module exposes a small API compatible with the high-level Unix-domain
socket helpers from asyncio:

    await open_unix_connection(path, ...)
    await start_unix_server(callback, path, ...)
    await create_unix_connection(protocol_factory, path, ...)
    await create_unix_server(protocol_factory, path, ...)

On POSIX platforms the functions delegate directly to asyncio's native
AF_UNIX implementation.  On Windows they use Winsock AF_UNIX plus
WSAEventSelect/IOCP handle waits so the API can be used with the default
ProactorEventLoop.

Notes / intentional limits of the Windows fallback:
- SSL over AF_UNIX is not implemented here.
- Windows AF_UNIX does not support Linux abstract namespace addresses.
- The returned server is asyncio.Server-like, not an actual asyncio.Server.
"""

from __future__ import annotations

import asyncio
import inspect
import os
import socket
import sys
from typing import Any, Callable, Iterable

_DEFAULT_LIMIT = 2**16
_IS_WINDOWS = sys.platform == "win32"

__all__ = [
    "open_unix_connection",
    "start_unix_server",
    "create_unix_connection",
    "create_unix_server",
    "install",
]


# ---------------------------------------------------------------------------
# POSIX implementation: defer to the stdlib implementation.
# ---------------------------------------------------------------------------

if not _IS_WINDOWS:

    async def open_unix_connection(
        path: os.PathLike[str] | str | bytes | None = None,
        *,
        limit: int = _DEFAULT_LIMIT,
        **kwds: Any,
    ) -> tuple[asyncio.StreamReader, asyncio.StreamWriter]:
        return await asyncio.open_unix_connection(path=path, limit=limit, **kwds)

    async def start_unix_server(
        client_connected_cb: Callable[[asyncio.StreamReader, asyncio.StreamWriter], Any],
        path: os.PathLike[str] | str | bytes | None = None,
        *,
        limit: int = _DEFAULT_LIMIT,
        **kwds: Any,
    ) -> asyncio.AbstractServer:
        return await asyncio.start_unix_server(
            client_connected_cb, path=path, limit=limit, **kwds
        )

    async def create_unix_connection(
        protocol_factory: Callable[[], asyncio.BaseProtocol],
        path: os.PathLike[str] | str | bytes | None = None,
        **kwds: Any,
    ) -> tuple[asyncio.Transport, asyncio.BaseProtocol]:
        loop = asyncio.get_running_loop()
        return await loop.create_unix_connection(protocol_factory, path=path, **kwds)

    async def create_unix_server(
        protocol_factory: Callable[[], asyncio.BaseProtocol],
        path: os.PathLike[str] | str | bytes | None = None,
        **kwds: Any,
    ) -> asyncio.AbstractServer:
        loop = asyncio.get_running_loop()
        return await loop.create_unix_server(protocol_factory, path=path, **kwds)

    def install() -> None:
        """No-op on POSIX. asyncio already has native AF_UNIX helpers."""
        return None

else:
    import ctypes
    import ctypes.wintypes as wintypes

    if not hasattr(socket, "AF_UNIX"):
        # Winsock uses 1 for AF_UNIX.  Python normally exposes this on modern
        # Windows builds, but keep this for older/minimal builds.
        socket.AF_UNIX = 1  # type: ignore[attr-defined]

    UNIX_PATH_MAX = 108

    FD_READ_BIT = 0
    FD_WRITE_BIT = 1
    FD_OOB_BIT = 2
    FD_ACCEPT_BIT = 3
    FD_CONNECT_BIT = 4
    FD_CLOSE_BIT = 5

    FD_READ = 1 << FD_READ_BIT
    FD_WRITE = 1 << FD_WRITE_BIT
    FD_OOB = 1 << FD_OOB_BIT
    FD_ACCEPT = 1 << FD_ACCEPT_BIT
    FD_CONNECT = 1 << FD_CONNECT_BIT
    FD_CLOSE = 1 << FD_CLOSE_BIT

    WSAEINTR = 10004
    WSAEWOULDBLOCK = 10035
    WSAEINPROGRESS = 10036
    WSAEALREADY = 10037
    WSAEINVAL = 10022
    WSAECONNRESET = 10054
    WSAENOTCONN = 10057

    WSA_INFINITE = 0xFFFFFFFF
    WSA_WAIT_EVENT_0 = 0
    WSA_WAIT_FAILED = 0xFFFFFFFF

    SOCKET_T = ctypes.c_uint64 if ctypes.sizeof(ctypes.c_void_p) == 8 else ctypes.c_uint32
    INVALID_SOCKET = (1 << (ctypes.sizeof(SOCKET_T) * 8)) - 1

    class SockaddrUn(ctypes.Structure):
        _fields_ = [
            ("sun_family", wintypes.USHORT),
            ("sun_path", ctypes.c_char * UNIX_PATH_MAX),
        ]

    class WSAData(ctypes.Structure):
        _fields_ = [
            ("wVersion", wintypes.WORD),
            ("wHighVersion", wintypes.WORD),
            ("szDescription", ctypes.c_char * 257),
            ("szSystemStatus", ctypes.c_char * 129),
            ("iMaxSockets", wintypes.USHORT),
            ("iMaxUdpDg", wintypes.USHORT),
            ("lpVendorInfo", ctypes.c_char_p),
        ]

    class WSANetworkEvents(ctypes.Structure):
        _fields_ = [
            ("lNetworkEvents", ctypes.c_long),
            ("iErrorCode", ctypes.c_int * 10),
        ]

    ws2_32 = ctypes.WinDLL("ws2_32", use_last_error=True)

    ws2_32.WSAStartup.restype = ctypes.c_int
    ws2_32.WSAStartup.argtypes = (wintypes.WORD, ctypes.POINTER(WSAData))

    ws2_32.WSAGetLastError.restype = ctypes.c_int
    ws2_32.WSAGetLastError.argtypes = ()

    ws2_32.WSACreateEvent.restype = wintypes.HANDLE
    ws2_32.WSACreateEvent.argtypes = ()

    ws2_32.WSASetEvent.restype = wintypes.BOOL
    ws2_32.WSASetEvent.argtypes = (wintypes.HANDLE,)

    ws2_32.WSACloseEvent.restype = wintypes.BOOL
    ws2_32.WSACloseEvent.argtypes = (wintypes.HANDLE,)

    ws2_32.WSAEventSelect.restype = ctypes.c_int
    ws2_32.WSAEventSelect.argtypes = (SOCKET_T, wintypes.HANDLE, ctypes.c_long)

    ws2_32.WSAEnumNetworkEvents.restype = ctypes.c_int
    ws2_32.WSAEnumNetworkEvents.argtypes = (
        SOCKET_T,
        wintypes.HANDLE,
        ctypes.POINTER(WSANetworkEvents),
    )

    ws2_32.WSAWaitForMultipleEvents.restype = wintypes.DWORD
    ws2_32.WSAWaitForMultipleEvents.argtypes = (
        wintypes.DWORD,
        ctypes.POINTER(wintypes.HANDLE),
        wintypes.BOOL,
        wintypes.DWORD,
        wintypes.BOOL,
    )

    ws2_32.bind.restype = ctypes.c_int
    ws2_32.bind.argtypes = (SOCKET_T, ctypes.POINTER(SockaddrUn), ctypes.c_int)

    ws2_32.connect.restype = ctypes.c_int
    ws2_32.connect.argtypes = (SOCKET_T, ctypes.POINTER(SockaddrUn), ctypes.c_int)

    ws2_32.listen.restype = ctypes.c_int
    ws2_32.listen.argtypes = (SOCKET_T, ctypes.c_int)

    ws2_32.accept.restype = SOCKET_T
    ws2_32.accept.argtypes = (SOCKET_T, ctypes.POINTER(SockaddrUn), ctypes.POINTER(ctypes.c_int))

    ws2_32.recv.restype = ctypes.c_int
    ws2_32.recv.argtypes = (SOCKET_T, ctypes.c_void_p, ctypes.c_int, ctypes.c_int)

    ws2_32.send.restype = ctypes.c_int
    ws2_32.send.argtypes = (SOCKET_T, ctypes.c_void_p, ctypes.c_int, ctypes.c_int)

    _wsa_started = False

    def _ensure_winsock() -> None:
        global _wsa_started
        if _wsa_started:
            return
        data = WSAData()
        ret = ws2_32.WSAStartup(0x0202, ctypes.byref(data))
        if ret:
            raise ctypes.WinError(ret)
        _wsa_started = True

    def _last_wsa_error() -> int:
        return int(ws2_32.WSAGetLastError())

    def _raise_last_wsa_error() -> None:
        raise ctypes.WinError(_last_wsa_error())

    def _socket_handle(sock: socket.socket | int) -> int:
        if isinstance(sock, socket.socket):
            return int(sock.fileno())
        return int(sock)

    def _is_invalid_socket(value: int) -> bool:
        return value in (-1, INVALID_SOCKET)

    def _path_bytes(path: os.PathLike[str] | str | bytes) -> bytes:
        if isinstance(path, bytes):
            raw = path
        else:
            raw = os.fsencode(os.fspath(path))
        if b"\x00" in raw:
            raise ValueError("AF_UNIX path must not contain NUL bytes on Windows")
        if len(raw) >= UNIX_PATH_MAX:
            raise OSError(206, f"AF_UNIX path is too long: {len(raw)} bytes")
        return raw

    def _sockaddr(path: os.PathLike[str] | str | bytes) -> SockaddrUn:
        raw = _path_bytes(path)
        addr = SockaddrUn()
        addr.sun_family = socket.AF_UNIX  # type: ignore[attr-defined]
        addr.sun_path = raw
        return addr

    def _check_no_ssl(ssl: Any, server_hostname: Any = None) -> None:
        if ssl is not None:
            raise NotImplementedError("SSL over Windows AF_UNIX is not implemented")
        if server_hostname is not None:
            raise ValueError("server_hostname is only meaningful with ssl")

    async def _wait_for_handle(loop: asyncio.AbstractEventLoop, handle: int) -> None:
        proactor = getattr(loop, "_proactor", None)
        if proactor is not None and hasattr(proactor, "wait_for_handle"):
            await proactor.wait_for_handle(handle)
            return

        # Fallback for non-Proactor event loops.  This costs one executor thread
        # while waiting, but keeps the public API usable.
        def blocking_wait() -> None:
            handles = (wintypes.HANDLE * 1)(handle)
            rc = ws2_32.WSAWaitForMultipleEvents(
                1, handles, True, WSA_INFINITE, False
            )
            if rc == WSA_WAIT_FAILED:
                _raise_last_wsa_error()

        await loop.run_in_executor(None, blocking_wait)

    class _AsyncWSASocket:
        """One WSAEventSelect event handle plus an asyncio waiter multiplexer."""

        def __init__(
            self,
            sock: socket.socket,
            mask: int,
            *,
            loop: asyncio.AbstractEventLoop,
        ) -> None:
            _ensure_winsock()
            self.sock = sock
            self.loop = loop
            self.handle = _socket_handle(sock)
            self.event = int(ws2_32.WSACreateEvent())
            if not self.event:
                _raise_last_wsa_error()
            self._closed = False
            self._waiters: list[tuple[int, asyncio.Future[WSANetworkEvents]]] = []
            self._pump_task: asyncio.Task[None] | None = None

            if ws2_32.WSAEventSelect(self.handle, self.event, mask) != 0:
                err = _last_wsa_error()
                ws2_32.WSACloseEvent(self.event)
                raise ctypes.WinError(err)

        def start(self) -> None:
            if self._pump_task is None:
                self._pump_task = self.loop.create_task(self._pump())

        async def wait_for(self, mask: int) -> WSANetworkEvents:
            if self._closed:
                raise OSError("socket watcher is closed")
            self.start()
            fut: asyncio.Future[WSANetworkEvents] = self.loop.create_future()
            self._waiters.append((mask, fut))
            try:
                return await fut
            finally:
                if fut.cancelled():
                    self._waiters = [(m, f) for (m, f) in self._waiters if f is not fut]

        async def _pump(self) -> None:
            try:
                while not self._closed:
                    await _wait_for_handle(self.loop, self.event)
                    if self._closed:
                        break
                    ne = WSANetworkEvents()
                    rc = ws2_32.WSAEnumNetworkEvents(
                        self.handle, self.event, ctypes.byref(ne)
                    )
                    if rc != 0:
                        exc = ctypes.WinError(_last_wsa_error())
                        self._finish_waiters(exc=exc)
                        break
                    self._finish_waiters(events=ne)
            except asyncio.CancelledError:
                raise
            except BaseException as exc:
                self._finish_waiters(exc=exc)

        def _finish_waiters(
            self,
            *,
            events: WSANetworkEvents | None = None,
            exc: BaseException | None = None,
        ) -> None:
            if exc is not None:
                waiters, self._waiters = self._waiters, []
                for _, fut in waiters:
                    if not fut.done():
                        fut.set_exception(exc)
                return

            if events is None:
                return
            mask_seen = int(events.lNetworkEvents)
            if not mask_seen:
                return

            remaining: list[tuple[int, asyncio.Future[WSANetworkEvents]]] = []
            ready: list[asyncio.Future[WSANetworkEvents]] = []
            for mask, fut in self._waiters:
                if fut.done():
                    continue
                if mask_seen & mask:
                    ready.append(fut)
                else:
                    remaining.append((mask, fut))
            self._waiters = remaining
            for fut in ready:
                if not fut.done():
                    fut.set_result(events)

        def close(self) -> None:
            if self._closed:
                return
            self._closed = True
            self._finish_waiters(exc=ConnectionAbortedError("socket watcher closed"))
            try:
                ws2_32.WSAEventSelect(self.handle, 0, 0)
            except Exception:
                pass
            try:
                ws2_32.WSASetEvent(self.event)
            except Exception:
                pass
            if self._pump_task is not None:
                self._pump_task.cancel()
            try:
                ws2_32.WSACloseEvent(self.event)
            except Exception:
                pass

    def _event_error(events: WSANetworkEvents, bit: int) -> int:
        return int(events.iErrorCode[bit])

    def _recv_nowait(handle: int, max_bytes: int) -> bytes | None:
        buf = ctypes.create_string_buffer(max_bytes)
        n = ws2_32.recv(handle, buf, max_bytes, 0)
        if n > 0:
            return bytes(buf.raw[:n])
        if n == 0:
            return b""
        err = _last_wsa_error()
        if err == WSAEWOULDBLOCK:
            return None
        if err in (WSAECONNRESET, WSAENOTCONN):
            return b""
        raise ctypes.WinError(err)

    def _send_nowait(handle: int, data: memoryview) -> int | None:
        if not data:
            return 0
        # Winsock send() needs a stable C buffer for the duration of the call.
        chunk = data[: min(len(data), 256 * 1024)]
        buf = (ctypes.c_char * len(chunk)).from_buffer_copy(chunk)
        n = ws2_32.send(handle, buf, len(chunk), 0)
        if n >= 0:
            return int(n)
        err = _last_wsa_error()
        if err == WSAEWOULDBLOCK:
            return None
        if err in (WSAECONNRESET, WSAENOTCONN):
            raise ConnectionResetError(err, "connection reset")
        raise ctypes.WinError(err)

    async def _connect_socket(
        path: os.PathLike[str] | str | bytes,
        *,
        loop: asyncio.AbstractEventLoop,
        sock: socket.socket | None = None,
    ) -> _AsyncWSASocket:
        _ensure_winsock()
        if sock is None:
            sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)  # type: ignore[attr-defined]
        watcher = _AsyncWSASocket(sock, FD_READ | FD_WRITE | FD_CONNECT | FD_CLOSE, loop=loop)

        addr = _sockaddr(path)
        rc = ws2_32.connect(
            _socket_handle(sock), ctypes.byref(addr), ctypes.sizeof(addr)
        )
        if rc == 0:
            watcher.start()
            return watcher

        err = _last_wsa_error()
        if err not in (WSAEWOULDBLOCK, WSAEINPROGRESS, WSAEALREADY, WSAEINVAL):
            watcher.close()
            sock.close()
            raise ctypes.WinError(err)

        events = await watcher.wait_for(FD_CONNECT | FD_CLOSE)
        if int(events.lNetworkEvents) & FD_CONNECT:
            err = _event_error(events, FD_CONNECT_BIT)
            if err:
                watcher.close()
                sock.close()
                raise ctypes.WinError(err)
            return watcher

        watcher.close()
        sock.close()
        raise ConnectionRefusedError("AF_UNIX connection closed before connect completed")

    class _UnixSocketTransport(asyncio.Transport):
        def __init__(
            self,
            loop: asyncio.AbstractEventLoop,
            watcher: _AsyncWSASocket,
            protocol: asyncio.BaseProtocol,
            *,
            peername: Any = None,
            sockname: Any = None,
        ) -> None:
            super().__init__()
            self._loop = loop
            self._watcher = watcher
            self._sock = watcher.sock
            self._protocol = protocol
            self._peername = peername
            self._sockname = sockname
            self._closing = False
            self._conn_lost = False
            self._reading_paused = False
            self._write_buffer = bytearray()
            self._write_task: asyncio.Task[None] | None = None
            self._read_task: asyncio.Task[None] | None = None
            self._high_water = 64 * 1024
            self._low_water = 16 * 1024
            self._protocol_paused = False
            self._resume_reading_waiter: asyncio.Future[None] | None = None

            self._watcher.start()
            self._protocol.connection_made(self)
            self._read_task = self._loop.create_task(self._read_loop())

        def is_closing(self) -> bool:
            return self._closing or self._conn_lost

        def get_extra_info(self, name: str, default: Any = None) -> Any:
            if name == "socket":
                return self._sock
            if name == "peername":
                return self._peername
            if name == "sockname":
                return self._sockname
            return default

        def set_write_buffer_limits(
            self,
            high: int | None = None,
            low: int | None = None,
        ) -> None:
            if high is None:
                high = 64 * 1024
            if low is None:
                low = high // 4
            if high < low or low < 0:
                raise ValueError("high must be >= low >= 0")
            self._high_water = high
            self._low_water = low
            self._maybe_pause_protocol()

        def get_write_buffer_size(self) -> int:
            return len(self._write_buffer)

        def write(self, data: bytes | bytearray | memoryview) -> None:
            if self._conn_lost:
                return
            if self._closing:
                raise RuntimeError("cannot write to closing transport")
            if not isinstance(data, (bytes, bytearray, memoryview)):
                raise TypeError("data must be bytes-like")
            if not data:
                return
            self._write_buffer.extend(data)
            self._maybe_pause_protocol()
            if self._write_task is None or self._write_task.done():
                self._write_task = self._loop.create_task(self._write_loop())

        def writelines(self, list_of_data: Iterable[bytes]) -> None:
            for data in list_of_data:
                self.write(data)

        def can_write_eof(self) -> bool:
            return True

        def write_eof(self) -> None:
            try:
                self._sock.shutdown(socket.SHUT_WR)
            except OSError:
                pass

        def pause_reading(self) -> bool:
            if self._reading_paused:
                return False
            self._reading_paused = True
            return True

        def resume_reading(self) -> bool:
            if not self._reading_paused:
                return False
            self._reading_paused = False
            waiter = self._resume_reading_waiter
            if waiter is not None and not waiter.done():
                waiter.set_result(None)
            return True

        def close(self) -> None:
            if self._conn_lost or self._closing:
                return
            self._closing = True
            if not self._write_buffer:
                self._force_close(None)

        def abort(self) -> None:
            self._force_close(ConnectionAbortedError("transport aborted"))

        def _maybe_pause_protocol(self) -> None:
            if (
                not self._protocol_paused
                and len(self._write_buffer) > self._high_water
                and hasattr(self._protocol, "pause_writing")
            ):
                self._protocol_paused = True
                self._protocol.pause_writing()  # type: ignore[attr-defined]

        def _maybe_resume_protocol(self) -> None:
            if (
                self._protocol_paused
                and len(self._write_buffer) <= self._low_water
                and hasattr(self._protocol, "resume_writing")
            ):
                self._protocol_paused = False
                self._protocol.resume_writing()  # type: ignore[attr-defined]

        async def _read_loop(self) -> None:
            try:
                while not self._closing and not self._conn_lost:
                    if self._reading_paused:
                        self._resume_reading_waiter = self._loop.create_future()
                        try:
                            await self._resume_reading_waiter
                        finally:
                            self._resume_reading_waiter = None
                        continue

                    data = _recv_nowait(self._watcher.handle, _DEFAULT_LIMIT)
                    if data is None:
                        events = await self._watcher.wait_for(FD_READ | FD_CLOSE)
                        if int(events.lNetworkEvents) & FD_READ:
                            err = _event_error(events, FD_READ_BIT)
                            if err:
                                raise ctypes.WinError(err)
                        if int(events.lNetworkEvents) & FD_CLOSE:
                            err = _event_error(events, FD_CLOSE_BIT)
                            if err and err not in (WSAECONNRESET, WSAENOTCONN):
                                raise ctypes.WinError(err)
                        continue
                    if data == b"":
                        eof_accepted = False
                        if hasattr(self._protocol, "eof_received"):
                            eof_accepted = bool(self._protocol.eof_received())  # type: ignore[attr-defined]
                        if not eof_accepted:
                            self._force_close(None)
                        return
                    if hasattr(self._protocol, "data_received"):
                        self._protocol.data_received(data)  # type: ignore[attr-defined]
            except asyncio.CancelledError:
                raise
            except BaseException as exc:
                self._force_close(exc)

        async def _write_loop(self) -> None:
            try:
                while self._write_buffer and not self._conn_lost:
                    view = memoryview(self._write_buffer)
                    try:
                        n = _send_nowait(self._watcher.handle, view)
                    finally:
                        view.release()
                    if n is None:
                        events = await self._watcher.wait_for(FD_WRITE | FD_CLOSE)
                        if int(events.lNetworkEvents) & FD_WRITE:
                            err = _event_error(events, FD_WRITE_BIT)
                            if err:
                                raise ctypes.WinError(err)
                        if int(events.lNetworkEvents) & FD_CLOSE:
                            err = _event_error(events, FD_CLOSE_BIT)
                            if err and err not in (WSAECONNRESET, WSAENOTCONN):
                                raise ctypes.WinError(err)
                        continue
                    if n == 0:
                        await self._watcher.wait_for(FD_WRITE | FD_CLOSE)
                        continue
                    del self._write_buffer[:n]
                    self._maybe_resume_protocol()
            except asyncio.CancelledError:
                raise
            except BaseException as exc:
                self._force_close(exc)
                return

            if self._closing:
                self._force_close(None)

        def _force_close(self, exc: BaseException | None) -> None:
            if self._conn_lost:
                return
            self._conn_lost = True
            self._closing = True
            self._watcher.close()
            try:
                self._sock.close()
            except OSError:
                pass

            read_task = self._read_task
            if read_task is not None and read_task is not asyncio.current_task(self._loop):
                read_task.cancel()
            write_task = self._write_task
            if write_task is not None and write_task is not asyncio.current_task(self._loop):
                write_task.cancel()

            if hasattr(self._protocol, "connection_lost"):
                self._loop.call_soon(self._protocol.connection_lost, exc)

    class _WindowsUnixServer:
        def __init__(
            self,
            loop: asyncio.AbstractEventLoop,
            sock: socket.socket,
            watcher: _AsyncWSASocket,
            protocol_factory: Callable[[], asyncio.BaseProtocol],
            *,
            cleanup_path: os.PathLike[str] | str | bytes | None = None,
        ) -> None:
            self._loop = loop
            self._sock = sock
            self._watcher = watcher
            self._protocol_factory = protocol_factory
            self._cleanup_path = cleanup_path
            self._serving = False
            self._closed = False
            self._accept_task: asyncio.Task[None] | None = None
            self._closed_fut: asyncio.Future[None] = loop.create_future()

        @property
        def sockets(self) -> tuple[socket.socket, ...]:
            return (self._sock,)

        def get_loop(self) -> asyncio.AbstractEventLoop:
            return self._loop

        def is_serving(self) -> bool:
            return self._serving and not self._closed

        async def start_serving(self) -> None:
            if self._closed:
                raise RuntimeError("server is closed")
            if self._serving:
                return
            self._serving = True
            self._watcher.start()
            self._accept_task = self._loop.create_task(self._accept_loop())

        async def serve_forever(self) -> None:
            await self.start_serving()
            fut = self._loop.create_future()
            try:
                await fut
            except asyncio.CancelledError:
                self.close()
                await self.wait_closed()
                raise

        def close(self) -> None:
            if self._closed:
                return
            self._closed = True
            self._serving = False
            if self._accept_task is not None:
                self._accept_task.cancel()
            self._watcher.close()
            try:
                self._sock.close()
            except OSError:
                pass
            if self._cleanup_path is not None:
                try:
                    os.unlink(self._cleanup_path)
                except FileNotFoundError:
                    pass
                except OSError:
                    # Match asyncio's best-effort cleanup behaviour.
                    pass
            if not self._closed_fut.done():
                self._closed_fut.set_result(None)

        async def wait_closed(self) -> None:
            await self._closed_fut

        async def _accept_loop(self) -> None:
            try:
                while self._serving and not self._closed:
                    client_handle = self._accept_nowait()
                    if client_handle is None:
                        events = await self._watcher.wait_for(FD_ACCEPT | FD_CLOSE)
                        if int(events.lNetworkEvents) & FD_ACCEPT:
                            err = _event_error(events, FD_ACCEPT_BIT)
                            if err:
                                raise ctypes.WinError(err)
                        if int(events.lNetworkEvents) & FD_CLOSE:
                            return
                        continue
                    self._start_client(client_handle)
            except asyncio.CancelledError:
                raise
            except BaseException as exc:
                self._loop.call_exception_handler(
                    {
                        "message": "AF_UNIX server accept loop failed",
                        "exception": exc,
                        "server": self,
                    }
                )
                self.close()

        def _accept_nowait(self) -> int | None:
            addr = SockaddrUn()
            addr_len = ctypes.c_int(ctypes.sizeof(addr))
            client = int(
                ws2_32.accept(
                    _socket_handle(self._sock), ctypes.byref(addr), ctypes.byref(addr_len)
                )
            )
            if not _is_invalid_socket(client):
                return client
            err = _last_wsa_error()
            if err == WSAEWOULDBLOCK:
                return None
            raise ctypes.WinError(err)

        def _start_client(self, client_handle: int) -> None:
            try:
                client_sock = socket.socket(
                    socket.AF_UNIX, socket.SOCK_STREAM, fileno=client_handle  # type: ignore[attr-defined]
                )
                watcher = _AsyncWSASocket(
                    client_sock, FD_READ | FD_WRITE | FD_CLOSE, loop=self._loop
                )
                protocol = self._protocol_factory()
                _UnixSocketTransport(
                    self._loop,
                    watcher,
                    protocol,
                    peername="",
                    sockname=self._cleanup_path,
                )
            except BaseException as exc:
                self._loop.call_exception_handler(
                    {
                        "message": "failed to initialize AF_UNIX client transport",
                        "exception": exc,
                        "server": self,
                    }
                )
                try:
                    socket.socket(
                        socket.AF_UNIX, socket.SOCK_STREAM, fileno=client_handle  # type: ignore[attr-defined]
                    ).close()
                except Exception:
                    pass

    async def create_unix_connection(
        protocol_factory: Callable[[], asyncio.BaseProtocol],
        path: os.PathLike[str] | str | bytes | None = None,
        *,
        ssl: Any = None,
        sock: socket.socket | None = None,
        server_hostname: str | None = None,
        ssl_handshake_timeout: float | None = None,
        ssl_shutdown_timeout: float | None = None,
    ) -> tuple[asyncio.Transport, asyncio.BaseProtocol]:
        _check_no_ssl(ssl, server_hostname)
        if path is None and sock is None:
            raise TypeError("path was not specified, and no sock specified")
        if path is not None and sock is not None:
            raise ValueError("path and sock can not be specified at the same time")

        loop = asyncio.get_running_loop()
        if sock is not None:
            watcher = _AsyncWSASocket(sock, FD_READ | FD_WRITE | FD_CLOSE, loop=loop)
        else:
            watcher = await _connect_socket(path, loop=loop)  # type: ignore[arg-type]

        protocol = protocol_factory()
        transport = _UnixSocketTransport(
            loop,
            watcher,
            protocol,
            peername=path,
            sockname=None,
        )
        return transport, protocol

    async def open_unix_connection(
        path: os.PathLike[str] | str | bytes | None = None,
        *,
        limit: int = _DEFAULT_LIMIT,
        ssl: Any = None,
        sock: socket.socket | None = None,
        server_hostname: str | None = None,
        ssl_handshake_timeout: float | None = None,
        ssl_shutdown_timeout: float | None = None,
    ) -> tuple[asyncio.StreamReader, asyncio.StreamWriter]:
        _check_no_ssl(ssl, server_hostname)
        loop = asyncio.get_running_loop()
        reader = asyncio.StreamReader(limit=limit)
        protocol = asyncio.StreamReaderProtocol(reader)

        transport, _ = await create_unix_connection(
            lambda: protocol,
            path=path,
            ssl=ssl,
            sock=sock,
            server_hostname=server_hostname,
            ssl_handshake_timeout=ssl_handshake_timeout,
            ssl_shutdown_timeout=ssl_shutdown_timeout,
        )
        writer = asyncio.StreamWriter(transport, protocol, reader, loop)
        return reader, writer

    async def create_unix_server(
        protocol_factory: Callable[[], asyncio.BaseProtocol],
        path: os.PathLike[str] | str | bytes | None = None,
        *,
        sock: socket.socket | None = None,
        backlog: int = 100,
        ssl: Any = None,
        ssl_handshake_timeout: float | None = None,
        ssl_shutdown_timeout: float | None = None,
        start_serving: bool = True,
        cleanup_socket: bool = True,
    ) -> _WindowsUnixServer:
        _check_no_ssl(ssl)
        if path is None and sock is None:
            raise TypeError("path was not specified, and no sock specified")
        if path is not None and sock is not None:
            raise ValueError("path and sock can not be specified at the same time")

        _ensure_winsock()
        loop = asyncio.get_running_loop()
        cleanup_path: os.PathLike[str] | str | bytes | None = None

        if sock is None:
            sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)  # type: ignore[attr-defined]
            addr = _sockaddr(path)  # type: ignore[arg-type]
            if ws2_32.bind(_socket_handle(sock), ctypes.byref(addr), ctypes.sizeof(addr)) != 0:
                err = _last_wsa_error()
                sock.close()
                raise ctypes.WinError(err)
            if ws2_32.listen(_socket_handle(sock), int(backlog)) != 0:
                err = _last_wsa_error()
                sock.close()
                raise ctypes.WinError(err)
            if cleanup_socket:
                cleanup_path = path
        else:
            if ws2_32.listen(_socket_handle(sock), int(backlog)) != 0:
                err = _last_wsa_error()
                raise ctypes.WinError(err)

        watcher = _AsyncWSASocket(sock, FD_ACCEPT | FD_CLOSE, loop=loop)
        server = _WindowsUnixServer(
            loop,
            sock,
            watcher,
            protocol_factory,
            cleanup_path=cleanup_path,
        )
        if start_serving:
            await server.start_serving()
        return server

    async def start_unix_server(
        client_connected_cb: Callable[[asyncio.StreamReader, asyncio.StreamWriter], Any],
        path: os.PathLike[str] | str | bytes | None = None,
        *,
        limit: int = _DEFAULT_LIMIT,
        sock: socket.socket | None = None,
        backlog: int = 100,
        ssl: Any = None,
        ssl_handshake_timeout: float | None = None,
        ssl_shutdown_timeout: float | None = None,
        start_serving: bool = True,
        cleanup_socket: bool = True,
    ) -> _WindowsUnixServer:
        _check_no_ssl(ssl)
        loop = asyncio.get_running_loop()

        def factory() -> asyncio.StreamReaderProtocol:
            reader = asyncio.StreamReader(limit=limit)
            protocol: asyncio.StreamReaderProtocol

            def connected_cb(
                reader: asyncio.StreamReader,
                writer: asyncio.StreamWriter,
            ) -> None:
                result = client_connected_cb(reader, writer)
                if inspect.isawaitable(result):
                    asyncio.ensure_future(result)

            protocol = asyncio.StreamReaderProtocol(reader, connected_cb)
            return protocol

        return await create_unix_server(
            factory,
            path=path,
            sock=sock,
            backlog=backlog,
            ssl=ssl,
            ssl_handshake_timeout=ssl_handshake_timeout,
            ssl_shutdown_timeout=ssl_shutdown_timeout,
            start_serving=start_serving,
            cleanup_socket=cleanup_socket,
        )

    def install() -> None:
        """
        Monkey-patch asyncio's module-level stream helpers on Windows.

        This intentionally patches only asyncio.open_unix_connection and
        asyncio.start_unix_server.  Event-loop instance methods are not patched
        because their implementation classes differ between Python versions.
        Use this module's create_unix_connection/create_unix_server directly
        when protocol-level APIs are needed.
        """
        asyncio.open_unix_connection = open_unix_connection  # type: ignore[assignment]
        asyncio.start_unix_server = start_unix_server  # type: ignore[assignment]

How to Force bcache to Recognize a Backing Device Without a Cache Device

vast cow — Thu, 18 Jun 2026 09:14:43 +0000

In an environment using bcache, if the cache device fails, disappears, or is disconnected, the bcache device (such as /dev/bcache0) may not be created automatically from the backing device alone.

In this situation, you can register the backing device with bcache and then write 1 to running to force the backing device to start even without a cache device.

Prerequisites

This article assumes the following backing device:

/dev/sdd4

Replace /dev/sdd4 with the appropriate device for your environment.

Procedure

1. Load the bcache kernel module

sudo modprobe bcache

2. Register the backing device with bcache

echo /dev/sdd4 | sudo tee /sys/fs/bcache/register

This makes bcache recognize the backing device.

3. Force startup without a cache device

echo 1 | sudo tee /sys/class/block/sdd/sdd4/bcache/running

This creates a bcache device such as /dev/bcache0.

Verify it:

lsblk
ls -l /dev/bcache*

The created device will typically be named /dev/bcache0, /dev/bcache1, and so on.

Which Device to Mount

Mount the generated bcache device, not the backing device itself.

For example, if /dev/bcache0 is created:

sudo mount /dev/bcache0 /mnt

Do not mount /dev/sdd4 directly; always use the bcache device such as /dev/bcache0.

About the sysfs Path

When a partition is used as the backing device, running is located at:

/sys/class/block/sdd/sdd4/bcache/running

In this example, the disk is /dev/sdd and the partition is /dev/sdd4, so this is the correct path.

For NVMe devices, if the backing device is /dev/nvme0n1p3, the path would be:

/sys/class/block/nvme0n1/nvme0n1p3/bcache/running

Summary

The minimum steps required to force bcache to recognize a backing device when no cache device is available are:

sudo modprobe bcache
echo /dev/sdd4 | sudo tee /sys/fs/bcache/register
echo 1 | sudo tee /sys/class/block/sdd/sdd4/bcache/running

Afterward, mount the generated bcache device:

sudo mount /dev/bcache0 /mnt

Important Notes

If writeback caching was being used, dirty data may still exist on the cache device.

Forcing the backing device to start without the cache device in that situation can result in filesystem inconsistencies or data corruption.

Be especially careful if:

The cache mode was writeback
The cache device failed unexpectedly
Dirty data was not flushed before shutdown
The consistency between the backing device and cache device is unknown

If you want to inspect the data safely, mount it read-only:

sudo mount -o ro /dev/bcache0 /mnt

For data recovery purposes, it is safest to mount the filesystem read-only first and copy important data elsewhere before attempting any repairs or modifications.

Settings to Save a Longer bash History, Reduce Duplicates, and Make It Easier to Use Across Multiple Terminals

vast cow — Thu, 18 Jun 2026 08:55:33 +0000

If you use bash regularly, you often want to find commands you ran earlier. With the default settings, the number of saved history entries may be small, or history may not be shared well when you work with multiple terminals open.

This article introduces basic settings for saving bash history for a longer period and making it easier to handle across multiple sessions.

Settings

Add the following settings to ~/.bashrc.

export HISTSIZE=100000
export HISTFILESIZE=200000
shopt -s histappend
export PROMPT_COMMAND='history -a; history -c; history -r'

After adding the settings, apply them with the following command.

source ~/.bashrc

Purpose

The main purpose of these settings is to make bash command history more convenient to use.

By increasing HISTSIZE and HISTFILESIZE, you can keep commands you ran in the past for longer. This makes it easier to reuse frequently used commands or complex commands you ran a little while ago.

Also, enabling histappend makes bash append to the history file instead of overwriting it. Even when you are working with multiple terminals open, your history is less likely to disappear.

In addition, the PROMPT_COMMAND setting saves history again each time you run a command and also reloads history executed in other terminals. This makes it easier to share history across multiple sessions.

Meaning of Each Setting

`HISTSIZE`

export HISTSIZE=100000

Specifies the number of history entries kept in the current shell.

In this example, up to 100,000 history entries are kept in memory.

`HISTFILESIZE`

export HISTFILESIZE=200000

Specifies the number of entries saved in the history file.

In this example, up to 200,000 history entries are saved in the history file.

`histappend`

shopt -s histappend

This setting appends to the history file when bash exits instead of overwriting it.

When using multiple terminals, this helps reduce problems such as only the history from the terminal closed last being preserved.

`PROMPT_COMMAND`

export PROMPT_COMMAND='history -a; history -c; history -r'

Specifies commands to run each time the prompt is displayed.

With this setting, the following processes are performed in order.

`history -a`

Appends the history added in the current session to the history file.

`history -c`

Clears the history in the current shell once.

`history -r`

Reloads history from the history file.

This makes commands executed in other terminals more likely to be reflected in the current terminal’s history.

Usage

After configuring these settings, just use bash as usual.

To search past commands, press Ctrl + r and use history search. Enter part of a command to search for commands you ran in the past.

Ctrl + r

You can also display the history list with the history command.

history

Since the history becomes long, it is useful to combine it with grep.

history | grep ssh
history | grep docker
history | grep rsync

Notes

Because these settings save history for a long time, you need to be careful not to enter confidential information such as passwords or tokens directly on the command line.

If you accidentally enter confidential information, you need to delete the relevant line from the history file. Normally, the bash history file is located here.

~/.bash_history

Summary

With these settings in place, you can save bash history for a long period and more easily share history when working with multiple terminals.

If you often use the terminal, this can improve work efficiency by allowing you to quickly reuse past commands.

When did WSL's --import add support for xz and zstd?

vast cow — Thu, 18 Jun 2026 07:50:42 +0000

The conclusion is: I could not find an official release note that says "supported from this version onward." However, there are lower bounds that can be verified from publicly available information.

Format	Earliest verifiable evidence from public sources	Confidence
`.tar.xz`	WSL 1.1.3.0, as of 2023-03-12	Relatively high
`.tar.zst`	WSL 2.1.5, as of 2024-04-11	Moderate

`.tar.xz` Was Working at Least in WSL 1.1.3.0

In GitHub issue #9778, the reported environment is WSL Version 1.1.3.0, and the reproduction steps include:

wsl --import ... lunar-server-cloudimg-amd64-root.tar.xz

The issue itself is not about import failures but about problems opening the imported distribution from VS Code afterward. That suggests that using .tar.xz with wsl --import was already a valid workflow at that time. (GitHub)

`.tar.zst` Was Working at Least in WSL 2.1.5

In a comment on GitHub issue #6056, dated 2024-04-11, a user reported that on WSL 2.1.5 (latest at the time), importing .tar.gz, .tar.xz, and .tar.zst all worked.

This is a community report rather than an official release note, but it is still useful as the earliest confirmation I could find for .tar.zst support. (GitHub)

As of 2020, Support Was Not Yet an Explicitly Supported Feature

Issue #6056, opened in October 2020, was a feature request asking for .xz support in wsl.exe --import and the WSL Distro Launcher ecosystem.

The issue notes that the expected formats at the time were install.tar and gzip-compressed archives such as install.tar.gz or install.tgz, and requests support for XZ-compressed archives.

In other words, as of 2020, .tar.xz was at least not an explicitly documented or officially assumed format. (GitHub)

Internally, Import Does Not Appear to Hard-Code Compression Formats

Looking at the current WSL source code, the wsl --import path appears to reject .vhd and .vhdx when passed without --vhd, then opens the file and passes it to RegisterDistribution.

The wsl.exe client code does not appear to explicitly branch on .gz, .xz, or .zst extensions. (GitHub)

By contrast, the --export side exposes explicit format options such as:

tar.gz
tar.xz
tar
vhd

This suggests that export explicitly enumerates output formats, while import largely hands a tar-based archive to backend extraction logic and lets that layer handle the compression format. (GitHub)

Summary

Strictly speaking:

.tar.xz was being used with wsl --import no later than WSL 1.1.3.0 (March 2023).

.tar.zst has confirmed working reports no later than WSL 2.1.5 (April 2024).

However, no official changelog has been found that explicitly states when support for either format was first introduced.

Therefore, support did not first appear in WSL 2.5.x. In particular, .tar.xz appears to have been working significantly earlier.

Reverse Proxying Local Applications with User-Mode nginx and AF_UNIX Sockets

vast cow — Thu, 18 Jun 2026 07:11:45 +0000

When running multiple local web applications, it is often useful to route requests to different applications based on URL prefixes.

For example:

http://localhost:8080/app1/

might need to be reverse proxied to an application running in a separate process.

This is easy to achieve with nginx. However, instead of using a system-wide nginx instance, this setup follows a different approach:

Run nginx as an unprivileged user
Use AF_UNIX (Unix domain) sockets instead of TCP ports for backend connections
Make the URL-to-application mapping obvious from the nginx configuration itself

Approach

The configuration runs nginx listening on localhost:8080.

Requests under /app1/ are forwarded to a Unix domain socket exposed by the application.

browser
  |
  | http://localhost:8080/app1/
  v
nginx
  |
  | AF_UNIX socket
  v
/home/user/app1/sock.sock

Assigning a dedicated TCP port to each application is also possible, but for local development reverse proxies, Unix domain sockets often provide a cleaner layout.

In the configuration, the mapping is explicit:

location /app1/ {
    proxy_pass http://unix:/home/user/app1/sock.sock:/;
}

From this alone, it is immediately clear which application corresponds to /app1/.

Directory Layout

Create a working directory for nginx:

nginx-local/
├── conf/
│   └── nginx.conf
├── logs/
└── temp/
    ├── client_body/
    ├── proxy/
    ├── fastcgi/
    ├── uwsgi/
    └── scgi/

Example setup:

mkdir -p ./conf
mkdir -p ./logs
mkdir -p ./temp/client_body
mkdir -p ./temp/proxy
mkdir -p ./temp/fastcgi
mkdir -p ./temp/uwsgi
mkdir -p ./temp/scgi

Starting nginx

This nginx instance runs in the foreground:

nginx -p . -c ./conf/nginx.conf -e ./logs/error.log -g 'daemon off;'

The -p . option sets the nginx prefix to the current directory.

Relative paths in the configuration, such as ./logs/error.log or ./temp/proxy, are resolved relative to this prefix.

Because -g 'daemon off;' is specified, nginx does not daemonize and instead runs in the foreground. This is generally more convenient for development and testing. Stop it with Ctrl-C.

Configuration File

conf/nginx.conf:

worker_processes 1;

error_log ./logs/error.log;
pid       ./nginx.pid;

events {
    worker_connections 1024;
}

http {
    access_log ./logs/access.log;

    client_body_temp_path ./temp/client_body;
    proxy_temp_path       ./temp/proxy;
    fastcgi_temp_path     ./temp/fastcgi;
    uwsgi_temp_path       ./temp/uwsgi;
    scgi_temp_path        ./temp/scgi;

    map $http_upgrade $connection_upgrade {
        default upgrade;
        ''      close;
    }

    server {
        listen 8080;
        server_name localhost;

        location = /app1 {
            return 301 /app1/;
        }

        location /app1/ {
            proxy_pass http://unix:/home/user/app1/sock.sock:/;

            proxy_http_version 1.1;

            proxy_set_header Host              $host;
            proxy_set_header X-Real-IP         $remote_addr;
            proxy_set_header X-Forwarded-For   $proxy_add_x_forwarded_for;
            proxy_set_header X-Forwarded-Proto $scheme;

            proxy_set_header Upgrade           $http_upgrade;
            proxy_set_header Connection        $connection_upgrade;

            proxy_buffering off;
            proxy_cache off;

            proxy_read_timeout 3600s;
            proxy_send_timeout 3600s;
        }
    }
}

Important Considerations for Running nginx as an Unprivileged User

When running nginx without root privileges, several points require attention.

First, use a listening port above 1024:

listen 8080;

Attempting to bind directly to privileged ports such as 80 or 443 as a regular user will fail.

Second, ensure that log files, PID files, and temporary files are written to locations the user can access:

error_log ./logs/error.log;
pid       ./nginx.pid;

Also explicitly configure temporary directories used by HTTP proxying:

client_body_temp_path ./temp/client_body;
proxy_temp_path       ./temp/proxy;
fastcgi_temp_path     ./temp/fastcgi;
uwsgi_temp_path       ./temp/uwsgi;
scgi_temp_path        ./temp/scgi;

Without these settings, nginx may attempt to use directories such as /var/log/nginx/ or /var/lib/nginx/, which are typically not writable by unprivileged users.

Reverse Proxying to AF_UNIX Sockets

To proxy requests to a Unix domain socket, proxy_pass can be written as:

proxy_pass http://unix:/path/to/sock.sock:/;

In this example:

proxy_pass http://unix:/home/user/app1/sock.sock:/;

the /app1/ prefix is stripped before forwarding.

Therefore:

http://localhost:8080/app1/foo

is forwarded to the backend application as:

/foo

If the application needs to receive /app1/foo unchanged, the proxy_pass URI handling must be adjusted accordingly.

For this setup, nginx uses /app1/ purely as a routing prefix, while applications receive paths relative to /.

WebSocket Support

To proxy WebSocket connections correctly, HTTP/1.1 and the Upgrade / Connection headers must be configured:

proxy_http_version 1.1;

proxy_set_header Upgrade    $http_upgrade;
proxy_set_header Connection $connection_upgrade;

Since $connection_upgrade is not a built-in nginx variable, it is defined via map inside the http block:

map $http_upgrade $connection_upgrade {
    default upgrade;
    ''      close;
}

Always setting Connection: upgrade would incorrectly affect normal HTTP requests and SSE connections. This mapping ensures that upgrade is only used when an Upgrade header is present.

Server-Sent Events (SSE) / HTTP Event Streams

For SSE and streaming HTTP responses, nginx buffering can prevent events from being delivered immediately.

Disable proxy buffering:

proxy_buffering off;
proxy_cache off;

Use generous timeouts to allow long-lived connections:

proxy_read_timeout 3600s;
proxy_send_timeout 3600s;

With these settings, standard HTTP traffic, WebSockets, and SSE can all use the same reverse proxy configuration.

Verification

Validate the configuration:

nginx -p . -c ./conf/nginx.conf -e ./logs/error.log -t

Start nginx:

nginx -p . -c ./conf/nginx.conf -e ./logs/error.log -g 'daemon off;'

Test HTTP:

curl -i http://localhost:8080/app1/

For SSE, disable curl buffering:

curl -N http://localhost:8080/app1/events

For WebSocket testing, tools such as websocat are useful:

websocat ws://localhost:8080/app1/ws

Summary

This approach provides two major advantages.

First, nginx can run entirely in user space.

There is no need to modify system nginx configuration or obtain root privileges. Everything can be managed within a local working directory, making it convenient for development and personal tooling.

Second, using AF_UNIX sockets makes application mappings explicit in the configuration:

location /app1/ {
    proxy_pass http://unix:/home/user/app1/sock.sock:/;
}

From the nginx configuration alone, it is easy to determine which application serves a given URL prefix.

For managing multiple local web applications behind a single endpoint, the combination of user-mode nginx and AF_UNIX sockets is a practical and maintainable solution.

A Simple Tool for Converting WebArchive and MHTML Files

vast cow — Thu, 18 Jun 2026 04:43:25 +0000

Purpose

This tool converts files between Apple .webarchive format and MHTML/MHT format.

It is useful when you have saved a web page in one format but need to open, share, or store it in another format. For example, Safari often uses .webarchive, while many browsers and email-based tools use .mhtml or .mht.

The converter works in both directions:

.mhtml or .mht to .webarchive
.webarchive to .mhtml

It is designed to be simple to run from the command line and does not require any external Python packages.

Main Benefits

The tool helps users:

Convert saved web pages between common archive formats
Keep the main page and related resources together
Choose an output file name when needed
Let the tool automatically detect the conversion direction
Use a few optional settings for special cases

Because it uses only Python’s standard library, setup is minimal.

Requirements

To use this tool, you need:

Python 3.9 or newer
The webarchive_mhtml_converter.py script
A .webarchive, .mhtml, or .mht file to convert

No additional libraries need to be installed.

Basic Usage

The simplest way to use the tool is to provide the input file.

For an MHTML file:

python webarchive_mhtml_converter.py page.mhtml

This creates:

page.webarchive

For a WebArchive file:

python webarchive_mhtml_converter.py page.webarchive

This creates:

page.mhtml

The tool detects the input file type from the extension and automatically chooses the opposite output format.

Choosing an Output File Name

You can choose the output file path with -o or --output.

Example:

python webarchive_mhtml_converter.py page.mhtml -o converted.webarchive

Another example:

python webarchive_mhtml_converter.py page.webarchive -o converted.mhtml

This is useful when you want to keep the original file name unchanged or save the converted file in another folder.

Choosing the Output Format Manually

If the input file extension is unclear, you can specify the target format with --to.

To create a WebArchive file:

python webarchive_mhtml_converter.py page.dat --to webarchive -o page.webarchive

To create an MHTML file:

python webarchive_mhtml_converter.py page.dat --to mhtml -o page.mhtml

This avoids confusion when the tool cannot determine the conversion direction automatically.

Optional Settings

Keeping CID CSS Links

When converting from MHTML to WebArchive, the tool normally adjusts certain stylesheet links to improve compatibility.

To keep those links unchanged, use:

python webarchive_mhtml_converter.py page.mhtml --keep-cid-css-link

Choosing the WebArchive Plist Format

When converting from MHTML to WebArchive, the default output format is binary.

To create an XML plist instead, use:

python webarchive_mhtml_converter.py page.mhtml --plist-format xml

Excluding Subframes

When converting from WebArchive to MHTML, the tool normally includes subframe archives.

To exclude them, use:

python webarchive_mhtml_converter.py page.webarchive --no-subframes

Error Handling

If the tool cannot convert the file, it prints an error message.

Common reasons include:

The input file does not exist
The input path is not a file
The file extension does not match the requested output type
The conversion direction cannot be detected
An option is used with the wrong conversion direction

These messages help identify what needs to be fixed before running the command again.

Summary

webarchive_mhtml_converter.py is a small command-line tool for converting saved web pages between Apple WebArchive and MHTML/MHT formats.

It is best suited for users who need a practical way to move archived web pages between different browsers, systems, or workflows. Basic conversion requires only one command, while optional settings provide more control when needed.

#!/usr/bin/env python3
"""
Unified converter for Apple .webarchive and MHTML/MHT files.

This combines the behavior of the two separate tools:
  - MHTML/MHT -> Apple WebArchive
  - Apple WebArchive -> MHTML/MHT

Requirements:
  - Python 3.9+
  - No external dependencies; uses only the Python standard library.

Examples:
  # Auto-detect direction from input extension
  python webarchive_mhtml_converter.py page.mhtml
  python webarchive_mhtml_converter.py page.webarchive

  # Explicit output path
  python webarchive_mhtml_converter.py page.mhtml -o page.webarchive
  python webarchive_mhtml_converter.py page.webarchive -o page.mhtml

  # Explicit conversion target
  python webarchive_mhtml_converter.py page.dat --to webarchive -o page.webarchive
  python webarchive_mhtml_converter.py page.dat --to mhtml -o page.mhtml

  # Direction-specific options
  python webarchive_mhtml_converter.py page.mhtml --keep-cid-css-link
  python webarchive_mhtml_converter.py page.webarchive --no-subframes
"""

from __future__ import annotations

import argparse
import base64
import email
import mimetypes
import plistlib
import re
import sys
import uuid
from email import policy
from email.message import EmailMessage, Message
from email.parser import BytesParser
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple
from urllib.parse import quote, urljoin, urlparse


Resource = Dict[str, Any]
Archive = Dict[str, Any]

MHTML_SUFFIXES = {".mhtml", ".mht"}
WEBARCHIVE_SUFFIXES = {".webarchive"}

TEXT_LIKE_MIME_PREFIXES = ("text/",)
TEXT_LIKE_MIME_TYPES = {
    "application/javascript",
    "application/ecmascript",
    "application/json",
    "application/xml",
    "application/xhtml+xml",
    "image/svg+xml",
}

DEFAULT_BINARY_MIME = "application/octet-stream"


# ---------------------------------------------------------------------------
# Shared helpers
# ---------------------------------------------------------------------------


def normalize_mime_type(mime_type: str) -> str:
    return (mime_type or DEFAULT_BINARY_MIME).split(";", 1)[0].strip().lower()


def is_text_like_mime(mime_type: str) -> bool:
    mt = normalize_mime_type(mime_type)
    return mt.startswith(TEXT_LIKE_MIME_PREFIXES) or mt in TEXT_LIKE_MIME_TYPES


def kind_from_suffix(path: Path) -> Optional[str]:
    suffix = path.suffix.lower()
    if suffix in WEBARCHIVE_SUFFIXES:
        return "webarchive"
    if suffix in MHTML_SUFFIXES:
        return "mhtml"
    return None


def default_output_path(input_path: Path, target: str) -> Path:
    if target == "mhtml":
        return input_path.with_suffix(".mhtml")
    if target == "webarchive":
        return input_path.with_suffix(".webarchive")
    raise ValueError(f"Unsupported target: {target}")


def infer_target(input_path: Path, output_path: Optional[Path], requested_target: str) -> str:
    """
    Return the output target: "mhtml" or "webarchive".

    In auto mode, the input extension is authoritative. If the input extension is
    unknown, the output extension is used as a fallback.
    """
    if requested_target != "auto":
        return requested_target

    input_kind = kind_from_suffix(input_path)
    if input_kind == "webarchive":
        return "mhtml"
    if input_kind == "mhtml":
        return "webarchive"

    if output_path is not None:
        output_kind = kind_from_suffix(output_path)
        if output_kind in {"mhtml", "webarchive"}:
            return output_kind

    raise ValueError(
        "Could not infer conversion direction. Use --to mhtml or --to webarchive."
    )


def validate_paths_and_direction(
    input_path: Path,
    output_path: Path,
    target: str,
    *,
    requested_target: str,
) -> None:
    if not input_path.exists():
        raise FileNotFoundError(f"Input file does not exist: {input_path}")
    if not input_path.is_file():
        raise ValueError(f"Input path is not a file: {input_path}")

    input_kind = kind_from_suffix(input_path)
    output_kind = kind_from_suffix(output_path)

    if input_kind == target:
        raise ValueError(
            f"Input extension already looks like the requested output type ({target}). "
            "Check --to or the input file extension."
        )

    if output_kind is not None and output_kind != target:
        raise ValueError(
            f"Output extension implies {output_kind}, but target is {target}: {output_path}"
        )

    if requested_target == "auto" and input_kind is None and output_kind is None:
        raise ValueError(
            "Could not infer conversion direction from extensions. Use --to explicitly."
        )


# ---------------------------------------------------------------------------
# MHTML/MHT -> Apple WebArchive
# ---------------------------------------------------------------------------


def normalize_cid(value: Optional[str]) -> Optional[str]:
    if not value:
        return None
    value = value.strip()
    if value.startswith("<") and value.endswith(">"):
        value = value[1:-1]
    return value


def cid_url_from_part(part: EmailMessage) -> Optional[str]:
    cid = normalize_cid(part.get("Content-ID"))
    return f"cid:{cid}" if cid else None


def is_absolute_or_cid(url: str) -> bool:
    if url.startswith("cid:"):
        return True
    parsed = urlparse(url)
    return bool(parsed.scheme)


def text_encoding_from_content_type(content_type: str) -> str:
    msg = Message()
    msg["content-type"] = content_type
    return msg.get_content_charset() or "utf-8"


def fix_mime_type(url: str, mime: str) -> str:
    mime = normalize_mime_type(mime)
    path = urlparse(url).path.lower()
    guessed, _ = mimetypes.guess_type(path)

    if path.endswith(".css") and mime not in {"text/css", DEFAULT_BINARY_MIME}:
        return "text/css"
    if path.endswith(".css") and mime == DEFAULT_BINARY_MIME:
        return "text/css"
    if path.endswith(".js") and mime in {"text/plain", DEFAULT_BINARY_MIME}:
        return "application/javascript"
    if guessed and mime == DEFAULT_BINARY_MIME:
        return normalize_mime_type(guessed)
    return mime


def payload_bytes(part: EmailMessage) -> bytes:
    data = part.get_payload(decode=True)
    if data is not None:
        return data

    raw = part.get_payload()
    if isinstance(raw, str):
        enc = part.get_content_charset() or "utf-8"
        return raw.encode(enc, errors="replace")

    return b""


def choose_root_part(msg: EmailMessage, parts: List[EmailMessage]) -> EmailMessage:
    start = normalize_cid(msg.get_param("start"))
    if start:
        for part in parts:
            if normalize_cid(part.get("Content-ID")) == start:
                return part

    # RFC-compatible fallback for typical MHTML saved by browsers.
    for part in parts:
        if part.get_content_type() in {"text/html", "application/xhtml+xml"}:
            return part

    raise ValueError("No HTML root part found in MHTML")


def mhtml_part_url(part: EmailMessage, base_url: Optional[str], index: int) -> str:
    loc = (part.get("Content-Location") or "").strip()
    if loc:
        if is_absolute_or_cid(loc):
            return loc
        if base_url:
            return urljoin(base_url, loc)
        return loc

    cid = cid_url_from_part(part)
    if cid:
        return cid

    return f"mhtml-resource-{index}"


def make_web_resource(url: str, mime: str, data: bytes, *, content_type_header: Optional[str] = None) -> Resource:
    fixed_mime = fix_mime_type(url, mime)

    if is_text_like_mime(fixed_mime):
        data = data.rstrip(b"\x00")

    resource: Resource = {
        "WebResourceURL": url,
        "WebResourceMIMEType": fixed_mime,
        "WebResourceData": data,
    }

    if is_text_like_mime(fixed_mime):
        resource["WebResourceTextEncodingName"] = text_encoding_from_content_type(
            content_type_header or fixed_mime
        )

    return resource


def strip_inline_charset(css: str) -> str:
    # @charset is only meaningful as a stylesheet byte-stream marker; remove it for <style>.
    return re.sub(r'^\s*@charset\s+["\'][^"\']+["\']\s*;\s*', "", css, flags=re.I)


def replace_stylesheet_link(html: str, url: str, css_text: str) -> str:
    style_tag = '<style type="text/css">\n' + strip_inline_charset(css_text) + "\n</style>"

    def repl(match: re.Match[str]) -> str:
        tag = match.group(0)
        href_pat = r'\bhref\s*=\s*(["\'])' + re.escape(url) + r'\1'
        rel_pat = r'\brel\s*=\s*(["\'])[^"\']*stylesheet[^"\']*\1'
        if re.search(href_pat, tag, flags=re.I) and re.search(rel_pat, tag, flags=re.I):
            return style_tag
        return tag

    return re.sub(r"<link\b[^>]*>", repl, html, flags=re.I)


def inline_cid_stylesheets(archive: Archive) -> None:
    resources = [archive["WebMainResource"], *archive.get("WebSubresources", [])]

    cid_css: List[Tuple[str, str]] = []
    for resource in resources:
        if resource.get("WebResourceMIMEType") == "text/css" and str(
            resource.get("WebResourceURL", "")
        ).startswith("cid:"):
            enc = resource.get("WebResourceTextEncodingName") or "utf-8"
            css = (
                resource.get("WebResourceData", b"")
                .rstrip(b"\x00")
                .decode(enc, errors="replace")
            )
            cid_css.append((resource["WebResourceURL"], css))

    if not cid_css:
        return

    for resource in resources:
        if resource.get("WebResourceMIMEType") not in {"text/html", "application/xhtml+xml"}:
            continue
        enc = resource.get("WebResourceTextEncodingName") or "utf-8"
        html = (
            resource.get("WebResourceData", b"")
            .rstrip(b"\x00")
            .decode(enc, errors="replace")
        )
        for url, css in cid_css:
            html = replace_stylesheet_link(html, url, css)
        resource["WebResourceData"] = html.encode(enc, errors="replace")


def parse_mhtml(path: Path, *, inline_cid_css: bool = True) -> Archive:
    msg = BytesParser(policy=policy.default).parsebytes(path.read_bytes())
    if not msg.is_multipart():
        raise ValueError("Input is not multipart MHTML")

    parts = [p for p in msg.walk() if not p.is_multipart()]
    root = choose_root_part(msg, parts)

    snapshot_url = (msg.get("Snapshot-Content-Location") or "").strip() or None
    root_loc = (root.get("Content-Location") or "").strip() or snapshot_url
    root_url = root_loc or cid_url_from_part(root) or "about:blank"

    main_data = payload_bytes(root)
    main_mime = root.get_content_type() or "text/html"
    main_resource = make_web_resource(
        root_url,
        main_mime,
        main_data,
        content_type_header=root.get("Content-Type"),
    )

    subresources: List[Resource] = []
    seen_main_identity = id(root)

    for idx, part in enumerate(parts, start=1):
        if id(part) == seen_main_identity:
            # Add a cid: alias for the root only if the HTML might refer to it.
            cid = cid_url_from_part(part)
            if cid and cid != root_url:
                subresources.append(
                    make_web_resource(
                        cid,
                        main_mime,
                        main_data,
                        content_type_header=part.get("Content-Type"),
                    )
                )
            continue

        url = mhtml_part_url(part, root_url, idx)
        mime = part.get_content_type() or DEFAULT_BINARY_MIME
        data = payload_bytes(part)
        subresources.append(
            make_web_resource(
                url,
                mime,
                data,
                content_type_header=part.get("Content-Type"),
            )
        )

        # Alias Content-ID as cid:... when Content-Location differs.
        cid = cid_url_from_part(part)
        if cid and cid != url:
            subresources.append(
                make_web_resource(
                    cid,
                    mime,
                    data,
                    content_type_header=part.get("Content-Type"),
                )
            )

        # Keep raw relative Content-Location as an alias for snapshots that use it verbatim.
        raw_loc = (part.get("Content-Location") or "").strip()
        if raw_loc and raw_loc != url and not is_absolute_or_cid(raw_loc):
            subresources.append(
                make_web_resource(
                    raw_loc,
                    mime,
                    data,
                    content_type_header=part.get("Content-Type"),
                )
            )

    archive: Archive = {
        "WebMainResource": main_resource,
        "WebSubresources": subresources,
    }

    if inline_cid_css:
        inline_cid_stylesheets(archive)

    return archive


def convert_mhtml_to_webarchive(
    input_path: Path,
    output_path: Path,
    *,
    inline_cid_css: bool = True,
    plist_format: str = "binary",
) -> None:
    archive = parse_mhtml(input_path, inline_cid_css=inline_cid_css)
    fmt = plistlib.FMT_BINARY if plist_format == "binary" else plistlib.FMT_XML
    with output_path.open("wb") as f:
        plistlib.dump(archive, f, fmt=fmt, sort_keys=False)


# ---------------------------------------------------------------------------
# Apple WebArchive -> MHTML/MHT
# ---------------------------------------------------------------------------


def load_webarchive(path: Path) -> Archive:
    """
    Load .webarchive with plistlib.

    plistlib supports both XML plist and binary plist. In .webarchive files,
    WebResourceData is preserved as bytes.
    """
    with path.open("rb") as f:
        archive = plistlib.load(f)

    if not isinstance(archive, dict):
        raise ValueError("Invalid webarchive: root object is not a dictionary")

    if not isinstance(archive.get("WebMainResource"), dict):
        raise ValueError("Invalid webarchive: missing WebMainResource")

    return archive


def decode_webresource_data(value: Any) -> bytes:
    """
    Decode WebResourceData.

    With plistlib this should normally be bytes. Other forms are handled
    defensively for compatibility with unusual plist representations.
    """
    if value is None:
        return b""

    if isinstance(value, bytes):
        return value

    if isinstance(value, bytearray):
        return bytes(value)

    if isinstance(value, str):
        s = value.strip()

        # Defensive support for base64 strings.
        try:
            padded = s + ("=" * ((4 - len(s) % 4) % 4))
            decoded = base64.b64decode(padded, validate=True)
            if base64.b64encode(decoded).decode("ascii").rstrip("=") == s.rstrip("="):
                return decoded
        except Exception:
            pass

        # Defensive support for plain text strings.
        return value.encode("utf-8")

    if isinstance(value, dict):
        for key in ("bytes", "data", "base64", "WebResourceData"):
            if key in value:
                return decode_webresource_data(value[key])

    raise TypeError(f"Unsupported WebResourceData type: {type(value).__name__}")


def guess_mime_type(resource: Resource, data: bytes, is_main: bool) -> str:
    mime = resource.get("WebResourceMIMEType")
    if isinstance(mime, str) and mime.strip():
        return normalize_mime_type(mime)

    url = resource.get("WebResourceURL")
    if isinstance(url, str):
        guessed, _ = mimetypes.guess_type(url)
        if guessed:
            return normalize_mime_type(guessed)

    if is_main:
        return "text/html"

    # Basic signature detection.
    if data.startswith(b"\x89PNG\r\n\x1a\n"):
        return "image/png"
    if data.startswith(b"\xff\xd8\xff"):
        return "image/jpeg"
    if data.startswith(b"GIF87a") or data.startswith(b"GIF89a"):
        return "image/gif"
    if data.startswith(b"RIFF") and data[8:12] == b"WEBP":
        return "image/webp"
    if data.startswith(b"\x00\x00\x00") and b"ftypavif" in data[:32]:
        return "image/avif"
    if data.startswith(b"wOFF"):
        return "font/woff"
    if data.startswith(b"wOF2"):
        return "font/woff2"
    if data.lstrip().startswith((b"<svg", b"<?xml")):
        return "image/svg+xml"

    return DEFAULT_BINARY_MIME


def get_charset(resource: Resource, mime_type: str) -> Optional[str]:
    encoding = resource.get("WebResourceTextEncodingName")
    if isinstance(encoding, str) and encoding.strip():
        return encoding.strip()

    if is_text_like_mime(mime_type):
        return "utf-8"

    return None


def content_location(resource: Resource, fallback: str) -> str:
    url = resource.get("WebResourceURL")
    if isinstance(url, str) and url.strip():
        return url.strip()
    return fallback


def sanitize_header_value(value: str) -> str:
    """
    Keep Content-Location browser-friendly.

    Do not use email.header/Header or RFC 2047 encoded-word for URLs. Remove
    CR/LF to prevent header injection. Percent-encode non-ASCII only.
    """
    value = value.replace("\r", "").replace("\n", "")

    try:
        value.encode("ascii")
        return value
    except UnicodeEncodeError:
        # Keep normal URL punctuation readable; encode non-ASCII characters.
        return quote(value, safe=":/?#[]@!$&'()*+,;=%")


def fold_header_line(name: str, value: str, limit: int = 998) -> bytes:
    """
    Fold long header lines without RFC 2047 encoding.

    RFC 5322 hard limit is 998 octets per line. MHTML readers are usually
    happier with raw folded URLs than encoded-word URLs.
    """
    prefix = f"{name}: "
    raw = sanitize_header_value(value)

    line = prefix + raw
    encoded = line.encode("utf-8")

    if len(encoded) <= limit:
        return encoded + b"\r\n"

    # Fold on safe URL boundary characters where possible.
    out: List[bytes] = []
    current = prefix

    for token in re.split(r"([/?&=#.;,:_-])", raw):
        if token == "":
            continue

        candidate = current + token
        if len(candidate.encode("utf-8")) <= limit:
            current = candidate
            continue

        out.append(current.encode("utf-8") + b"\r\n")
        current = " " + token

    if current:
        out.append(current.encode("utf-8") + b"\r\n")

    return b"".join(out)


def make_content_type(mime_type: str, charset: Optional[str]) -> str:
    if charset and is_text_like_mime(mime_type):
        return f'{mime_type}; charset="{charset}"'
    return mime_type


def wrap_base64(data: bytes) -> bytes:
    """Base64-wrap to MIME's conventional 76-character lines."""
    return base64.encodebytes(data).replace(b"\n", b"\r\n")


def make_mhtml_part(
    *,
    data: bytes,
    mime_type: str,
    charset: Optional[str],
    location: str,
    content_id: Optional[str] = None,
) -> bytes:
    headers = bytearray()
    headers += fold_header_line("Content-Type", make_content_type(mime_type, charset))
    headers += b"Content-Transfer-Encoding: base64\r\n"
    headers += fold_header_line("Content-Location", location)

    if content_id:
        headers += fold_header_line("Content-ID", f"<{content_id}>")

    return bytes(headers) + b"\r\n" + wrap_base64(data)


def iter_archive_resources(
    archive: Archive,
    *,
    include_subframes: bool,
    prefix: str = "",
) -> Iterable[Tuple[Resource, bool, str]]:
    """
    Yield (resource, is_main, fallback_location).

    Main resource is yielded before subresources for each archive. Subframe
    archives are recursively included after subresources.
    """
    main = archive.get("WebMainResource")
    if isinstance(main, dict):
        yield main, True, f"{prefix}main-resource"

    subresources = archive.get("WebSubresources") or archive.get("WebSubResources") or []
    if isinstance(subresources, list):
        for i, resource in enumerate(subresources, start=1):
            if isinstance(resource, dict):
                yield resource, False, f"{prefix}resource-{i}"

    if include_subframes:
        subframes = archive.get("WebSubframeArchives") or []
        if isinstance(subframes, list):
            for frame_index, subarchive in enumerate(subframes, start=1):
                if isinstance(subarchive, dict):
                    frame_prefix = f"{prefix}frame-{frame_index}-"
                    yield from iter_archive_resources(
                        subarchive,
                        include_subframes=True,
                        prefix=frame_prefix,
                    )


def build_mhtml(
    archive: Archive,
    *,
    source_name: str,
    include_subframes: bool = True,
) -> bytes:
    boundary = f"----=_NextPart_{uuid.uuid4().hex}"
    boundary_bytes = boundary.encode("ascii")

    resources = list(
        iter_archive_resources(
            archive,
            include_subframes=include_subframes,
        )
    )

    if not resources:
        raise ValueError("Invalid webarchive: no resources found")

    main_resource, _, _ = resources[0]
    main_data = decode_webresource_data(main_resource.get("WebResourceData"))
    main_mime = guess_mime_type(main_resource, main_data, is_main=True)

    lines = bytearray()
    lines += b"MIME-Version: 1.0\r\n"
    lines += fold_header_line("Subject", f"Converted from {source_name}")
    lines += fold_header_line(
        "Content-Type",
        f'multipart/related; type="{main_mime}"; start="<main-resource>"; boundary="{boundary}"',
    )
    lines += b"\r\n"
    lines += b"This is a multi-part message in MIME format.\r\n"

    seen_locations = set()

    for resource_index, (resource, is_main, fallback_location) in enumerate(resources):
        data = decode_webresource_data(resource.get("WebResourceData"))
        if not data:
            continue

        mime_type = guess_mime_type(resource, data, is_main=is_main)
        charset = get_charset(resource, mime_type)
        location = content_location(resource, fallback_location)

        if location in seen_locations:
            continue
        seen_locations.add(location)

        content_id = "main-resource" if resource_index == 0 else None

        lines += b"\r\n--" + boundary_bytes + b"\r\n"
        lines += make_mhtml_part(
            data=data,
            mime_type=mime_type,
            charset=charset,
            location=location,
            content_id=content_id,
        )

    lines += b"\r\n--" + boundary_bytes + b"--\r\n"
    return bytes(lines)


def convert_webarchive_to_mhtml(
    input_path: Path,
    output_path: Path,
    *,
    include_subframes: bool = True,
) -> None:
    archive = load_webarchive(input_path)
    mhtml = build_mhtml(
        archive,
        source_name=input_path.name,
        include_subframes=include_subframes,
    )
    output_path.write_bytes(mhtml)


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Convert between Apple .webarchive and .mhtml/.mht files."
    )
    parser.add_argument("input", type=Path, help="Input .webarchive, .mhtml, or .mht file")
    parser.add_argument(
        "-o",
        "--output",
        type=Path,
        help="Output path. Default: input name with the opposite extension.",
    )
    parser.add_argument(
        "--to",
        choices=("auto", "mhtml", "webarchive"),
        default="auto",
        help=(
            "Output format. Default auto: .webarchive input becomes .mhtml; "
            ".mhtml/.mht input becomes .webarchive."
        ),
    )
    parser.add_argument(
        "--keep-cid-css-link",
        action="store_true",
        help=(
            "For MHTML -> WebArchive only: do not inline cid: stylesheet links. "
            "Default is to inline them for better iOS WebArchive compatibility."
        ),
    )
    parser.add_argument(
        "--plist-format",
        choices=("binary", "xml"),
        default="binary",
        help="For MHTML -> WebArchive only: plist output format. Default: binary.",
    )
    parser.add_argument(
        "--no-subframes",
        action="store_true",
        help="For WebArchive -> MHTML only: do not include WebSubframeArchives recursively.",
    )
    return parser


def run(args: argparse.Namespace) -> int:
    input_path: Path = args.input
    target = infer_target(input_path, args.output, args.to)
    output_path: Path = args.output or default_output_path(input_path, target)

    validate_paths_and_direction(
        input_path,
        output_path,
        target,
        requested_target=args.to,
    )

    if target == "webarchive":
        if args.no_subframes:
            raise ValueError("--no-subframes applies only to WebArchive -> MHTML conversion")
        convert_mhtml_to_webarchive(
            input_path,
            output_path,
            inline_cid_css=not args.keep_cid_css_link,
            plist_format=args.plist_format,
        )
    elif target == "mhtml":
        if args.keep_cid_css_link:
            raise ValueError("--keep-cid-css-link applies only to MHTML -> WebArchive conversion")
        if args.plist_format != "binary":
            raise ValueError("--plist-format applies only to MHTML -> WebArchive conversion")
        convert_webarchive_to_mhtml(
            input_path,
            output_path,
            include_subframes=not args.no_subframes,
        )
    else:
        raise ValueError(f"Unsupported target: {target}")

    print(f"wrote: {output_path}")
    return 0


def main(argv: Optional[Sequence[str]] = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)

    try:
        return run(args)
    except Exception as e:
        print(f"error: {e}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())

Activate the Window for the Specified HWND

vast cow — Mon, 15 Jun 2026 11:13:54 +0000

In Windows applications, there are cases where you want to bring a specific window to the front and make it the active target for user interaction.

This code is designed to activate the window associated with a specified HWND and, when possible, make it the foreground window.

Purpose

The purpose of this function is to make the specified window available for user interaction.

For example, it can be used when you want to bring an application hidden behind other windows to the front, or restore and display a minimized window.

Usage

The primary function is activate_window.

bool result = activate_window(hwnd);

Pass the handle of the window you want to activate in hwnd.

The function returns true if the operation succeeds and the window becomes the foreground window.

It returns false if the operation fails.

Processing Flow

The code first converts the specified HWND into its top-level window.

This ensures that even if a child window handle is provided, the operation targets the parent top-level window.

Next, it verifies that the target window exists and can be displayed.

If the target window is already in the foreground, the function immediately returns true.

When the Window Is Minimized

If the target window is minimized, it is restored using ShowWindowAsync.

ShowWindowAsync(hwnd, SW_RESTORE);

If the window is not minimized, it is shown normally.

ShowWindowAsync(hwnd, SW_SHOW);

This makes hidden or minimized windows visible again.

Bringing the Window to the Front

After displaying the window, the code attempts to bring it to the foreground using operations such as:

SetWindowPos(...);
BringWindowToTop(hwnd);
SetForegroundWindow(hwnd);
SetActiveWindow(hwnd);
SetFocus(hwnd);

By combining these APIs, the code tries to move the specified window to the front and assign keyboard focus as reliably as possible.

Using AttachThreadInput

Windows may prevent a window owned by another thread from being activated directly.

To address this, the code temporarily associates the current thread with the target window's thread using AttachThreadInput.

This increases the likelihood that SetForegroundWindow and SetFocus will succeed.

After the operation is complete, the thread association is removed.

AttachThreadInput(current_tid, target_tid, FALSE);

In other words, the threads are temporarily linked, the activation work is performed, and then the connection is released.

ALT Key Fallback

If the normal activation process fails, the code sends an ALT key press as a fallback.

tap_alt_key();

On Windows, sending an ALT key event can sometimes make SetForegroundWindow succeed afterward.

This can be particularly effective in scenarios involving UWP applications or ApplicationFrameHost.

The code then retries displaying the window and bringing it to the foreground.

Return Value

At the end of the process, activate_window checks whether the target window is actually the foreground window.

return is_foreground(hwnd);

It returns true if successful and false otherwise.

Example Use Cases

This function can be useful in situations such as:

Bringing an external application's window to the foreground
Restoring a minimized window
Making a specific tool window the target of user interaction
Switching to a designated window in a multi-window application

Notes

Windows includes restrictions that prevent applications from freely forcing other windows into the foreground.

As a result, this code cannot guarantee success in every situation.

However, by combining top-level window normalization, thread input attachment, minimized-window restoration, and an ALT-key fallback mechanism, it achieves a practical and generally reliable approach.

Summary

This code provides a practical helper function for activating a window specified by an HWND.

Rather than relying solely on SetForegroundWindow, it includes window-state validation, restoration, thread-input attachment, and fallback handling to improve reliability.

If you need to bring a specific window to the foreground on Windows, this function serves as a convenient and robust utility.

#define WIN32_LEAN_AND_MEAN
#include <windows.h>
#include <stdbool.h>

static HWND normalize_top_level(HWND hwnd)
{
    HWND root;

    if (!hwnd) return NULL;

    root = GetAncestor(hwnd, GA_ROOT);
    if (root && IsWindow(root)) return root;

    return hwnd;
}

static bool same_top_level(HWND a, HWND b)
{
    return normalize_top_level(a) == normalize_top_level(b);
}

static bool is_foreground(HWND hwnd)
{
    HWND fg = GetForegroundWindow();
    return fg && same_top_level(fg, hwnd);
}

static void tap_alt_key(void)
{
    INPUT inputs[2] = {0};

    inputs[0].type = INPUT_KEYBOARD;
    inputs[0].ki.wVk = VK_MENU;

    inputs[1].type = INPUT_KEYBOARD;
    inputs[1].ki.wVk = VK_MENU;
    inputs[1].ki.dwFlags = KEYEVENTF_KEYUP;

    SendInput(2, inputs, sizeof(INPUT));
}

bool activate_window(HWND hwnd)
{
    DWORD current_tid;
    DWORD foreground_tid = 0;
    DWORD target_tid = 0;
    DWORD dummy_pid = 0;
    HWND foreground_hwnd;
    BOOL attached_foreground = FALSE;
    BOOL attached_target = FALSE;
    MSG msg;

    hwnd = normalize_top_level(hwnd);

    if (!hwnd) return false;
    if (!IsWindow(hwnd)) return false;
    if (!IsWindowVisible(hwnd)) return false;

    if (is_foreground(hwnd)) return true;

    /*
        AttachThreadInput 用に自スレッドの message queue を作る。
    */
    PeekMessage(&msg, NULL, 0, 0, PM_NOREMOVE);

    current_tid = GetCurrentThreadId();

    foreground_hwnd = GetForegroundWindow();
    if (foreground_hwnd && IsWindow(foreground_hwnd)) {
        foreground_tid = GetWindowThreadProcessId(foreground_hwnd, &dummy_pid);
    }

    target_tid = GetWindowThreadProcessId(hwnd, &dummy_pid);

    if (foreground_tid && foreground_tid != current_tid) {
        attached_foreground = AttachThreadInput(current_tid, foreground_tid, TRUE);
    }

    if (target_tid && target_tid != current_tid && target_tid != foreground_tid) {
        attached_target = AttachThreadInput(current_tid, target_tid, TRUE);
    }

    if (IsIconic(hwnd)) {
        ShowWindowAsync(hwnd, SW_RESTORE);
    } else {
        ShowWindowAsync(hwnd, SW_SHOW);
    }

    SetWindowPos(
        hwnd,
        HWND_TOP,
        0,
        0,
        0,
        0,
        SWP_NOMOVE |
        SWP_NOSIZE |
        SWP_SHOWWINDOW |
        SWP_ASYNCWINDOWPOS
    );

    BringWindowToTop(hwnd);
    SetForegroundWindow(hwnd);
    SetActiveWindow(hwnd);
    SetFocus(hwnd);

    if (attached_target) {
        AttachThreadInput(current_tid, target_tid, FALSE);
    }

    if (attached_foreground) {
        AttachThreadInput(current_tid, foreground_tid, FALSE);
    }

    if (is_foreground(hwnd)) {
        return true;
    }

    tap_alt_key();

    if (IsIconic(hwnd)) {
        ShowWindowAsync(hwnd, SW_RESTORE);
    } else {
        ShowWindowAsync(hwnd, SW_SHOW);
    }

    SetWindowPos(
        hwnd,
        HWND_TOP,
        0,
        0,
        0,
        0,
        SWP_NOMOVE |
        SWP_NOSIZE |
        SWP_SHOWWINDOW |
        SWP_ASYNCWINDOWPOS
    );

    BringWindowToTop(hwnd);
    SetForegroundWindow(hwnd);

    return is_foreground(hwnd);
}

Using ffplay as a Lightweight Presentation Preview Window

vast cow — Mon, 15 Jun 2026 07:33:58 +0000

When giving a presentation over Zoom or a similar video conferencing tool, it is often useful to show the presentation screen inside a normal window. This makes it easier to check what is being presented, arrange windows on the desktop, and share only the intended content.

I used to do this with a virtual display driver and raiton. The virtual display acted as an additional monitor, and raiton displayed that monitor in a window. This setup worked, but after looking into it again, I found that ffplay, which is included with FFmpeg, can be a better alternative.

Purpose

The goal is simple: display a secondary or virtual monitor inside a window.

This is useful when presenting slides in PowerPoint or similar software. For example, PowerPoint can show the presentation on a second display, while ffplay shows that display as a window on the main desktop. That window can then be monitored or shared in Zoom.

Using ffplay Instead of raiton

The following command displays the second screen in a window:

ffplay -fflags nobuffer -flags low_delay -framedrop -f lavfi "ddagrab=output_idx=1:framerate=60,hwdownload,format=bgra,format=bgr0"

This uses FFmpeg’s ddagrab source to capture a desktop output and display it with ffplay.

In this command, output_idx=1 means that the second display is captured. If the target display is different, this number may need to be changed.

The options nobuffer, low_delay, and framedrop help reduce preview latency, which is important when using the window during a live presentation.

Performance

In my environment, raiton could display the virtual display at around 15 fps. With ffplay, I was able to display it at 60 fps.

This makes ffplay feel much smoother, especially when slides include animations, cursor movement, or other visual changes.

Virtual Display Driver

For the virtual display, I use ge9/IddSampleDriver.

This provides an additional display output on Windows. Presentation software can use that virtual display as if it were a real monitor, and ffplay can then show that display in a normal window.

Summary

ffplay can be used as a practical replacement for raiton when you want to preview or share a presentation display inside a window.

It requires only FFmpeg, works well with a virtual display driver, and can provide smoother output than raiton in this use case. For presentations over Zoom or similar tools, this is a simple and effective way to handle a separate presentation screen.

Tool for Monitoring Cumulative CPU, Memory, and I/O Usage by Process

vast cow — Mon, 15 Jun 2026 03:43:51 +0000

Overview

This is a simple process monitoring tool that runs on Linux.

While the standard top command shows instantaneous values, this tool can display cumulative CPU time and read/write statistics.

Purpose

The main purpose of this tool is to make it easy to inspect system load on a per-process basis.

In addition to CPU and memory usage, it can also display read volume, write volume, and the number of read/write operations, making it useful for investigating the causes of system load.

For example, it can be used in situations such as:

When you want to identify processes consuming a large amount of CPU.
When you want to identify processes consuming a large amount of memory.
When you want to find processes performing heavy disk or file I/O.
When you want to check how much I/O has increased over a certain period of time.

Basic Usage

Run the tool as a Python script.

python3 mytop.py

When executed, a list of processes is displayed in the terminal. The display is updated at regular intervals, allowing you to monitor CPU, memory, and I/O information for each process.

Main Displayed Fields

Information for each process is shown in columns.

The main fields are as follows.

PID

The process ID.

USER

The user running the process.

%CPU

The percentage of CPU used by the process during the most recent update interval.

%MEM

The percentage of total system memory used by the process.

DTIME+

The CPU time consumed by the process since this tool was started.

TIME+

The total CPU time consumed by the process since it started.

RBYTES / WBYTES / RWBYTES

The number of bytes read, bytes written, and the total number of bytes read and written by the process.

RCHAR / WCHAR / RWCHAR

Counters representing the number of characters read and written by the process.

SYSCR / SYSCW / SYSCRW

The number of read and write system calls, and their total.

COMMAND

The command name or command line being executed.

Sorting Operations

You can change the sort order by pressing keys while the tool is running.

The main keys are as follows.

Key	Sort Order
`d`	CPU time used since the tool started
`p`	CPU usage percentage
`m`	Memory usage percentage
`t`	Total CPU time
`n`	PID
`R`	Total read/write bytes
`B`	Read bytes
`W`	Written bytes
`C`	Cancelled write bytes
`S`	Total read/write system calls
`H`	Total read/write character count
`q`	Quit

Changing the Refresh Interval

By default, the screen is refreshed every second.

To change the refresh interval, use --delay or -d.

python3 mytop.py --delay 2

In this example, the display is updated every 2 seconds.

Manual Refresh Mode

If you want updates only when a key is pressed rather than automatic updates, use --manual-refresh.

python3 mytop.py --manual-refresh

In this mode, the display is refreshed when you press the Space key or Enter key. This is useful when you want to reduce unnecessary updates.

Selecting Displayed Columns

You can specify which columns to display using --columns.

python3 mytop.py --columns pid,user,pcpu,pmem,read_bytes,write_bytes,command

In this example, only PID, user, CPU usage, memory usage, read volume, write volume, and command are displayed.

To see the available columns, run:

python3 mytop.py --list-columns

Summary

This is a lightweight monitoring tool for viewing Linux process status in a terminal.

In addition to CPU and memory usage, it can display per-process I/O information, making it useful for investigating the causes of system load. It supports changing the refresh interval, manual refresh mode, column selection, and key-based sorting, allowing you to focus on the information you need.

#!/usr/bin/env python3
import argparse
import curses
import os
import pwd
import signal
import time
from dataclasses import dataclass
from typing import Callable, Dict, List, Optional, Tuple


CLK_TCK = os.sysconf(os.sysconf_names["SC_CLK_TCK"])
PAGE_SIZE = os.sysconf("SC_PAGE_SIZE")

IO_FIELDS = (
    "rchar",
    "wchar",
    "syscr",
    "syscw",
    "read_bytes",
    "write_bytes",
    "cancelled_write_bytes",
)

IO_SORT_KEYS = (
    "rchar",
    "wchar",
    "rwchar",
    "syscr",
    "syscw",
    "syscrw",
    "read_bytes",
    "write_bytes",
    "rw_bytes",
    "cancelled_write_bytes",
)


@dataclass
class ProcSample:
    pid: int
    user: str
    uid: int
    cpu_ticks: int          # utime + stime
    starttime: int          # /proc/<pid>/stat field 22
    rss_bytes: int
    command: str
    io: Dict[str, int]


@dataclass
class ProcRow:
    pid: int
    user: str
    pcpu: float
    pmem: float
    dtime_ticks: int
    total_ticks: int
    command: str

    # Deltas since this tool started.
    rchar: int
    wchar: int
    syscr: int
    syscw: int
    read_bytes: int
    write_bytes: int
    cancelled_write_bytes: int

    # Derived deltas since this tool started.
    syscrw: int
    rwchar: int
    rw_bytes: int


@dataclass(frozen=True)
class ColumnSpec:
    key: str
    title: str
    width: Optional[int]   # None means variable width
    align: str             # "left" or "right"
    formatter: Callable[[ProcRow], str]


def read_mem_total_bytes() -> int:
    with open("/proc/meminfo", "r", encoding="utf-8") as f:
        for line in f:
            if line.startswith("MemTotal:"):
                # kB
                return int(line.split()[1]) * 1024
    return 1


def uid_to_user(uid: int, cache: Dict[int, str]) -> str:
    if uid in cache:
        return cache[uid]
    try:
        name = pwd.getpwuid(uid).pw_name
    except KeyError:
        name = str(uid)
    cache[uid] = name
    return name


def read_uid(pid: int) -> Optional[int]:
    try:
        with open(f"/proc/{pid}/status", "r", encoding="utf-8", errors="replace") as f:
            for line in f:
                if line.startswith("Uid:"):
                    return int(line.split()[1])  # effective uid
    except OSError:
        return None
    return None


def read_command(pid: int, fallback_comm: str) -> str:
    try:
        with open(f"/proc/{pid}/cmdline", "rb") as f:
            data = f.read()
        if data:
            parts = data.rstrip(b"\0").split(b"\0")
            cmd = " ".join(p.decode("utf-8", errors="replace") for p in parts)
            return cmd
    except OSError:
        pass
    return fallback_comm


def read_proc_io(pid: int) -> Dict[str, int]:
    """
    Read /proc/<pid>/io.

    Missing or unreadable fields are treated as 0.
    /proc/<pid>/io can be unreadable depending on permissions,
    hidepid mount options, or process lifetime races.
    """
    values = {k: 0 for k in IO_FIELDS}

    try:
        with open(f"/proc/{pid}/io", "r", encoding="utf-8", errors="replace") as f:
            for line in f:
                if ":" not in line:
                    continue

                key, value = line.split(":", 1)
                key = key.strip()

                if key not in values:
                    continue

                try:
                    values[key] = int(value.strip())
                except ValueError:
                    values[key] = 0
    except OSError:
        pass

    return values


def parse_proc_stat(pid: int) -> Optional[Tuple[str, int, int, int]]:
    """
    Return:
      comm, cpu_ticks, starttime, rss_pages

    /proc/<pid>/stat fields:
      14 utime
      15 stime
      22 starttime
      24 rss
    """
    try:
        with open(f"/proc/{pid}/stat", "r", encoding="utf-8", errors="replace") as f:
            s = f.read()
    except OSError:
        return None

    # comm is inside parentheses and may contain spaces or ')'.
    rparen = s.rfind(")")
    if rparen == -1:
        return None

    lparen = s.find("(")
    if lparen == -1:
        return None

    comm = s[lparen + 1:rparen]
    rest = s[rparen + 2:].split()

    # rest[0] is field 3: state
    try:
        utime = int(rest[11])       # field 14
        stime = int(rest[12])       # field 15
        starttime = int(rest[19])   # field 22
        rss_pages = int(rest[21])   # field 24
    except (IndexError, ValueError):
        return None

    return comm, utime + stime, starttime, rss_pages


def list_pids() -> List[int]:
    pids = []
    for name in os.listdir("/proc"):
        if name.isdigit():
            pids.append(int(name))
    return pids


def read_processes(user_cache: Dict[int, str]) -> Dict[int, ProcSample]:
    samples: Dict[int, ProcSample] = {}

    for pid in list_pids():
        parsed = parse_proc_stat(pid)
        if parsed is None:
            continue

        comm, cpu_ticks, starttime, rss_pages = parsed

        uid = read_uid(pid)
        if uid is None:
            continue

        user = uid_to_user(uid, user_cache)
        command = read_command(pid, comm)
        io = read_proc_io(pid)

        samples[pid] = ProcSample(
            pid=pid,
            user=user,
            uid=uid,
            cpu_ticks=cpu_ticks,
            starttime=starttime,
            rss_bytes=max(0, rss_pages) * PAGE_SIZE,
            command=command,
            io=io,
        )

    return samples


def format_time_plus(ticks: int) -> str:
    """
    Similar to top's TIME+.
    CPU ticks -> H:MM:SS.hh
    """
    total_seconds = ticks / CLK_TCK
    hours = int(total_seconds // 3600)
    minutes = int((total_seconds % 3600) // 60)
    seconds = int(total_seconds % 60)
    hundredths = int((total_seconds - int(total_seconds)) * 100)

    if hours > 0:
        return f"{hours}:{minutes:02d}:{seconds:02d}.{hundredths:02d}"
    return f"{minutes:02d}.{seconds:02d}"


def format_count(n: int) -> str:
    """
    Compact integer formatter for counters.
    """
    sign = "-" if n < 0 else ""
    n = abs(n)

    units = (
        (1_000_000_000_000, "T"),
        (1_000_000_000, "G"),
        (1_000_000, "M"),
        (1_000, "K"),
    )

    for base, suffix in units:
        if n >= base:
            value = n / base
            if value >= 100:
                return f"{sign}{value:.0f}{suffix}"
            if value >= 10:
                return f"{sign}{value:.1f}{suffix}"
            return f"{sign}{value:.2f}{suffix}"

    return f"{sign}{n}"


def truncate(s: str, width: int) -> str:
    if width <= 0:
        return ""
    if len(s) <= width:
        return s
    if width == 1:
        return "…"
    return s[:width - 1] + "…"


def format_cell(value: str, width: int, align: str) -> str:
    value = truncate(value, width)
    if align == "left":
        return f"{value:<{width}}"
    return f"{value:>{width}}"


def get_column_specs() -> Dict[str, ColumnSpec]:
    return {
        "pid": ColumnSpec("pid", "PID", 7, "right", lambda r: str(r.pid)),
        "user": ColumnSpec("user", "USER", 10, "left", lambda r: r.user),
        "pcpu": ColumnSpec("pcpu", "%CPU", 6, "right", lambda r: f"{r.pcpu:.1f}"),
        "pmem": ColumnSpec("pmem", "%MEM", 6, "right", lambda r: f"{r.pmem:.1f}"),
        "dtime": ColumnSpec("dtime", "DTIME+", 12, "right", lambda r: format_time_plus(r.dtime_ticks)),
        "time": ColumnSpec("time", "TIME+", 12, "right", lambda r: format_time_plus(r.total_ticks)),

        "rchar": ColumnSpec("rchar", "RCHAR", 9, "right", lambda r: format_count(r.rchar)),
        "wchar": ColumnSpec("wchar", "WCHAR", 9, "right", lambda r: format_count(r.wchar)),
        "rwchar": ColumnSpec("rwchar", "RWCHAR", 9, "right", lambda r: format_count(r.rwchar)),

        "syscr": ColumnSpec("syscr", "SYSCR", 9, "right", lambda r: format_count(r.syscr)),
        "syscw": ColumnSpec("syscw", "SYSCW", 9, "right", lambda r: format_count(r.syscw)),
        "syscrw": ColumnSpec("syscrw", "SYSCRW", 9, "right", lambda r: format_count(r.syscrw)),

        "read_bytes": ColumnSpec("read_bytes", "RBYTES", 9, "right", lambda r: format_count(r.read_bytes)),
        "write_bytes": ColumnSpec("write_bytes", "WBYTES", 9, "right", lambda r: format_count(r.write_bytes)),
        "rw_bytes": ColumnSpec("rw_bytes", "RWBYTES", 9, "right", lambda r: format_count(r.rw_bytes)),
        "cancelled_write_bytes": ColumnSpec(
            "cancelled_write_bytes",
            "CANCEL",
            9,
            "right",
            lambda r: format_count(r.cancelled_write_bytes),
        ),

        "command": ColumnSpec("command", "COMMAND", None, "left", lambda r: r.command),
    }


DEFAULT_COLUMNS = [
    "pid",
    "user",
    "pcpu",
    "pmem",
    "dtime",
    "time",
    "read_bytes",
    "write_bytes",
    "rw_bytes",
    "rchar",
    "wchar",
    "rwchar",
    "syscr",
    "syscw",
    "syscrw",
    "cancelled_write_bytes",
    "command",
]


def parse_columns(value: str, specs: Dict[str, ColumnSpec]) -> List[str]:
    columns = [c.strip() for c in value.split(",") if c.strip()]

    if not columns:
        raise argparse.ArgumentTypeError("--columns must contain at least one column")

    unknown = [c for c in columns if c not in specs]
    if unknown:
        valid = ", ".join(specs.keys())
        raise argparse.ArgumentTypeError(
            f"unknown column(s): {', '.join(unknown)}. Valid columns: {valid}"
        )

    return columns


def delta_io(cur: ProcSample, base_io: Dict[str, int]) -> Dict[str, int]:
    """
    Calculate /proc/<pid>/io deltas since this tool started.
    Counter regressions are clamped to 0.
    """
    d = {}
    for key in IO_FIELDS:
        d[key] = max(0, cur.io.get(key, 0) - base_io.get(key, 0))
    return d


def build_rows(
    current: Dict[int, ProcSample],
    previous: Dict[int, ProcSample],
    baselines: Dict[int, Tuple[int, int, Dict[str, int]]],
    interval: float,
    mem_total: int,
) -> List[ProcRow]:
    rows: List[ProcRow] = []

    for pid, cur in current.items():
        # PID reuse detection.
        # If the same PID has a different start time, it's a different process,
        # so reset the baseline.
        baseline = baselines.get(pid)
        if baseline is None or baseline[0] != cur.starttime:
            baselines[pid] = (cur.starttime, cur.cpu_ticks, dict(cur.io))
            base_ticks = cur.cpu_ticks
            base_io = cur.io
        else:
            base_ticks = baseline[1]
            base_io = baseline[2]

        dtime_ticks = max(0, cur.cpu_ticks - base_ticks)
        dio = delta_io(cur, base_io)

        prev = previous.get(pid)
        if prev is not None and prev.starttime == cur.starttime and interval > 0:
            delta_ticks = max(0, cur.cpu_ticks - prev.cpu_ticks)
            pcpu = (delta_ticks / CLK_TCK) / interval * 100.0
        else:
            pcpu = 0.0

        pmem = cur.rss_bytes / mem_total * 100.0 if mem_total > 0 else 0.0

        syscrw = dio["syscr"] + dio["syscw"]
        rwchar = dio["rchar"] + dio["wchar"]
        rw_bytes = dio["read_bytes"] + dio["write_bytes"]

        rows.append(
            ProcRow(
                pid=pid,
                user=cur.user,
                pcpu=pcpu,
                pmem=pmem,
                dtime_ticks=dtime_ticks,
                total_ticks=cur.cpu_ticks,
                command=cur.command,

                rchar=dio["rchar"],
                wchar=dio["wchar"],
                syscr=dio["syscr"],
                syscw=dio["syscw"],
                read_bytes=dio["read_bytes"],
                write_bytes=dio["write_bytes"],
                cancelled_write_bytes=dio["cancelled_write_bytes"],

                syscrw=syscrw,
                rwchar=rwchar,
                rw_bytes=rw_bytes,
            )
        )

    # Clean up baselines for terminated PIDs.
    live_pids = set(current.keys())
    for pid in list(baselines.keys()):
        if pid not in live_pids:
            baselines.pop(pid, None)

    return rows


def draw(
    stdscr,
    rows: List[ProcRow],
    refresh_sec: float,
    sort_key: str,
    start_mono: float,
    columns: List[str],
    manual_refresh: bool,
):
    stdscr.erase()
    height, width = stdscr.getmaxyx()

    specs = get_column_specs()
    selected_specs = [specs[c] for c in columns]

    elapsed = time.monotonic() - start_mono
    now = time.strftime("%H:%M:%S")

    if manual_refresh:
        refresh_label = "refresh: manual"
        refresh_hint = "Space/Enter:update "
    else:
        refresh_label = f"delay: {refresh_sec:g}s"
        refresh_hint = ""

    title = (
        f"dtop - {now}  "
        f"uptime-of-tool: {elapsed:,.1f}s  "
        f"{refresh_label}  "
        f"sort: {sort_key}  "
        f"q:quit {refresh_hint}"
        f"d:DTIME p:CPU m:MEM t:TIME n:PID "
        f"R:RWBYTES B:RBYTES W:WBYTES C:CANCEL S:SYSCRW H:RWCHAR"
    )
    stdscr.addnstr(0, 0, title, width - 1, curses.A_BOLD)

    screen_width = max(0, width - 1)
    separators_width = max(0, len(selected_specs) - 1)

    fixed_width = 0
    variable_count = 0
    for spec in selected_specs:
        if spec.width is None:
            variable_count += 1
        else:
            fixed_width += spec.width

    remaining_width = max(0, screen_width - fixed_width - separators_width)
    variable_width = remaining_width // variable_count if variable_count > 0 else 0

    def effective_width(spec: ColumnSpec) -> int:
        if spec.width is not None:
            return spec.width
        return variable_width

    header_parts = []
    for spec in selected_specs:
        w = effective_width(spec)
        header_parts.append(format_cell(spec.title, w, spec.align))

    header = " ".join(header_parts)
    stdscr.addnstr(1, 0, truncate(header, screen_width), screen_width, curses.A_REVERSE)

    max_rows = max(0, height - 2)

    for i, r in enumerate(rows[:max_rows], start=2):
        parts = []
        for spec in selected_specs:
            w = effective_width(spec)
            value = spec.formatter(r)
            parts.append(format_cell(value, w, spec.align))

        line = " ".join(parts)

        try:
            stdscr.addnstr(i, 0, truncate(line, screen_width), screen_width)
        except curses.error:
            pass

    stdscr.refresh()


def sort_rows(rows: List[ProcRow], key: str) -> List[ProcRow]:
    if key == "pid":
        return sorted(rows, key=lambda r: r.pid)
    if key == "cpu":
        return sorted(rows, key=lambda r: r.pcpu, reverse=True)
    if key == "mem":
        return sorted(rows, key=lambda r: r.pmem, reverse=True)
    if key == "time":
        return sorted(rows, key=lambda r: r.total_ticks, reverse=True)

    if key in IO_SORT_KEYS:
        return sorted(rows, key=lambda r: getattr(r, key), reverse=True)

    # Default: cumulative CPU time since this tool started.
    return sorted(rows, key=lambda r: r.dtime_ticks, reverse=True)


def key_to_sort_key(ch: int, current_sort_key: str) -> Tuple[str, bool]:
    """
    Return:
      new_sort_key, should_quit
    """
    if ch in (ord("q"), ord("Q")):
        return current_sort_key, True
    if ch in (ord("d"), ord("D")):
        return "dtime", False
    if ch in (ord("p"), ord("P")):
        return "cpu", False
    if ch in (ord("m"), ord("M")):
        return "mem", False
    if ch in (ord("t"), ord("T")):
        return "time", False
    if ch in (ord("n"), ord("N")):
        return "pid", False

    # I/O sort shortcuts.
    if ch == ord("R"):
        return "rw_bytes", False
    if ch == ord("B"):
        return "read_bytes", False
    if ch == ord("W"):
        return "write_bytes", False
    if ch == ord("C"):
        return "cancelled_write_bytes", False
    if ch == ord("S"):
        return "syscrw", False
    if ch == ord("H"):
        return "rwchar", False

    # Lowercase variants for raw counters.
    if ch == ord("r"):
        return "rchar", False
    if ch == ord("w"):
        return "wchar", False
    if ch == ord("s"):
        return "syscr", False
    if ch == ord("c"):
        return "syscw", False

    # In manual-refresh mode, Space/Enter or any other non-quit key can be used
    # as an explicit refresh trigger without changing the current sort key.
    return current_sort_key, False


def redraw_existing_rows(
    stdscr,
    args,
    rows: List[ProcRow],
    sort_key: str,
    start_mono: float,
) -> None:
    """
    Redraw already sampled rows without reading /proc again.

    This is used for curses-side repaint requests such as KEY_RESIZE.
    It must not update CPU or I/O statistics.
    """
    draw(
        stdscr=stdscr,
        rows=rows,
        refresh_sec=args.delay,
        sort_key=sort_key,
        start_mono=start_mono,
        columns=args.columns,
        manual_refresh=args.manual_refresh,
    )


def sample_and_draw(
    stdscr,
    args,
    user_cache: Dict[int, str],
    mem_total: int,
    baselines: Dict[int, Tuple[int, int, Dict[str, int]]],
    previous: Dict[int, ProcSample],
    sort_key: str,
    start_mono: float,
    last_sample_mono: float,
) -> Tuple[Dict[int, ProcSample], float, List[ProcRow]]:
    now = time.monotonic()
    current = read_processes(user_cache)
    interval = max(0.001, now - last_sample_mono)

    rows = build_rows(
        current=current,
        previous=previous,
        baselines=baselines,
        interval=interval,
        mem_total=mem_total,
    )
    rows = sort_rows(rows, sort_key)

    draw(
        stdscr=stdscr,
        rows=rows,
        refresh_sec=args.delay,
        sort_key=sort_key,
        start_mono=start_mono,
        columns=args.columns,
        manual_refresh=args.manual_refresh,
    )

    return current, now, rows


def curses_main(stdscr, args):
    curses.curs_set(0)

    if args.manual_refresh:
        stdscr.nodelay(False)
        stdscr.timeout(-1)
    else:
        stdscr.nodelay(True)
        stdscr.timeout(100)

    user_cache: Dict[int, str] = {}
    mem_total = read_mem_total_bytes()

    baselines: Dict[int, Tuple[int, int, Dict[str, int]]] = {}
    previous: Dict[int, ProcSample] = {}

    sort_key = args.sort
    start_mono = time.monotonic()
    last_sample_mono = start_mono

    # Initial sample: use this as the DTIME+ and /proc/<pid>/io baseline.
    current = read_processes(user_cache)
    for pid, p in current.items():
        baselines[pid] = (p.starttime, p.cpu_ticks, dict(p.io))
    previous = current

    # Initial draw. Deltas are zero at startup.
    rows = build_rows(
        current=current,
        previous=previous,
        baselines=baselines,
        interval=0.001,
        mem_total=mem_total,
    )
    rows = sort_rows(rows, sort_key)
    draw(
        stdscr=stdscr,
        rows=rows,
        refresh_sec=args.delay,
        sort_key=sort_key,
        start_mono=start_mono,
        columns=args.columns,
        manual_refresh=args.manual_refresh,
    )

    next_sample = time.monotonic() + args.delay

    while True:
        if args.manual_refresh:
            ch = stdscr.getch()

            if ch == curses.KEY_RESIZE:
                # Terminal resize / curses repaint request:
                # redraw using already sampled rows only.
                redraw_existing_rows(
                    stdscr=stdscr,
                    args=args,
                    rows=rows,
                    sort_key=sort_key,
                    start_mono=start_mono,
                )
                continue

            sort_key, should_quit = key_to_sort_key(ch, sort_key)
            if should_quit:
                break

            previous, last_sample_mono, rows = sample_and_draw(
                stdscr=stdscr,
                args=args,
                user_cache=user_cache,
                mem_total=mem_total,
                baselines=baselines,
                previous=previous,
                sort_key=sort_key,
                start_mono=start_mono,
                last_sample_mono=last_sample_mono,
            )
            continue

        ch = stdscr.getch()
        if ch != -1:
            if ch == curses.KEY_RESIZE:
                # Terminal resize / curses repaint request:
                # redraw using already sampled rows only. Do not reset
                # next_sample and do not read /proc here.
                redraw_existing_rows(
                    stdscr=stdscr,
                    args=args,
                    rows=rows,
                    sort_key=sort_key,
                    start_mono=start_mono,
                )
                continue

            sort_key, should_quit = key_to_sort_key(ch, sort_key)
            if should_quit:
                break

            # In auto mode, a normal key-driven sort/refresh should redraw
            # immediately using a fresh sample instead of waiting for the next
            # interval. curses.KEY_RESIZE is handled above and is redraw-only.
            previous, last_sample_mono, rows = sample_and_draw(
                stdscr=stdscr,
                args=args,
                user_cache=user_cache,
                mem_total=mem_total,
                baselines=baselines,
                previous=previous,
                sort_key=sort_key,
                start_mono=start_mono,
                last_sample_mono=last_sample_mono,
            )
            next_sample = last_sample_mono + args.delay

        now = time.monotonic()
        if now >= next_sample:
            previous, last_sample_mono, rows = sample_and_draw(
                stdscr=stdscr,
                args=args,
                user_cache=user_cache,
                mem_total=mem_total,
                baselines=baselines,
                previous=previous,
                sort_key=sort_key,
                start_mono=start_mono,
                last_sample_mono=last_sample_mono,
            )
            next_sample = last_sample_mono + args.delay

        time.sleep(0.02)


def parse_args():
    specs = get_column_specs()

    sort_choices = [
        "dtime",
        "cpu",
        "mem",
        "time",
        "pid",
        *IO_SORT_KEYS,
    ]

    parser = argparse.ArgumentParser(
        description=(
            "top-like process viewer with DTIME+ and /proc/<pid>/io deltas "
            "since this tool started; optional manual refresh mode"
        )
    )
    parser.add_argument(
        "-d",
        "--delay",
        type=float,
        default=1.0,
        help="refresh interval seconds, default: 1.0",
    )
    parser.add_argument(
        "--manual-refresh",
        action="store_true",
        help=(
            "disable periodic updates and refresh only when a key is pressed. "
            "Space/Enter refresh without changing the current sort key."
        ),
    )
    parser.add_argument(
        "-s",
        "--sort",
        choices=sort_choices,
        default="dtime",
        help="initial sort key, default: dtime",
    )
    parser.add_argument(
        "--columns",
        type=lambda s: parse_columns(s, specs),
        default=DEFAULT_COLUMNS,
        help=(
            "comma-separated columns to display. "
            "Use --list-columns to show available columns."
        ),
    )
    parser.add_argument(
        "--list-columns",
        action="store_true",
        help="list available column keys and exit",
    )

    args = parser.parse_args()

    if args.list_columns:
        for key, spec in specs.items():
            print(f"{key:<24} {spec.title}")
        raise SystemExit(0)

    return args


def main():
    signal.signal(signal.SIGINT, signal.SIG_DFL)
    args = parse_args()
    curses.wrapper(curses_main, args)


if __name__ == "__main__":
    main()

OCG Configuration Tool for Aligning PDF Display and Print Output

vast cow — Fri, 12 Jun 2026 08:28:40 +0000

The content of a PDF can sometimes differ between what you see on screen and what gets printed.

One common cause is OCG (Optional Content Groups), which are layer definitions in a PDF that can be shown or hidden depending on viewing, printing, or export settings.

This tool is designed to help when OCG settings cause discrepancies between on-screen display and printed output by making the print state closely match the current display state.

Purpose

The purpose of this tool is to synchronize the display, print, and export states of each layer in a PDF.

In many PDFs, layers that are visible on screen may be hidden during printing. As a result, content that appears correctly while viewing the PDF may disappear when printed, or content hidden on screen may unexpectedly appear in the printed output.

Using this tool, the current layer state of the PDF is used to configure the print and export settings so that they follow the same visibility state as the display configuration.

Typical Use Cases

This tool is useful in situations such as:

Printing a PDF exactly as it appears on screen
When certain text, diagrams, annotations, or backgrounds disappear during printing
When layer settings cause display and print results to differ
Preparing OCG-enabled PDFs for more predictable output behavior

Usage

The basic usage is to specify an input PDF and an output PDF.

python sync_pdf_ocg_print_state.py input.pdf output.pdf

This command examines the OCG configuration of input.pdf, applies the current display state to the print/view/export settings, and saves the result as output.pdf.

Listing Layers

To see which OCG layers are present in a PDF, run:

python sync_pdf_ocg_print_state.py input.pdf --list-layers

This displays the layer numbers and layer names contained in the PDF.

It is useful when you want to verify whether a PDF contains OCG layers before processing it.

Preview Changes Without Modifying the PDF

To see what changes would be applied without creating a new PDF, use --dry-run:

python sync_pdf_ocg_print_state.py input.pdf --dry-run

In this mode, no output PDF is generated.

The tool simply reports which layers would be treated as ON or OFF.

Display Detailed Information

To view information about the original layer configuration during processing, add -v or --verbose:

python sync_pdf_ocg_print_state.py input.pdf output.pdf --verbose

This displays a summary of the effective BaseState and related layer settings used by the PDF.

Notes

This tool is only effective for PDFs that contain OCG layers.

If a PDF does not contain any OCGs, there are no layers to process and the tool cannot make any changes.

It is also recommended to save the processed file as a new PDF rather than overwriting the original file. Keeping the original PDF allows you to revert if necessary.

Summary

This is a simple utility for situations where OCG settings cause differences between PDF display and printed output.

By propagating the current display state to the print and export settings, it helps ensure that printed output more closely matches what you see on screen.

#!/usr/bin/env python3
"""sync_pdf_ocg_print_state.py

Set OCG Usage state metadata for all Optional Content Groups.

Default behavior:
- Resolve each OCG effective BaseState from OCProperties / get_layer(-1).
- Write that state to:
    /Usage << /Print << /PrintState ... >> /View << /ViewState ... >> /Export << /ExportState ... >> >>
- Applies to all OCGs by default; layer selection options are intentionally removed.
"""

from __future__ import annotations

import argparse
from pathlib import Path
from typing import Optional

try:
    import pymupdf as fitz  # type: ignore
except ImportError:  # pragma: no cover
    import fitz  # type: ignore


def _normalize_state(raw: str | None) -> str:
    if not raw:
        return "/ON"
    s = raw.strip()
    return s if s.startswith("/") else f"/{s}"


def _parse_refs(raw: str | None) -> set[int]:
    if not raw:
        return set()
    out: set[int] = set()
    parts = raw.replace("\r", " ").replace("\n", " ").split()
    for i in range(0, len(parts) - 2, 3):
        if parts[i + 1] == "0" and parts[i + 2] == "R" and parts[i].isdigit():
            out.add(int(parts[i]))
    return out


def _get_catalog_xref(doc: fitz.Document) -> Optional[int]:
    try:
        c = doc.pdf_catalog()
        return int(c) if c else None
    except Exception:
        for xref in range(1, doc.xref_length()):
            try:
                obj = doc.xref_object(xref, compressed=False)
            except Exception:
                continue
            if "/Type /Catalog" in obj or "/Type/Catalog" in obj:
                return xref
    return None


def _get_basestate_map(
    doc: fitz.Document,
    ocg_xrefs: set[int],
) -> tuple[str, set[int], set[int]]:
    """Return (base_state, on_set, off_set) for effective BaseState evaluation."""
    base_state = "/ON"
    on_set: set[int] = set()
    off_set: set[int] = set()

    if hasattr(doc, "get_layer"):
        try:
            cfg = doc.get_layer(-1)
            if isinstance(cfg, dict):
                raw_base = cfg.get("basestate") or cfg.get("base_state") or cfg.get("BaseState")
                if raw_base:
                    base_state = _normalize_state(str(raw_base).strip("'\""))
                on_raw = cfg.get("on") or cfg.get("ON") or []
                off_raw = cfg.get("off") or cfg.get("OFF") or []
                on_set = {
                    int(v) for v in on_raw if isinstance(v, int) or (isinstance(v, str) and v.isdigit())
                }
                off_set = {
                    int(v) for v in off_raw if isinstance(v, int) or (isinstance(v, str) and v.isdigit())
                }
                return base_state, on_set & ocg_xrefs, off_set & ocg_xrefs
        except Exception:
            pass

    catalog = _get_catalog_xref(doc)
    if not catalog:
        return base_state, on_set, off_set

    try:
        t_base, v_base = doc.xref_get_key(catalog, "OCProperties/D/BaseState")
        if t_base != "null" and v_base:
            base_state = _normalize_state(v_base)
        t_on, v_on = doc.xref_get_key(catalog, "OCProperties/D/ON")
        t_off, v_off = doc.xref_get_key(catalog, "OCProperties/D/OFF")
        if t_on != "null":
            on_set = _parse_refs(v_on)
        if t_off != "null":
            off_set = _parse_refs(v_off)
    except Exception:
        pass

    return base_state, (on_set & ocg_xrefs), (off_set & ocg_xrefs)


def _is_ocg_on_by_basestate(
    xref: int,
    base_state: str,
    on_set: set[int],
    off_set: set[int],
) -> bool:
    if base_state == "/OFF":
        return xref in on_set
    return xref not in off_set


def _collect_ocgs(doc: fitz.Document) -> dict[int, str]:
    ocgs: dict[int, str] = {}

    if hasattr(doc, "get_ocgs"):
        try:
            raw = doc.get_ocgs()
            for xref, info in raw.items():
                if isinstance(info, dict):
                    ocgs[int(xref)] = str(info.get("name", f"(xref {int(xref)})"))
        except Exception:
            ocgs = {}

    if ocgs:
        return ocgs

    for xref in range(1, doc.xref_length()):
        try:
            obj = doc.xref_object(xref, compressed=False)
        except Exception:
            continue
        if "/Type /OCG" not in obj and "/Type/OCG" not in obj:
            continue
        marker = "/Name"
        pos = obj.find(marker)
        if pos < 0:
            ocgs[xref] = f"(xref {xref})"
            continue
        start = obj.find("(", pos)
        if start < 0:
            ocgs[xref] = f"(xref {xref})"
            continue
        end = obj.find(")", start + 1)
        ocgs[xref] = obj[start + 1 : end] if end > start else f"(xref {xref})"
    return ocgs


def _ensure_usage_dict(doc: fitz.Document, ocg_xref: int) -> int:
    usage_typ, usage_val = doc.xref_get_key(ocg_xref, "Usage")

    if usage_typ == "xref" and usage_val:
        return int(usage_val.split(" ")[0])

    if usage_typ in {"null", "none"}:
        doc.xref_set_key(ocg_xref, "Usage", "<<>>")
        return ocg_xref

    if usage_typ:
        return ocg_xref

    doc.xref_set_key(ocg_xref, "Usage", "<<>>")
    return ocg_xref


def _set_usage_states(
    doc: fitz.Document,
    ocg_xref: int,
    *,
    state: str,
) -> None:
    target = _ensure_usage_dict(doc, ocg_xref)
    doc.xref_set_key(target, "Print/PrintState", state)
    doc.xref_set_key(target, "View/ViewState", state)
    doc.xref_set_key(target, "Export/ExportState", state)


def _list_layers(ocgs: dict[int, str]) -> None:
    if not ocgs:
        print("No OCG layers found.")
        return
    print("xref\tname")
    for xref in sorted(ocgs):
        print(f"{xref}\t{ocgs[xref]}")


def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(
        description=(
            "Apply OCG Usage state metadata to all layers. "
            "Default mode derives state from BaseState and applies Print/View/Export states."
        )
    )
    p.add_argument("input", help="Input PDF")
    p.add_argument("output", nargs="?", help="Output PDF")
    p.add_argument(
        "--list-layers",
        action="store_true",
        help="List OCG xref and name, then exit",
    )
    p.add_argument(
        "--dry-run",
        action="store_true",
        help="Do not write output; show planned state for each OCG",
    )
    p.add_argument(
        "-v",
        "--verbose",
        action="store_true",
        help="Show effective BaseState summary",
    )
    return p


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)

    if not args.output and not (args.list_layers or args.dry_run):
        raise SystemExit("output is required unless --list-layers or --dry-run is used")

    doc = fitz.open(Path(args.input))
    try:
        ocgs = _collect_ocgs(doc)

        if args.list_layers:
            _list_layers(ocgs)
            return 0

        if not ocgs:
            raise SystemExit("No OCG layers found.")

        base_state, base_on, base_off = _get_basestate_map(doc, set(ocgs.keys()))
        if args.verbose:
            print(f"BaseState: {base_state}, ON refs: {sorted(base_on)}, OFF refs: {sorted(base_off)}")

        if args.dry_run:
            for xref in sorted(ocgs.keys()):
                state = "/ON" if _is_ocg_on_by_basestate(xref, base_state, base_on, base_off) else "/OFF"
                print(f"Would update xref {xref} ({ocgs[xref]}) state = {state} for Print/View/Export")
            return 0

        for xref in sorted(ocgs.keys()):
            state = "/ON" if _is_ocg_on_by_basestate(xref, base_state, base_on, base_off) else "/OFF"
            _set_usage_states(doc, xref, state=state)

        doc.save(Path(args.output), garbage=4, deflate=True, clean=True, incremental=False)
        print(f"Wrote: {Path(args.output)}")
        return 0
    finally:
        doc.close()


if __name__ == "__main__":
    raise SystemExit(main())

PDF Page Combining Tool

vast cow — Fri, 12 Jun 2026 07:15:53 +0000

Overview

This tool combines multiple pages from a PDF into a grid layout on each output page. It is useful when you want to place several PDF pages together, such as arranging four pages into a 2 × 2 layout.

The tool can be used to reduce the number of pages, prepare handouts, create overview sheets, or make compact versions of PDF documents.

Main Purpose

The main purpose of this tool is to create an n-up PDF.

“N-up” means placing multiple pages onto one sheet. For example, a 2 × 2 layout places four original PDF pages on one output page.

The tool can:

Select all pages or only specific pages from the input PDF
Arrange pages in rows and columns
Create larger output pages without shrinking the original pages
Shrink pages to fit into the original page size
Add margins around the output page
Add gaps between the arranged pages

How It Works

The tool reads an input PDF, selects the requested pages, and places them into a grid. Each group of pages becomes one output page.

For example, if the layout is set to 2 columns and 2 rows, the tool places up to four source pages on each output page.

Pages are placed from left to right and from top to bottom.

Selecting Pages

You can choose which pages to include by using a page range.

If no page range is given, the tool uses all pages in the PDF.

Examples:

Page Range	Meaning
`1-4`	Use pages 1 through 4
`1,3,5-8`	Use pages 1, 3, and 5 through 8
`2-`	Use page 2 through the last page
`-10`	Use page 1 through page 10
empty	Use all pages

Page numbers are written in the normal 1-based style, so page 1 means the first page of the PDF.

Basic Usage

Run the tool from the command line by giving an input PDF and an output PDF:

python nup_pdf.py input.pdf output.pdf

By default, this creates a 2 × 2 layout using all pages.

Choosing Rows and Columns

You can change the layout with --cols and --rows.

For example, this creates a 3 × 2 layout:

python nup_pdf.py input.pdf output.pdf --cols 3 --rows 2

This places up to six original pages on each output page.

Using a Page Range

To use only selected pages, add --range.

python nup_pdf.py input.pdf output.pdf --range "1-4"

This uses only pages 1 through 4.

You can also combine separate pages and ranges:

python nup_pdf.py input.pdf output.pdf --range "1,3,5-8"

Keeping the Original Page Size

By default, the tool does not shrink the source pages. Instead, it creates a larger output page large enough to hold the full grid.

For example, if four A4 pages are arranged in a 2 × 2 layout, the output page will be roughly A2-sized.

This is useful when you want to preserve the original page size and avoid reducing text or image quality.

Scaling Pages to Fit

Use --scale-to-fit when you want the output page to stay the same size as the first page of the input PDF.

python nup_pdf.py input.pdf output.pdf --scale-to-fit

In this mode, each source page is reduced so that it fits inside its grid cell.

This is closer to ordinary printer-style n-up output, where multiple pages are placed on one sheet.

Adding Margins and Gaps

You can add an outer margin with --margin and spacing between grid cells with --gap.

python nup_pdf.py input.pdf output.pdf --margin 20 --gap 10

Both values are measured in PDF points.

One point is 1/72 inch.

Margins are added around the outside of the output page. Gaps are added between the placed pages.

Example Commands

Create a 2 × 2 PDF using all pages

python nup_pdf.py input.pdf output.pdf

Create a 2 × 2 PDF using only pages 1 to 8

python nup_pdf.py input.pdf output.pdf --range "1-8"

Create a 3 × 2 layout

python nup_pdf.py input.pdf output.pdf --cols 3 --rows 2

Create a printer-style compact version

python nup_pdf.py input.pdf output.pdf --scale-to-fit

Add margin and spacing

python nup_pdf.py input.pdf output.pdf --margin 24 --gap 12

When to Use This Tool

This tool is helpful when you need to:

Make handouts from slide PDFs
Combine several document pages into one overview page
Reduce the number of printed sheets
Create compact review materials
Arrange selected PDF pages into a clean grid

Summary

This PDF page combining tool creates a new PDF by arranging selected pages from an existing PDF into a grid. It supports flexible page selection, custom rows and columns, optional scaling, margins, and gaps.

Use the default mode when you want to preserve original page sizes. Use --scale-to-fit when you want several pages to fit onto a page of the original size.

from __future__ import annotations

from pathlib import Path

from pypdf import PdfReader, PdfWriter, PageObject, Transformation


def parse_page_range(page_range: str | None, total_pages: int) -> list[int]:
    """
    Convert a 1-based page range string to a list of 0-based page indices.

    If page_range is None or empty, all pages are selected.

    Examples
    --------
    "1-4"     -> [0, 1, 2, 3]
    "1,3,5-7" -> [0, 2, 4, 5, 6]
    "2-"      -> from page 2 to the last page
    "-5"      -> from page 1 to page 5
    "" or None -> all pages
    """
    if page_range is None or not page_range.strip():
        return list(range(total_pages))

    pages: list[int] = []

    for part in page_range.split(","):
        part = part.strip()
        if not part:
            continue

        if "-" in part:
            start_s, end_s = part.split("-", 1)

            start = int(start_s) if start_s else 1
            end = int(end_s) if end_s else total_pages

            if start < 1 or end > total_pages or start > end:
                raise ValueError(f"Invalid page range: {part}")

            pages.extend(range(start - 1, end))
        else:
            p = int(part)
            if p < 1 or p > total_pages:
                raise ValueError(f"Page does not exist: {p}")
            pages.append(p - 1)

    if not pages:
        return list(range(total_pages))

    return pages


def nup_pdf(
    input_pdf: str | Path,
    output_pdf: str | Path,
    page_range: str | None = None,
    cols: int = 2,
    rows: int = 2,
    *,
    scale_to_fit: bool = False,
    margin: float = 0,
    gap: float = 0,
) -> None:
    """
    Combine selected PDF pages into a cols x rows grid on each output page.

    Parameters
    ----------
    input_pdf:
        Input PDF file.
    output_pdf:
        Output PDF file.
    page_range:
        Page range to use, expressed with 1-based page numbers.
        If omitted, all pages are used.
        Examples: "1-4", "1,3,5-8", "2-", "-10"
    cols:
        Number of pages in the horizontal direction. Use 2 for a 2x2 layout.
    rows:
        Number of pages in the vertical direction. Use 2 for a 2x2 layout.
    scale_to_fit:
        False:
            Preserve the original page size and create a larger output page.
            For example, a 2x2 layout of A4 pages is roughly A2-sized.
        True:
            Keep the output page size equal to the first input page and shrink
            each source page into its grid cell. This is closer to ordinary
            n-up printing behavior.
    margin:
        Outer margin of the output page, in PDF points.
        1 point = 1/72 inch.
    gap:
        Gap between cells, in PDF points.
    """
    if cols <= 0 or rows <= 0:
        raise ValueError("cols and rows must be 1 or greater.")

    reader = PdfReader(str(input_pdf))
    writer = PdfWriter()

    selected_indices = parse_page_range(page_range, len(reader.pages))
    if not selected_indices:
        raise ValueError("No pages were selected.")

    pages_per_sheet = cols * rows

    # Reference page size.
    first_page = reader.pages[selected_indices[0]]
    base_width = float(first_page.mediabox.width)
    base_height = float(first_page.mediabox.height)

    if scale_to_fit:
        # Keep the output size equal to one original PDF page.
        out_width = base_width
        out_height = base_height

        cell_width = (out_width - 2 * margin - gap * (cols - 1)) / cols
        cell_height = (out_height - 2 * margin - gap * (rows - 1)) / rows
    else:
        # Do not shrink pages; create a larger page for the full grid.
        cell_width = base_width
        cell_height = base_height

        out_width = cols * cell_width + gap * (cols - 1) + 2 * margin
        out_height = rows * cell_height + gap * (rows - 1) + 2 * margin

    for group_start in range(0, len(selected_indices), pages_per_sheet):
        group = selected_indices[group_start : group_start + pages_per_sheet]

        output_page = PageObject.create_blank_page(
            width=out_width,
            height=out_height,
        )

        for slot, page_index in enumerate(group):
            src_page = reader.pages[page_index]

            src_width = float(src_page.mediabox.width)
            src_height = float(src_page.mediabox.height)

            col = slot % cols
            row = slot // cols

            # The PDF coordinate system starts at the lower-left corner.
            # Flip the y-position calculation so row=0 is placed on the top row.
            x0 = margin + col * (cell_width + gap)
            y0 = margin + (rows - 1 - row) * (cell_height + gap)

            if scale_to_fit:
                scale = min(cell_width / src_width, cell_height / src_height)
            else:
                scale = 1.0

            placed_width = src_width * scale
            placed_height = src_height * scale

            # Center the source page within the cell.
            tx = x0 + (cell_width - placed_width) / 2
            ty = y0 + (cell_height - placed_height) / 2

            # Also support PDFs whose mediabox lower-left corner is not (0, 0).
            src_left = float(src_page.mediabox.left)
            src_bottom = float(src_page.mediabox.bottom)

            transform = (
                Transformation()
                .translate(tx=-src_left, ty=-src_bottom)
                .scale(scale, scale)
                .translate(tx=tx, ty=ty)
            )

            output_page.merge_transformed_page(
                src_page,
                transform,
                over=True,
            )

        writer.add_page(output_page)

    with open(output_pdf, "wb") as f:
        writer.write(f)


import argparse


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Combine PDF pages into an n-up grid layout."
    )
    parser.add_argument("input_pdf")
    parser.add_argument("output_pdf")
    parser.add_argument(
        "--range",
        dest="page_range",
        default=None,
        help="1-based page range to use, such as '1-4', '1,3,5-8', '2-', or '-10'. If omitted, all pages are used.",
    )
    parser.add_argument("--cols", type=int, default=2)
    parser.add_argument("--rows", type=int, default=2)
    parser.add_argument("--scale-to-fit", action="store_true")
    parser.add_argument("--margin", type=float, default=0)
    parser.add_argument("--gap", type=float, default=0)

    args = parser.parse_args()

    nup_pdf(
        input_pdf=args.input_pdf,
        output_pdf=args.output_pdf,
        page_range=args.page_range,
        cols=args.cols,
        rows=args.rows,
        scale_to_fit=args.scale_to_fit,
        margin=args.margin,
        gap=args.gap,
    )


if __name__ == "__main__":
    main()

DEV Community: vast cow

How to Disable Suspend on Lid Close in Rocky Linux

Recommended Configuration

Verification

If Editing /etc/systemd/logind.conf Directly

If the System Still Suspends

Use Unix Domain Sockets on Windows Python: Building an AF_UNIX Compatibility API

Goal

What Was Built

1. Asyncio Version

2. Synchronous Socket Version

Using the Asyncio Version

Using the Synchronous Socket Version

The Role of install()

Implementation Strategy

On Unix-Like Systems

On Windows

API Compatibility

Limitations

SSL Is Not Supported

Abstract Namespace Sockets Are Not Supported

Windows Path Length Restrictions

Use install() Carefully

Choosing Between the Asyncio and Synchronous Versions

Conclusion

How to Force bcache to Recognize a Backing Device Without a Cache Device

Prerequisites

Procedure

1. Load the bcache kernel module

2. Register the backing device with bcache

3. Force startup without a cache device

Which Device to Mount

About the sysfs Path

Summary

Important Notes

Settings to Save a Longer bash History, Reduce Duplicates, and Make It Easier to Use Across Multiple Terminals

Settings

Purpose

Meaning of Each Setting

HISTSIZE

HISTFILESIZE

histappend

PROMPT_COMMAND

history -a

history -c

history -r

Usage

Notes

Summary

When did WSL's --import add support for xz and zstd?

.tar.xz Was Working at Least in WSL 1.1.3.0

.tar.zst Was Working at Least in WSL 2.1.5

As of 2020, Support Was Not Yet an Explicitly Supported Feature

Internally, Import Does Not Appear to Hard-Code Compression Formats

Summary

Reverse Proxying Local Applications with User-Mode nginx and AF_UNIX Sockets

Approach

Directory Layout

Starting nginx

Configuration File

Important Considerations for Running nginx as an Unprivileged User

Reverse Proxying to AF_UNIX Sockets

WebSocket Support

Server-Sent Events (SSE) / HTTP Event Streams

Verification

Summary

A Simple Tool for Converting WebArchive and MHTML Files

Purpose

Main Benefits

Requirements

Basic Usage

Choosing an Output File Name

Choosing the Output Format Manually

Optional Settings

Keeping CID CSS Links

Choosing the WebArchive Plist Format

Excluding Subframes

Error Handling

Summary

Activate the Window for the Specified HWND

If Editing `/etc/systemd/logind.conf` Directly

The Role of `install()`

Use `install()` Carefully

`HISTSIZE`

`HISTFILESIZE`

`histappend`

`PROMPT_COMMAND`

`history -a`

`history -c`

`history -r`

`.tar.xz` Was Working at Least in WSL 1.1.3.0

`.tar.zst` Was Working at Least in WSL 2.1.5