Command.get_file() - Code Metrics - Inspection of "Update LICENSE.txt" - usnistgov/ocr-pipeline - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Branch — master (e214b7)

by Philippe

created 2017-07-17 12:35 UTC

Command.get_file() A

↳ Parent: Command

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	1

Metric	Value
cc	1
c	1
b	0
f	1
dl	0
loc	9
rs	9.6666

"""Command object

.. Authors:
    Philippe Dessauw
    [email protected]

.. Sponsor:
    Alden Dima
    [email protected]
    Information Systems Group
    Software and Systems Division
    Information Technology Laboratory
    National Institute of Standards and Technology
    http://www.nist.gov/itl/ssd/is
"""
from apputils.fileop import zip_directory, unzip_directory
from pipeline.files import FileManager


class Command(object):
    """Main command object
    """

    def __init__(self, filename, logger, app_config):
        self.logger = logger

        ip = app_config["machines"]["master"][0].split('@')
        master_ip = ip[-1:][0]
        master_queue_port = app_config["redis"]["port"]
        self.fman = FileManager(master_ip, master_queue_port)

        self.filename = filename
        self.unzipped = None
        self.config = app_config

    def get_file(self):
        """Retrieve file from redis and unzip it to the local filesystem
        """
        # Get hash from redis
        self.logger.debug("Retrieving "+self.filename+"...")
        self.fman.retrieve_file(self.filename)

        # Write it in the tmp folder
        self._unzip_file()

    def store_file(self):
        """Zip file on the local filesystem and store it to redis
        """
        self._zip_file()

        # Store it in redis
        self.fman.store_file(self.filename)

    def _zip_file(self):
        """Check if the file can be zipped and zip it
        """
        if self.unzipped is None:
            self.logger.error("Zipped directory has not been unzipped")
            return

        zip_directory(self.unzipped)
        self.unzipped = None

    def _unzip_file(self):
        """Check if the file can be unzipped and unzip it
        """
        if self.unzipped is not None:
            self.logger.error("Archive already unzipped")
            return

        self.unzipped = unzip_directory(self.filename)


1			"""Command object
2
3			.. Authors:
4			Philippe Dessauw
5			[email protected]
6
7			.. Sponsor:
8			Alden Dima
9			[email protected]
10			Information Systems Group
11			Software and Systems Division
12			Information Technology Laboratory
13			National Institute of Standards and Technology
14			http://www.nist.gov/itl/ssd/is
15			"""
16			from apputils.fileop import zip_directory, unzip_directory
17			from pipeline.files import FileManager
18
19
20			class Command(object):
21			"""Main command object
22			"""
23
24			def __init__(self, filename, logger, app_config):
25			self.logger = logger
26
27			ip = app_config["machines"]["master"][0].split('@')
28			master_ip = ip[-1:][0]
29			master_queue_port = app_config["redis"]["port"]
30			self.fman = FileManager(master_ip, master_queue_port)
31
32			self.filename = filename
33			self.unzipped = None
34			self.config = app_config
35
36			def get_file(self):
37			"""Retrieve file from redis and unzip it to the local filesystem
38			"""
39			# Get hash from redis
40			self.logger.debug("Retrieving "+self.filename+"...")
41			self.fman.retrieve_file(self.filename)
42
43			# Write it in the tmp folder
44			self._unzip_file()
45
46			def store_file(self):
47			"""Zip file on the local filesystem and store it to redis
48			"""
49			self._zip_file()
50
51			# Store it in redis
52			self.fman.store_file(self.filename)
53
54			def _zip_file(self):
55			"""Check if the file can be zipped and zip it
56			"""
57			if self.unzipped is None:
58			self.logger.error("Zipped directory has not been unzipped")
59			return
60
61			zip_directory(self.unzipped)
62			self.unzipped = None
63
64			def _unzip_file(self):
65			"""Check if the file can be unzipped and unzip it
66			"""
67			if self.unzipped is not None:
68			self.logger.error("Archive already unzipped")
69			return
70
71			self.unzipped = unzip_directory(self.filename)
72

usnistgov / ocr-pipeline

Branch — master (e214b7)

Command.get_file() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like