### Retrieve spectral data using SparclClient Source: https://sparclclient.readthedocs.io/en/latest/_modules/sparcl/client.html Examples of initializing the client and retrieving spectral data for specific IDs. ```python specids = [394069118933821440, 1355741587413428224, 1355617892355303424, 1355615143576233984, 1355661872820414464, 1355755331308775424, 1355716848401803264] client = client.SparclClient(url='http://localhost:8030/sparc') client.retrieve(specids)[0].keys() # >> dict_keys(['flux','loglam']) data0 = client.retrieve(specids,columns='flux') f'{len(str(data0)):,}' # -> '3,435,687' dataall = client.retrieve(specids,columns=allc) f'{len(str(dataall)):,}' # -> '27,470,052' ``` -------------------------------- ### Check SPARCL version via HTTPie Source: https://sparclclient.readthedocs.io/en/latest/_modules/sparcl/client.html Example command to check the server version using HTTPie. ```bash http :8030/sparc/version ``` -------------------------------- ### Get Available Fields Source: https://sparclclient.readthedocs.io/en/latest/sparcl.html Retrieves a list of all fields available from data sets in the SPARCL database. ```python >>> client = SparclClient(announcement=False) >>> sorted(client.get_available_fields()) ['data_release', 'datasetgroup', 'dateobs', 'dateobs_center', 'dec', 'exptime', 'extra_files', 'file', 'flux', 'instrument', 'ivar', 'mask', 'model', 'ra', 'redshift', 'redshift_err', 'redshift_warning', 'site', 'sparcl_id', 'specid', 'specprimary', 'spectype', 'survey', 'targetid', 'telescope', 'updated', 'wave_sigma', 'wavelength', 'wavemax', 'wavemin'] ``` -------------------------------- ### Get Default Fields Source: https://sparclclient.readthedocs.io/en/latest/sparcl.html Retrieves fields tagged as 'default' that are used for the default include parameter in retrieval operations. ```python >>> client = SparclClient(announcement=False) >>> client.get_default_fields() ['dec', 'flux', 'ra', 'sparcl_id', 'specid', 'wavelength'] ``` -------------------------------- ### Importing specutils and dependencies Source: https://sparclclient.readthedocs.io/en/latest/_modules/sparcl/specutils.html Initializes the module by importing necessary libraries, including handling version differences for the Spectrum class. ```python """Functions for converting SPARCL results to specutils objects. """ import numpy as np try: # specutils >= 2.0 from specutils import Spectrum except ImportError: from specutils import Spectrum1D as Spectrum from specutils import SpectrumCollection, SpectrumList from astropy.nddata import InverseVariance import astropy.units as u ``` -------------------------------- ### get_default_fields(dataset_list=None) Source: https://sparclclient.readthedocs.io/en/latest/sparcl.html Get fields tagged as 'default' that are in DATASET_LIST. ```APIDOC ## get_default_fields(dataset_list=None) ### Description Get fields tagged as 'default' that are in DATASET_LIST. These are the fields used for the DEFAULT value of the include parameter of client.retrieve(). ### Parameters - **dataset_list** (list) - Optional - List of data sets from which to get the default fields. Defaults to None. ### Returns - **list** - List of fields tagged as 'default'. ``` -------------------------------- ### Run tests against PAT server Source: https://sparclclient.readthedocs.io/en/latest/_modules/sparcl/client.html Commands to set the server URL and execute tests against the PAT server. ```bash export serverurl=https://sparc1.datalab.noirlab.edu/ python -m unittest tests.tests_api ``` -------------------------------- ### Initialize Counter Objects Source: https://sparclclient.readthedocs.io/en/latest/_modules/collections.html Demonstrates various ways to instantiate a Counter object using iterables, mappings, or keyword arguments. ```python >>> c = Counter() # a new, empty counter >>> c = Counter('gallahad') # a new counter from an iterable >>> c = Counter({'a': 4, 'b': 2}) # a new counter from a mapping >>> c = Counter(a=4, b=2) # a new counter from keyword args ``` -------------------------------- ### Retrieve spectra by SPARCL ID Source: https://sparclclient.readthedocs.io/en/latest/sparcl.html Demonstrates how to initialize the client and retrieve records using a list of SPARCL IDs. ```python >>> client = SparclClient(announcement=False) >>> ids = client.find(limit=1).ids >>> inc = ['sparcl_id', 'flux', 'wavelength', 'model'] >>> ret = client.retrieve(uuid_list=ids, include=inc) >>> type(ret.records[0].wavelength) ``` -------------------------------- ### Initialize SparclClient Source: https://sparclclient.readthedocs.io/en/latest/sparcl.html Instantiate the client with announcements disabled. ```python >>> client = SparclClient(announcement=False) ``` -------------------------------- ### Define namedtuple factory function Source: https://sparclclient.readthedocs.io/en/latest/_modules/collections.html The implementation of namedtuple, which creates a subclass of tuple with named fields. Includes examples of usage in the docstring. ```python def namedtuple(typename, field_names, *, rename=False, defaults=None, module=None): """Returns a new subclass of tuple with named fields. >>> Point = namedtuple('Point', ['x', 'y']) >>> Point.__doc__ # docstring for the new class 'Point(x, y)' >>> p = Point(11, y=22) # instantiate with positional args or keywords >>> p[0] + p[1] # indexable like a plain tuple 33 >>> x, y = p # unpack like a regular tuple >>> x, y (11, 22) >>> p.x + p.y # fields also accessible by name 33 >>> d = p._asdict() # convert to a dictionary >>> d['x'] 11 >>> Point(**d) # convert from a dictionary Point(x=11, y=22) >>> p._replace(x=100) # _replace() is like str.replace() but targets named fields Point(x=100, y=22) """ # Validate the field names. At the user's option, either generate an error # message or automatically replace the field name with a valid name. if isinstance(field_names, str): field_names = field_names.replace(',', ' ').split() field_names = list(map(str, field_names)) typename = _sys.intern(str(typename)) if rename: seen = set() for index, name in enumerate(field_names): if (not name.isidentifier() or _iskeyword(name) or name.startswith('_') or name in seen): field_names[index] = f'_{index}' seen.add(name) for name in [typename] + field_names: if type(name) is not str: raise TypeError('Type names and field names must be strings') if not name.isidentifier(): raise ValueError('Type names and field names must be valid ' f'identifiers: {name!r}') if _iskeyword(name): raise ValueError('Type names and field names cannot be a ' f'keyword: {name!r}') seen = set() for name in field_names: if name.startswith('_') and not rename: raise ValueError('Field names cannot start with an underscore: ' f'{name!r}') if name in seen: raise ValueError(f'Encountered duplicate field name: {name!r}') seen.add(name) field_defaults = {} if defaults is not None: defaults = tuple(defaults) if len(defaults) > len(field_names): raise TypeError('Got more default values than field names') field_defaults = dict(reversed(list(zip(reversed(field_names), reversed(defaults))))) # Variables used in the methods and docstrings field_names = tuple(map(_sys.intern, field_names)) num_fields = len(field_names) arg_list = ', '.join(field_names) if num_fields == 1: arg_list += ',' repr_fmt = '(' + ', '.join(f'{name}=%r' for name in field_names) + ')' tuple_new = tuple.__new__ _dict, _tuple, _len, _map, _zip = dict, tuple, len, map, zip # Create all the named tuple methods to be added to the class namespace namespace = { '_tuple_new': tuple_new, '__builtins__': {}, '__name__': f'namedtuple_{typename}', } code = f'lambda _cls, {arg_list}: _tuple_new(_cls, ({arg_list}))' __new__ = eval(code, namespace) __new__.__name__ = '__new__' __new__.__doc__ = f'Create new instance of {typename}({arg_list})' if defaults is not None: __new__.__defaults__ = defaults @classmethod def _make(cls, iterable): result = tuple_new(cls, iterable) if _len(result) != num_fields: raise TypeError(f'Expected {num_fields} arguments, got {len(result)}') return result ``` -------------------------------- ### Run tests against DEV server Source: https://sparclclient.readthedocs.io/en/latest/_modules/sparcl/client.html Command to execute tests against a local development server. ```bash serverurl=http://localhost:8050 python -m unittest tests.tests_api ``` -------------------------------- ### Retrieve spectra by specid Source: https://sparclclient.readthedocs.io/en/latest/sparcl.html Demonstrates how to retrieve specific spectra records using a list of specids. ```python >>> client = SparclClient(announcement=False) >>> sids = [4753625089450465280, 1254253099313293312] >>> inc = ['specid', 'flux', 'wavelength', 'model'] >>> ret = client.retrieve_by_specid(specid_list=sids, include=inc) >>> len(ret.records[0].wavelength) 4617 ``` -------------------------------- ### Initialize Retrieved class Source: https://sparclclient.readthedocs.io/en/latest/_modules/sparcl/Results.html Constructor for the Retrieved class which holds spectral records. ```python def __init__(self, dict_list, client=None): super().__init__(dict_list, client=client) ``` -------------------------------- ### SparclClient Constructor Source: https://sparclclient.readthedocs.io/en/latest/sparcl.html Initializes a new instance of the SparclClient to interface with the SPARCL Server. ```APIDOC ## SparclClient(url='https://astrosparcl.datalab.noirlab.edu', verbose=False, connect_timeout=1.1, read_timeout=5400, announcement=True) ### Description Initializes the client connection to the SPARCL server. Throws an exception if the client version is incompatible with the server. ### Parameters - **url** (str) - Optional - Base URL of SPARCL Server. - **verbose** (bool) - Optional - Set to True for detailed output. - **connect_timeout** (float) - Optional - Seconds to wait for connection. - **read_timeout** (float) - Optional - Seconds to wait for server response. - **announcement** (bool) - Optional - Enable or disable SPARCL announcements. ``` -------------------------------- ### Check API version Source: https://sparclclient.readthedocs.io/en/latest/sparcl.html Demonstrates how to retrieve the version of the Server REST API currently in use. ```python >>> client = SparclClient(announcement=False) >>> client.version 13.0 ``` -------------------------------- ### version Source: https://sparclclient.readthedocs.io/en/latest/_modules/sparcl/client.html Returns the version of the Server REST API used by the client. ```APIDOC ## version ### Description Returns the version of the Server REST API currently in use. If the version is not cached, it fetches it from the server. ### Returns - **float** - The API version number. ``` -------------------------------- ### to_SpectrumList(results, *, flux_unit=None, wave_unit=None) Source: https://sparclclient.readthedocs.io/en/latest/_modules/sparcl/specutils.html Converts retrieved SPARCL results into a specutils.SpectrumList object. ```APIDOC ## to_SpectrumList ### Description Converts a `sparcl.Results.Retrieved` object into a `specutils.SpectrumList` object. If units are not provided, they are automatically resolved. ### Parameters - **results** (sparcl.Results.Retrieved) - Required - The retrieved results to convert. - **flux_unit** (str) - Optional - Unit string for flux. If None, resolved via _get_units(). - **wave_unit** (str) - Optional - Unit string for wavelength. If None, resolved via _get_units(). ### Returns - **SpectrumList** (specutils.SpectrumList) - The converted spectrum list object. ``` -------------------------------- ### Run doctests Source: https://sparclclient.readthedocs.io/en/latest/_modules/sparcl/client.html Standard boilerplate for executing doctests when the module is run as a script. ```python if __name__ == "__main__": import doctest doctest.testmod() ``` -------------------------------- ### login(email, password=None) Source: https://sparclclient.readthedocs.io/en/latest/sparcl.html Login to the SPARCL service. ```APIDOC ## login(email, password=None) ### Description Login to the SPARCL service. ### Parameters - **email** (str) - Required - User login email. - **password** (str) - Optional - User SSO password. If not given, the output will prompt the user to enter in their SSO password. ### Returns - **None** ``` -------------------------------- ### OrderedDict Metadata and Views Source: https://sparclclient.readthedocs.io/en/latest/_modules/collections.html Methods for calculating memory usage and providing dictionary views. ```python def __sizeof__(self): sizeof = _sys.getsizeof n = len(self) + 1 # number of links including root size = sizeof(self.__dict__) # instance dictionary size += sizeof(self.__map) * 2 # internal dict and inherited dict size += sizeof(self.__hardroot) * n # link objects size += sizeof(self.__root) * n # proxy objects return size ``` ```python def keys(self): "D.keys() -> a set-like object providing a view on D's keys" return _OrderedDictKeysView(self) ``` ```python def items(self): "D.items() -> a set-like object providing a view on D's items" return _OrderedDictItemsView(self) ``` ```python def values(self): "D.values() -> an object providing a view on D's values" return _OrderedDictValuesView(self) ``` -------------------------------- ### OrderedDict Serialization and Creation Source: https://sparclclient.readthedocs.io/en/latest/_modules/collections.html Methods for pickling, copying, and creating instances from iterables. ```python def __reduce__(self): 'Return state information for pickling' inst_dict = vars(self).copy() for k in vars(OrderedDict()): inst_dict.pop(k, None) return self.__class__, (), inst_dict or None, None, iter(self.items()) ``` ```python def copy(self): 'od.copy() -> a shallow copy of od' return self.__class__(self) ``` ```python @classmethod def fromkeys(cls, iterable, value=None): '''Create a new ordered dictionary with keys from iterable and values set to value. ''' self = cls() for key in iterable: self[key] = value return self ``` -------------------------------- ### OrderedDict Utility Methods Source: https://sparclclient.readthedocs.io/en/latest/_modules/collections.html Methods for popping keys, setting defaults, and representation. ```python def pop(self, key, default=__marker): '''od.pop(k[,d]) -> v, remove specified key and return the corresponding value. If key is not found, d is returned if given, otherwise KeyError is raised. ''' if key in self: result = self[key] del self[key] return result if default is self.__marker: raise KeyError(key) return default ``` ```python def setdefault(self, key, default=None): '''Insert key with a value of default if key is not in the dictionary. Return the value for key if key is in the dictionary, else default. ''' if key in self: return self[key] self[key] = default return default ``` ```python @_recursive_repr() def __repr__(self): 'od.__repr__() <==> repr(od)' if not self: return '%s()' % (self.__class__.__name__,) return '%s(%r)' % (self.__class__.__name__, list(self.items())) ``` -------------------------------- ### retrieve_by_specid(specid_list, svc='spectras', format='pkl', include='DEFAULT', dataset_list=None, limit=500, units=True, fmt=None, verbose=False) Source: https://sparclclient.readthedocs.io/en/latest/sparcl.html Retrieves spectra records from the SPARCL database by a list of specids. ```APIDOC ## retrieve_by_specid(specid_list, svc='spectras', format='pkl', include='DEFAULT', dataset_list=None, limit=500, units=True, fmt=None, verbose=False) ### Description Retrieve spectra records from the SPARCL database by list of specids. ### Parameters - **specid_list** (list) - Required - List of specids. - **include** (list) - Optional - List of field names to include in each record. Defaults to 'DEFAULT'. - **dataset_list** (list) - Optional - List of data sets from which to retrieve spectra data. Defaults to None. - **limit** (int) - Optional - Maximum number of records to return. Defaults to 500. - **units** (bool) - Optional - Set to True to include units in the header for each applicable field. Defaults to True. - **fmt** (str) - Optional - Output format for the results. Defaults to None. - **verbose** (bool) - Optional - Set to True for in-depth return statement. Defaults to False. ### Returns - **Retrieved** (object) - Contains header and records, or a specutils/Pandas object if fmt is specified. ``` -------------------------------- ### Internal Spectrum construction logic Source: https://sparclclient.readthedocs.io/en/latest/_modules/sparcl/specutils.html This snippet demonstrates the internal logic for mapping SPARCL result attributes to a specutils Spectrum object. ```python redshift = r.redshift if 'redshift' in r else None uncertainty = InverseVariance(r.ivar) if 'ivar' in r else None mask = r.mask if 'mask' in r else None meta = {} for attribute in r: if attribute not in ('flux', 'wavelength', 'ivar', 'redshift', 'mask'): value = r[attribute] unit_str = dr_units.get(attribute) if unit_str is not None: try: value = value * u.Unit(unit_str) except (ValueError, TypeError): pass meta[attribute] = value s1 = Spectrum(flux=r.flux * u.Unit(flux_unit), spectral_axis=r.wavelength * u.Unit(wave_unit), uncertainty=uncertainty, redshift=redshift, mask=mask, meta=meta) s.append(s1) return s ``` -------------------------------- ### SparclClient.login() / SparclClient.logout() Source: https://sparclclient.readthedocs.io/en/latest/sparcl.html Manages user authentication sessions for the SPARCL client. ```APIDOC ## SparclClient.login() / SparclClient.logout() ### Description Handles authentication tokens to establish or terminate a session with the SPARCL API. ``` -------------------------------- ### UserList Method Implementations Source: https://sparclclient.readthedocs.io/en/latest/_modules/collections.html Internal methods for managing data within a UserList-like structure. ```python def count(self, item): return self.data.count(item) def index(self, item, *args): return self.data.index(item, *args) def reverse(self): self.data.reverse() def sort(self, /, *args, **kwds): self.data.sort(*args, **kwds) def extend(self, other): if isinstance(other, UserList): self.data.extend(other.data) else: self.data.extend(other) ``` -------------------------------- ### Retrieve All Fields Source: https://sparclclient.readthedocs.io/en/latest/sparcl.html Fetch fields tagged as 'all' from the SPARCL database. ```python >>> client = SparclClient(announcement=False) >>> client.get_all_fields() ['data_release', 'datasetgroup', 'dateobs', 'dateobs_center', 'dec', 'exptime', 'flux', 'instrument', 'ivar', 'mask', 'model', 'ra', 'redshift', 'redshift_err', 'redshift_warning', 'site', 'sparcl_id', 'specid', 'specprimary', 'spectype', 'survey', 'targetid', 'telescope', 'wave_sigma', 'wavelength', 'wavemax', 'wavemin'] ``` -------------------------------- ### UserDict Method Implementations Source: https://sparclclient.readthedocs.io/en/latest/_modules/collections.html Core methods for a UserDict wrapper, including iteration, containment, representation, and dictionary-like merge operations. ```python def __iter__(self): return iter(self.data) # Modify __contains__ to work correctly when __missing__ is present def __contains__(self, key): return key in self.data # Now, add the methods in dicts but not in MutableMapping def __repr__(self): return repr(self.data) def __or__(self, other): if isinstance(other, UserDict): return self.__class__(self.data | other.data) if isinstance(other, dict): return self.__class__(self.data | other) return NotImplemented def __ror__(self, other): if isinstance(other, UserDict): return self.__class__(other.data | self.data) if isinstance(other, dict): return self.__class__(other | self.data) return NotImplemented def __ior__(self, other): if isinstance(other, UserDict): self.data |= other.data else: self.data |= other return self def __copy__(self): inst = self.__class__.__new__(self.__class__) inst.__dict__.update(self.__dict__) # Create a copy and avoid triggering descriptors inst.__dict__["data"] = self.__dict__["data"].copy() return inst def copy(self): if self.__class__ is UserDict: return UserDict(self.data.copy()) import copy data = self.data try: self.data = {} c = copy.copy(self) finally: self.data = data c.update(self) return c @classmethod def fromkeys(cls, iterable, value=None): d = cls() for key in iterable: d[key] = value return d ``` -------------------------------- ### SparclClient.find(outfields=None, constraints={}, limit=500, sort=None, units=True, fmt=None, verbose=False) Source: https://sparclclient.readthedocs.io/en/latest/_modules/sparcl/client.html Queries the SPARC database for records based on specified constraints and output fields. Returns a Found object or a Pandas DataFrame. ```APIDOC ## Method: SparclClient.find ### Description Queries the SPARC database for records. Users can filter results using constraints, limit the number of records returned, and specify the output format. ### Parameters - **outfields** (list, optional) - List of fields to return. Defaults to None (returns sparcl_id and _dr). - **constraints** (dict, optional) - Key-Value pairs of constraints for record selection. Key is the field name, Value is a list of values. - **limit** (int, optional) - Maximum number of records to return. Defaults to 500. - **sort** (list, optional) - Comma separated list of fields to sort by. - **units** (bool, optional) - Include units in the header. Defaults to True. - **fmt** (str, optional) - Output format. Use 'pandas' for a DataFrame, otherwise returns a Found object. - **verbose** (bool, optional) - Set to True for in-depth return statements. Defaults to False. ### Returns - **Found** (object) - Contains header and records, or a Pandas DataFrame if fmt='pandas' is specified. ### Example ```python client = SparclClient(announcement=False) outs = ['sparcl_id', 'ra', 'dec'] cons = {'spectype': ['GALAXY'], 'redshift': [0.5, 0.9]} found = client.find(outfields=outs, constraints=cons) ``` ``` -------------------------------- ### retrieve(uuid_list, include=None, dataset_list=None, limit=500, units=True, fmt=None, verbose=False) Source: https://sparclclient.readthedocs.io/en/latest/sparcl.html Retrieves spectra records from the SPARCL database using a list of sparcl_ids. ```APIDOC ## retrieve(uuid_list, include=None, dataset_list=None, limit=500, units=True, fmt=None, verbose=False) ### Description Retrieves spectra records from the SPARCL database by a list of sparcl_ids. ### Parameters - **uuid_list** (list) - Required - List of sparcl_ids. - **include** (list) - Optional - List of field names to include in each record. Defaults to 'DEFAULT'. - **dataset_list** (list) - Optional - List of data sets from which to retrieve spectra data. Defaults to None. - **limit** (int) - Optional - Maximum number of records to return. Defaults to 500. Maximum allowed is 24,000. - **units** (bool) - Optional - Set to True to include units in the header for each applicable field. Defaults to True. - **fmt** (str) - Optional - Output format for the results ('specutils', 'pandas', or None). Defaults to None. - **verbose** (bool) - Optional - Set to True for in-depth return statement. Defaults to False. ### Returns - **Retrieved** (object) - Contains header and records, or a specutils/Pandas object if fmt is specified. ``` -------------------------------- ### retrieve(uuid_list, include='DEFAULT', dataset_list=None, limit=500, units=True, fmt=None, verbose=False) Source: https://sparclclient.readthedocs.io/en/latest/_modules/sparcl/client.html Retrieves spectra records from the SPARCL database by a list of sparcl_ids. ```APIDOC ## retrieve(uuid_list, include='DEFAULT', dataset_list=None, limit=500, units=True, fmt=None, verbose=False) ### Description Retrieve spectra records from the SPARCL database by list of sparcl_ids. ### Parameters - **uuid_list** (list) - Required - List of sparcl_ids. - **include** (list) - Optional - List of field names to include in each record. Defaults to 'DEFAULT'. - **dataset_list** (list) - Optional - List of data sets from which to retrieve spectra data. Defaults to None (all data sets). - **limit** (int) - Optional - Maximum number of records to return. Defaults to 500. Maximum allowed is 24,000. - **units** (bool) - Optional - Set to True to include units in the header for each applicable field. Defaults to True. - **fmt** (str) - Optional - Output format for the results. Options: 'specutils', 'pandas', or None (default). - **verbose** (bool) - Optional - Set to True for in-depth return statement. Defaults to False. ### Returns - **Results.Retrieved** (object) - Contains header and records, unless fmt='specutils' or fmt='pandas' is specified, in which case a specutils or Pandas Dataframe object is returned. ``` -------------------------------- ### Data Retrieval and Request Handling Source: https://sparclclient.readthedocs.io/en/latest/_modules/sparcl/client.html This snippet handles the logic for validating fields, constructing the API request, executing the POST request, and parsing the response in JSON or pickle format. ```python if (include == DEFAULT) or (include is None) or include == []: include_list = self.get_default_fields(dataset_list=dataset_list) elif include == ALL: include_list = self.get_all_fields(dataset_list=dataset_list) else: include_list = include self._validate_include(include_list, dataset_list) # Ensure 'sparcl_id' and 'specid' are always included for required in ("sparcl_id", "specid"): if required not in include_list: include_list.append(required) req_num = min(len(uuid_list), (limit or len(uuid_list))) #! print(f'DBG: req_num = {req_num:,d}' #! f' len(uuid_list)={len(uuid_list):,d}' #! f' limit={limit}' #! f' MAX_NUM_RECORDS_RETRIEVED={MAX_NUM_RECORDS_RETRIEVED:,d}') if req_num > MAX_NUM_RECORDS_RETRIEVED: msg = ( f"Too many records asked for with client.retrieve()." f" {len(uuid_list):,d} IDs provided," f" limit={limit}." f" But the maximum allowed is" f" {MAX_NUM_RECORDS_RETRIEVED:,d}." ) raise ex.TooManyRecords(msg) com_include = self._common_internal( science_fields=include_list, dataset_list=dataset_list ) uparams = { "include": ",".join(com_include), # limit=limit, # altered uuid_list to reflect limit #! "chunk_len": chunk, "format": format, #! "1thread": "yes", # @@@ 7.3.2023 #!"dataset_list": ",".join(dataset_list), } if not units: uparams["units"] = False # Do not put dataset_list in server call if it wasn't in client call. # Else will interfer with defaulting behavior of AUTH (DLS-504). if orig_dataset_list is not None: uparams["dataset_list"] = ",".join(dataset_list) qstr = urlencode(uparams) #!url = f'{self.apiurl}/retrieve/?{qstr}' url = f"{self.apiurl}/{svc}/?{qstr}" if verbose: print(f'Using url="{url}"') ut.tic() ids = list(uuid_list) if limit is None else list(uuid_list)[:limit] if self.show_curl: cmd = ut.curl_retrieve_str(ids, self.rooturl, svc=svc, qstr=qstr) print(cmd) try: auth = TokenAuth(self.token, self.token_expired) if self.token else None # noqa: E501 res = requests.post(url, json=ids, auth=auth, timeout=self.timeout) except requests.exceptions.ConnectTimeout as reCT: raise ex.UnknownSparcl(f"ConnectTimeout: {reCT}") except requests.exceptions.ReadTimeout as reRT: msg = ( f'Try increasing the value of the "read_timeout" parameter' f' to "SparclClient()".' f" The current values is: {self.r_timeout} (seconds)" f"{reRT}" ) raise ex.ReadTimeout(msg) from None except requests.exceptions.ConnectionError as reCE: raise ex.UnknownSparcl(f"ConnectionError: {reCE}") except requests.exceptions.TooManyRedirects as reTMR: raise ex.UnknownSparcl(f"TooManyRedirects: {reTMR}") except requests.exceptions.HTTPError as reHTTP: raise ex.UnknownSparcl(f"HTTPError: {reHTTP}") except requests.exceptions.URLRequired as reUR: raise ex.UnknownSparcl(f"URLRequired: {reUR}") except requests.exceptions.RequestException as reRE: raise ex.UnknownSparcl(f"RequestException: {reRE}") except Exception as err: # fall through raise ex.UnknownSparcl(err) if verbose: elapsed = ut.toc() print(f"Got response to post in {elapsed} seconds") if res.status_code != 200: if verbose: print(f"DBG: Server response=\n{res.text}") # @@@ FAILS on invalid JSON. Maybe not json at all !!! if verbose and ("traceback" in res.json()): print(f"DBG: Server traceback=\n{res.json()['traceback']}") raise ex.genSparclException(res, verbose=verbose) if format == "json": results = res.json() elif format == "pkl": # Read chunked binary file (representing pickle file) from # server response. Load pickle into python data structure. # Python structure is list of records where first element # is a header. with tempfile.TemporaryFile(mode="w+b") as fp: for idx, chunk in enumerate(res.iter_content(chunk_size=None)): fp.write(chunk) # Position to start of file for pickle reading (load) fp.seek(0) results = pickle.load(fp) else: results = res.json() ``` -------------------------------- ### Counter utility methods Source: https://sparclclient.readthedocs.io/en/latest/_modules/collections.html Standard utility methods for copying, reducing, and deleting items from a Counter. ```python def copy(self): 'Return a shallow copy.' return self.__class__(self) def __reduce__(self): return self.__class__, (dict(self),) def __delitem__(self, elem): 'Like dict.__delitem__() but does not raise KeyError for missing values.' if elem in self: super().__delitem__(elem) ``` -------------------------------- ### Bin spectra records Source: https://sparclclient.readthedocs.io/en/latest/_modules/sparcl/Results.html Aligns flux from all records by a common wavelength bin, returning a 2D flux array and 1D wavelength array. ```python >>> client = sparcl.client.SparclClient() >>> found = client.find( constraints={"data_release": ['BOSS-DR16']}, limit=10) >>> got = client.retrieve(found.ids) >>> flux2d,wavs = got.bin_spectra() ``` -------------------------------- ### Defining default units Source: https://sparclclient.readthedocs.io/en/latest/_modules/sparcl/specutils.html Sets the fallback units for flux and wavelength when header information is missing. ```python # Default units used as fallback when results have no header (e.g. bare # dict or _AttrDict records passed directly to conversion functions). DEFAULT_FLUX_UNIT = '10**-17 erg cm-2 s-1 AA-1' DEFAULT_WAVE_UNIT = 'AA' ``` -------------------------------- ### Retrieve records from SPARCL database Source: https://sparclclient.readthedocs.io/en/latest/_modules/sparcl/client.html This snippet demonstrates the internal logic for finding and retrieving records based on provided constraints and identifiers. ```python # Science Field Name for uuid. dr = list(self.fields.all_drs)[0] idfld = self.fields._science_name("sparcl_id", dr) found = self.find([idfld], constraints=constraints, limit=limit) if verbose: print(f"Found {found.count} matches.") if found.count < len(specid_list): usrcount = len(specid_list) dbcount = found.count warn = ( f"UserWarning: Some SPECIDs were not found. " f"{usrcount - dbcount} out of the {usrcount} requested " f"uuids have no records available in the SPARCL database " f"associated with DataSets {dr_list}." ) print(warn) res = self.retrieve( found.ids, #! svc=svc, #! format=format, include=include, dataset_list=dataset_list, limit=limit, units=units, fmt=fmt, verbose=verbose, ) if verbose: print(f"Got {res.count} records.") return res ``` -------------------------------- ### Implement TokenAuth for requests Source: https://sparclclient.readthedocs.io/en/latest/_modules/sparcl/client.html Custom authentication class that attaches a token to request headers and checks for renewal. ```python class TokenAuth(AuthBase): """Attaches HTTP Token Authentication to the given Request object.""" def __init__(self, token, renew_check): # setup any auth-related data here self.token = token self.renew_check = renew_check # SparcClient Method def __call__(self, request): # modify and return the request # check and renew token if needed self.renew_check(renew=True) request.headers["Authorization"] = self.token return request ``` -------------------------------- ### Convert field names between internal and science formats Source: https://sparclclient.readthedocs.io/en/latest/_modules/sparcl/Results.html Methods to map record keys between internal and science-specific naming conventions. Note that to_science_fields modifies the internal records list in-place. ```python # Convert Internal field names to Science field names. # SIDE-EFFECT: modifies self.recs def to_science_fields(self): # from_orig newrecs = list() for rec in self.recs: newrec = dict() dr = rec["_dr"] keep = True for orig in rec.keys(): if orig == "_dr": # keep DR around unchanged. We need it to rename back # to Internal Field Names later. newrec[orig] = rec[orig] else: new = self.fields._science_name(orig, dr) if new is None: keep = False # We don't have name mapping, toss rec newrec[new] = rec[orig] if keep: newrecs.append(_AttrDict(newrec)) self.recs = newrecs ``` ```python # Convert Science field names to Internal field names. def to_internal_fields(self): for rec in self.recs: dr = rec.get("_dr") for new in rec.keys(): if new == "_dr": # keep DR around unchanged. We need it to rename back # to Internal Field Names later. continue new = self.fields._internal_name(new, dr) rec[new] = rec.pop(new) ``` ```python def science_to_internal_fields(self): newrecs = list() for rec in self.recs: newrec = dict() dr = rec["_dr"] keep = True for sci_name in rec.keys(): if sci_name == "_dr": # keep DR around unchanged. We need it to rename back # to Internal Field Names later. newrec[sci_name] = rec[sci_name] else: new = self.fields._internal_name(sci_name, dr) if new is None: keep = False newrec[new] = rec[sci_name] if keep: newrecs.append(_AttrDict(newrec)) self.recs = newrecs return self.recs ``` -------------------------------- ### SparclClient.login Source: https://sparclclient.readthedocs.io/en/latest/_modules/sparcl/client.html Authenticates the user with the SPARCL service using an email and password to obtain an access token. ```APIDOC ## login(email, password=None) ### Description Authenticates the user with the SPARCL service. If the password is not provided, it will be prompted via standard input. ### Parameters - **email** (str) - Required - The user's email address. - **password** (str) - Optional - The user's password. ### Returns - **None** (None) - Returns None on success. - **str** (str) - Returns an error message if authentication fails. ``` -------------------------------- ### Login to SPARCL Source: https://sparclclient.readthedocs.io/en/latest/sparcl.html Authenticates the user with the SPARCL service using an email and password. ```python >>> >> client = SparclClient(announcement=False) >> client.login('test_user@noirlab.edu', 'testpw') Logged in successfully with email='test_user@noirlab.edu' ``` -------------------------------- ### Retrieve field units from SPARCL results Source: https://sparclclient.readthedocs.io/en/latest/_modules/sparcl/Results.html Demonstrates how to look up unit strings for specific fields, optionally filtered by data release. ```python >>> results.unit_for('flux') '1e-17 erg cm-2 s-1 AA-1' >>> results.unit_for('wave_sigma') {'SDSS-DR17': 'pixel', 'DESI-DR1': 'AA'} >>> results.unit_for('dec', data_release='DESI-DR1') 'deg' ``` -------------------------------- ### Define OrderedDict and internal helper classes Source: https://sparclclient.readthedocs.io/en/latest/_modules/collections.html Implementation of OrderedDict and its associated view and link classes for maintaining insertion order. ```python class _OrderedDictKeysView(_collections_abc.KeysView): def __reversed__(self): yield from reversed(self._mapping) class _OrderedDictItemsView(_collections_abc.ItemsView): def __reversed__(self): for key in reversed(self._mapping): yield (key, self._mapping[key]) class _OrderedDictValuesView(_collections_abc.ValuesView): def __reversed__(self): for key in reversed(self._mapping): yield self._mapping[key] class _Link(object): __slots__ = 'prev', 'next', 'key', '__weakref__' class OrderedDict(dict): 'Dictionary that remembers insertion order' # An inherited dict maps keys to values. # The inherited dict provides __getitem__, __len__, __contains__, and get. # The remaining methods are order-aware. # Big-O running times for all methods are the same as regular dictionaries. # The internal self.__map dict maps keys to links in a doubly linked list. # The circular doubly linked list starts and ends with a sentinel element. # The sentinel element never gets deleted (this simplifies the algorithm). # The sentinel is in self.__hardroot with a weakref proxy in self.__root. # The prev links are weakref proxies (to prevent circular references). # Individual links are kept alive by the hard reference in self.__map. # Those hard references disappear when a key is deleted from an OrderedDict. def __init__(self, other=(), /, **kwds): '''Initialize an ordered dictionary. The signature is the same as regular dictionaries. Keyword argument order is preserved. ''' try: self.__root except AttributeError: self.__hardroot = _Link() self.__root = root = _proxy(self.__hardroot) root.prev = root.next = root self.__map = {} self.__update(other, **kwds) def __setitem__(self, key, value, dict_setitem=dict.__setitem__, proxy=_proxy, Link=_Link): 'od.__setitem__(i, y) <==> od[i]=y' # Setting a new item creates a new link at the end of the linked list, # and the inherited dictionary is updated with the new key/value pair. if key not in self: self.__map[key] = link = Link() root = self.__root last = root.prev link.prev, link.next, link.key = last, root, key last.next = link root.prev = proxy(link) dict_setitem(self, key, value) def __delitem__(self, key, dict_delitem=dict.__delitem__): 'od.__delitem__(y) <==> del od[y]' # Deleting an existing item uses self.__map to find the link which gets # removed by updating the links in the predecessor and successor nodes. dict_delitem(self, key) link = self.__map.pop(key) link_prev = link.prev link_next = link.next link_prev.next = link_next link_next.prev = link_prev link.prev = None link.next = None ``` -------------------------------- ### SparclClient.find() Source: https://sparclclient.readthedocs.io/en/latest/sparcl.html Searches for datasets or records within the SPARCL system. ```APIDOC ## SparclClient.find() ### Description Performs a search operation to find records or datasets based on specified constraints. ### Method Python SDK Method ### Parameters - **args/kwargs** - Search constraints and query parameters as defined by the SPARCL service. ``` -------------------------------- ### Counter representation Source: https://sparclclient.readthedocs.io/en/latest/_modules/collections.html Returns a string representation of the Counter instance. ```python def __repr__(self): if not self: return f'{self.__class__.__name__}()' try: # dict() preserves the ordering returned by most_common() d = dict(self.most_common()) except TypeError: # handle case where values are not orderable d = dict(self) return f'{self.__class__.__name__}({d!r})' ``` -------------------------------- ### to_Spectrum Source: https://sparclclient.readthedocs.io/en/latest/_modules/sparcl/specutils.html Converts retrieved results into a specutils Spectrum or SpectrumCollection object. ```APIDOC ## to_Spectrum(results, *, collection=False, flux_unit=None, wave_unit=None) ### Description Converts retrieved results into a specutils Spectrum or SpectrumCollection object. ### Parameters - **results** (sparcl.Results.Retrieved) - Required - Retrieved results, or a single record from a set of results. - **collection** (bool) - Optional - If True, attempt to convert to a specutils.SpectrumCollection instead. - **flux_unit** (str) - Optional - Unit string for flux. If None, resolved via _get_units(). - **wave_unit** (str) - Optional - Unit string for wavelength. If None, resolved via _get_units(). ### Returns - **Spectrum or SpectrumCollection** - The requested spectral object. ``` -------------------------------- ### get_available_fields() Source: https://sparclclient.readthedocs.io/en/latest/sparcl.html Retrieves a list of all available fields for specified datasets. ```APIDOC ## get_available_fields(dataset_list=None) ### Description Returns the intersection of all available fields across the provided dataset list, or all datasets if none are specified. ### Parameters - **dataset_list** (list) - Optional - List of data sets to query. ``` -------------------------------- ### Convert SPARCL results to SpectrumList Source: https://sparclclient.readthedocs.io/en/latest/_modules/sparcl/specutils.html Converts retrieved SPARCL results into a specutils SpectrumList object. ```python [docs] def to_SpectrumList(results, *, flux_unit=None, wave_unit=None): """Convert `results` to :class:`specutils.SpectrumList`. Parameters ---------- results : :class:`sparcl.Results.Retrieved` Retrieved results. flux_unit : str, optional Unit string for flux. If None, resolved via _get_units(). wave_unit : str, optional Unit string for wavelength. If None, resolved via _get_units(). Returns ------- :class:`~specutils.SpectrumList` The requested object. """ if flux_unit is None or wave_unit is None: flux_unit, wave_unit = _get_units(results) s = SpectrumList() if isinstance(results, dict): records = [results] else: records = results.records for r in records: dr_units = {} if hasattr(results, 'hdr'): try: dr = r.get('_dr') dr_units = results.hdr.get('UNITS', {}).get(dr, {}) except (AttributeError, KeyError): pass ``` -------------------------------- ### get_all_fields() Source: https://sparclclient.readthedocs.io/en/latest/sparcl.html Retrieves a list of fields tagged as 'all' for specified datasets. ```APIDOC ## get_all_fields(dataset_list=None) ### Description Returns the intersection of all fields tagged as 'all' across the provided dataset list, or all datasets if none are specified. ### Parameters - **dataset_list** (list) - Optional - List of data sets to query. ``` -------------------------------- ### Convert results to specutils Source: https://sparclclient.readthedocs.io/en/latest/_modules/sparcl/Results.html Converts the current results object into a specutils Spectrum object. ```python def to_specutils(self): """Convert results to a `specutils` object. Returns: to_specutils (:class:`~specutils.Spectrum`): a `specutils` object. """ return su.to_specutils(self) ``` -------------------------------- ### Results.to_specutils() Source: https://sparclclient.readthedocs.io/en/latest/_modules/sparcl/Results.html Converts the results object into a specutils Spectrum object. ```APIDOC ## Results.to_specutils() ### Description Converts the current results into a `specutils.Spectrum` object. ### Returns - **to_specutils** (specutils.Spectrum) - A specutils object containing the spectral data. ```