import os
import requests
from loguru import logger
from requests.exceptions import ConnectionError
from panoptes.utils.error import InvalidConfig
from panoptes.utils.serializers import from_json, to_json
[docs]
def server_is_running(*args, **kwargs): # pragma: no cover
"""Thin-wrapper to check server."""
try:
return get_config(endpoint="heartbeat", verbose=False, *args, **kwargs)
except Exception as e:
logger.warning(f"server_is_running error (ignore if just starting server): {e!r}")
return False
[docs]
def get_config(
key=None, default=None, host=None, port=None, endpoint="get-config", parse=True, verbose=False
):
"""Get a config item from the config server.
Return the config entry for the given ``key``. If ``key=None`` (default), return
the entire config.
Nested keys can be specified as a string, as per `scalpl <https://pypi.org/project/scalpl/>`_.
Examples:
.. doctest::
>>> get_config(key='name')
'Testing PANOPTES Unit'
>>> get_config(key='location.horizon')
<Quantity 30. deg>
>>> # With no parsing, the raw string (including quotes) is returned.
>>> get_config(key='location.horizon', parse=False)
'"30 deg"'
>>> get_config(key='cameras.devices[1].model')
'canon_gphoto2'
>>> # Returns `None` if key is not found.
>>> foobar = get_config(key='foobar')
>>> foobar is None
True
>>> # But you can supply a default.
>>> get_config(key='foobar', default='baz')
'baz'
>>> # key and default are first two parameters.
>>> get_config('foobar', 'baz')
'baz'
>>> # Can use Quantities as well.
>>> from astropy import units as u
>>> get_config('foobar', 42 * u.meter)
<Quantity 42. m>
Notes:
By default all calls to this function will log at the `trace` level because
there are some calls (e.g. during POCS operation) that will be quite noisy.
Setting `verbose=True` changes those to `debug` log levels for an individual
call.
Args:
key (str): The key to update, see Examples in :func:`get_config` for details.
default (str, optional): The config server port, defaults to 6563.
host (str, optional): The config server host. First checks for PANOPTES_CONFIG_HOST
env var, defaults to 'localhost'.
port (str or int, optional): The config server port. First checks for PANOPTES_CONFIG_HOST
env var, defaults to 6563.
endpoint (str, optional): The relative url endpoint to use for getting
the config items, default 'get-config'. See `server_is_running()`
for example of usage.
parse (bool, optional): If response should be parsed by
:func:`panoptes.utils.serializers.from_json`, default True.
verbose (bool, optional): Determines the output log level, defaults to
True (i.e. `debug` log level). See notes for details.
Returns:
dict: The corresponding config entry.
Raises:
Exception: Raised if the config server is not available.
"""
log_level = "DEBUG" if verbose else "TRACE"
host = host or os.getenv("PANOPTES_CONFIG_HOST", "localhost")
port = port or os.getenv("PANOPTES_CONFIG_PORT", 6563)
url = f"http://{host}:{port}/{endpoint}"
config_entry = default
try:
logger.log(log_level, f"Calling get_config on url={url!r} with key={key!r}")
response = requests.post(url, json={"key": key, "verbose": verbose})
if not response.ok: # pragma: no cover
raise InvalidConfig(f"Config server returned invalid JSON: {response.content!r}")
except ConnectionError:
logger.log(log_level, "Bad connection to config-server. Check to make sure it is running.")
except Exception as e: # pragma: no cover
logger.warning(f"Problem with get_config: {e!r}")
else:
response_text = response.text.strip()
logger.log(log_level, f"Decoded response_text={response_text!r}")
if response_text != "null":
logger.log(log_level, f"Received config key={key!r} response_text={response_text!r}")
if parse:
logger.log(log_level, f"Parsing config results: response_text={response_text!r}")
config_entry = from_json(response_text)
else:
config_entry = response_text
if config_entry is None:
logger.log(log_level, f"No config entry found, returning default={default!r}")
config_entry = default
logger.log(log_level, f"Config key={key!r}: config_entry={config_entry!r}")
return config_entry
[docs]
def set_config(key, new_value, host=None, port=None, parse=True):
"""Set config item in config server.
Given a `key` entry, update the config to match. The `key` is a dot accessible
string, as given by `scalpl <https://pypi.org/project/scalpl/>`_. See Examples in
:func:`get_config` for details.
Examples:
.. doctest::
>>> from astropy import units as u
>>> # Can use astropy units.
>>> set_config('location.horizon', 35 * u.degree)
{'location.horizon': <Quantity 35. deg>}
>>> get_config(key='location.horizon')
<Quantity 35. deg>
>>> # String equivalent works for 'deg', 'm', 's'.
>>> set_config('location.horizon', '30 deg')
{'location.horizon': <Quantity 30. deg>}
Args:
key (str): The key to update, see Examples in :func:`get_config` for details.
new_value (scalar|object): The new value for the key, can be any serializable object.
host (str, optional): The config server host. First checks for PANOPTES_CONFIG_HOST
env var, defaults to 'localhost'.
port (str or int, optional): The config server port. First checks for PANOPTES_CONFIG_HOST
env var, defaults to 6563.
parse (bool, optional): If response should be parsed by
:func:`panoptes.utils.serializers.from_json`, default True.
Returns:
dict: The updated config entry.
Raises:
Exception: Raised if the config server is not available.
"""
host = host or os.getenv("PANOPTES_CONFIG_HOST", "localhost")
port = port or os.getenv("PANOPTES_CONFIG_PORT", 6563)
url = f"http://{host}:{port}/set-config"
json_str = to_json({key: new_value})
config_entry = None
try:
# We use our own serializer so pass as `data` instead of `json`.
logger.debug(f"Calling set_config on url={url!r}")
response = requests.post(url, data=json_str, headers={"Content-Type": "application/json"})
if not response.ok: # pragma: no cover
raise Exception(f"Cannot access config server: {response.text}")
except Exception as e:
logger.warning(f"Problem with set_config: {e!r}")
else:
if parse:
config_entry = from_json(response.content.decode("utf8"))
else:
config_entry = response.json()
return config_entry