Skip to content

Instantly share code, notes, and snippets.

@onyb
Last active November 8, 2015 13:59
  • Star 0 You must be signed in to star a gist
  • Fork 1 You must be signed in to fork a gist
Star You must be signed in to star a gist
Embed
What would you like to do?
Open Cosmics scripts
from pprint import pprint
from astropy.io import fits
hdulist = fits.open('new.fits', mode='update')
hdulist[0].header['ANI'] = "latitude longitude altitude"
pprint(hdulist[0].header)
pprint(hdulist[0].data)
hdulist.flush()
hdulist.info()
from PIL import Image
from astropy.io import fits
import numpy as np
import cv2
from pprint import pprint
img = cv2.imread('2AdUsE75QiDi')
#data = np.asarray(img, dtype="int64")
hdu = fits.PrimaryHDU(img)
hdulist = fits.HDUList([hdu])
hdulist[0].header['LOCATION'] = "46.2323538 6.0557501 417.6"
hdulist[0].header['WINDOW'] = 0
hdulist.writeto('new.fits')
#hdulist.info()
#pprint(hdulist[0].header)
hdulist = fits.open('new.fits')
scidata = hdulist[0].data
cv2.imwrite('foo.jpg', scidata)
pymongo
astropy
numpy
enki
pandas
from pymongo import MongoClient
from bson.objectid import ObjectId
import requests
import urllib
from pprint import pprint
from datetime import datetime
from astropy.io import fits
import cv2
import enki
connection = MongoClient('mongodb://localhost:27017')
db=connection.opencosmics
stations = db.stations
class Station(object):
station = {}
name = None
picture = None
cluster = None
subcluster = None
country = None
location = {}
livetime = None
status = None
free = []
events = []
def populate_station(self):
raise Exception("Not implemented!")
def build_events(self):
raise Exception("Not implemented!")
class SCool(Station):
def __init__(self):
self.station = {}
self.station["name"] = "Master Cloud Chamber"
self.station["picture"] = "https://files.slack.com/files-pri/T08EPG8G1-F08FPUB9S/bmbiabyigaakxed.jpg"
self.station["cluster"] = "CERN, Geneva"
self.station["subcluster"] = "S'Cool Lab"
self.station["country"] = "Switzerland"
self.station["location"] = {
"latitude" : 46.2323538,
"longitude" : 6.0557501,
"altitude" : 417.6}
self.station["livetime"] = datetime(2015, 7, 31, 18, 43, 33, 4313)
self.station["status"] = "alive"
self.station["free"] = []
self.station["events"] = self.build_events()
def populate_station(self):
stations.insert(self.station)
def build_events(self):
_events = requests.get('http://crowdcrafting.org/api/task?project_id=3211').json()
events = []
for _event in _events:
event = {
'taskid': _event['id'],
'images': {"thumb": _event['info']['url_m'], "original": _event['info']['url_b']},
'window': False,
'time': [datetime.now()],
'filePath': str(_event['id']) + '.fits'
}
events.append(event)
pprint(events)
return events
class FITS_maker(object):
@staticmethod
def get_station_info(station_id):
station = stations.find_one({"_id": ObjectId(station_id)})
return station
@staticmethod
def get_event_info(event_id):
task = requests.get('http://crowdcrafting.org/api/task/%s'%(event_id)).json()
event = stations.find_one({"events.taskid": event_id}, {"events.$":1})['events'][0]
return event
@staticmethod
def generate_FITS(event_id, station_id):
station = FITS_maker.get_station_info(station_id)
event = FITS_maker.get_event_info(event_id)
img_url = event['images']['original']
urllib.urlretrieve(img_url, "%s.jpg"%(event_id))
img_na = cv2.imread("%s.jpg"%(event_id))
hdu = fits.PrimaryHDU(img_na)
hdulist = fits.HDUList([hdu])
hdulist[0].header['STATION_ID'] = str(station['_id'])
hdulist[0].header['STATION_NAME'] = station['name']
hdulist[0].header['CLUSTER'] = station['cluster']
hdulist[0].header['SUBCLUSTER'] = station['subcluster']
hdulist[0].header['COUNTRY'] = station['country']
hdulist[0].header['LOCATION'] = " ".join(station['location'])
hdulist[0].header['LIVETIME'] = station['livetime'].isoformat()
hdulist[0].header['STATUS'] = station['status']
hdulist[0].header['EVENT_ID'] = str(event['taskid'])
hdulist[0].header['EVENT_WINDOW'] = event['window']
event['time'] = [each.isoformat() for each in event['time']]
hdulist[0].header['TIME'] = " ".join(event['time'])
hdulist.writeto('%s.fits'%(event_id))
scool_station = SCool()
scool_station.populate_station()
#cosmic_obj = FITS_maker()
#cosmic_obj.generate_FITS(1069294, "55bdda20c7ac216d2f8d34db")
class Daemon(object):
@staticmethod
def update_FITS(station_id):
_events = requests.get('http://crowdcrafting.org/api/task?project_id=3211').json()
#_event = stations.find({"events.taskid": event_id}, {"events.$":1})['events'][0]
for _event in _events:
print("Updating event: ", _event['id'])
cosmic_obj = FITS_maker()
cosmic_obj.generate_FITS(_event['id'], station_id)
d = Daemon()
d.update_FITS("55be1ec25770e66e74d4504e")
d = Daemon()
d.update_FITS("55be0996c7ac21737be13b61")
d = Daemon()
d.update_FITS("55be249d7f792872b40237b0")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment