ArmHelper._fetch() - Code Metrics - fuzeman/Plex-Trakt-Scrobbler - Measure and Improve Code Quality continuously with Scrutinizer

ArmHelper._fetch() B
last analyzed 2018-03-05 01:11 UTC

↳ Parent: ArmHelper

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	10
CRAP Score	6.8395

Importance

Changes

Metric	Value
c	0
b	0
f	0
dl	0
loc	24
ccs	10
cts	14
cp	0.7143
rs	7.6129
cc	6
crap	6.8395

from plugin.core.helpers.variable import try_convert

import logging
import os

log = logging.getLogger(__name__)


class ArmHelper(object):
    _cpuinfo_cache = None
    _cpuinfo_lines = None

    @classmethod
    def attributes(cls, force_refresh=False):
        if not force_refresh and cls._cpuinfo_cache is not None:
            return cls._cpuinfo_cache

        # Retrieve data from "/proc/cpuinfo"
        lines = cls._fetch(force_refresh)

        if lines:
            # Parse lines
            cls._cpuinfo_cache = cls._parse(lines)
        else:
            # Clear cache
            cls._cpuinfo_cache = None

        return cls._cpuinfo_cache or (None, None)

    @classmethod
    def identifier(cls, force_refresh=False):
        # Retrieve CPU information
        processors, extra = cls.attributes(force_refresh)


        # Lookup CPU in table
        return cls.lookup(processors, extra)

    @classmethod
    def lookup(cls, processors, extra):
        # Find processor identifier
        cpu_implementer, cpu_part = cls.cpu_identifier(processors, extra)

        if not cpu_implementer or not cpu_part:
            log.info('Unable to retrieve processor identifier from "/proc/cpuinfo"')
            return None, None, None

        # Try map CPU to a known name
        identifier = CPU_TABLE.get((cpu_implementer, cpu_part))

        if identifier is None:
            log.warn('Unknown CPU - implementer: 0x%X, part: 0x%X, hardware: %r' % (cpu_implementer, cpu_part, extra.get('hardware')))

            return None, None, None

        if len(identifier) < 3:
            # Pad identifier with `None` values
            identifier = tuple(list(identifier) + [None for x in xrange(3 - len(identifier))])


        if len(identifier) > 3:
            log.warn('Invalid identifier format returned: %r', identifier)
            return None, None, None

        return identifier

    @classmethod
    def cpu_identifier(cls, processors, extra = None):

        if not processors:
            return None, None

        for key, processor in processors.items():

            if not processor:
                continue

            # Check `processor` has identifiers
            cpu_implementer = cls._cast_hex(processor.get('cpu_implementer'))
            cpu_part = cls._cast_hex(processor.get('cpu_part'))

            if cpu_implementer and cpu_part:
                # Valid identifiers found
                return cpu_implementer, cpu_part

        if extra:
            # Check `extra` has identifiers
            cpu_implementer = cls._cast_hex(extra.get('cpu_implementer'))
            cpu_part = cls._cast_hex(extra.get('cpu_part'))

            if cpu_implementer and cpu_part:
                # Valid identifiers found
                return cpu_implementer, cpu_part

        # Couldn't find CPU identifiers
        return None, None

    @classmethod
    def _cast_hex(cls, value):
        if not value:
            return None

        try:
            return int(value, 0)
        except Exception:

            log.warn('Unable to cast %r to an integer', value, exc_info=True)
            return None

    @classmethod
    def _fetch(cls, force_refresh=False):
        if not force_refresh and cls._cpuinfo_lines is not None:
            return cls._cpuinfo_lines

        # Ensure cpuinfo is available
        if not os.path.exists('/proc/cpuinfo'):
            log.info('Unable to retrieve information from "/proc/cpuinfo", path doesn\'t exist')
            return None

        # Fetch cpuinfo from procfs
        log.debug('Fetching processor information from "/proc/cpuinfo"...')

        with open('/proc/cpuinfo') as fp:
            data = fp.read()

        # Split `data` into lines
        if data:
            cls._cpuinfo_lines = data.split('\n')
        else:
            cls._cpuinfo_lines = None

        # Return lines
        return cls._cpuinfo_lines

    @classmethod
    def _parse(cls, lines):
        processors = {}
        extra = {}

        # Parse lines into `processors` and `extra`
        section = None
        current = {}

        for line in lines:
            # Handle section break
            if line == '':
                # Store current attributes
                if section == 'processor':
                    num = try_convert(current.pop('processor', None), int)

                    if num is None:
                        num = len(processors)

                    processors[num] = current
                elif section == 'extra':
                    extra.update(current)
                elif current:
                    log.debug('Discarding unknown attributes: %r', current)

                # Reset state
                section = None
                current = {}

                # Continue with next line
                continue

            # Parse attribute from line
            parts = [part.strip() for part in line.split(':', 1)]

            if len(parts) < 2:
                log.debug('Unable to parse attribute from line: %r', line)
                continue

            # Retrieve attribute components
            key, value = parts[0], parts[1]

            if not key:
                log.debug('Invalid key returned for line: %r', line)
                continue

            # Transform `key`
            key = key.lower()
            key = key.replace(' ', '_')

            # Check for section-identifier
            if not section:
                if key == 'processor':
                    section = 'processor'
                else:
                    section = 'extra'

            # Store attribute in current dictionary
            current[key] = value

        # Store any leftover extra attributes
        if section == 'extra' and current:
            extra.update(current)

        # Return result
        return processors, extra


CPU_TABLE = {
    # Format: (<implementer>, <part>): (<vendor>, <name>[, <type>])

    # ARM
    (0x41, 0xB02): ('arm', '11-MPCore'),
    (0x41, 0xB36): ('arm', '1136'),
    (0x41, 0xB56): ('arm', '1156'),
    (0x41, 0xB76): ('arm', '1176'),                                 # (Raspberry Pi 1)
    (0x41, 0xC05): ('arm', 'cortex-a5'),
    (0x41, 0xC08): ('arm', 'cortex-a8'),
    (0x41, 0xC07): ('arm', 'cortex-a7'),                            # (Raspberry Pi 2)
    (0x41, 0xC0F): ('arm', 'cortex-a15'),                           # (DS215+)
    (0x41, 0xD03): ('arm', 'cortex-a53'),                           # (NVIDIA SHIELD, Raspberry Pi 3)
    (0x41, 0xD07): ('arm', 'cortex-a57'),                           # (NVIDIA SHIELD)

    # Marvell
    (0x56, 0x131): ('marvell', 'kirkwood-88f6281'),                 # (DS112j, DS212j, DS211j, DS411j, DS110j, DS210j, DS410j, DS109, DS209, DS409, DS409slim, RS409)

    (0x56, 0x581): ('marvell', 'armada-370/XP', 'marvell-pj4'),     # (DS213j)
    (0x56, 0x584): ('marvell', 'armada-370/XP', 'marvell-pj4'),     # (DS114, DS115j, DS214se, DS216se, DS414slim, RS214)

    (0x41, 0xC09): ('marvell', 'armada-375',    'marvell-pj4')      # (DS215j)

}


1	1	from plugin.core.helpers.variable import try_convert
2
3	1	import logging
4	1	import os
5
6	1	log = logging.getLogger(__name__)
7
8
9	1	class ArmHelper(object):
10	1	_cpuinfo_cache = None
11	1	_cpuinfo_lines = None
12
13	1	@classmethod
14	1	def attributes(cls, force_refresh=False):
15	1	if not force_refresh and cls._cpuinfo_cache is not None:
16		return cls._cpuinfo_cache
17
18		# Retrieve data from "/proc/cpuinfo"
19	1	lines = cls._fetch(force_refresh)
20
21	1	if lines:
22		# Parse lines
23	1	cls._cpuinfo_cache = cls._parse(lines)
24		else:
25		# Clear cache
26		cls._cpuinfo_cache = None
27
28	1	return cls._cpuinfo_cache or (None, None)
29
30	1	@classmethod
31	1	def identifier(cls, force_refresh=False):
32		# Retrieve CPU information
33	1	processors, extra = cls.attributes(force_refresh)
		0 ignored issues – show Bug Best Practice introduced 2016-03-01 12:17 UTC by Report Bug Copy Issue Report It seems like you are trying to unpack a non-sequence, which was defined at line 10. Loading history... Bug Best Practice introduced 2016-03-01 12:17 UTC by Report Bug Copy Issue Report It seems like you are trying to unpack a non-sequence, which was defined at line 26. Loading history...
34
35		# Lookup CPU in table
36	1	return cls.lookup(processors, extra)
37
38	1	@classmethod
39		def lookup(cls, processors, extra):
40		# Find processor identifier
41	1	cpu_implementer, cpu_part = cls.cpu_identifier(processors, extra)
42
43	1	if not cpu_implementer or not cpu_part:
44	1	log.info('Unable to retrieve processor identifier from "/proc/cpuinfo"')
45	1	return None, None, None
46
47		# Try map CPU to a known name
48	1	identifier = CPU_TABLE.get((cpu_implementer, cpu_part))
49
50	1	if identifier is None:
51		log.warn('Unknown CPU - implementer: 0x%X, part: 0x%X, hardware: %r' % (cpu_implementer, cpu_part, extra.get('hardware')))
		0 ignored issues – show Coding Style introduced 2016-03-01 12:17 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (134/120). This check looks for lines that are too long. You can specify the maximum line length. Loading history... Coding Style Best Practice introduced 2016-03-14 20:48 UTC by Report Bug Copy Issue Report Specify string format arguments as logging function parameters Loading history...
52		return None, None, None
53
54	1	if len(identifier) < 3:
55		# Pad identifier with `None` values
56	1	identifier = tuple(list(identifier) + [None for x in xrange(3 - len(identifier))])
		0 ignored issues – show Comprehensibility Best Practice introduced 2016-03-01 12:17 UTC by Report Bug Copy Issue Report Undefined variable 'xrange' Loading history...
57
58	1	if len(identifier) > 3:
59		log.warn('Invalid identifier format returned: %r', identifier)
60		return None, None, None
61
62	1	return identifier
63
64	1	@classmethod
65	1	def cpu_identifier(cls, processors, extra = None):
		0 ignored issues – show Coding Style introduced 2016-04-10 23:04 UTC by Report Bug Copy Issue Report No space allowed around keyword argument assignment def cpu_identifier(cls, processors, extra = None): ^ Loading history...
66	1	if not processors:
67	1	return None, None
68
69	1	for key, processor in processors.items():
		0 ignored issues – show Unused Code introduced 2016-04-10 23:04 UTC by Report Bug Copy Issue Report The variable `key` seems to be unused. Loading history...
70	1	if not processor:
71	1	continue
72
73		# Check `processor` has identifiers
74	1	cpu_implementer = cls._cast_hex(processor.get('cpu_implementer'))
75	1	cpu_part = cls._cast_hex(processor.get('cpu_part'))
76
77	1	if cpu_implementer and cpu_part:
78		# Valid identifiers found
79	1	return cpu_implementer, cpu_part
80
81	1	if extra:
82		# Check `extra` has identifiers
83	1	cpu_implementer = cls._cast_hex(extra.get('cpu_implementer'))
84	1	cpu_part = cls._cast_hex(extra.get('cpu_part'))
85
86	1	if cpu_implementer and cpu_part:
87		# Valid identifiers found
88	1	return cpu_implementer, cpu_part
89
90		# Couldn't find CPU identifiers
91	1	return None, None
92
93	1	@classmethod
94		def _cast_hex(cls, value):
95	1	if not value:
96	1	return None
97
98	1	try:
99	1	return int(value, 0)
100		except Exception:
		0 ignored issues – show Best Practice introduced 2016-03-01 12:17 UTC by Report Bug Copy Issue Report Catching very general exceptions such as `Exception` is usually not recommended. Generally, you would want to handle very specific errors in the exception handler. This ensure that you do not hide other types of errors which should be fixed. So, unless you specifically plan to handle any error, consider adding a more specific exception. Loading history...
101		log.warn('Unable to cast %r to an integer', value, exc_info=True)
102		return None
103
104	1	@classmethod
105	1	def _fetch(cls, force_refresh=False):
106	1	if not force_refresh and cls._cpuinfo_lines is not None:
107		return cls._cpuinfo_lines
108
109		# Ensure cpuinfo is available
110	1	if not os.path.exists('/proc/cpuinfo'):
111		log.info('Unable to retrieve information from "/proc/cpuinfo", path doesn\'t exist')
112		return None
113
114		# Fetch cpuinfo from procfs
115	1	log.debug('Fetching processor information from "/proc/cpuinfo"...')
116
117	1	with open('/proc/cpuinfo') as fp:
118	1	data = fp.read()
119
120		# Split `data` into lines
121	1	if data:
122	1	cls._cpuinfo_lines = data.split('\n')
123		else:
124		cls._cpuinfo_lines = None
125
126		# Return lines
127	1	return cls._cpuinfo_lines
128
129	1	@classmethod
130		def _parse(cls, lines):
131	1	processors = {}
132	1	extra = {}
133
134		# Parse lines into `processors` and `extra`
135	1	section = None
136	1	current = {}
137
138	1	for line in lines:
139		# Handle section break
140	1	if line == '':
141		# Store current attributes
142	1	if section == 'processor':
143	1	num = try_convert(current.pop('processor', None), int)
144
145	1	if num is None:
146		num = len(processors)
147
148	1	processors[num] = current
149	1	elif section == 'extra':
150	1	extra.update(current)
151	1	elif current:
152		log.debug('Discarding unknown attributes: %r', current)
153
154		# Reset state
155	1	section = None
156	1	current = {}
157
158		# Continue with next line
159	1	continue
160
161		# Parse attribute from line
162	1	parts = [part.strip() for part in line.split(':', 1)]
163
164	1	if len(parts) < 2:
165		log.debug('Unable to parse attribute from line: %r', line)
166		continue
167
168		# Retrieve attribute components
169	1	key, value = parts[0], parts[1]
170
171	1	if not key:
172		log.debug('Invalid key returned for line: %r', line)
173		continue
174
175		# Transform `key`
176	1	key = key.lower()
177	1	key = key.replace(' ', '_')
178
179		# Check for section-identifier
180	1	if not section:
181	1	if key == 'processor':
182	1	section = 'processor'
183		else:
184	1	section = 'extra'
185
186		# Store attribute in current dictionary
187	1	current[key] = value
188
189		# Store any leftover extra attributes
190	1	if section == 'extra' and current:
191	1	extra.update(current)
192
193		# Return result
194	1	return processors, extra
195
196
197	1	CPU_TABLE = {
198		# Format: (<implementer>, <part>): (<vendor>, <name>[, <type>])
199
200		# ARM
201		(0x41, 0xB02): ('arm', '11-MPCore'),
202		(0x41, 0xB36): ('arm', '1136'),
203		(0x41, 0xB56): ('arm', '1156'),
204		(0x41, 0xB76): ('arm', '1176'), # (Raspberry Pi 1)
205		(0x41, 0xC05): ('arm', 'cortex-a5'),
206		(0x41, 0xC08): ('arm', 'cortex-a8'),
207		(0x41, 0xC07): ('arm', 'cortex-a7'), # (Raspberry Pi 2)
208		(0x41, 0xC0F): ('arm', 'cortex-a15'), # (DS215+)
209		(0x41, 0xD03): ('arm', 'cortex-a53'), # (NVIDIA SHIELD, Raspberry Pi 3)
210		(0x41, 0xD07): ('arm', 'cortex-a57'), # (NVIDIA SHIELD)
211
212		# Marvell
213		(0x56, 0x131): ('marvell', 'kirkwood-88f6281'), # (DS112j, DS212j, DS211j, DS411j, DS110j, DS210j, DS410j, DS109, DS209, DS409, DS409slim, RS409)
		0 ignored issues – show Coding Style introduced 2016-03-14 20:48 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (165/120). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
214		(0x56, 0x581): ('marvell', 'armada-370/XP', 'marvell-pj4'), # (DS213j)
215		(0x56, 0x584): ('marvell', 'armada-370/XP', 'marvell-pj4'), # (DS114, DS115j, DS214se, DS216se, DS414slim, RS214)
		0 ignored issues – show Coding Style introduced 2016-03-01 12:17 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (121/120). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
216		(0x41, 0xC09): ('marvell', 'armada-375', 'marvell-pj4') # (DS215j)
		0 ignored issues – show Coding Style introduced 2016-03-01 12:17 UTC by Report Bug Copy Issue Report Exactly one space required after comma (0x41, 0xC09): ('marvell', 'armada-375', 'marvell-pj4') # (DS215j) ^ Loading history...
217		}
218

fuzeman / Plex-Trakt-Scrobbler

ArmHelper._fetch() B last analyzed 2018-03-05 01:11 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

ArmHelper._fetch() B
last analyzed 2018-03-05 01:11 UTC