#--------------------------------------------------------------------- # Python example to create continuously updated CSV logfiles from # Fusion MicroGC calibrated method runs. # # Copyright(c) 2014 INFICON This open source example may be freely # used and modified for any purpose under terms of the MIT Licence # http://opensource.org/licenses/MIT # from __future__ import print_function import json import requests import time import sys import os.path import dateutil.parser from dateutil import tz import logging #--------------------------------------------------------------------- # IP address of MicroGC Fusion instrument. host = '10.210.107.4' #--------------------------------------------------------------------- # Folder for CSV files filePath = 'C:\\Users\\swilson1\\Desktop\\CSV_Export' #--------------------------------------------------------------------- csvName = '' #Add a name here to catch all files, regardless of sample name. This will not make a new file when changing sample names #--------------------------------------------------------------------- def getLastRunLocation(): lastRunLocation = 'http://' + host + '/v1/lastRun' try: response = requests.get(lastRunLocation) lastRun = response.json() return lastRun['dataLocation'] except KeyboardInterrupt: print("Exiting on interrupt.") sys.exit() except: # catchall e = sys.exc_info()[0] print("Error retreiving lastRunLocation from " + lastRunLocation) print(e) return "" def exportCSV(runData, sampleName, filePath, csvName, tags): fileLocation = filePath+"\\"+sampleName+'.csv' #creates the file location and name of file if csvName != '': fileLocation = filePath+"\\"+csvName+'.csv' #If the csv Name is added, the file should change, but the sample name remains the same. # Headers: TimeStamp,Concentration, Normalized Concentration, Area, RT headers = ['Timestamp', 'Sample Name','Tags'] #These are the starting headers at the first row, and first 3 columns utcTime = dateutil.parser.parse(str(runData['runTimeStamp'])) # Timestamp from run data file localTime = utcTime.astimezone(tz.tzlocal()) #timezone conversion timestamp = localTime.strftime('%Y/%m/%d %H:%M:%S') #Timestamp conversion in excel readable format data = [timestamp, sampleName,tags] #Add first 3 columns of data to the appended row (sample name, timestamp and tags) detectors = [] #Creates a list of the possible detectors in the system for detector in runData['detectors']: #search the run data['detectors'] for the names of the detectors detectors.append(detector) #Add the detector name to the list of detectors detectors.sort() #Sort by detector names so that they will always be in the same format for detector in detectors: #Search for peaks under the first detector try: peaks = runData['detectors'][detector]['analysis']['peaks'] #If a peak is found, it will be represented at this location except: print('No peaks found') #Print a message if peak location is empty break for peak in peaks: #search through individual peaks if 'label' in peak: #If "label" is found, this is a calibrated peak by the user try: label = peak['label'] #name this peak via the user input value height = round(peak['height'],0) #Get the peak height, round the value area = round(peak['area'],0) # Get the peak area, round the value rt = peak['top'] #Get the "top", which represent the retention time of the peak try: concentration = round(peak['concentration'],6) #Try to get the calculated peak concentration, round the value except: concentration = '-' #If the peak concentration is missing (not calibrated properly), enter a "-" as the value try: normConc = round(peak['normalizedConcentration'],6) #Try to get the unNormalized Concentration, round the value except: normConc = '-' # if the peak Normalized concentration is missing, enter a "-" as the value except: print("Error retrieving information from " + peak['label']) #catch all if an error occurs when attempting to gether the peak data continue #Continue trying to gather data on the rest of the peaks headers.append(label+' Concentration,'+label+' Normalized Concentration,'+label+' Area,'+label+' RT') #Create the data headers, which includes the peak name and data description data.append(str(concentration)) #Add the concentration data to the row data data.append(str(normConc)) #Add the normalized Concentration to the row data data.append(str(area)) #Add the raw area to the row data data.append(str(rt)) #Add the Retention Time to the row data # pprint.pprint(data) #uncomment this out if you want the data to print out into the command window after each run. if os.path.isfile(fileLocation): #Check to see if the file path / file exists try: with open(fileLocation, 'a') as file: #if the file already exists, open the file file.write(",".join(data)) #Write the data to the file file.write('\n') #Append the row of the file (for next time) file.close() #Close the file print('File appended') #Print a notification that the file was appended except: print('File might be open, trying again in 15 seconds') #exception error, if the file can not be written to, the file is likely open. time.sleep(15) #Allow the users 15 seconds to close the file try: with open(fileLocation, 'a') as file: #Try to write the data again file.write(",".join(data)) file.write('\n') file.close() print('File appended') except Exception as e: #Print the error and continue exit the function, if the error happens again print("Error: "+str(e)) return else: #Else, meaning the file does not exist yet try: with open(fileLocation, 'w') as file: #Try to create a new file file.write(','.join(headers)) #Add the headers to row 1 file.write('\n') #Move to the next line in the CSV file file.write(",".join(data)) #Write the data to row 2 file.write('\n') #Move down to the next line for next time file.close() #Close the file print('New file created') #Notify the user that a new file was created except Exception as e: print('Error creating new file') #Send an error if the file can not be created for some reason print(e) #Print the error into the command window return #Return to the main loop, continue monitoring #Main Program starts here newRunLocation = getLastRunLocation() oldRunLocation = newRunLocation print('Logging analysis from instrument ' + host) while True: newRunLocation = getLastRunLocation() if newRunLocation == oldRunLocation: #Check here for new runs continue #No new runs, continue in loop oldRunLocation = newRunLocation #there is a new run, set the old location to the new location runData = requests.get('http://' + host +newRunLocation).json() #get all of the run data from the system try: sampleName = runData['annotations']['name'] #Try to get sample name from run data tags = str(runData['annotations']['tags']).replace(',','/') #Get sample tags from run data, replace the "," in the list with a "/" so that it will not mess up the CSV file except: sampleName = runData['methodName'] #if "annotations do not exist in JSON file, default the sample name to the method name" tags = '' #if annotations do no exist, no tags were added exportCSV(runData,sampleName,filePath,csvName,tags) #Send the run data to the "Export CSV" function so that the new data can be processed and written to the csv file time.sleep(3) #Sleep for 3 seconds until pinging the system for a new run