#!/usr/bin/env python import datetime import logging import os import re import shm import signal import struct import time import xmlrpclib class tc08: """ Class providing an python interface to a Pico Technology Ltd. TC-08 temperature logger using the tc08lnx Linux driver from Pico Technology. Requires tc08lnx to be installed in /usr/local/bin. Also requires the non-standard python module shm (System V shared memory). """ # Format of the C struct used in the shared memory. __format = "ii8ll" def __init__(self, channels, channel_names, portname, \ units='C', interval=60, name='tc08', id=42, master=True): """ Arguments: name: A string with which to label this instance. Used for logging purposes channels: Type of thermocouple connected to each channel, encoded as a string of eight characters, one per channel. Available types are B, E, J, K, R, S, T and X (microvolts). For unused channels enter '-'. channel_names: String containing a comma seperated list of labels for the channels. portname: String containing the device name of the serial port that the TC-08 is connected to, e.g. '/dev/ttyS0' units: Specifies the units, 'C' for celsius, 'K' for Kelvin. interval: Interval, in seconds, at which entries are added to the log name: String to indentify the particlar instance for logging purposes. id: Unique ID number used to generate the shared memory key. master: True is this instance is the 'master', and should start the tc08lnx driver. False if this is a slave instance, which should just connect to a pre-existing shared memory area and log file. """ # Begin logging self.logger = logging.getLogger(name) # Store the channel config string self.__channels = channels # Store the list of channel names self.__channel_names = channel_names.split(',') # Also create a dictionary for reverse lookups of channel numbers self.__channel_dict = dict(zip(self.channel_names, \ range(1, len(self.channel_names)+1))) # Determine what units to use. if units == 'C' or 'c': self.__units = 'C' elif units == 'K' or 'k': self.__units = 'K' else: self.logger.warning("Invalid units '%s' given! Using Celsius." \ % units) self.__units = 'C' # Generate a key for the shared memory self.__id = int(id) key = shm.ftok("/usr/local/bin/tc08lnx", self.__id) # Construct a unique file name for the temperature log based on the # date and the name of this tc08 instance. date = datetime.date.fromtimestamp(time.time()) filename = '%s_%s.log' % (date.strftime('%Y%m%d'), name) if os.path.exists(filename): i = 1 while os.path.exists(filename + '-%i' % i): i += 1 filename += '-%i' % i # Add the relevant config info to the log file as initial comment line try: logfile = file(filename, 'w') logfile.write("# %s %s %s\n" % (name, channels, channel_names)) logfile.close except IOError: self.logger.critical("Couldn't access log file %s for writing!" % \ filename) # Start the driver running self.logger.info('Connecting to TC-08 on port %s...' % portname) self.__pid = os.spawnlp(os.P_NOWAIT, "/usr/local/bin/tc08lnx", \ "tc08lnx", \ "-u", "%s" % units, \ "-t", channels, \ "-k", "%i" % key, \ "-o", "%s" % filename, \ "-s", "%i" % int(interval), \ portname) self.__filename = filename # Check the state of the tc08lnx process. Should be either # running or sleeping. If the tc08 isn't connected to the port # specified, or some other problem, it'll be a zombie by now. ps_fd = os.popen('ps -p %i -o state=' % self.__pid) state = ps_fd.read(1) ps_fd.close() if state == 'Z' or state == 'X' or state == '': # tc08lnx is a zombie, or dead, or non-existant. # Noisily log this terrible event self.logger.critical('Driver failure for TC-08 on port %s!' \ % portname) # Get the id of the shared memory area # This sometimes fails for mysterious reasons. try: id = shm.getshmid(key) except KeyError: # Noisily log this disaster. self.logger.critical('Shm key failure for TC-08 on port %s!'\ % portname) # Create a shared memory object try: self.__memory = shm.memory(id) except shm.error: # Noisily log this disaster. self.logger.critical('Shm error for TC-08 on port %s!'\ % portname) # Store the size of the shared memory object self.__size = self.__memory.size # Attach the mem to address space of this process self.__memory.attach() def __del__(self): """ Ensure the tc08lnx driver gets killed. """ self.logger.debug('__del__() called') os.kill(self.__pid, signal.SIGINT) def getChannels(self): """ Returns the channel config string. """ self.logger.debug('getChannels() called.') return self.__channels def getColdJunction(self): """ Returns the current cold junction temperature. """ self.logger.debug('getColdJunction() called.') cjt = struct.unpack(tc08.__format, \ self.__memory.read(self.__size))[10]\ /100.0 if self.__units == 'K' or self.__units == 'k': cjt += 273.15 return cjt def getID(self): """ Returns the unique ID used to identify the shared memory block. """ self.logger.debug('getID() called') return self.__id def getName(self, number): """ Returns the channel label for a given channel number """ self.logger.debug('getName(%i) called.' % number) return self.__channel_names[number-1] def getNames(self): """ Returns the list of channel labels """ self.logger.debug('getNames() called.') return self.__channel_names def getNumber(self, name): """ Returns the channel number corresponding to a given label """ self.logger.debug('getNumber("%s") called.' % name) return self.channel_dict[name] def getTemps(self): """ Returns the current temperatures as an list. """ self.logger.debug('getTemps() called.') num_temps = len(self.channel_names) return [t/100.0 for t in \ struct.unpack(tc08.__format, \ self.__memory.read(self.__size))[2:2+num_temps]] def getUnits(self): """ Returns the temperature units in use, either 'C' for celsius or 'K' for kelvin. """ def isActive(self): self.logger.debug('isActive() called') if struct.unpack(tc08.__format, self.__memory.read(self.__size))[1]: return True else: return False def isInitialised(self): self.logger.debug('isInitialised() called') if struct.unpack(tc08.__format, self.__memory.read(self.__size))[0]: return True else: return False