4
\$\begingroup\$

I'm trying to create a filewatcher program that can watch different files. The files might have subfiles that are being imported into the main file, which is being represented by the file object. I'm feeling the code is a little messy and I would like to know how I can improve it.

The input_watcher() is not there to stay, it is being replaced by a GUI eventually, but it's just there for testing.

import multiprocessing
import threading
import os
import time


file_lock = threading.Lock()
update_interval = 0.1


class FileMethods(object):
    def a_import(self):
        # do some sort of file manipulation

    # all different kinds of filewatching / compiling types
    # will be added as methods and also added to the compiler_methods




class FileObject(FileMethods):
    def __init__(self, full_name, compiler):

        self.full_name = os.path.abspath(full_name)
        self.file_name = os.path.basename(self.full_name)
        self.path_name = os.path.dirname(self.full_name)

        name, exstention = os.path.splitext(full_name)
        self.concat_name = name + '-concat' + exstention

        self.compiler = compiler
        self.compiler_methods = {'aImport': self.a_import}

        self.last_updated = os.path.getatime(self.full_name)

        self.subfiles = []
        self.last_subfiles_mod = {}

    def exists(self):
        return os.path.isfile(self.full_name)

    def mod_check(self):
        if self.last_updated < os.path.getmtime(self.full_name):
            self.last_updated = os.path.getmtime(self.full_name)
            print '%s updated' % self.file_name
            return True
        else:
            return False

    def sub_mod_check(self):
        for s in self.subfiles:
            if self.last_subfiles_mod.get(s) < os.path.getmtime(s):
                self.last_subfiles_mod[s] = os.path.getmtime(s)
                return True

        return False


files = []


def input_watcher():
    while True:
        input_file = os.path.abspath(raw_input('Input file name: '))
        compiler = raw_input('Choose compiler: ')

        if os.path.isfile(input_file):

            obj = FileObject(input_file, compiler)

            with file_lock:
                files.append(obj)

            print 'Adding %s with %s as compiler' % (obj.file_name, obj.compiler)
        else:
            print 'File does not exists'


def file_manipulation():
    for f in files:
        p = multiprocessing.Process(target=f.compiler_methods.get(f.compiler)())
        p.start()


def file_watcher():
    while True:
        with file_lock:
            file_manipulation()
        time.sleep(update_interval)

if __name__ == '__main__':
    iw = threading.Thread(target=input_watcher)
    fw = threading.Thread(target=file_watcher)

    iw.start()
    fw.start()
\$\endgroup\$
  • \$\begingroup\$ I think what you're trying to do is similar to peat, so I'd recommend taking a look at it. Once you're done with the check part, you can have a look at doit for the part about tracking dependencies (that's what I understand it's the goal of having subfiles). \$\endgroup\$ – jcollado Jul 27 '14 at 3:19
1
\$\begingroup\$

Sort your imports alphabetically. Make it easier to find. - Core libs first, then 3rd party.

import multiprocessing
import threading
import os
import time

Constants are typically in all caps.

file_lock = threading.Lock()
update_interval = 0.1

If you have some TODOs, a common practice is to use code tags.

# FIXME:
# TODO:

It's not a must standard, but it is something that is easily recognizable.

class FileMethods(object):
    def a_import(self):
        # do some sort of file manipulation

Ditto on the docstring. Include a desc of your attributes as well.

class FileObject(FileMethods):
    def __init__(self, full_name, compiler):

I would change

self.concat_name = name + '-concat' + exstention

self.concat_name = ''.join([name, '-concat', extension]) # Assuming extension was a typo.

I don't see this method being used anywhere. You also have an if statement further down in your code. I would remove this if your not using it. Also, I probably wouldn't create a method for this unless I had to write this out more than 2 times. I only see one instance in your code.

def exists(self):
    return os.path.isfile(self.full_name)

Is this script meant to run interactively? If so, I would suggest adding an interactive flag.

So you can run this in automated fashion (maybe with a config file), or a -i so I can be prompted for input. Also, I would change your print statements to the nicer print statements. They also add more flexibility. Consider using argparse for this.

print '%s updated' % self.file_name

print '{} updated'.format(self.file_name)

Docstrings

def sub_mod_check(self):

Ditto on the print statements:

Use .format

        print 'Adding %s with %s as compiler' % (obj.file_name, obj.compiler)
    else:
        print 'File does not exists'

Consider using a main function or method to wrap your logic into when calling your script.

if __name__ == '__main__':
    iw = threading.Thread(target=input_watcher)
   s fw = threading.Thread(target=file_watcher)

if __name__ == '__main__':
    main()
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.