<commit_before>import abc
import collections
import dmidecode
import logging
import multiprocessing
import os
import re
import subprocess

from lxml import etree

from . import benchmark
from . import utils


def get_subsection_value(output, section_name, subsection_name):
    """Extract data from tabulated output like lshw and dmidecode."""
    section = output.find(section_name)
    ## end_section = output.find("*-", section)
    ## XXX WIP try to limit context to a section (how to detect end on dmidecode?)
    ## NOTE will be replaced with dmidecode Python2 library
    subsection = output.find(subsection_name, section)#, end_section)
    end = output.find("\n", subsection)
    return output[subsection:end].split(':')[1].strip()

Connector = collections.namedtuple('Connector', ['name', 'count', 'verbose_name'])


def get_xpath_text(node, path, default=None):
    try:
        return node.xpath("{0}/text()".format(path))[0]
    except IndexError:
        return default


class Device(object):
    __metaclass__ = abc.ABCMeta
    
    LSHW_NODE_ID = None
    
    @classmethod
    def retrieve(cls, lshw_xml):
        assert cls.LSHW_NODE_ID is not None, "LSHW_NODE_ID should be defined on the subclass."
        
        objects = []
        for node in lshw_xml.xpath('//node[@id="{0}"]'.format(cls.LSHW_NODE_ID)):
            objects.append(cls(node))
        
        return objects
    

class Motherboard(object):
    CONNECTORS = (
        ("USB", "usb"),
        ("FireWire", "firewire"),
        ("Serial Port", "serial"),
        ("PCMCIA", "pcmcia"),
    )
    
    def __init__(self, lshw_xml, dmi):
        self.serialNumber = get_subsection_value(dmi, "Base Board Information", "Serial Number")
        self.manufacturer =  get_subsection_value(dmi, "Base Board Information", "Manufacturer")
        self.model =  get_subsection_value(dmi, "Base Board Information", "Product Name")
        
        self.connectors = []
        for verbose, value in self.CONNECTORS:
            count = self.number_of_connectors(lshw_xml, value)
            self.connectors.append(
                Connector(name=value, count=count, verbose_name=verbose)
            )
    
    def number_of_connectors(self, root, name):
        for i in range(10):
            if not root.xpath('//node[@id="{0}:{1}"]'.format(name, i)):
                return i


class HardDrive(Device):
    # TODO USB and (S)ATA subclasses
    CAPACITY_UNITS = "MB"
    LSHW_NODE_ID = "disk"
    
    def __init__(self, node):
        self.serialNumber = get_xpath_text(node, 'serial')
        self.manufacturer = get_xpath_text(node, 'vendor')
        
        logical_name = get_xpath_text(node, 'logicalname')
        self.interface = utils.run("udevadm info --query=all --name={0} | grep ID_BUS | cut -c 11-".format(logical_name))
        
        # TODO implement method for USB disk
        if self.interface == "usb":
            self.model = self.serial = self.size = "Unknown"
        
        else:
            # (S)ATA disk
            self.model = utils.run("hdparm -I {0} | grep 'Model\ Number' | cut -c 22-".format(logical_name))
            #self.serial = utils.run("hdparm -I {0} | grep 'Serial\ Number' | cut -c 22-".format(logical_name))
            self.size = utils.run("hdparm -I {0} | grep 'device\ size\ with\ M' | head -n1 | awk '{{print $7}}'".format(logical_name))
        
        # TODO read config to know if we should run SMART
        self.smart = self.run_smart(logical_name)
    
    def run_smart(self, logical_name):  # TODO allow choosing short or extended
        return benchmark.hard_disk_smart(disk=logical_name)


class GraphicCard(Device):
    CAPACITY_UNITS = "MB"
    LSHW_NODE_ID = "display"
     
    def __init__(self, node):
        self.serialNumber = None  # TODO could be retrieved?
        self.manufacturer = get_xpath_text(node, 'vendor')
        self.model = get_xpath_text(node, 'product')
        
        # Find VGA memory
        bus_info = get_xpath_text(node, 'businfo').split("@")[1]
        mem = utils.run("lspci -v -s {bus} | grep 'prefetchable' | grep -v 'non-prefetchable' | egrep -o '[0-9]{{1,3}}[KMGT]+'".format(bus=bus_info)).splitlines()
        
        # Get max memory value
        max_size = 0
        for value in mem:
            unit = re.split('\d+', value)[1]
            size = int(value.rstrip(unit))
            
            # convert all values to KB before compare
            size_kb = utils.convert_base(size, unit, 'K', distance=1024)
            if size_kb > max_size:
                max_size = size_kb

        self.memory = utils.convert_capacity(max_size, 'KB', 'MB')
    
    @property
    def score(self):
        return benchmark.score_vga(self.model)


class NetworkAdapter(Device):
    SPEED_UNIT = "Mbps"
    LSHW_NODE_ID = "network"
    
    def __init__(self, node):
        self.serialNumber = self.get_serial(node)
        self.model = get_xpath_text(node, 'product')
        self.manufacturer = get_xpath_text(node, 'vendor')
        self.speed = get_xpath_text(node, 'capacity')
        
        if self.speed is not None:
            units = "bps"  # net.xpath('capacity/@units')[0]
            self.speed = utils.convert_speed(self.speed, units, self.SPEED_UNIT)
        
    def get_serial(self, node):
        serial = get_xpath_text(node, 'serial')
        
        # TODO get serialNumber of USB NetworkAdapter!!
        # lshw has other id="network:1"...
        if serial is None:
            logical_name = get_xpath_text(node, 'logicalname')
            if logical_name is None:
                logging.error("Error retrieving MAC: '{0}'".format(etree.tostring(node)))
            else:
                serial = utils.get_hw_addr(logical_name)
        
        return serial


class OpticalDrive(Device):
    LSHW_NODE_ID = "cdrom"
    
    def __init__(self, node):
        self.serialNumber = None  # TODO could be retrieved?
        self.model = get_xpath_text(node, 'product')
        self.manufacturer = get_xpath_text(node, 'vendor')
        # TODO normalize values?
        self.description = get_xpath_text(node, 'description')


class Processor(Device):
    CLOCK_UNIT = 'MHz'<commit_msg>Use different verbosity according to logging level.<commit_after>import abc
import collections
import dmidecode
import logging
import multiprocessing
import os
import re
import subprocess

from lxml import etree

from . import benchmark
from . import utils


def get_subsection_value(output, section_name, subsection_name):
    """Extract data from tabulated output like lshw and dmidecode."""
    section = output.find(section_name)
    ## end_section = output.find("*-", section)
    ## XXX WIP try to limit context to a section (how to detect end on dmidecode?)
    ## NOTE will be replaced with dmidecode Python2 library
    subsection = output.find(subsection_name, section)#, end_section)
    end = output.find("\n", subsection)
    return output[subsection:end].split(':')[1].strip()

Connector = collections.namedtuple('Connector', ['name', 'count', 'verbose_name'])


def get_xpath_text(node, path, default=None):
    try:
        return node.xpath("{0}/text()".format(path))[0]
    except IndexError:
        return default


class Device(object):
    __metaclass__ = abc.ABCMeta
    
    LSHW_NODE_ID = None
    
    @classmethod
    def retrieve(cls, lshw_xml):
        assert cls.LSHW_NODE_ID is not None, "LSHW_NODE_ID should be defined on the subclass."
        
        objects = []
        for node in lshw_xml.xpath('//node[<EMAIL>="{0}"]'.format(cls.LSHW_NODE_ID)):
            objects.append(cls(node))
        
        return objects
    

class Motherboard(object):
    CONNECTORS = (
        ("USB", "usb"),
        ("FireWire", "firewire"),
        ("Serial Port", "serial"),
        ("PCMCIA", "pcmcia"),
    )
    
    def __init__(self, lshw_xml, dmi):
        self.serialNumber = get_subsection_value(dmi, "Base Board Information", "Serial Number")
        self.manufacturer =  get_subsection_value(dmi, "Base Board Information", "Manufacturer")
        self.model =  get_subsection_value(dmi, "Base Board Information", "Product Name")
        
        self.connectors = []
        for verbose, value in self.CONNECTORS:
            count = self.number_of_connectors(lshw_xml, value)
            self.connectors.append(
                Connector(name=value, count=count, verbose_name=verbose)
            )
    
    def number_of_connectors(self, root, name):
        for i in range(10):
            if not root.xpath('//node[@id="{0}:{1}"]'.format(name, i)):
                return i


class HardDrive(Device):
    # TODO USB and (S)ATA subclasses
    CAPACITY_UNITS = "MB"
    LSHW_NODE_ID = "disk"
    
    def __init__(self, node):
        self.serialNumber = get_xpath_text(node, 'serial')
        self.manufacturer = get_xpath_text(node, 'vendor')
        
        logical_name = get_xpath_text(node, 'logicalname')
        self.interface = utils.run("udevadm info --query=all --name={0} | grep ID_BUS | cut -c 11-".format(logical_name))
        
        # TODO implement method for USB disk
        if self.interface == "usb":
            self.model = self.serial = self.size = "Unknown"
        
        else:
            # (S)ATA disk
            self.model = utils.run("hdparm -I {0} | grep 'Model\ Number' | cut -c 22-".format(logical_name))
            #self.serial = utils.run("hdparm -I {0} | grep 'Serial\ Number' | cut -c 22-".format(logical_name))
            self.size = utils.run("hdparm -I {0} | grep 'device\ size\ with\ M' | head -n1 | awk '{{print $7}}'".format(logical_name))
        
        # TODO read config to know if we should run SMART
        self.smart = self.run_smart(logical_name)
    
    def run_smart(self, logical_name):  # TODO allow choosing short or extended
        return benchmark.hard_disk_smart(disk=logical_name)


class GraphicCard(Device):
    CAPACITY_UNITS = "MB"
    LSHW_NODE_ID = "display"
     
    def __init__(self, node):
        self.serialNumber = None  # TODO could be retrieved?
        self.manufacturer = get_xpath_text(node, 'vendor')
        self.model = get_xpath_text(node, 'product')
        
        # Find VGA memory
        bus_info = get_xpath_text(node, 'businfo').split("@")[1]
        mem = utils.run("lspci -v -s {bus} | grep 'prefetchable' | grep -v 'non-prefetchable' | egrep -o '[0-9]{{1,3}}[KMGT]+'".format(bus=bus_info)).splitlines()
        
        # Get max memory value
        max_size = 0
        for value in mem:
            unit = re.split('\d+', value)[1]
            size = int(value.rstrip(unit))
            
            # convert all values to KB before compare
            size_kb = utils.convert_base(size, unit, 'K', distance=1024)
            if size_kb > max_size:
                max_size = size_kb

        self.memory = utils.convert_capacity(max_size, 'KB', 'MB')
    
    @property
    def score(self):
        return benchmark.score_vga(self.model)


class NetworkAdapter(Device):
    SPEED_UNIT = "Mbps"
    LSHW_NODE_ID = "network"
    
    def __init__(self, node):
        self.model = get_xpath_text(node, 'product')
        self.manufacturer = get_xpath_text(node, 'vendor')
        self.serialNumber = self.get_serial(node)
        self.speed = get_xpath_text(node, 'capacity')
        
        if self.speed is not None:
            units = "bps"  # net.xpath('capacity/@units')[0]
            self.speed = utils.convert_speed(self.speed, units, self.SPEED_UNIT)
        
    def get_serial(self, node):
        serial = get_xpath_text(node, 'serial')
        
        # TODO get serialNumber of USB NetworkAdapter!!
        # lshw has other id="network:1"...
        if serial is None:
            logical_name = get_xpath_text(node, 'logicalname')
            if logical_name is None:
                error = "Error retrieving MAC: '%s'"
                logging.error(error, self.model)
                logging.debug(error, etree.tostring(node))
            else:
                serial = utils.get_hw_addr(logical_name)
        
        return serial


class OpticalDrive(Device):
    LSHW_NODE_ID = "cdrom"
    
    def __init__(self, node):
        self.serialNumber = None  # TODO could be retrieved?
        self.model = get_xpath_text(node, 'product')
        self.manufacturer = get_xpath_text(node, 'vendor')
        # TODO normalize values?
        self.description = get_xpath_text(node, 'description')


class Processor(Device):
    CLOCK_UNIT = 'MHz'
{'fraction_non_alphanumeric': 0.11827269092363055, 'fraction_numerical': 0.005517792882846861, 'mean_word_length': 2.4825953773322196, 'pattern_counts': {'":': 2, '<': 4, '<?xml version=': 0, '>': 6, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 10}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4013319', 'n_tokens_mistral': 4024, 'n_tokens_neox': 3926, 'n_words': 948}