get_cpu_info() - Code Metrics - Inspection of "Add a basic example for orchestra runner" - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#4197)

by W

created 2018-06-25 04:33 UTC

get_cpu_info() C

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	9
dl	0
loc	41
rs	6.5626
c	0
b	0
f	0

# Licensed to the StackStorm, Inc ('StackStorm') under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import re

from st2common.util.shell import run_command
from st2common.util.shell import quote_unix

__all__ = [
    'get_cpu_info',
    'get_memory_info',
    'get_package_list',
    'get_deb_package_list',
    'get_rpm_package_list'
]

CPU_INFO_PATH = '/proc/cpuinfo'
MEMORY_INFO_PATH = '/proc/meminfo'


def get_cpu_info():
    """
    Retrieve CPU information.

    :return: List which contain dictionary with information for each core / CPU.
    :rtype: ``list`` of ``dict``
    """
    try:
        with open(CPU_INFO_PATH) as fp:
            content = fp.read()
    except IOError:
        return {}

    lines = content.split('\n')

    result = []
    item = None
    lines_count = len(lines)
    for index, line in enumerate(lines):
        line = line.strip()

        if not line:
            if item and index != lines_count:
                result.append(item)
            continue

        split = line.split(':')

        if len(split) != 2:
            continue

        name = split[0].replace('\t', '').strip().replace(' ', '_')
        value = split[1].replace('\t', '').strip()

        if name == 'processor':
            # Info about new core / CPU
            item = {}

        item[name] = value

    return result


def get_memory_info():
    """
    Retrieve memory information.

    :rtype: ``dict``
    """
    try:
        with open(MEMORY_INFO_PATH) as fp:
            content = fp.read()
    except IOError:
        return {}

    lines = content.split('\n')

    result = {}
    for line in lines:
        line = line.strip()

        if not line:
            continue

        split = line.split(':')
        name = split[0].strip()
        value = split[1].replace('kB', '').strip()

        try:
            value = int(value)
        except Exception:
            continue

        result[name] = value

    return result


def get_package_list(name_startswith):
    """
    Retrieve system packages which name matches the provided startswith filter.

    Note: This function only supports Debian and RedHat based systems.

    :param name_startswith: Package name startswith filter string.
    :type name_startswith: ``str``

    :rtype: ``list`` of ``dict``
    """
    dpkg_exit_code, _, _ = run_command(cmd='dpkg', shell=True)
    rpm_exit_code, _, _ = run_command(cmd='rpm', shell=True)

    if dpkg_exit_code != 127:
        result = get_deb_package_list(name_startswith=name_startswith)
    elif rpm_exit_code != 127:
        result = get_rpm_package_list(name_startswith=name_startswith)
    else:
        raise Exception('Unsupported platform (dpkg or rpm binary not available)')

    return result


def get_deb_package_list(name_startswith):
    cmd = 'dpkg -l | grep %s' % (quote_unix(name_startswith))
    exit_code, stdout, _ = run_command(cmd=cmd, shell=True)

    lines = stdout.split('\n')

    packages = []
    for line in lines:
        line = line.strip()

        if not line:
            continue

        split = re.split('\s+', line)

        name = split[1]
        version = split[2]

        if not name.startswith(name_startswith):
            continue

        item = {
            'name': name,
            'version': version
        }
        packages.append(item)

    return packages


def get_rpm_package_list(name_startswith):
    cmd = 'rpm -qa | grep %s' % (quote_unix(name_startswith))
    exit_code, stdout, _ = run_command(cmd=cmd, shell=True)

    lines = stdout.split('\n')

    packages = []
    for line in lines:
        line = line.strip()

        if not line:
            continue

        split = line.rsplit('.', 1)
        split = split[0].split('-', 1)
        name = split[0]
        version = split[1]

        if not name.startswith(name_startswith):
            continue

        item = {
            'name': name,
            'version': version
        }
        packages.append(item)

    return packages


1			# Licensed to the StackStorm, Inc ('StackStorm') under one or more
2			# contributor license agreements. See the NOTICE file distributed with
3			# this work for additional information regarding copyright ownership.
4			# The ASF licenses this file to You under the Apache License, Version 2.0
5			# (the "License"); you may not use this file except in compliance with
6			# the License. You may obtain a copy of the License at
7			#
8			# http://www.apache.org/licenses/LICENSE-2.0
9			#
10			# Unless required by applicable law or agreed to in writing, software
11			# distributed under the License is distributed on an "AS IS" BASIS,
12			# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13			# See the License for the specific language governing permissions and
14			# limitations under the License.
15
16			import re
17
18			from st2common.util.shell import run_command
19			from st2common.util.shell import quote_unix
20
21			__all__ = [
22			'get_cpu_info',
23			'get_memory_info',
24			'get_package_list',
25			'get_deb_package_list',
26			'get_rpm_package_list'
27			]
28
29			CPU_INFO_PATH = '/proc/cpuinfo'
30			MEMORY_INFO_PATH = '/proc/meminfo'
31
32
33			def get_cpu_info():
34			"""
35			Retrieve CPU information.
36
37			:return: List which contain dictionary with information for each core / CPU.
38			:rtype: ``list`` of ``dict``
39			"""
40			try:
41			with open(CPU_INFO_PATH) as fp:
42			content = fp.read()
43			except IOError:
44			return {}
45
46			lines = content.split('\n')
47
48			result = []
49			item = None
50			lines_count = len(lines)
51			for index, line in enumerate(lines):
52			line = line.strip()
53
54			if not line:
55			if item and index != lines_count:
56			result.append(item)
57			continue
58
59			split = line.split(':')
60
61			if len(split) != 2:
62			continue
63
64			name = split[0].replace('\t', '').strip().replace(' ', '_')
65			value = split[1].replace('\t', '').strip()
66
67			if name == 'processor':
68			# Info about new core / CPU
69			item = {}
70
71			item[name] = value
72
73			return result
74
75
76			def get_memory_info():
77			"""
78			Retrieve memory information.
79
80			:rtype: ``dict``
81			"""
82			try:
83			with open(MEMORY_INFO_PATH) as fp:
84			content = fp.read()
85			except IOError:
86			return {}
87
88			lines = content.split('\n')
89
90			result = {}
91			for line in lines:
92			line = line.strip()
93
94			if not line:
95			continue
96
97			split = line.split(':')
98			name = split[0].strip()
99			value = split[1].replace('kB', '').strip()
100
101			try:
102			value = int(value)
103			except Exception:
104			continue
105
106			result[name] = value
107
108			return result
109
110
111			def get_package_list(name_startswith):
112			"""
113			Retrieve system packages which name matches the provided startswith filter.
114
115			Note: This function only supports Debian and RedHat based systems.
116
117			:param name_startswith: Package name startswith filter string.
118			:type name_startswith: ``str``
119
120			:rtype: ``list`` of ``dict``
121			"""
122			dpkg_exit_code, _, _ = run_command(cmd='dpkg', shell=True)
123			rpm_exit_code, _, _ = run_command(cmd='rpm', shell=True)
124
125			if dpkg_exit_code != 127:
126			result = get_deb_package_list(name_startswith=name_startswith)
127			elif rpm_exit_code != 127:
128			result = get_rpm_package_list(name_startswith=name_startswith)
129			else:
130			raise Exception('Unsupported platform (dpkg or rpm binary not available)')
131
132			return result
133
134
135			def get_deb_package_list(name_startswith):
136			cmd = 'dpkg -l \| grep %s' % (quote_unix(name_startswith))
137			exit_code, stdout, _ = run_command(cmd=cmd, shell=True)
138
139			lines = stdout.split('\n')
140
141			packages = []
142			for line in lines:
143			line = line.strip()
144
145			if not line:
146			continue
147
148			split = re.split('\s+', line)
			0 ignored issues – show Bug introduced 2015-11-20 23:04 UTC by Report Bug Copy Issue Report A suspicious escape sequence `\s` was found. Did you maybe forget to add an `r` prefix? Escape sequences in Python are generally interpreted according to rules similar to standard C. Only if strings are prefixed with `r` or `R` are they interpreted as regular expressions. The escape sequence that was used indicates that you might have intended to write a regular expression. Learn more about the available escape sequences. in the Python documentation. Loading history...
149			name = split[1]
150			version = split[2]
151
152			if not name.startswith(name_startswith):
153			continue
154
155			item = {
156			'name': name,
157			'version': version
158			}
159			packages.append(item)
160
161			return packages
162
163
164			def get_rpm_package_list(name_startswith):
165			cmd = 'rpm -qa \| grep %s' % (quote_unix(name_startswith))
166			exit_code, stdout, _ = run_command(cmd=cmd, shell=True)
167
168			lines = stdout.split('\n')
169
170			packages = []
171			for line in lines:
172			line = line.strip()
173
174			if not line:
175			continue
176
177			split = line.rsplit('.', 1)
178			split = split[0].split('-', 1)
179			name = split[0]
180			version = split[1]
181
182			if not name.startswith(name_startswith):
183			continue
184
185			item = {
186			'name': name,
187			'version': version
188			}
189			packages.append(item)
190
191			return packages
192

StackStorm / st2

Pull Request — master (#4197)

get_cpu_info() C

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like