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

LibrariesManager.reset() B
last analyzed 2018-03-05 01:11 UTC

↳ Parent: LibrariesManager

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	10
CRAP Score	4.0119

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
dl	0
loc	25
ccs	10
cts	11
cp	0.9091
rs	8.5806
c	1
b	0
f	0
cc	4
crap	4.0119

from plugin.core.configuration import Configuration
from plugin.core.environment import Environment
from plugin.core.message import InterfaceMessages
from plugin.core.helpers.variable import merge
from plugin.core.libraries.cache import CacheManager
from plugin.core.libraries.constants import CONTENTS_PATH, NATIVE_DIRECTORIES, UNICODE_MAP
from plugin.core.libraries.helpers import PathHelper, StorageHelper, SystemHelper
from plugin.core.libraries.tests import LIBRARY_TESTS

import json
import logging
import os
import platform
import sys

log = logging.getLogger(__name__)


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

        :param cache: Enable native library caching
        :type cache: bool
        """

        # Read distribution metadata
        distribution = cls._read_distribution_metadata()

        # Use `cache` value from advanced configuration
        cache = Configuration.advanced['libraries'].get_boolean('cache', cache)

        # Retrieve libraries path (and cache libraries, if enabled)
        libraries_path = cls._libraries_path(cache)

        if not libraries_path:
            return

        log.info('Using native libraries at %r', StorageHelper.to_relative_path(libraries_path))

        # Remove current native library directories from `sys.path`
        cls.reset()

        # Insert platform specific library paths
        cls._insert_paths(distribution, libraries_path)

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

            if StorageHelper.is_framework_path(path):
                continue

            log.info('[PATH] %s', StorageHelper.to_relative_path(path))

    @staticmethod
    def test():
        """Test native libraries to ensure they can be correctly loaded"""
        log.info('Testing native library support...')

        # Retrieve library directories
        search_paths = []

        for path in sys.path:
            path_lower = path.lower()

            if 'trakttv.bundle' not in path_lower and 'com.plexapp.plugins.trakttv' not in path_lower:
                continue

            search_paths.append(path)

        # Run library tests
        metadata = {}

        for test in LIBRARY_TESTS:
            # Run tests
            result = test.run(search_paths)

            if not result.get('success'):
                log_func = logging.warn if test.optional else logging.error

                # Write message to logfile
                if 'traceback' in result:
                    log_func('%s: unavailable - %s\n%%s' % (test.name, result.get('message')), result['traceback'])
                else:
                    log_func('%s: unavailable - %s' % (test.name, result.get('message')), exc_info=result.get('exc_info'))


                if not test.optional:
                    return

                continue

            # Test successful
            t_metadata = result.get('metadata') or {}
            t_versions = t_metadata.get('versions')

            if t_versions:
                expanded = len(t_versions) > 1 or (
                    t_versions and t_versions.keys()[0] != test.name
                )

                if expanded:
                    log.info('%s: available (%s)', test.name, ', '.join([
                        '%s: %s' % (key, value)
                        for key, value in t_versions.items()
                    ]))
                else:
                    key = t_versions.keys()[0]

                    log.info('%s: available (%s)', test.name, t_versions[key])
            else:
                log.info('%s: available', test.name)

            # Merge result into `metadata`
            merge(metadata, t_metadata, recursive=True)

        # Include versions in error reports
        versions = metadata.get('versions') or {}



    @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(CONTENTS_PATH.lower()):
                continue

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

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

            # Convert to unix-style separators (/)
            path_rel = path_rel.replace('\\', '/')

            # Ignore non-native library directories
            if path_rel not in NATIVE_DIRECTORIES:
                continue

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

    @classmethod
    def _read_distribution_metadata(cls):
        metadata_path = os.path.join(Environment.path.contents, 'distribution.json')

        if not os.path.exists(metadata_path):
            return None

        # Read distribution metadata
        try:
            with open(metadata_path, 'r') as fp:
                distribution = json.load(fp)
        except Exception as ex:

            log.warn('Unable to read distribution metadata: %s', ex, exc_info=True)
            return None

        if not distribution or not distribution.get('name'):
            return

        return distribution

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

        :param cache: Enable native library caching
        :type cache: bool
        """

        # Use specified libraries path (from "advanced.ini')
        libraries_path = Configuration.advanced['libraries'].get('libraries_path')

        if libraries_path and os.path.exists(libraries_path):
            log.info('Using libraries at %r', StorageHelper.to_relative_path(libraries_path))
            return libraries_path

        # Use system libraries (if bundled libraries have been disabled in "advanced.ini")
        if not Configuration.advanced['libraries'].get_boolean('bundled', True):
            log.info('Bundled libraries have been disabled, using system libraries')
            return None

        # Cache libraries (if enabled)
        if cache:
            return cls._cache_libraries()

        return Environment.path.libraries

    @classmethod
    def _cache_libraries(cls):
        cache_path = Configuration.advanced['libraries'].get('cache_path')

        # Try cache libraries to `cache_path`
        libraries_path = CacheManager.sync(cache_path)

        if not libraries_path:
            log.info('Unable to cache libraries, using bundled libraries directly')
            return Environment.path.libraries

        log.info('Cached libraries to %r', StorageHelper.to_relative_path(libraries_path))
        return libraries_path

    @classmethod
    def _insert_paths(cls, distribution, libraries_path):
        # Display platform details
        p_bits, _ = platform.architecture()
        p_machine = platform.machine()

        log.debug('Bits: %r, Machine: %r', p_bits, p_machine)

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

        if not architecture:
            InterfaceMessages.add(60, 'Unable to retrieve system architecture')
            return False

        log.debug('System: %r, Architecture: %r', system, architecture)

        # Build architecture list
        architectures = [architecture]

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

        # Insert library paths
        found = False

        for arch in architectures + ['universal']:
            if cls._insert_architecture_paths(libraries_path, system, arch):
                log.debug('Inserted libraries path for system: %r, arch: %r', system, arch)
                found = True

        # Display interface message if no libraries were found
        if not found:
            if distribution and distribution.get('name'):
                message = 'Unable to find compatible native libraries in the %s distribution' % distribution['name']
            else:
                message = 'Unable to find compatible native libraries'

            InterfaceMessages.add(60, '%s (system: %r, architecture: %r)', message, system, architecture)

        return found

    @classmethod
    def _insert_architecture_paths(cls, libraries_path, system, architecture):
        architecture_path = os.path.join(libraries_path, system, architecture)

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

        # Architecture libraries
        PathHelper.insert(libraries_path, system, architecture)

        # System libraries
        if system == 'Windows':
            # Windows libraries (VC++ specific)
            cls._insert_paths_windows(libraries_path, system, architecture)
        else:
            # Darwin/FreeBSD/Linux libraries
            cls._insert_paths_unix(libraries_path, system, architecture)

        return True

    @staticmethod
    def _insert_paths_unix(libraries_path, system, architecture):
        # UCS specific libraries
        ucs = UNICODE_MAP.get(sys.maxunicode)
        log.debug('UCS: %r', ucs)

        if ucs:
            PathHelper.insert(libraries_path, system, architecture, ucs)

        # CPU specific libraries
        cpu_type = SystemHelper.cpu_type()
        page_size = SystemHelper.page_size()

        log.debug('CPU Type: %r', cpu_type)
        log.debug('Page Size: %r', page_size)

        if cpu_type:
            PathHelper.insert(libraries_path, system, architecture, cpu_type)

            if page_size:
                PathHelper.insert(libraries_path, system, architecture, '%s_%s' % (cpu_type, page_size))

        # UCS + CPU specific libraries
        if cpu_type and ucs:
            PathHelper.insert(libraries_path, system, architecture, cpu_type, ucs)

            if page_size:
                PathHelper.insert(libraries_path, system, architecture, '%s_%s' % (cpu_type, page_size), ucs)

    @staticmethod
    def _insert_paths_windows(libraries_path, system, architecture):
        vcr = SystemHelper.vcr_version() or 'vc12'  # Assume "vc12" if call fails
        ucs = UNICODE_MAP.get(sys.maxunicode)

        log.debug('VCR: %r, UCS: %r', vcr, ucs)

        # VC++ libraries
        PathHelper.insert(libraries_path, system, architecture, vcr)

        # UCS libraries
        if ucs:
            PathHelper.insert(libraries_path, system, architecture, vcr, ucs)


1	1	from plugin.core.configuration import Configuration
2	1	from plugin.core.environment import Environment
3	1	from plugin.core.message import InterfaceMessages
4	1	from plugin.core.helpers.variable import merge
5	1	from plugin.core.libraries.cache import CacheManager
6	1	from plugin.core.libraries.constants import CONTENTS_PATH, NATIVE_DIRECTORIES, UNICODE_MAP
7	1	from plugin.core.libraries.helpers import PathHelper, StorageHelper, SystemHelper
8	1	from plugin.core.libraries.tests import LIBRARY_TESTS
9
10	1	import json
11	1	import logging
12	1	import os
13	1	import platform
14	1	import sys
15
16	1	log = logging.getLogger(__name__)
17
18
19	1	class LibrariesManager(object):
20	1	@classmethod
21	1	def setup(cls, cache=False):
22		"""Setup native library directories
23
24		:param cache: Enable native library caching
25		:type cache: bool
26		"""
27
28		# Read distribution metadata
29	1	distribution = cls._read_distribution_metadata()
30
31		# Use `cache` value from advanced configuration
32	1	cache = Configuration.advanced['libraries'].get_boolean('cache', cache)
33
34		# Retrieve libraries path (and cache libraries, if enabled)
35	1	libraries_path = cls._libraries_path(cache)
36
37	1	if not libraries_path:
38		return
39
40	1	log.info('Using native libraries at %r', StorageHelper.to_relative_path(libraries_path))
41
42		# Remove current native library directories from `sys.path`
43	1	cls.reset()
44
45		# Insert platform specific library paths
46	1	cls._insert_paths(distribution, libraries_path)
47
48		# Display library paths in logfile
49	1	for path in sys.path:
50	1	path = os.path.abspath(path)
51
52	1	if StorageHelper.is_framework_path(path):
53		continue
54
55	1	log.info('[PATH] %s', StorageHelper.to_relative_path(path))
56
57	1	@staticmethod
58		def test():
59		"""Test native libraries to ensure they can be correctly loaded"""
60	1	log.info('Testing native library support...')
61
62		# Retrieve library directories
63	1	search_paths = []
64
65	1	for path in sys.path:
66	1	path_lower = path.lower()
67
68	1	if 'trakttv.bundle' not in path_lower and 'com.plexapp.plugins.trakttv' not in path_lower:
69	1	continue
70
71	1	search_paths.append(path)
72
73		# Run library tests
74	1	metadata = {}
75
76	1	for test in LIBRARY_TESTS:
77		# Run tests
78	1	result = test.run(search_paths)
79
80	1	if not result.get('success'):
81		log_func = logging.warn if test.optional else logging.error
82
83		# Write message to logfile
84		if 'traceback' in result:
85		log_func('%s: unavailable - %s\n%%s' % (test.name, result.get('message')), result['traceback'])
86		else:
87		log_func('%s: unavailable - %s' % (test.name, result.get('message')), exc_info=result.get('exc_info'))
		0 ignored issues – show Coding Style introduced 2016-06-06 12:03 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (122/120). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
88
89		if not test.optional:
90		return
91
92		continue
93
94		# Test successful
95	1	t_metadata = result.get('metadata') or {}
96	1	t_versions = t_metadata.get('versions')
97
98	1	if t_versions:
99	1	expanded = len(t_versions) > 1 or (
100		t_versions and t_versions.keys()[0] != test.name
101		)
102
103	1	if expanded:
104	1	log.info('%s: available (%s)', test.name, ', '.join([
105		'%s: %s' % (key, value)
106		for key, value in t_versions.items()
107		]))
108		else:
109	1	key = t_versions.keys()[0]
110
111	1	log.info('%s: available (%s)', test.name, t_versions[key])
112		else:
113	1	log.info('%s: available', test.name)
114
115		# Merge result into `metadata`
116	1	merge(metadata, t_metadata, recursive=True)
117
118		# Include versions in error reports
119	1	versions = metadata.get('versions') or {}
		0 ignored issues – show Unused Code introduced 2018-03-04 22:11 UTC by Report Bug Copy Issue Report The variable `versions` seems to be unused. Loading history...
120
121
122	1	@classmethod
123		def reset(cls):
124		"""Remove all the native library directives from `sys.path`"""
125
126	1	for path in sys.path:
127	1	path = os.path.abspath(path)
128
129	1	if not path.lower().startswith(CONTENTS_PATH.lower()):
130	1	continue
131
132		# Convert to relative path
133	1	path_rel = os.path.relpath(path, CONTENTS_PATH)
134
135		# Take the first two fragments
136	1	path_rel = os.path.sep.join(path_rel.split(os.path.sep)[:2])
137
138		# Convert to unix-style separators (/)
139	1	path_rel = path_rel.replace('\\', '/')
140
141		# Ignore non-native library directories
142	1	if path_rel not in NATIVE_DIRECTORIES:
143	1	continue
144
145		# Remove from `sys.path`
146		PathHelper.remove(path)
147
148	1	@classmethod
149		def _read_distribution_metadata(cls):
150	1	metadata_path = os.path.join(Environment.path.contents, 'distribution.json')
151
152	1	if not os.path.exists(metadata_path):
153		return None
154
155		# Read distribution metadata
156	1	try:
157	1	with open(metadata_path, 'r') as fp:
158	1	distribution = json.load(fp)
159		except Exception as ex:
		0 ignored issues – show Best Practice introduced 2016-12-30 03:05 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...
160		log.warn('Unable to read distribution metadata: %s', ex, exc_info=True)
161		return None
162
163	1	if not distribution or not distribution.get('name'):
164		return
165
166	1	return distribution
167
168	1	@classmethod
169	1	def _libraries_path(cls, cache=False):
170		"""Retrieve the native libraries base directory (and cache the libraries if enabled)
171
172		:param cache: Enable native library caching
173		:type cache: bool
174		"""
175
176		# Use specified libraries path (from "advanced.ini')
177	1	libraries_path = Configuration.advanced['libraries'].get('libraries_path')
178
179	1	if libraries_path and os.path.exists(libraries_path):
180		log.info('Using libraries at %r', StorageHelper.to_relative_path(libraries_path))
181		return libraries_path
182
183		# Use system libraries (if bundled libraries have been disabled in "advanced.ini")
184	1	if not Configuration.advanced['libraries'].get_boolean('bundled', True):
185		log.info('Bundled libraries have been disabled, using system libraries')
186		return None
187
188		# Cache libraries (if enabled)
189	1	if cache:
190		return cls._cache_libraries()
191
192	1	return Environment.path.libraries
193
194	1	@classmethod
195		def _cache_libraries(cls):
196		cache_path = Configuration.advanced['libraries'].get('cache_path')
197
198		# Try cache libraries to `cache_path`
199		libraries_path = CacheManager.sync(cache_path)
200
201		if not libraries_path:
202		log.info('Unable to cache libraries, using bundled libraries directly')
203		return Environment.path.libraries
204
205		log.info('Cached libraries to %r', StorageHelper.to_relative_path(libraries_path))
206		return libraries_path
207
208	1	@classmethod
209		def _insert_paths(cls, distribution, libraries_path):
210		# Display platform details
211	1	p_bits, _ = platform.architecture()
212	1	p_machine = platform.machine()
213
214	1	log.debug('Bits: %r, Machine: %r', p_bits, p_machine)
215
216		# Retrieve system details
217	1	system = SystemHelper.name()
218	1	architecture = SystemHelper.architecture()
219
220	1	if not architecture:
221		InterfaceMessages.add(60, 'Unable to retrieve system architecture')
222		return False
223
224	1	log.debug('System: %r, Architecture: %r', system, architecture)
225
226		# Build architecture list
227	1	architectures = [architecture]
228
229	1	if architecture == 'i686':
230		# Fallback to i386
231		architectures.append('i386')
232
233		# Insert library paths
234	1	found = False
235
236	1	for arch in architectures + ['universal']:
237	1	if cls._insert_architecture_paths(libraries_path, system, arch):
238	1	log.debug('Inserted libraries path for system: %r, arch: %r', system, arch)
239	1	found = True
240
241		# Display interface message if no libraries were found
242	1	if not found:
243		if distribution and distribution.get('name'):
244		message = 'Unable to find compatible native libraries in the %s distribution' % distribution['name']
245		else:
246		message = 'Unable to find compatible native libraries'
247
248		InterfaceMessages.add(60, '%s (system: %r, architecture: %r)', message, system, architecture)
249
250	1	return found
251
252	1	@classmethod
253		def _insert_architecture_paths(cls, libraries_path, system, architecture):
254	1	architecture_path = os.path.join(libraries_path, system, architecture)
255
256	1	if not os.path.exists(architecture_path):
257	1	return False
258
259		# Architecture libraries
260	1	PathHelper.insert(libraries_path, system, architecture)
261
262		# System libraries
263	1	if system == 'Windows':
264		# Windows libraries (VC++ specific)
265		cls._insert_paths_windows(libraries_path, system, architecture)
266		else:
267		# Darwin/FreeBSD/Linux libraries
268	1	cls._insert_paths_unix(libraries_path, system, architecture)
269
270	1	return True
271
272	1	@staticmethod
273		def _insert_paths_unix(libraries_path, system, architecture):
274		# UCS specific libraries
275	1	ucs = UNICODE_MAP.get(sys.maxunicode)
276	1	log.debug('UCS: %r', ucs)
277
278	1	if ucs:
279	1	PathHelper.insert(libraries_path, system, architecture, ucs)
280
281		# CPU specific libraries
282	1	cpu_type = SystemHelper.cpu_type()
283	1	page_size = SystemHelper.page_size()
284
285	1	log.debug('CPU Type: %r', cpu_type)
286	1	log.debug('Page Size: %r', page_size)
287
288	1	if cpu_type:
289		PathHelper.insert(libraries_path, system, architecture, cpu_type)
290
291		if page_size:
292		PathHelper.insert(libraries_path, system, architecture, '%s_%s' % (cpu_type, page_size))
293
294		# UCS + CPU specific libraries
295	1	if cpu_type and ucs:
296		PathHelper.insert(libraries_path, system, architecture, cpu_type, ucs)
297
298		if page_size:
299		PathHelper.insert(libraries_path, system, architecture, '%s_%s' % (cpu_type, page_size), ucs)
300
301	1	@staticmethod
302		def _insert_paths_windows(libraries_path, system, architecture):
303		vcr = SystemHelper.vcr_version() or 'vc12' # Assume "vc12" if call fails
304		ucs = UNICODE_MAP.get(sys.maxunicode)
305
306		log.debug('VCR: %r, UCS: %r', vcr, ucs)
307
308		# VC++ libraries
309		PathHelper.insert(libraries_path, system, architecture, vcr)
310
311		# UCS libraries
312		if ucs:
313		PathHelper.insert(libraries_path, system, architecture, vcr, ucs)
314

fuzeman / Plex-Trakt-Scrobbler

LibrariesManager.reset() B last analyzed 2018-03-05 01:11 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

LibrariesManager.reset() B
last analyzed 2018-03-05 01:11 UTC