ocrd.resource_manager.OcrdResourceManager.__init__() - Code Metrics - Inspection of "Resmgr tess config" - OCR-D/core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#680)

by Konstantin

created 2021-03-05 18:11 UTC

OcrdResourceManager.init() A

↳ Parent: ocrd.resource_manager

Complexity

Conditions

Size

Total Lines	11
Code Lines	11

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	11
dl	0
loc	11
rs	9.85
c	0
b	0
f	0
cc	4
nop	1

from pathlib import Path
from os.path import join
from os import environ, listdir, getcwd
import re
from shutil import copytree
from datetime import datetime
from tarfile import open as open_tarfile
from urllib.parse import urlparse, unquote

import requests
from yaml import safe_load, safe_dump

from ocrd_validators import OcrdResourceListValidator
from ocrd_utils import getLogger
from ocrd_utils.constants import HOME, XDG_DATA_HOME, XDG_CONFIG_HOME
from ocrd_utils.os import list_all_resources, pushd_popd

from .constants import RESOURCE_LIST_FILENAME, RESOURCE_USER_LIST_COMMENT

class OcrdResourceManager():

    """
    Managing processor resources
    """
    def __init__(self):
        self.log = getLogger('ocrd.resource_manager')
        self.database = {}
        self.load_resource_list(Path(RESOURCE_LIST_FILENAME))
        self.user_list = Path(XDG_CONFIG_HOME, 'ocrd', 'resources.yml')
        if not self.user_list.exists():
            if not self.user_list.parent.exists():
                self.user_list.parent.mkdir(parents=True)
            with open(str(self.user_list), 'w', encoding='utf-8') as f:
                f.write(RESOURCE_USER_LIST_COMMENT)
        self.load_resource_list(self.user_list)

    def load_resource_list(self, list_filename, database=None):
        if not database:
            database = self.database
        if list_filename.is_file():
            with open(list_filename, 'r', encoding='utf-8') as f:
                list_loaded = safe_load(f) or {}
            report = OcrdResourceListValidator.validate(list_loaded)
            if not report.is_valid:
                self.log.error('\n'.join(report.errors))
                raise ValueError("Resource list %s is invalid!" % (list_filename))
            for executable, resource_list in list_loaded.items():
                if executable not in database:
                    database[executable] = []
                # Prepend, so user provided is sorted before builtin
                database[executable] = list_loaded[executable] + database[executable]
        return database

    def list_available(self, executable=None):
        """
        List models available for download by processor
        """
        if executable:
            return [(executable, self.database[executable])]
        return [(x, y) for x, y in self.database.items()]

    def list_installed(self, executable=None):
        """
        List installed resources, matching with registry by ``name``
        """
        ret = []
        if executable:
            all_executables = [executable]
        else:
            # resources we know about
            all_executables = list(self.database.keys())
            # resources in the file system
            parent_dirs = [join(x, 'ocrd-resources') for x in [XDG_DATA_HOME, '/usr/local/share']]
            for parent_dir in parent_dirs:
                if Path(parent_dir).exists():
                    all_executables += [x for x in listdir(parent_dir) if x.startswith('ocrd-')]
        for this_executable in set(all_executables):
            reslist = []
            for res_filename in list_all_resources(this_executable):
                res_name = Path(res_filename).name
                resdict = [x for x in self.database.get(this_executable, []) if x['name'] == res_name]
                if not resdict:
                    self.log.info("%s resource '%s' (%s) not a known resource, creating stub in %s'" % (this_executable, res_name, res_filename, self.user_list))
                    resdict = [self.add_to_user_database(this_executable, res_filename)]
                resdict[0]['path'] = res_filename
                reslist.append(resdict[0])
            ret.append((this_executable, reslist))
        return ret

    def add_to_user_database(self, executable, res_filename, url=None):
        """
        Add a stub entry to the user resource.yml
        """
        res_name = Path(res_filename).name
        res_size = Path(res_filename).stat().st_size
        with open(self.user_list, 'r', encoding='utf-8') as f:
            user_database = safe_load(f) or {}
        if executable not in user_database:
            user_database[executable] = []
        if not self.find_resources(executable=executable, name=res_name, database=user_database):
            resdict = {
                'name': res_name,
                'url': url if url else '???',
                'description': 'Found at %s on %s' % (self.resource_dir_to_location(res_filename), datetime.now()),
                'version_range': '???',
                'size': res_size
            }
            user_database[executable].append(resdict)
        with open(self.user_list, 'w', encoding='utf-8') as f:
            f.write(RESOURCE_USER_LIST_COMMENT)
            f.write('\n')
            f.write(safe_dump(user_database))
        return resdict


    def find_resources(self, executable=None, name=None, url=None, database=None):
        """
        Find resources in the registry
        """
        if not database:
            database = self.database
        ret = []
        if executable and executable not in database.keys():
            return ret
        for executable in [executable] if executable else database.keys():
            for resdict in database[executable]:
                if not name and not url:
                    ret.append((executable, resdict))
                elif url and url == resdict['url']:
                    ret.append((executable, resdict))
                elif name and name == resdict['name']:
                    ret.append((executable, resdict))
        return ret

    @property
    def default_resource_dir(self):
        return self.location_to_resource_dir('data')

    def location_to_resource_dir(self, location):
        return '/usr/local/share/ocrd-resources' if location == 'system' else \
                join(XDG_DATA_HOME, 'ocrd-resources') if location == 'data' else \
                getcwd()

    def resource_dir_to_location(self, resource_path):
        resource_path = str(resource_path)
        return 'system' if resource_path.startswith('/usr/local/share/ocrd-resources') else \
               'data' if resource_path.startswith(join(XDG_DATA_HOME, 'ocrd-resources')) else \
               'cwd' if resource_path.startswith(getcwd()) else \
               resource_path

    def parameter_usage(self, name, usage='as-is'):
        if usage == 'as-is':
            return name
        if usage == 'without-extension':
            return Path(name).stem

    def _download_impl(self, url, filename, progress_cb=None, size=None):
        log = getLogger('ocrd.resource_manager._download_impl')
        log.info("Downloading %s to %s" % (url, filename))
        with open(filename, 'wb') as f:
            with requests.get(url, stream=True) as r:
                total = size if size else int(r.headers.get('content-length'))
                for data in r.iter_content(chunk_size=4096):
                    if progress_cb:
                        progress_cb(len(data))
                    f.write(data)

    def _copy_impl(self, src_filename, filename, progress_cb=None):
        log = getLogger('ocrd.resource_manager._copy_impl')
        log.info("Copying %s" % src_filename)
        with open(filename, 'wb') as f_out, open(src_filename, 'rb') as f_in:
            while True:
                chunk = f_in.read(4096)
                if chunk:
                    f_out.write(chunk)
                    if progress_cb:
                        progress_cb(len(chunk))
                else:
                    break

    # TODO Proper caching (make head request for size, If-Modified etc)
    def download(
        self,
        executable,
        url,
        basedir,
        overwrite=False,
        no_subdir=False,
        name=None,
        resource_type='file',
        path_in_archive='.',
        progress_cb=None,
        size=None,
    ):
        """
        Download a resource by URL
        """
        log = getLogger('ocrd.resource_manager.download')
        destdir = Path(basedir) if no_subdir else Path(basedir, executable)
        if not name:
            url_parsed = urlparse(url)
            name = Path(unquote(url_parsed.path)).name
        fpath = Path(destdir, name)
        is_url = url.startswith('https://') or url.startswith('http://')
        if fpath.exists() and not overwrite:
            log.info("%s to be %s to %s which already exists and overwrite is False" % (url, 'downloaded' if is_url else 'copied', fpath))
            return fpath
        destdir.mkdir(parents=True, exist_ok=True)
        if resource_type == 'file':
            if is_url:
                self._download_impl(url, fpath, progress_cb)
            else:
                self._copy_impl(url, fpath, progress_cb)
        elif resource_type == 'tarball':
            with pushd_popd(tempdir=True):
                if is_url:
                    self._download_impl(url, 'download.tar.xx', progress_cb, size)
                else:
                    self._copy_impl(url, 'download.tar.xx', progress_cb)
                Path('out').mkdir()
                with pushd_popd('out'):
                    log.info("Extracting tarball")
                    with open_tarfile('../download.tar.xx', 'r:*') as tar:
                        tar.extractall()
                    log.info("Copying '%s' from tarball to %s" % (path_in_archive, fpath))
                    copytree(path_in_archive, str(fpath))
        # TODO
        # elif resource_type == 'github-dir':
        return fpath


1			from pathlib import Path
2			from os.path import join
3			from os import environ, listdir, getcwd
4			import re
5			from shutil import copytree
6			from datetime import datetime
7			from tarfile import open as open_tarfile
8			from urllib.parse import urlparse, unquote
9
10			import requests
11			from yaml import safe_load, safe_dump
12
13			from ocrd_validators import OcrdResourceListValidator
14			from ocrd_utils import getLogger
15			from ocrd_utils.constants import HOME, XDG_DATA_HOME, XDG_CONFIG_HOME
16			from ocrd_utils.os import list_all_resources, pushd_popd
17
18			from .constants import RESOURCE_LIST_FILENAME, RESOURCE_USER_LIST_COMMENT
19
20			class OcrdResourceManager():
21
22			"""
23			Managing processor resources
24			"""
25			def __init__(self):
26			self.log = getLogger('ocrd.resource_manager')
27			self.database = {}
28			self.load_resource_list(Path(RESOURCE_LIST_FILENAME))
29			self.user_list = Path(XDG_CONFIG_HOME, 'ocrd', 'resources.yml')
30			if not self.user_list.exists():
31			if not self.user_list.parent.exists():
32			self.user_list.parent.mkdir(parents=True)
33			with open(str(self.user_list), 'w', encoding='utf-8') as f:
34			f.write(RESOURCE_USER_LIST_COMMENT)
35			self.load_resource_list(self.user_list)
36
37			def load_resource_list(self, list_filename, database=None):
38			if not database:
39			database = self.database
40			if list_filename.is_file():
41			with open(list_filename, 'r', encoding='utf-8') as f:
42			list_loaded = safe_load(f) or {}
43			report = OcrdResourceListValidator.validate(list_loaded)
44			if not report.is_valid:
45			self.log.error('\n'.join(report.errors))
46			raise ValueError("Resource list %s is invalid!" % (list_filename))
47			for executable, resource_list in list_loaded.items():
48			if executable not in database:
49			database[executable] = []
50			# Prepend, so user provided is sorted before builtin
51			database[executable] = list_loaded[executable] + database[executable]
52			return database
53
54			def list_available(self, executable=None):
55			"""
56			List models available for download by processor
57			"""
58			if executable:
59			return [(executable, self.database[executable])]
60			return [(x, y) for x, y in self.database.items()]
61
62			def list_installed(self, executable=None):
63			"""
64			List installed resources, matching with registry by ``name``
65			"""
66			ret = []
67			if executable:
68			all_executables = [executable]
69			else:
70			# resources we know about
71			all_executables = list(self.database.keys())
72			# resources in the file system
73			parent_dirs = [join(x, 'ocrd-resources') for x in [XDG_DATA_HOME, '/usr/local/share']]
74			for parent_dir in parent_dirs:
75			if Path(parent_dir).exists():
76			all_executables += [x for x in listdir(parent_dir) if x.startswith('ocrd-')]
77			for this_executable in set(all_executables):
78			reslist = []
79			for res_filename in list_all_resources(this_executable):
80			res_name = Path(res_filename).name
81			resdict = [x for x in self.database.get(this_executable, []) if x['name'] == res_name]
82			if not resdict:
83			self.log.info("%s resource '%s' (%s) not a known resource, creating stub in %s'" % (this_executable, res_name, res_filename, self.user_list))
84			resdict = [self.add_to_user_database(this_executable, res_filename)]
85			resdict[0]['path'] = res_filename
86			reslist.append(resdict[0])
87			ret.append((this_executable, reslist))
88			return ret
89
90			def add_to_user_database(self, executable, res_filename, url=None):
91			"""
92			Add a stub entry to the user resource.yml
93			"""
94			res_name = Path(res_filename).name
95			res_size = Path(res_filename).stat().st_size
96			with open(self.user_list, 'r', encoding='utf-8') as f:
97			user_database = safe_load(f) or {}
98			if executable not in user_database:
99			user_database[executable] = []
100			if not self.find_resources(executable=executable, name=res_name, database=user_database):
101			resdict = {
102			'name': res_name,
103			'url': url if url else '???',
104			'description': 'Found at %s on %s' % (self.resource_dir_to_location(res_filename), datetime.now()),
105			'version_range': '???',
106			'size': res_size
107			}
108			user_database[executable].append(resdict)
109			with open(self.user_list, 'w', encoding='utf-8') as f:
110			f.write(RESOURCE_USER_LIST_COMMENT)
111			f.write('\n')
112			f.write(safe_dump(user_database))
113			return resdict
			0 ignored issues – show introduced 2021-01-05 13:16 UTC by Report Bug Copy Issue Report The variable `resdict` does not seem to be defined in case `BooleanNotNode` on line `100` is `False`. Are you sure this can never be the case? Loading history...
114
115			def find_resources(self, executable=None, name=None, url=None, database=None):
116			"""
117			Find resources in the registry
118			"""
119			if not database:
120			database = self.database
121			ret = []
122			if executable and executable not in database.keys():
123			return ret
124			for executable in [executable] if executable else database.keys():
125			for resdict in database[executable]:
126			if not name and not url:
127			ret.append((executable, resdict))
128			elif url and url == resdict['url']:
129			ret.append((executable, resdict))
130			elif name and name == resdict['name']:
131			ret.append((executable, resdict))
132			return ret
133
134			@property
135			def default_resource_dir(self):
136			return self.location_to_resource_dir('data')
137
138			def location_to_resource_dir(self, location):
139			return '/usr/local/share/ocrd-resources' if location == 'system' else \
140			join(XDG_DATA_HOME, 'ocrd-resources') if location == 'data' else \
141			getcwd()
142
143			def resource_dir_to_location(self, resource_path):
144			resource_path = str(resource_path)
145			return 'system' if resource_path.startswith('/usr/local/share/ocrd-resources') else \
146			'data' if resource_path.startswith(join(XDG_DATA_HOME, 'ocrd-resources')) else \
147			'cwd' if resource_path.startswith(getcwd()) else \
148			resource_path
149
150			def parameter_usage(self, name, usage='as-is'):
151			if usage == 'as-is':
152			return name
153			if usage == 'without-extension':
154			return Path(name).stem
155
156			def _download_impl(self, url, filename, progress_cb=None, size=None):
157			log = getLogger('ocrd.resource_manager._download_impl')
158			log.info("Downloading %s to %s" % (url, filename))
159			with open(filename, 'wb') as f:
160			with requests.get(url, stream=True) as r:
161			total = size if size else int(r.headers.get('content-length'))
162			for data in r.iter_content(chunk_size=4096):
163			if progress_cb:
164			progress_cb(len(data))
165			f.write(data)
166
167			def _copy_impl(self, src_filename, filename, progress_cb=None):
168			log = getLogger('ocrd.resource_manager._copy_impl')
169			log.info("Copying %s" % src_filename)
170			with open(filename, 'wb') as f_out, open(src_filename, 'rb') as f_in:
171			while True:
172			chunk = f_in.read(4096)
173			if chunk:
174			f_out.write(chunk)
175			if progress_cb:
176			progress_cb(len(chunk))
177			else:
178			break
179
180			# TODO Proper caching (make head request for size, If-Modified etc)
181			def download(
182			self,
183			executable,
184			url,
185			basedir,
186			overwrite=False,
187			no_subdir=False,
188			name=None,
189			resource_type='file',
190			path_in_archive='.',
191			progress_cb=None,
192			size=None,
193			):
194			"""
195			Download a resource by URL
196			"""
197			log = getLogger('ocrd.resource_manager.download')
198			destdir = Path(basedir) if no_subdir else Path(basedir, executable)
199			if not name:
200			url_parsed = urlparse(url)
201			name = Path(unquote(url_parsed.path)).name
202			fpath = Path(destdir, name)
203			is_url = url.startswith('https://') or url.startswith('http://')
204			if fpath.exists() and not overwrite:
205			log.info("%s to be %s to %s which already exists and overwrite is False" % (url, 'downloaded' if is_url else 'copied', fpath))
206			return fpath
207			destdir.mkdir(parents=True, exist_ok=True)
208			if resource_type == 'file':
209			if is_url:
210			self._download_impl(url, fpath, progress_cb)
211			else:
212			self._copy_impl(url, fpath, progress_cb)
213			elif resource_type == 'tarball':
214			with pushd_popd(tempdir=True):
215			if is_url:
216			self._download_impl(url, 'download.tar.xx', progress_cb, size)
217			else:
218			self._copy_impl(url, 'download.tar.xx', progress_cb)
219			Path('out').mkdir()
220			with pushd_popd('out'):
221			log.info("Extracting tarball")
222			with open_tarfile('../download.tar.xx', 'r:*') as tar:
223			tar.extractall()
224			log.info("Copying '%s' from tarball to %s" % (path_in_archive, fpath))
225			copytree(path_in_archive, str(fpath))
226			# TODO
227			# elif resource_type == 'github-dir':
228			return fpath
229

OCR-D / core

Pull Request — master (#680)

OcrdResourceManager.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

OcrdResourceManager.init() A