trustchain-oc1-orchestral
/
devicehub-teal
Archived
10
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

add parsing from lshw and reuse code of worbench computer for parse

This commit is contained in:
Cayo Puigdefabregas 2022-03-28 19:14:19 +02:00
parent 883c01053f
commit 7a09127de9
6 changed files with 204 additions and 135 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

60
ereuse_devicehub/parser/computer.py
View File

@ -90,9 +90,8 @@ class Processor(Component):
self.cores = int(node['configuration']['cores']) self.cores = int(node['configuration']['cores'])
self.threads = int(node['configuration']['threads']) self.threads = int(node['configuration']['threads'])
except KeyError: except KeyError:
self.threads = os.cpu_count() self.threads = 1
if self.threads == 1: self.cores = 1
self.cores = 1 # If there is only one thread there is only one core
self.serial_number = None # Processors don't have valid SN :-( self.serial_number = None # Processors don't have valid SN :-(
self.brand, self.generation = self.processor_brand_generation(self.model) self.brand, self.generation = self.processor_brand_generation(self.model)
@ -282,17 +281,18 @@ class GraphicCard(Component):
def _memory(bus_info): def _memory(bus_info):
"""The size of the memory of the gpu.""" """The size of the memory of the gpu."""
try: try:
lines = cmd.run( # lines = cmd.run(
'lspci', # 'lspci',
'-v -s {bus} | ', # '-v -s {bus} | ',
'grep \'prefetchable\' | ', # 'grep \'prefetchable\' | ',
'grep -v \'non-prefetchable\' | ', # 'grep -v \'non-prefetchable\' | ',
'egrep -o \'[0-9]{{1,3}}[KMGT]+\''.format(bus=bus_info), # 'egrep -o \'[0-9]{{1,3}}[KMGT]+\''.format(bus=bus_info),
shell=True, # shell=True,
).stdout.splitlines() # ).stdout.splitlines()
return max( # return max(
(base2.Quantity(value).to('MiB') for value in lines), default=None # (base2.Quantity(value).to('MiB') for value in lines), default=None
) # )
return None
except subprocess.CalledProcessError: except subprocess.CalledProcessError:
return None return None
@ -307,6 +307,7 @@ class Motherboard(Component):
def new(cls, lshw, hwinfo, **kwargs) -> C: def new(cls, lshw, hwinfo, **kwargs) -> C:
node = next(get_nested_dicts_with_key_value(lshw, 'description', 'Motherboard')) node = next(get_nested_dicts_with_key_value(lshw, 'description', 'Motherboard'))
bios_node = next(get_nested_dicts_with_key_value(lshw, 'id', 'firmware')) bios_node = next(get_nested_dicts_with_key_value(lshw, 'id', 'firmware'))
# bios_node = '1'
memory_array = next( memory_array = next(
g.indents(hwinfo, 'Physical Memory Array', indent=' '), None g.indents(hwinfo, 'Physical Memory Array', indent=' '), None
) )
@ -321,16 +322,16 @@ class Motherboard(Component):
self.firewire = self.num_interfaces(node, 'firewire') self.firewire = self.num_interfaces(node, 'firewire')
self.serial = self.num_interfaces(node, 'serial') self.serial = self.num_interfaces(node, 'serial')
self.pcmcia = self.num_interfaces(node, 'pcmcia') self.pcmcia = self.num_interfaces(node, 'pcmcia')
self.slots = int( self.slots = int(2)
run( # run(
'dmidecode -t 17 | ' 'grep -o BANK | ' 'wc -l', # 'dmidecode -t 17 | ' 'grep -o BANK | ' 'wc -l',
check=True, # check=True,
universal_newlines=True, # universal_newlines=True,
shell=True, # shell=True,
stdout=PIPE, # stdout=PIPE,
).stdout # ).stdout
)
self.bios_date = dateutil.parser.parse(bios_node['date']) self.bios_date = dateutil.parser.parse(bios_node['date']).isoformat()
self.version = bios_node['version'] self.version = bios_node['version']
self.ram_slots = self.ram_max_size = None self.ram_slots = self.ram_max_size = None
if memory_array: if memory_array:
@ -435,7 +436,7 @@ class Display(Component):
g.kv(node, 'Year of Manufacture'), g.kv(node, 'Week of Manufacture') g.kv(node, 'Year of Manufacture'), g.kv(node, 'Week of Manufacture')
) )
# We assume it has been produced the first day of such week # We assume it has been produced the first day of such week
self.production_date = datetime.strptime(d, '%Y %W %w') self.production_date = datetime.strptime(d, '%Y %W %w').isoformat()
self._aspect_ratio = Fraction(self.resolution_width, self.resolution_height) self._aspect_ratio = Fraction(self.resolution_width, self.resolution_height)
def __str__(self) -> str: def __str__(self) -> str:
@ -465,12 +466,13 @@ class Battery(Component):
@classmethod @classmethod
def new(cls, lshw, hwinfo, **kwargs) -> Iterator[C]: def new(cls, lshw, hwinfo, **kwargs) -> Iterator[C]:
try: try:
uevent = cmd.run( # uevent = cmd.run(
'cat', '/sys/class/power_supply/BAT*/uevent', shell=True # 'cat', '/sys/class/power_supply/BAT*/uevent', shell=True
).stdout.splitlines() # ).stdout.splitlines()
return
except CalledProcessError: except CalledProcessError:
return return
yield cls(uevent) # yield cls(uevent)
def __init__(self, node: List[str]) -> None: def __init__(self, node: List[str]) -> None:
super().__init__(node) super().__init__(node)

255
ereuse_devicehub/parser/parser.py
View File

@ -1,13 +1,12 @@
import json import json
import logging
from enum import Enum, unique
from dmidecode import DMIParse from dmidecode import DMIParse
from ereuse_devicehub.parser.computer import ( from ereuse_devicehub.parser.computer import Computer
Display,
GraphicCard, logger = logging.getLogger(__name__)
NetworkAdapter,
SoundCard,
)
class ParseSnapshot: class ParseSnapshot:
@ -57,7 +56,7 @@ class ParseSnapshot:
for cpu in self.dmi.get('Processor'): for cpu in self.dmi.get('Processor'):
self.components.append( self.components.append(
{ {
"actions": [], "actions": set(),
"type": "Processor", "type": "Processor",
"speed": self.get_cpu_speed(cpu), "speed": self.get_cpu_speed(cpu),
"cores": int(cpu.get('Core Count', 1)), "cores": int(cpu.get('Core Count', 1)),
@ -76,17 +75,15 @@ class ParseSnapshot:
for ram in self.dmi.get("Memory Device"): for ram in self.dmi.get("Memory Device"):
self.components.append( self.components.append(
{ {
"actions": [], "actions": set(),
"type": "RamModule", "type": "RamModule",
"size": self.get_ram_size(ram), "size": self.get_ram_size(ram),
"speed": self.get_ram_speed(ram), "speed": self.get_ram_speed(ram),
"manufacturer": ram.get("Manufacturer", self.default), "manufacturer": ram.get("Manufacturer", self.default),
"serialNumber": ram.get("Serial Number", self.default), "serialNumber": ram.get("Serial Number", self.default),
"interface": ram.get("Type", "DDR"), "interface": self.get_ram_type(ram),
"format": ram.get("Format", "DIMM"), # "DIMM", "format": self.get_ram_format(ram),
"model": ram.get( "model": ram.get("Part Number", self.default),
"Model", self.default
), # "48594D503131325336344350362D53362020",
} }
) )
@ -95,7 +92,7 @@ class ParseSnapshot:
for moder_board in self.dmi.get("Baseboard"): for moder_board in self.dmi.get("Baseboard"):
self.components.append( self.components.append(
{ {
"actions": [], "actions": set(),
"type": "Motherboard", "type": "Motherboard",
"version": moder_board.get("Version"), "version": moder_board.get("Version"),
"serialNumber": moder_board.get("Serial Number"), "serialNumber": moder_board.get("Serial Number"),
@ -163,6 +160,16 @@ class ParseSnapshot:
size = ram.get("Speed", "0") size = ram.get("Speed", "0")
return int(size.split(" ")[0]) return int(size.split(" ")[0])
def get_ram_type(self, ram):
TYPES = {'ddr', 'sdram', 'sodimm'}
for t in TYPES:
if t in ram.get("Type", "DDR"):
return t
def get_ram_format(self, ram):
channel = ram.get("Locator", "DIMM")
return 'SODIMM' if 'SODIMM' in channel else 'DIMM'
def get_cpu_speed(self, cpu): def get_cpu_speed(self, cpu):
speed = cpu.get('Max Speed', "0") speed = cpu.get('Max Speed', "0")
return float(speed.split(" ")[0]) / 1024 return float(speed.split(" ")[0]) / 1024
@ -228,7 +235,7 @@ class ParseSnapshot:
self.components.append( self.components.append(
{ {
"actions": [], "actions": set(),
"type": self.get_data_storage_type(sm), "type": self.get_data_storage_type(sm),
"model": model, "model": model,
"manufacturer": manufacturer, "manufacturer": manufacturer,
@ -265,7 +272,7 @@ class ParseSnapshot:
for line in self.hwinfo: for line in self.hwinfo:
iface = { iface = {
"variant": "1", "variant": "1",
"actions": [], "actions": set(),
"speed": 100.0, "speed": 100.0,
"type": "NetworkAdapter", "type": "NetworkAdapter",
"wireless": False, "wireless": False,
@ -295,102 +302,162 @@ class ParseSnapshot:
return x return x
class LsHw: class ParseSnapshotLsHw:
def __init__(self, dmi, jshw, hwinfo, default="n/a"): @unique
class DataStorageInterface(Enum):
ATA = 'ATA'
USB = 'USB'
PCI = 'PCI'
NVME = 'NVME'
def __str__(self):
return self.value
def __init__(self, snapshot, default="n/a"):
self.default = default self.default = default
self.hw = self.loads(jshw) self.dmidecode_raw = snapshot["data"]["dmidecode"]
self.hwinfo = hwinfo.splitlines() self.smart_raw = snapshot["data"]["smart"]
self.childrens = self.hw.get('children', []) self.hwinfo_raw = snapshot["data"]["hwinfo"]
self.dmi = dmi self.lshw_raw = snapshot["data"]["lshw"]
self.components = dmi.components self.device = {"actions": []}
self.device = dmi.device self.components = []
self.add_components() self.components_obj = []
def add_components(self): self.dmi = DMIParse(self.dmidecode_raw)
self.get_cpu_addr() self.smart = self.loads(self.smart_raw)
self.get_networks() self.hwinfo = self.parse_hwinfo()
self.get_display() self.lshw = self.loads(self.lshw_raw)
self.get_sound_card()
self.get_graphic_card()
def get_cpu_addr(self): self.set_basic_datas()
for cpu in self.components: self.set_components()
if not cpu['type'] == "Processor":
continue
cpu["address"] = self.hw.get("width") self.snapshot_json = {
"device": self.device,
"software": "Workbench",
"components": self.components,
"uuid": snapshot['uuid'],
"type": snapshot['type'],
"version": snapshot["version"],
"endTime": snapshot["timestamp"],
"elapsed": 1,
}
# import pdb; pdb.set_trace()
def get_networks(self): def parse_hwinfo(self):
networks = NetworkAdapter.new(self.lshw, self.hwinfo) hw_blocks = self.hwinfo_raw.split("\n\n")
return [x.split("\n") for x in hw_blocks]
for x in networks: def loads(self, x):
if isinstance(x, str):
return json.loads(x)
return x
def set_basic_datas(self):
pc, self.components_obj = Computer.run(self.lshw_raw, self.hwinfo_raw)
# import pdb; pdb.set_trace()
self.device = pc.dump()
def set_components(self):
memory = None
for x in self.components_obj:
if x.type == 'RamModule':
memory = 1
if x.type == 'Motherboard':
x.slots = self.get_ram_slots()
self.components.append(x.dump())
if not memory:
self.get_ram()
self.get_data_storage()
def get_ram(self):
for ram in self.dmi.get("Memory Device"):
self.components.append( self.components.append(
{ {
"actions": [], "actions": set(),
"type": "NetworkAdapter", "type": "RamModule",
"serialNumber": x.serial_number, "size": self.get_ram_size(ram),
"speed": x.speed, "speed": self.get_ram_speed(ram),
"model": x.model, "manufacturer": ram.get("Manufacturer", self.default),
"manufacturer": x.manufacturer, "serialNumber": ram.get("Serial Number", self.default),
"variant": x.variant, "interface": self.get_ram_type(ram),
"wireless": x.wireless, "format": self.get_ram_format(ram),
"model": ram.get("Part Number", self.default),
} }
) )
def get_display(self): def get_ram_size(self, ram):
if not self.device['type'] == 'Laptop': size = ram.get("Size", "0")
return return int(size.split(" ")[0])
displays = Display.new(self.lshw, self.hwinfo) def get_ram_speed(self, ram):
size = ram.get("Speed", "0")
return int(size.split(" ")[0])
def get_ram_slots(self):
slots = 0
for x in self.dmi.get("Physical Memory Array"):
slots += int(x.get("Number Of Devices", 0))
return slots
def get_ram_type(self, ram):
TYPES = {'ddr', 'sdram', 'sodimm'}
for t in TYPES:
if t in ram.get("Type", "DDR"):
return t
def get_ram_format(self, ram):
channel = ram.get("Locator", "DIMM")
return 'SODIMM' if 'SODIMM' in channel else 'DIMM'
def get_data_storage(self):
for sm in self.smart:
# import pdb; pdb.set_trace()
model = sm.get('model_name')
manufacturer = None
if len(model.split(" ")) > 1:
mm = model.split(" ")
model = mm[-1]
manufacturer = " ".join(mm[:-1])
for x in displays:
self.components.append( self.components.append(
{ {
"actions": [], "actions": set(),
"type": "Display", "type": self.get_data_storage_type(sm),
"model": x.model, "model": model,
"manufacturer": x.manufacturer, "manufacturer": manufacturer,
"serialNumber": x.serial_number, "serialNumber": sm.get('serial_number'),
"resolutionWidth": x.resolution_width, "size": self.get_data_storage_size(sm),
"resolutionHeight": x.resolution_height, "variant": sm.get("firmware_version"),
"refreshRate": x.refresh_rate, "interface": self.get_data_storage_interface(sm),
"technology": x.technology,
"productionDate": x.production_date,
"size": x.size,
} }
) )
def get_sound_card(self): def get_data_storage_type(self, x):
soundcards = SoundCard.new(self.lshw, self.hwinfo) # TODO @cayop add more SSDS types
SSDS = ["nvme"]
SSD = 'SolidStateDrive'
HDD = 'HardDrive'
type_dev = x.get('device', {}).get('type')
return SSD if type_dev in SSDS else HDD
for x in soundcards: def get_data_storage_interface(self, x):
self.components.append( interface = x.get('device', {}).get('protocol', 'ATA')
{ try:
"actions": [], self.DataStorageInterface(interface.upper())
"type": "SoundCard", except ValueError as err:
"model": x.model, logger.error(
"manufacturer": x.manufacturer, "interface {} is not in DataStorageInterface Enum".format(interface)
"serialNumber": x.serial_number,
}
) )
raise err
def get_graphic_card(self): def get_data_storage_size(self, x):
# TODO @cayop memory get info from lspci on fly type_dev = x.get('device', {}).get('type')
graphicards = GraphicCard.new(self.lshw, self.hwinfo) total_capacity = "{type}_total_capacity".format(type=type_dev)
# convert bytes to Mb
for x in graphicards: return x.get(total_capacity) / 1024**2
self.components.append(
{
"actions": [],
"type": "GraphicCard",
"model": x.model,
"manufacturer": x.manufacturer,
"serialNumber": x.serial_number,
"memory": x.memory,
}
)
def loads(jshw):
if isinstance(jshw, dict):
return jshw
return json.loads(jshw)

1
ereuse_devicehub/resources/action/schemas.py
View File

@ -420,6 +420,7 @@ class Snapshot_lite_data(MarshmallowSchema):
dmidecode = String(required=False) dmidecode = String(required=False)
hwinfo = String(required=False) hwinfo = String(required=False)
smart = String(required=False) smart = String(required=False)
lshw = String(required=False)
class Snapshot_lite(MarshmallowSchema): class Snapshot_lite(MarshmallowSchema):

14
ereuse_devicehub/resources/action/views/snapshot.py
View File

@ -11,7 +11,7 @@ from flask.json import jsonify
from sqlalchemy.util import OrderedSet from sqlalchemy.util import OrderedSet
from ereuse_devicehub.db import db from ereuse_devicehub.db import db
from ereuse_devicehub.parser.parser import ParseSnapshot from ereuse_devicehub.parser.parser import ParseSnapshotLsHw, ParseSnapshot
from ereuse_devicehub.resources.action.models import RateComputer, Snapshot from ereuse_devicehub.resources.action.models import RateComputer, Snapshot
from ereuse_devicehub.resources.action.rate.v1_0 import CannotRate from ereuse_devicehub.resources.action.rate.v1_0 import CannotRate
from ereuse_devicehub.resources.action.schemas import Snapshot_lite from ereuse_devicehub.resources.action.schemas import Snapshot_lite
@ -79,9 +79,9 @@ class SnapshotView:
self.tmp_snapshots = app.config['TMP_SNAPSHOTS'] self.tmp_snapshots = app.config['TMP_SNAPSHOTS']
self.path_snapshot = save_json(snapshot_json, self.tmp_snapshots, g.user.email) self.path_snapshot = save_json(snapshot_json, self.tmp_snapshots, g.user.email)
snapshot_json.pop('debug', None) snapshot_json.pop('debug', None)
if snapshot_json.get('version') in ["14.0.0"]: if snapshot_json.get('version') in ["2022.03"]:
self.validate_json(snapshot_json) self.validate_json(snapshot_json)
self.response = self.build2() self.response = self.build_lite()
else: else:
self.snapshot_json = resource_def.schema.load(snapshot_json) self.snapshot_json = resource_def.schema.load(snapshot_json)
self.response = self.build() self.response = self.build()
@ -161,8 +161,8 @@ class SnapshotView:
self.schema2 = Snapshot_lite() self.schema2 = Snapshot_lite()
self.snapshot_json = self.schema2.load(snapshot_json) self.snapshot_json = self.schema2.load(snapshot_json)
def build2(self): def build_lite(self):
snap = ParseSnapshot(self.snapshot_json) snap = ParseSnapshotLsHw(self.snapshot_json)
snap_json = snap.snapshot_json # snap = ParseSnapshot(self.snapshot_json)
self.snapshot_json = self.resource_def.schema.load(snap_json) self.snapshot_json = self.resource_def.schema.load(snap.snapshot_json)
return self.build() return self.build()

2
tests/files/example_wb14_x1.json
View File

File diff suppressed because one or more lines are too long

7
tests/test_snapshot.py
View File

@ -1030,13 +1030,12 @@ def test_min_validate_fields(user: UserClient):
@pytest.mark.mvp @pytest.mark.mvp
def test_snapshot_wb_lite(user: UserClient): def test_snapshot_wb_lite(user: UserClient):
"""This test check the minimum validation of json that come from snapshot""" """This test check the minimum validation of json that come from snapshot"""
# import pdb; pdb.set_trace()
snapshot = file_json("example_wb14_x1.json") snapshot = file_json("example_wb14_x1.json")
body, res = user.post(snapshot, res=Snapshot) body, res = user.post(snapshot, res=Snapshot)
a = [x['type'] for x in body['components']] # a = [x['type'] for x in body['components']]
import pdb
pdb.set_trace()
ssd = [x for x in body['components'] if x['type'] == 'SolidStateDrive'][0] ssd = [x for x in body['components'] if x['type'] == 'SolidStateDrive'][0]