将物理 USB 插槽与逻辑磁盘设备关联

Question

根据@EduardoBissi 的提示，我找到了一种方法来做到这一点。

我连接到netlink套接字只是为了udev获得我需要的信息（比我在中找到的更好klog）。

我写了两个小classes：

import os
import socket

NETLINK_KOBJECT_UEVENT = 15


class KernelEventMonitor(object):

    def __init__(self):
        self.socket = socket.socket(socket.AF_NETLINK, socket.SOCK_DGRAM, NETLINK_KOBJECT_UEVENT)

    def __enter__(self):
        self.socket.bind((os.getpid(), -1))
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.socket.close()

    def __iter__(self):
        while True:
            data = self.socket.recv(16384)
            if data.startswith(b'libudev'):
                event = {}
                for item in data.split(b'\x00'):
                    try:
                        k, v = item.split(b'=', 1)
                        event[k.decode('ascii')] = v.decode('ascii')
                    except ValueError:
                        pass
                yield event


if __name__ == '__main__':
    import re

    r2 = re.compile(r'/devices/[^/]+/[^/]+/[^/]+/(?:[0-9-.]+/)+([0-9-.]+):[0-9.]+/[^/]+/[^/]+/[^/]+/'
                    r'block/(sd[a-z])/sd[a-z]1$')

    with KernelEventMonitor() as monitor:
        for evt in monitor:
            if evt['SUBSYSTEM'] == 'block' and (m := r2.search(evt['DEVPATH'])):
                print(f'Device {m.group(2)} went {evt["ACTION"]} on usb:{m.group(1)}')

注意：出现“相同”事件两次在AF_NETLINK套接字中，一个在处理它之前，一个在udev处理它之后。我选择使用“udev 之后”的那个，这样我就可以确保在生成事件时，任何所需的设备都已经到位。

第二个class只是过滤和重塑事件，并且更具体到我的 USB 配置（我有几个相同的 7 端口 USB 集线器，实际上它们是几个 4 端口集线器，第二个集线器级联在第一个的端口 1 上；这显然非常具体，但我把它留在里面以展示如何处理类似的情况）：

import re

import netlink


class SDMonitor(netlink.KernelEventMonitor):
    r2 = re.compile(r'/devices/[^/]+/[^/]+/[^/]+/(?:[0-9-.]+/)+[0-9]+-(\d+)\.([0-9.]+):'
                    r'[0-9.]+/[^/]+/[^/]+/[^/]+/block/(sd[a-z])/sd[a-z]1$')

    def __iter__(self):
        def _get_index(t):
            return

        for event in super().__iter__():
            if event['SUBSYSTEM'] == 'block':
                m = SDMonitor.r2.search(event['DEVPATH'])
                if m:
                    try:
                        act = {'add': '+', 'remove': '-'}[event["ACTION"]]
                        dev = m.group(3)
                        row = int(m.group(1))
                        col = {'1.1': 1, '1.2': 2, '1.3': 3, '1.4': 4, '2': 5, '3': 6, '4': 7}[m.group(2)]
                        evt = (act, dev, row, col)
                        yield evt
                    except (KeyError, ValueError):
                        print(f'** Bad entry ** {m.group(3)} went {event["ACTION"]} on usb:{m.group(1)}.{m.group(2)}')


if __name__ == '__main__':
    with SDMonitor() as monitor:
        for a, d, r, c in monitor:
            print(f'{a} {d} --> ({r}, {c})')

我对结果非常满意，但几天内我不会接受我自己的答案，看看是否有人能想出更好的解决方案。

Answer 1