#!/usr/bin/python # usb_drive_sensor.py # --copyright-- Copyright 2011 (C) Tranzoa, Co. All rights reserved. Warranty: You're free and on your own here. This code is not necessarily up-to-date or of public quality. # --url-- http://www.tranzoa.net/tzpython/ # --email-- pycode is the name to send to. tranzoa.com is the place to send to. # --bodstamps-- # November 5, 2011 bar # November 6, 2011 bar get_thumb_drive_path() # November 7, 2011 bar glob # November 19, 2011 bar pause() and resume() # November 26, 2011 bar tighter error control and such # November 29, 2011 bar pyflake cleanup # April 11, 2012 bar fail_cnt # May 13, 2012 bar don't find "./floppy..." - treat it like "./cdrom..." # --eodstamps-- ## \file # # # This logic senses USB thumb drives and allows the caller to write files to them. # # The "drive" is sensed by detecting something in the PATH directory. # If PATH (as optionally given at instantiate time) has a trailing slash, then this logic looks for the given REQUIRED_DIR in sub-directories of PATH. # Or, put another way, under Unix, this logic looks for /media/xxxxx/usb_data and under Windows it looks for F:/usb_data, # or, if the caller spec's a PATH of ".", this logic looks for "./usb_data". # # import glob import os import Queue import sys import threading import time import replace_file import tzlib if sys.platform == 'win32' : PATH = "F:" else : PATH = "/media/" # note: slash on the end means that we'll look for our thumb drive data dir in 1st level sub-dirs of this directory (unless we find the REQUIRED_DIR in this dir) POLL_TIME = 3.14146 # how often we check for a drive REQUIRED_DIR = "usb_data" # the drive must have this directory in its root directory for us to see the drive class a_usb_sensor(object) : # # Note: These callbacks are called from a thread. # def thumb_drive(me, td) : # stub for callback for when there is a change in the state of thumb-drive-present-ness. 'td' is "" if no thumb drive is present, or 'td' is the path to the thumb drive's data directory. pass def log(me, s, flush = False) : # stub for log output of string callback pass def sync(me) : # stub for callback to "sync" the drives - flush them out pass def fail(me, s) : # stub for callback to notify caller that something failed pass def __init__(me, path = None, required_dir = None, on_thumb_drive = None, on_sync = None, on_fail = None, on_log = None, make_required_dir = False) : me.path = path or PATH me.required_dir = REQUIRED_DIR if (required_dir is None) else required_dir me.on_thumb_drive = on_thumb_drive or me.thumb_drive me.on_sync = on_sync or me.sync me.on_log = on_log or me.log me.on_fail = on_fail or me.fail me.make_required_dir = make_required_dir or False me.lock = threading.RLock() me._stop = 0 me._pause = False me.busy = False me.td = None me.th = None me.q = Queue.Queue() me.fail_cnt = 0 def _write_file(me, fn, fd = None) : t = time.time() if fd == None : if os.path.isfile(fn) : try : fd = tzlib.read_whole_binary_file(fn) t = os.path.getmtime(fn) except ( IOError, OSError ) : me.fail_cnt += 1 fd = None # this can happen if there is no .png file s = "Copy to usb drive failed to read [%s]" % fn me.on_log(s) me.on_fail(s) me.on_thumb_drive("") pass pass if fd : fn = os.path.join(me.td, os.path.basename(fn)) if not os.path.isfile(fn) : try : tfn = fn + ".tmp" tzlib.write_whole_binary_file(tfn, fd) replace_file.replace_file(fn, tfn, fn + ".bak") # note: since it doesn't exist, this is not needed. But it's here if the code changes in the future. os.utime(fn, ( os.path.getatime(fn), int(t) ) ) s = "Copy to usb drive wrote [%s]:%u" % ( fn, len(fd) ) me.on_log(s) except ( IOError, OSError ) : me.fail_cnt += 1 s = "Copy to usb drive failed to write [%s]:%u" % ( fn, len(fd) ) me.on_log(s) me.on_fail(s) me.on_thumb_drive("") pass pass pass def poll_thumb_drive(me) : """ Run every so often to sense the drive and to write any files out that are in our queue. """ me.lock.acquire() me.th = None if (not me._stop) and (not me._pause) : try : td = me.td if (not td) or (not os.path.isdir(td)) : td = "" d = os.path.join(me.path, me.required_dir) if os.path.isdir(d) : td = d elif me.path.endswith("/") : bd = None bt = -(sys.maxint - 1) dd = None dt = -(sys.maxint - 1) da = [ d for d in glob.glob(me.path + "*") if os.path.isdir(d) and (not os.path.basename(d).startswith(".")) and (not os.path.basename(d).startswith("cdrom")) and (not os.path.basename(d).startswith("floppy")) ] for d in da : rd = os.path.join(d, me.required_dir) if os.path.isdir(rd) : if bt < os.path.getmtime(rd) : bt = os.path.getmtime(rd) # try to find the newest required_dir on the mounted drives bd = rd elif not bd : if dt < os.path.getmtime(d) : dt = os.path.getmtime(d) # try to find the newestly mounted or something usb_dir subdirectory (better: with newest sub-dir there? or with fewest subdirs?) dd = rd pass pass bd = bd or dd # drives with required_dir take precedence over drives with no required_dir if bd : td = bd if td and (not os.path.isdir(td)) : if me.make_required_dir : me.on_log("USB sensor making %s" % td) os.makedirs(td) else : td = "" pass if td and (not os.path.isdir(td)) : td = "" pass pass while not me._stop : # do anything we might need to do with the mounted drive try : qd = me.q.get_nowait() if qd : # ignore nops me.busy = True if td and (td == me.td) : # we can't write a file when there's no drive or it's changed, so toss all the queued items me._write_file(qd[0], qd[1]) pass pass except Queue.Empty : if me.busy : me.on_sync() break # nothing to do just now pass if me.td != td : me.td = td me.on_thumb_drive(td) # tell our owner that the drive is in or out now pass except ( IOError, OSError ) : me.fail_cnt += 1 e = sys.exc_info() me.on_thumb_drive("") s = "usb drive operation failed [%s] line %u" % ( str(e[1]), e[2].tb_frame.f_lineno ) me.on_log(s) me.on_fail(s) me.on_thumb_drive("") pass me.busy = False me.lock.release() me.start() def put_file_to_write(me, file_name, fd = None) : me.q.put((file_name, fd)) def is_idle(me) : return((not me.busy) and me.q.empty()) def get_fail_count(me) : return(me.fail_cnt) def get_thumb_drive_path(me) : return(me.td) def start(me) : me.lock.acquire() if not me._stop : me.th = threading.Timer(POLL_TIME, me.poll_thumb_drive) me.th.setDaemon(True) me.th.start() me.lock.release() def pause(me) : me._pause = True def resume(me) : me._pause = False def stop(me) : me.lock.acquire() me._stop = 1 ( me.th, th ) = ( None, me.th ) me.lock.release() if th : try : th.cancel() except RuntimeError : pass pass pass # a_usb_sensor if __name__ == '__main__' : def print_status(s) : print s print "Create ./%s in the current directory and I'll sense it." % REQUIRED_DIR me = a_usb_sensor(path = '.', on_thumb_drive = print_status) me.start() while True : pass me.stop() # # # eof