USB-PD pretty-printer for data exposed over Linux sysfs

usb-pd.py

#!/usr/bin/env python3
#
# Pretty-Printer for USB-PD data exposed in /sys/class/typec/
#
# Based on https://www.kernel.org/doc/Documentation/ABI/testing/sysfs-class-typec
#
# (C) Georg Lukas <georg@op-co.de>
#

from pathlib import Path
import re

def parse_brackets(value):
    match = re.search('\[(.*)\]', value)
    if match:
        return match.group(1)
    return value

def parse_yesno(value):
    return (value == "yes")

def parse_mA(value):
    return float(value.removesuffix("mA"))/1000

def parse_mV(value):
    return float(value.removesuffix("mV"))/1000

def parse_mW(value):
    return float(value.removesuffix("mW"))/1000

def read_file(fn, parser):
    if fn.exists():
        with fn.open() as f:
            response = f.readline().strip('\n')
            return parser(response)
    pass

def read_sysfs_dir(name, dir_path, FIELDS):
    result = {
            'name': name,
            'path': dir_path,
          }
    for fn, parser in FIELDS.items():
        abs_fn = dir_path / fn
        result[fn] = read_file(abs_fn, parser)
    return result

PARTNER_FIELDS = {
        'accessory_mode': str,
        'supports_usb_power_delivery': parse_yesno,
        'usb_power_delivery_revision': str,
        'type': str,
        }

def read_partner(partner):
    partner = read_sysfs_dir(partner.name, partner, PARTNER_FIELDS)
    #for ...

PORT_FIELDS = {
        'data_role': parse_brackets,
        'power_role': parse_brackets,
        'usb_power_delivery_revision': str,
        'usb_typec_revision': str
        }

def read_sinks_sources(port_path):
    sinks_path = port_path / 'usb_power_delivery' / 'sink-capabilities'
    result = {}
    result['sinks'] = []
    #print(sinks_path)
    for sink in sorted(sinks_path.glob('*:*')):
        result['sinks'].append(read_sink_source(sink))
    sources_path = port_path / 'usb_power_delivery' / 'source-capabilities'
    result['sources'] = []
    for source in sorted(sources_path.glob('*:*')):
        result['sources'].append(read_sink_source(source))
    return result


def read_port(port_path):
    port = read_sysfs_dir(int(port_path.name.removeprefix('port')), port_path, PORT_FIELDS)
    port['location'] = read_sysfs_dir('location', port_path / 'physical_location', LOCATION_FIELDS)
    port.update(read_sinks_sources(port_path))
    partner = port_path / (port_path.name + '-partner')
    if partner.exists():
        port['partner'] = read_sysfs_dir(partner.name, partner, PARTNER_FIELDS)
        port['partner'].update(read_sinks_sources(partner))
    return port

LOCATION_FIELDS = {
        'dock': parse_yesno,
        'horizontal_position': str,
        'lid': parse_yesno,
        'panel': str,
        'vertical_position': str,
        }

def read_location(loc_path):
    return read_sysfs_dir(loc_path.name, loc_path, LOCATION_FIELDS)

def format_location(location):
    if not location or not ('panel' in location) or location['panel'] == 'unknown':
        return "unknown location"
    dock = " on dock" if location['dock'] else ''
    lid = " on lid" if location['lid'] else ''
    return "{panel} panel {horizontal_position} {vertical_position}".format(**location) + dock + lid

SINK_SOURCE_FIELDS = {
        'dual_role_data': bool,
        'dual_role_power': bool,
        'maximum_current': parse_mA,
        'maximum_voltage': parse_mV,
        'minimum_voltage': parse_mV,
        'operational_current': parse_mA,
        'operational_power': parse_mW,
        'usb_communication_capable': bool,
        'voltage': parse_mV,
        }

def read_sink_source(s_path):
    return read_sysfs_dir(s_path.name, s_path, SINK_SOURCE_FIELDS)

def format_partner(partner):
    #accessory_mode = partner['accessory_mode' != 'none'
    if not partner['supports_usb_power_delivery']:
        return "Non-USB-PD partner"
    revision = partner['usb_power_delivery_revision']
    if revision == "0.0":
        revision = "(unknown revision)"
    is_dual_role = len(partner['sources']) > 0 and len(partner['sinks']) > 0 and (partner['sinks'][0]['dual_role_power'] or partner['sources'][0]['dual_role_power'])
    dual_role = " dual-role-power" if is_dual_role else ""
    return f"USB-PD {revision}{dual_role} partner"

def format_pps(v):
    v['voltage'] = v['maximum_voltage']
    if v['maximum_current']:
        v['power'] = v['maximum_voltage']*v['maximum_current']
        return "{minimum_voltage}-{maximum_voltage}V*{maximum_current}A ({power}W, PPS)".format(**v)
    else: # only a sink, no amperage
        return "{minimum_voltage}-{maximum_voltage}V (PPS)".format(**v)

def format_sink(sink):
    #print(sink)
    v = sink
    if v['operational_power']: # battery
        return "{minimum_voltage}-{maximum_voltage}V (battery)".format(**v)
    if v['maximum_voltage']: # programmable power supply
        return format_pps(v)
    v['power'] = v['voltage']*v['operational_current']
    return "{voltage}V*{operational_current}A ({power}W)".format(**v)

def format_source(source):
    #print(source)
    v = source
    if v['maximum_voltage']: # programmable power supply
        return format_pps(v)
    v['power'] = v['voltage']*v['maximum_current']
    return "{voltage}V*{maximum_current}A ({power}W)".format(**v)

def format_list(items, format):
    return ", ".join([format(s) for s in items])

def print_port(p):
    print(f"Port {p['name']}: USB-PD {p['usb_power_delivery_revision']} (USB-C {p['usb_typec_revision']}) {format_location(p['location'])}, data {p['data_role']}, power {p['power_role']}")
    if len(p['sinks']) > 0:
        print("    Requires", format_list(p['sinks'], format_sink))
    if len(p['sources']) > 0:
        print("    Provides", format_list(p['sources'], format_source))
    if 'partner' in p:
        print("   ", format_partner(p['partner']))
        if len(p['partner']['sinks']) > 0:
            print("        Requires", format_list(p['partner']['sinks'], format_sink))
        if len(p['partner']['sources']) > 0:
            print("        Provides", format_list(p['partner']['sources'], format_source))

def list_ports():
    for p in Path('/sys/class/typec').glob('port?'):
        port = read_port(p)
        print_port(port)


list_ports()