SystemHelper.arm() - Code Metrics - Inspection of "Merge branch 'issue/258' into develop" - fuzeman/Plex-Trakt-Scrobbler - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — develop ( 1e4ff5...af752c )

by Dean

created 2016-01-11 02:02 UTC

SystemHelper.arm() A

↳ Parent: SystemHelper

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	3
dl	0
loc	17
rs	9.4286

from plugin.core.environment import Environment

import os
import platform
import shutil
import sys


# Create dummy `Log` (for tests)
try:
    Log.Debug('Using framework "Log" handler')
except NameError:
    from plex_mock.framework import Logger

    Log = Logger()
    Log.Debug('Using dummy "Log" handler')


class Libraries(object):
    contents_path = os.path.abspath(os.path.join(Environment.path.code, '..'))

    native_directories = [
        'Libraries\\FreeBSD',
        'Libraries\\Linux',
        'Libraries\\MacOSX',
        'Libraries\\Windows'
    ]

    unicode_map = {
        65535:      'ucs2',
        1114111:    'ucs4'
    }

    @classmethod
    def cache(cls):
        """Cache native libraries into the plugin data directory"""

        # Retrieve library platforms
        libraries_path = os.path.join(cls.contents_path, 'Libraries')
        platforms = os.listdir(libraries_path)

        if 'Shared' in platforms:
            platforms.remove('Shared')

        # Create destination directory
        destination = os.path.join(Environment.path.plugin_data, 'Libraries')

        # Ensure destination exists
        StorageHelper.create_directories(destination)

        # Delete existing libraries
        for name in os.listdir(destination):
            path = os.path.join(destination, name)

            StorageHelper.delete_tree(path)

        # Copy libraries to directory
        for name in platforms:
            p_source = os.path.join(libraries_path, name)
            p_destination = os.path.join(destination, name)

            if not StorageHelper.copy_tree(p_source, p_destination):
                return None

        Log.Debug('Cached native libraries to %r', StorageHelper.to_relative_path(destination))
        return destination

    @classmethod
    def get_libraries_path(cls, cache=False):
        """Retrieve the native libraries base directory (and caching the libraries if enabled)"""

        if not cache:
            return Environment.path.libraries

        # Cache native libraries
        libraries_path = cls.cache()

        if libraries_path:
            # Reset native library directories in `sys.path`
            cls.reset()

            return libraries_path

        return Environment.path.libraries

    @classmethod
    def setup(cls, cache=False):
        """Setup native library directories"""

        # Retrieve libraries path
        libraries_path = cls.get_libraries_path(cache)

        Log.Info('Using native libraries at %r', StorageHelper.to_relative_path(libraries_path))

        # Retrieve system details
        system = SystemHelper.name()
        system_architecture = SystemHelper.architecture()

        if not system_architecture:
            return

        Log.Debug('System: %r, Architecture: %r', system, system_architecture)

        architectures = [system_architecture]

        if system_architecture == 'i686':
            # Fallback to i386
            architectures.append('i386')

        for architecture in reversed(architectures + ['universal']):
            # Common
            PathHelper.insert(libraries_path, system, architecture)

            # UCS
            if sys.maxunicode in cls.unicode_map:
                PathHelper.insert(libraries_path, system, architecture, cls.unicode_map[sys.maxunicode])

        # Log library paths
        for path in sys.path:
            path = os.path.abspath(path)

            if not StorageHelper.is_relative_path(path):
                continue

            Log.Info('[PATH] %s', StorageHelper.to_relative_path(path))

    @classmethod
    def reset(cls):
        """Remove all the native library directives from `sys.path`"""

        for path in sys.path:
            path = os.path.abspath(path)

            if not path.lower().startswith(cls.contents_path.lower()):
                continue

            # Convert to relative path
            path_rel = os.path.relpath(path, cls.contents_path)

            # Take the first two fragments
            path_rel = os.path.sep.join(path_rel.split(os.path.sep)[:2])

            # Ignore non-native library directories
            if path_rel not in cls.native_directories:
                continue

            # Remove from `sys.path`
            PathHelper.remove(path)


class PathHelper(object):
    @classmethod
    def insert(cls, base, system, architecture, *args):
        """Insert a new path into `sys.path` if it passes basic validation"""

        path = os.path.join(base, system, architecture, *args)

        if path in sys.path:
            return False

        if not os.path.exists(path):
            return False

        sys.path.insert(0, path)

        Log.Debug('Inserted path: %r', StorageHelper.to_relative_path(path))
        return True

    @classmethod
    def remove(cls, path):
        """Remove path from `sys.path` if it exists"""

        if path not in sys.path:
            return False

        sys.path.remove(path)

        Log.Debug('Removed path: %r', StorageHelper.to_relative_path(path))
        return True


class StorageHelper(object):
    base_names = [
        'plug-ins',
        'plug-in support',
        'trakttv.bundle'
    ]

    @classmethod
    def create_directories(cls, path, *args, **kwargs):
        """Create directory at `path` include any parent directories"""

        try:
            os.makedirs(path, *args, **kwargs)
            return True
        except OSError, ex:
            if ex.errno == 17:
                # Directory already exists
                return True

            Log.Warn('Unable to create directories: %r - (%s) %s', cls.to_relative_path(path), ex.errno, ex)
        except Exception, ex:
            Log.Warn('Unable to create directories: %r - (%s) %s', cls.to_relative_path(path), type(ex), ex)

        return False

    @classmethod
    def copy_tree(cls, source, destination):
        """Copy the directory at `source` to `destination`"""

        try:
            shutil.copytree(source, destination)

            Log.Debug('Copied %r to %r', cls.to_relative_path(source), cls.to_relative_path(destination))
            return True
        except Exception, ex:
            Log.Warn('Unable to copy %r to %r - %s', cls.to_relative_path(source), cls.to_relative_path(destination), ex)

        return False

    @classmethod
    def delete_tree(cls, path):
        """Delete the directory (at `path`"""

        try:
            shutil.rmtree(path)

            Log.Debug('Deleted %r', cls.to_relative_path(path))
            return True
        except Exception, ex:
            Log.Warn('Unable to delete directory: %r - %s', cls.to_relative_path(path), ex)

        return False

    @classmethod
    def to_relative_path(cls, path):
        """Convert `path` to be relative to `StorageHelper.base_names`"""

        path_lower = path.lower()

        # Find base path
        base_path = None

        for base in cls.base_names:
            if base not in path_lower:
                continue

            base_path = path[:path_lower.find(base)]
            break

        # Check if `base_path` was found
        if not base_path:
            Log.Warn('Unable to find base path in %r', path)
            return path

        # Return relative path
        return os.path.relpath(path, base_path)

    @classmethod
    def is_relative_path(cls, path):
        """Check if `path` is relative to `StorageHelper.base_names`"""

        path_lower = path.lower()

        # Ignore framework paths
        if 'framework.bundle' in path_lower:
            return False

        # Find base path
        for base in cls.base_names:
            if base not in path_lower:
                continue

            return True

        return False


class SystemHelper(object):
    bits_map = {
        '32bit': 'i386',
        '64bit': 'x86_64'
    }

    machine_map = {
        ('32bit', 'i686'): 'i686'
    }

    name_map = {
        'Darwin': 'MacOSX'
    }

    @classmethod
    def architecture(cls):
        """Retrieve system architecture (i386, i686, x86_64)"""

        bits, _ = platform.architecture()
        machine = platform.machine()

        # Check for ARM machine
        if bits == '32bit' and machine.startswith('armv'):
            return cls.arm(machine)

        # Check (bits, machine) map
        machine_key = (bits, machine)

        if machine_key in cls.machine_map:
            return cls.machine_map[machine_key]

        # Check (bits) map
        if bits in cls.bits_map:
            return cls.bits_map[bits]

        Log.Info('Unable to determine system architecture - bits: %r, machine: %r', bits, machine)
        return None

    @classmethod
    def name(cls):
        """Retrieve system name (Windows, Linux, FreeBSD, MacOSX)"""

        system = platform.system()

        # Apply system map
        if system in cls.name_map:
            system = cls.name_map[system]

        return system

    @classmethod
    def arm(cls, machine):
        # Determine floating-point type
        float_type = cls.arm_float_type()

        if float_type is None:
            Log.Warn('Unable to use ARM libraries, unsupported floating-point type?')
            return None

        # Determine ARM version
        version = cls.arm_version()

        if version is None:
            Log.Warn('Unable to use ARM libraries, unsupported ARM version (%r)?' % machine)
            return None

        return '%s_%s' % (version, float_type)

    @classmethod
    def arm_version(cls, machine=None):
        # Read `machine` name if not provided
        if machine is None:
            machine = platform.machine()

        # Ensure `machine` is valid
        if not machine:
            return None

        # ARMv6
        if machine.startswith('armv6'):
            return 'armv6'

        # ARMv7
        if machine.startswith('armv7'):
            return 'armv7'

        return None

    @classmethod
    def arm_float_type(cls):
        if os.path.exists('/lib/arm-linux-gnueabihf'):
            return 'hf'

        if os.path.exists('/lib/arm-linux-gnueabi'):
            return 'sf'

        return None


1			from plugin.core.environment import Environment
2
3			import os
4			import platform
5			import shutil
6			import sys
7
8
9			# Create dummy `Log` (for tests)
10			try:
11			Log.Debug('Using framework "Log" handler')
12			except NameError:
13			from plex_mock.framework import Logger
14
15			Log = Logger()
16			Log.Debug('Using dummy "Log" handler')
17
18
19			class Libraries(object):
20			contents_path = os.path.abspath(os.path.join(Environment.path.code, '..'))
21
22			native_directories = [
23			'Libraries\\FreeBSD',
24			'Libraries\\Linux',
25			'Libraries\\MacOSX',
26			'Libraries\\Windows'
27			]
28
29			unicode_map = {
30			65535: 'ucs2',
31			1114111: 'ucs4'
32			}
33
34			@classmethod
35			def cache(cls):
36			"""Cache native libraries into the plugin data directory"""
37
38			# Retrieve library platforms
39			libraries_path = os.path.join(cls.contents_path, 'Libraries')
40			platforms = os.listdir(libraries_path)
41
42			if 'Shared' in platforms:
43			platforms.remove('Shared')
44
45			# Create destination directory
46			destination = os.path.join(Environment.path.plugin_data, 'Libraries')
47
48			# Ensure destination exists
49			StorageHelper.create_directories(destination)
50
51			# Delete existing libraries
52			for name in os.listdir(destination):
53			path = os.path.join(destination, name)
54
55			StorageHelper.delete_tree(path)
56
57			# Copy libraries to directory
58			for name in platforms:
59			p_source = os.path.join(libraries_path, name)
60			p_destination = os.path.join(destination, name)
61
62			if not StorageHelper.copy_tree(p_source, p_destination):
63			return None
64
65			Log.Debug('Cached native libraries to %r', StorageHelper.to_relative_path(destination))
66			return destination
67
68			@classmethod
69			def get_libraries_path(cls, cache=False):
70			"""Retrieve the native libraries base directory (and caching the libraries if enabled)"""
71
72			if not cache:
73			return Environment.path.libraries
74
75			# Cache native libraries
76			libraries_path = cls.cache()
77
78			if libraries_path:
79			# Reset native library directories in `sys.path`
80			cls.reset()
81
82			return libraries_path
83
84			return Environment.path.libraries
85
86			@classmethod
87			def setup(cls, cache=False):
88			"""Setup native library directories"""
89
90			# Retrieve libraries path
91			libraries_path = cls.get_libraries_path(cache)
92
93			Log.Info('Using native libraries at %r', StorageHelper.to_relative_path(libraries_path))
94
95			# Retrieve system details
96			system = SystemHelper.name()
97			system_architecture = SystemHelper.architecture()
98
99			if not system_architecture:
100			return
101
102			Log.Debug('System: %r, Architecture: %r', system, system_architecture)
103
104			architectures = [system_architecture]
105
106			if system_architecture == 'i686':
107			# Fallback to i386
108			architectures.append('i386')
109
110			for architecture in reversed(architectures + ['universal']):
111			# Common
112			PathHelper.insert(libraries_path, system, architecture)
113
114			# UCS
115			if sys.maxunicode in cls.unicode_map:
116			PathHelper.insert(libraries_path, system, architecture, cls.unicode_map[sys.maxunicode])
117
118			# Log library paths
119			for path in sys.path:
120			path = os.path.abspath(path)
121
122			if not StorageHelper.is_relative_path(path):
123			continue
124
125			Log.Info('[PATH] %s', StorageHelper.to_relative_path(path))
126
127			@classmethod
128			def reset(cls):
129			"""Remove all the native library directives from `sys.path`"""
130
131			for path in sys.path:
132			path = os.path.abspath(path)
133
134			if not path.lower().startswith(cls.contents_path.lower()):
135			continue
136
137			# Convert to relative path
138			path_rel = os.path.relpath(path, cls.contents_path)
139
140			# Take the first two fragments
141			path_rel = os.path.sep.join(path_rel.split(os.path.sep)[:2])
142
143			# Ignore non-native library directories
144			if path_rel not in cls.native_directories:
145			continue
146
147			# Remove from `sys.path`
148			PathHelper.remove(path)
149
150
151			class PathHelper(object):
152			@classmethod
153			def insert(cls, base, system, architecture, *args):
154			"""Insert a new path into `sys.path` if it passes basic validation"""
155
156			path = os.path.join(base, system, architecture, *args)
157
158			if path in sys.path:
159			return False
160
161			if not os.path.exists(path):
162			return False
163
164			sys.path.insert(0, path)
165
166			Log.Debug('Inserted path: %r', StorageHelper.to_relative_path(path))
167			return True
168
169			@classmethod
170			def remove(cls, path):
171			"""Remove path from `sys.path` if it exists"""
172
173			if path not in sys.path:
174			return False
175
176			sys.path.remove(path)
177
178			Log.Debug('Removed path: %r', StorageHelper.to_relative_path(path))
179			return True
180
181
182			class StorageHelper(object):
183			base_names = [
184			'plug-ins',
185			'plug-in support',
186			'trakttv.bundle'
187			]
188
189			@classmethod
190			def create_directories(cls, path, args, *kwargs):
191			"""Create directory at `path` include any parent directories"""
192
193			try:
194			os.makedirs(path, args, *kwargs)
195			return True
196			except OSError, ex:
197			if ex.errno == 17:
198			# Directory already exists
199			return True
200
201			Log.Warn('Unable to create directories: %r - (%s) %s', cls.to_relative_path(path), ex.errno, ex)
202			except Exception, ex:
203			Log.Warn('Unable to create directories: %r - (%s) %s', cls.to_relative_path(path), type(ex), ex)
204
205			return False
206
207			@classmethod
208			def copy_tree(cls, source, destination):
209			"""Copy the directory at `source` to `destination`"""
210
211			try:
212			shutil.copytree(source, destination)
213
214			Log.Debug('Copied %r to %r', cls.to_relative_path(source), cls.to_relative_path(destination))
215			return True
216			except Exception, ex:
217			Log.Warn('Unable to copy %r to %r - %s', cls.to_relative_path(source), cls.to_relative_path(destination), ex)
218
219			return False
220
221			@classmethod
222			def delete_tree(cls, path):
223			"""Delete the directory (at `path`"""
224
225			try:
226			shutil.rmtree(path)
227
228			Log.Debug('Deleted %r', cls.to_relative_path(path))
229			return True
230			except Exception, ex:
231			Log.Warn('Unable to delete directory: %r - %s', cls.to_relative_path(path), ex)
232
233			return False
234
235			@classmethod
236			def to_relative_path(cls, path):
237			"""Convert `path` to be relative to `StorageHelper.base_names`"""
238
239			path_lower = path.lower()
240
241			# Find base path
242			base_path = None
243
244			for base in cls.base_names:
245			if base not in path_lower:
246			continue
247
248			base_path = path[:path_lower.find(base)]
249			break
250
251			# Check if `base_path` was found
252			if not base_path:
253			Log.Warn('Unable to find base path in %r', path)
254			return path
255
256			# Return relative path
257			return os.path.relpath(path, base_path)
258
259			@classmethod
260			def is_relative_path(cls, path):
261			"""Check if `path` is relative to `StorageHelper.base_names`"""
262
263			path_lower = path.lower()
264
265			# Ignore framework paths
266			if 'framework.bundle' in path_lower:
267			return False
268
269			# Find base path
270			for base in cls.base_names:
271			if base not in path_lower:
272			continue
273
274			return True
275
276			return False
277
278
279			class SystemHelper(object):
280			bits_map = {
281			'32bit': 'i386',
282			'64bit': 'x86_64'
283			}
284
285			machine_map = {
286			('32bit', 'i686'): 'i686'
287			}
288
289			name_map = {
290			'Darwin': 'MacOSX'
291			}
292
293			@classmethod
294			def architecture(cls):
295			"""Retrieve system architecture (i386, i686, x86_64)"""
296
297			bits, _ = platform.architecture()
298			machine = platform.machine()
299
300			# Check for ARM machine
301			if bits == '32bit' and machine.startswith('armv'):
302			return cls.arm(machine)
303
304			# Check (bits, machine) map
305			machine_key = (bits, machine)
306
307			if machine_key in cls.machine_map:
308			return cls.machine_map[machine_key]
309
310			# Check (bits) map
311			if bits in cls.bits_map:
312			return cls.bits_map[bits]
313
314			Log.Info('Unable to determine system architecture - bits: %r, machine: %r', bits, machine)
315			return None
316
317			@classmethod
318			def name(cls):
319			"""Retrieve system name (Windows, Linux, FreeBSD, MacOSX)"""
320
321			system = platform.system()
322
323			# Apply system map
324			if system in cls.name_map:
325			system = cls.name_map[system]
326
327			return system
328
329			@classmethod
330			def arm(cls, machine):
331			# Determine floating-point type
332			float_type = cls.arm_float_type()
333
334			if float_type is None:
335			Log.Warn('Unable to use ARM libraries, unsupported floating-point type?')
336			return None
337
338			# Determine ARM version
339			version = cls.arm_version()
340
341			if version is None:
342			Log.Warn('Unable to use ARM libraries, unsupported ARM version (%r)?' % machine)
343			return None
344
345			return '%s_%s' % (version, float_type)
346
347			@classmethod
348			def arm_version(cls, machine=None):
349			# Read `machine` name if not provided
350			if machine is None:
351			machine = platform.machine()
352
353			# Ensure `machine` is valid
354			if not machine:
355			return None
356
357			# ARMv6
358			if machine.startswith('armv6'):
359			return 'armv6'
360
361			# ARMv7
362			if machine.startswith('armv7'):
363			return 'armv7'
364
365			return None
366
367			@classmethod
368			def arm_float_type(cls):
369			if os.path.exists('/lib/arm-linux-gnueabihf'):
370			return 'hf'
371
372			if os.path.exists('/lib/arm-linux-gnueabi'):
373			return 'sf'
374
375			return None
376

fuzeman / Plex-Trakt-Scrobbler

Push — develop ( 1e4ff5...af752c )

SystemHelper.arm() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like