Coverage for sherlock/imports/ifs.py: 86%
69 statements
« prev ^ index » next coverage.py v7.2.2, created at 2023-10-10 13:58 +0000
« prev ^ index » next coverage.py v7.2.2, created at 2023-10-10 13:58 +0000
1#!/usr/local/bin/python
2# encoding: utf-8
3"""
4*Import Multi Unit Spectroscopic Explorer (MUSE) IFS galaxy stream into sherlock-catalogues database*
6:Author:
7 David Young
8"""
9from __future__ import print_function
10from ._base_importer import _base_importer
11from fundamentals.download import multiobject_download
12from astrocalc.coords import unit_conversion
13from docopt import docopt
14import re
15import requests
16import string
17import codecs
18import pickle
19import glob
20import readline
21import sys
22import os
23os.environ['TERM'] = 'vt100'
26class ifs(_base_importer):
27 """
28 *Importer for the Multi Unit Spectroscopic Explorer (MUSE) IFS galaxy catalogue stream*
30 **Key Arguments**
32 - ``log`` -- logger
33 - ``settings`` -- the settings dictionary
36 **Usage**
38 To import the IFS catalogue stream into the sherlock-catalogues database, run the following:
41 ```python
42 from sherlock.imports import IFS
43 ```
45 stream = IFS(
46 log=log,
47 settings=settings
48 )
49 stream.ingest()
51 .. todo ::
53 - abstract this module out into its own stand alone script
54 - check sublime snippet exists
55 """
56 # INITIALISATION
58 def ingest(self):
59 """*Import the IFS catalogue into the sherlock-catalogues database*
61 The method first generates a list of python dictionaries from the IFS datafile, imports this list of dictionaries into a database table and then generates the HTMIDs for that table.
63 **Usage**
65 See class docstring for usage
67 """
68 self.log.debug('starting the ``get`` method')
70 self.primaryIdColumnName = "primaryId"
71 self.raColName = "raDeg"
72 self.declColName = "decDeg"
73 self.dbTableName = "tcs_cat_ifs_stream"
74 self.databaseInsertbatchSize = 500
76 dictList = self._create_dictionary_of_IFS()
78 tableName = self.dbTableName
79 createStatement = """
80 CREATE TABLE IF NOT EXISTS `%(tableName)s` (
81 `primaryId` bigint(20) NOT NULL AUTO_INCREMENT COMMENT 'An internal counter',
82 `dateCreated` datetime DEFAULT CURRENT_TIMESTAMP,
83 `decDeg` double DEFAULT NULL,
84 `name` varchar(100) COLLATE utf8_unicode_ci DEFAULT NULL,
85 `raDeg` double DEFAULT NULL,
86 `z` double DEFAULT NULL,
87 `htm16ID` bigint(20) DEFAULT NULL,
88 `htm10ID` bigint(20) DEFAULT NULL,
89 `htm13ID` bigint(20) DEFAULT NULL,
90 `dateLastModified` datetime DEFAULT CURRENT_TIMESTAMP,
91 `updated` varchar(45) DEFAULT '0',
92 PRIMARY KEY (`primaryId`),
93 UNIQUE KEY `radeg_decdeg` (`raDeg`,`decDeg`),
94 KEY `idx_htm16ID` (`htm16ID`),
95 KEY `idx_htm10ID` (`htm10ID`),
96 KEY `idx_htm13ID` (`htm13ID`)
97 ) ENGINE=MyISAM AUTO_INCREMENT=0 DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci;
98""" % locals()
100 self.add_data_to_database_table(
101 dictList=dictList,
102 createStatement=createStatement
103 )
105 self.log.debug('completed the ``get`` method')
106 return None
108 def _create_dictionary_of_IFS(
109 self):
110 """*Generate the list of dictionaries containing all the rows in the IFS stream*
112 **Return**
114 - ``dictList`` - a list of dictionaries containing all the rows in the IFS stream
117 **Usage**
119 ```python
120 from sherlock.imports import IFS
121 stream = IFS(
122 log=log,
123 settings=settings
124 )
125 dictList = stream._create_dictionary_of_IFS()
126 ```
128 """
129 self.log.debug(
130 'starting the ``_create_dictionary_of_IFS`` method')
132 # GRAB THE CONTENT OF THE IFS CSV
133 try:
134 response = requests.get(
135 url=self.settings["ifs galaxies url"],
136 )
137 thisData = response.content
138 thisData = str(thisData).split("\n")
139 status_code = response.status_code
140 except requests.exceptions.RequestException:
141 print('HTTP Request failed')
142 sys.exit(0)
144 dictList = []
145 columns = ["name", "raDeg", "decDeg", "z"]
147 for line in thisData:
148 thisDict = {}
149 line = line.strip()
150 line = line.replace("\t", " ")
151 values = line.split("|")
152 if len(values) > 3:
153 thisDict["name"] = values[0].strip()
155 # ASTROCALC UNIT CONVERTER OBJECT
156 converter = unit_conversion(
157 log=self.log
158 )
159 try:
160 raDeg = converter.ra_sexegesimal_to_decimal(
161 ra=values[1].strip()
162 )
163 thisDict["raDeg"] = raDeg
164 decDeg = converter.dec_sexegesimal_to_decimal(
165 dec=values[2].strip()
166 )
167 thisDict["decDeg"] = decDeg
168 except:
169 name = thisDict["name"]
170 self.log.warning(
171 'Could not convert the coordinates for IFS source %(name)s. Skipping import of this source.' % locals())
172 continue
173 try:
174 z = float(values[3].strip())
175 if z > 0.:
176 thisDict["z"] = float(values[3].strip())
177 else:
178 thisDict["z"] = None
179 except:
180 thisDict["z"] = None
181 dictList.append(thisDict)
183 self.log.debug(
184 'completed the ``_create_dictionary_of_IFS`` method')
185 return dictList
187 # use the tab-trigger below for new method
188 # xt-class-method