panoptes.utils package

Subpackages

Submodules

panoptes.utils.error module

exception panoptes.utils.error.AlreadyExists(msg=None, exit=False)[source]

Bases: PanError

Generic already exists class

exception panoptes.utils.error.ArduinoDataError(msg=None, exit=False)[source]

Bases: PanError

PanError raised when there is something very wrong with Arduino information.

exception panoptes.utils.error.BadConnection(msg=None, exit=False)[source]

Bases: PanError

PanError raised when a connection is bad

exception panoptes.utils.error.BadSerialConnection(msg=None, exit=False)[source]

Bases: PanError

PanError raised when serial command is bad

exception panoptes.utils.error.CameraNotFound(msg=None, exit=False)[source]

Bases: NotFound

Camera cannot be imported

exception panoptes.utils.error.DomeNotFound(msg=None, exit=False)[source]

Bases: NotFound

Dome device not found.

exception panoptes.utils.error.GoogleCloudError(msg=None, exit=False)[source]

Bases: PanError

Errors related to google cloud

exception panoptes.utils.error.IllegalValue(msg=None, exit=False)[source]

Bases: PanError, ValueError

Errors from trying to hardware parameters to values not supported by a particular model

exception panoptes.utils.error.InvalidCommand(msg=None, exit=False)[source]

Bases: PanError

PanError raised if a system command does not run

exception panoptes.utils.error.InvalidConfig(msg=None, exit=False)[source]

Bases: PanError

PanError raised if config file is invalid

exception panoptes.utils.error.InvalidDeserialization(msg='Problem deserializing', **kwargs)[source]

Bases: PanError

Error for serialization errors

exception panoptes.utils.error.InvalidMountCommand(msg=None, exit=False)[source]

Bases: PanError

PanError raised if attempting to send command that doesn’t exist

exception panoptes.utils.error.InvalidObservation(msg=None, exit=False)[source]

Bases: NotFound

PanError raised if a field is invalid.

exception panoptes.utils.error.InvalidSerialization(msg='Problem Serializing', **kwargs)[source]

Bases: PanError

Error for serialization errors

exception panoptes.utils.error.InvalidSystemCommand(msg='Problem running system command', **kwargs)[source]

Bases: PanError

Error for a system level command malfunction

exception panoptes.utils.error.MountNotFound(msg='Mount Not Found', **kwargs)[source]

Bases: NotFound

Mount cannot be import

exception panoptes.utils.error.NoObservation(msg='No valid observations found.', **kwargs)[source]

Bases: PanError

Generic no Observation

exception panoptes.utils.error.NotFound(msg=None, exit=False)[source]

Bases: PanError

Generic not found class

exception panoptes.utils.error.NotSupported(msg=None, exit=False)[source]

Bases: PanError, NotImplementedError

Errors from trying to use hardware features not supported by a particular model

exception panoptes.utils.error.PanError(msg=None, exit=False)[source]

Bases: Exception

Base class for Panoptes errors

exit_program(msg=None)[source]

Kills running program

exception panoptes.utils.error.SolveError(msg=None, exit=False)[source]

Bases: NotFound

Camera cannot be imported

exception panoptes.utils.error.TheSkyXError(msg=None, exit=False)[source]

Bases: PanError

Errors from TheSkyX

exception panoptes.utils.error.TheSkyXKeyError(msg=None, exit=False)[source]

Bases: TheSkyXError

Errors from TheSkyX because bad key passed

exception panoptes.utils.error.TheSkyXTimeout(msg=None, exit=False)[source]

Bases: TheSkyXError

Errors from TheSkyX because bad key passed

exception panoptes.utils.error.Timeout(msg='Timeout waiting for event', **kwargs)[source]

Bases: PanError

Error called when an event times out

panoptes.utils.horizon module

class panoptes.utils.horizon.Horizon(obstructions=None, default_horizon=30)[source]

Bases: object

A simple class to define some coordinate points.

Accepts a list of lists where each list consists of two points corresponding to an altitude (0-90) and an azimuth (0-360).

The list of points for a given obstruction must be in clockwise ordering.

If azimuth is a negative number (but greater than -360) then 360 will be added to put it in the correct range.

The list are points that are obstruction points beyond the default horizon.

get_horizon(az)[source]

Get the horizon level in degrees at a given azimuth. :param az: The azimuth. If float, assumed in degrees. :type az: float or astropy.Quantity

Returns:

The angular horizon level.

Return type:

astropy.Quantity

class panoptes.utils.horizon.Obstruction(points_list)[source]

Bases: object

Define an obstruction on the horizon.

An obstruction is defined by a list of altitude/azimuth coordinate pairs that form a polygon representing the obstruction boundary. Points should be provided in clockwise ordering.

get_horizon(az)[source]

Get the horizon level in degrees at a given azimuth. :param az: The azimuth. If float, assumed in degrees. :type az: float or astropy.Quantity

Returns:

The angular horizon level.

Return type:

astropy.Quantity

panoptes.utils.library module

panoptes.utils.library.load_c_library(name, path=None, mode=0, **kwargs)[source]

Utility function to load a shared/dynamically linked library (.so/.dylib/.dll).

The name and location of the shared library can be manually specified with the library_path argument, otherwise the ctypes.util.find_library function will be used to try to locate based on library_name.

Parameters:

  • name (str) – name of the library (without ‘lib’ prefix or any suffixes, e.g. ‘fli’).

  • path (str, optional) – path to the library e.g. ‘/usr/local/lib/libfli.so’.

  • mode (int, optional) – mode in which to load the library, see dlopen(3) man page for details. Should be one of ctypes.RTLD_GLOBAL, ctypes.RTLD_LOCAL, or ctypes.DEFAULT_MODE. Default is ctypes.DEFAULT_MODE.

Returns:

ctypes.CDLL

Raises:

  • pocs.utils.error.NotFound – raised if library_path not given & find_library fails to locate the library.

  • OSError – raises if the ctypes.CDLL loader cannot load the library.

panoptes.utils.library.load_module(module_name)[source]

Dynamically load a module or a class from the module.

>>> from panoptes.utils.library import load_module
>>> error = load_module('panoptes.utils.error')
>>> error.__name__
'panoptes.utils.error'
>>> error.__package__
'panoptes.utils'
>>> PanError = load_module('panoptes.utils.error.PanError')
>>> PanError.__name__
'PanError'
Parameters:

module_name (str) – Name of module to import.

Returns:

an imported module name

Return type:

module

Raises:

error.NotFound – If module cannot be imported.

panoptes.utils.rs232 module

class panoptes.utils.rs232.SerialData(*args, **kwargs)[source]

Bases: object

SerialData wraps a PySerial instance for reading from and writing to a serial device.

Because POCS is intended to be very long running, and hardware may be turned off when unused or to force a reset, this wrapper may or may not have an open connection to the underlying serial device. Note that for most devices, is_connected will return true if the device is turned off/unplugged after a connection is opened; the code will only discover there is a problem when we attempt to interact with the device.

>>> from panoptes.utils.rs232 import SerialData
>>> # Connect to our fake buffered device
>>> device_listener = SerialData(port='loop://')
>>> device_listener.is_connected
True
>>> device_listener.port
'loop://'
>>> # Device sends event
>>> bytes = device_listener.write('Hello World')
>>> device_listener.read(bytes)
'Hello World'
connect()[source]

If disconnected, then connect to the serial port.

Raises:

error.BadSerialConnection if unable to open the connection. –

disconnect()[source]

Closes the serial connection.

Raises:

error.BadSerialConnection if unable to close the connection. –

get_and_parse_reading(retry_limit=5)[source]

Reads a line of JSON text and returns the decoded value, along with the current time.

Parameters:

retry_limit – Number of lines to read in an attempt to get one that parses as JSON.

Returns:

A pair (tuple) of (timestamp, decoded JSON line). The timestamp is the time of completion of the readline operation.

get_reading()[source]

Reads and returns a line, along with the timestamp of the read.

Returns:

A pair (tuple) of (timestamp, line). The timestamp is the time of completion of the readline operation.

property is_connected

True if serial port is open, False otherwise.

property port

Name of the port.

read(retry_limit=None, retry_delay=None)[source]

Reads next line of input using readline.

If no response is given, delay for retry_delay and then try to read again. Fail after retry_limit attempts.

read_bytes(size=1)[source]

Reads size bytes from the serial port.

If a read timeout is set on self.ser, this may return less characters than requested. With no timeout it will block until the requested number of bytes is read.

Parameters:

size – Number of bytes to read.

Returns:

Bytes read from the port.

reset_input_buffer()[source]

Clear buffered data from connected port/device.

Note that Wilfred reports that the input from an Arduino can seriously lag behind realtime (e.g. 10 seconds), and that clear_buffer may exist for that reason (i.e. toss out any buffered input from a device, and then read the next full line, which likely requires tossing out a fragment of a line).

write(value)[source]

Write value (a string) after encoding as bytes.

write_bytes(data)[source]

Write data of type bytes.

panoptes.utils.rs232.find_serial_port(vendor_id, product_id, return_all=False)[source]

Finds the serial port that matches the given vendor and product id.

>>> from panoptes.utils.rs232 import find_serial_port
>>> vendor_id = 0x2a03  # arduino
>>> product_id = 0x0043 # Uno Rev 3
>>> find_serial_port(vendor_id, product_id)
'/dev/ttyACM0'

>>> # Raises error when not found.
>>> find_serial_port(0x1234, 0x4321)
Traceback (most recent call last):
  ...
panoptes.utils.error.NotFound: NotFound: No serial ports for vendor_id=4660 and product_id=17185
Parameters:

  • vendor_id (int) – The vendor id, can be hex or int.

  • product_id (int) – The product id, can be hex or int.

  • return_all (bool) – If more than one serial port matches, return all devices, default False.

Returns:

Either the path to the detected port or a list of all comports that match.

Return type:

str or list

panoptes.utils.rs232.get_serial_port_info()[source]

Returns the serial ports defined on the system.

Returns: a list of PySerial’s ListPortInfo objects. See:

https://github.com/pyserial/pyserial/blob/master/serial/tools/list_ports_common.py

panoptes.utils.serializers module

class panoptes.utils.serializers.StringYAML(*, typ: List[str] | str | None = None, pure: Any = False, output: Any = None, plug_ins: Any = None)[source]

Bases: YAML

YAML class that can dump to a string.

By default the YAML parser doesn’t serialize directly to a string. This class is a small wrapper to output StreamIO as a string if no stream is provided.

dump(data, stream: IO | None = None, **kwargs) → str | None[source]

YAML class that can dump to a string.

By default the YAML parser doesn’t serialize directly to a string. This class is a small wrapper to output StreamIO as a string if no stream is provided.

See https://yaml.readthedocs.io/en/latest/example.html#output-of-dump-as-a-string.

Note

This class should not be used directly but instead is instantiated as part of the yaml convenience methods below.

Parameters:

  • data (object) – An object, usually dict-like.

  • stream (None | stream, optional) – A stream object to write the YAML. If default None, return value as string.

  • **kwargs – Keywords passed to the dump function.

Returns:

The serialized object string.

Return type:

str

panoptes.utils.serializers.deserialize_all_objects(obj)[source]

Recursively parse the incoming object for various data types.

This will currently attempt to parse and return, in the following order:

If obj is a dict with exactly two keys named unit and value, then attempt to parse into a valid astropy.unit.Quantity.

A boolean.

A datetime.datetime object as parsed by dateutil.parser.parse.

If a string ending with any of ['m', 'deg', 's'], an astropy.unit.Quantity

Note

See the to/from_json/yaml methods, which use this function.

Parameters:

obj (dict or str or object) – Object to check for quantities.

Returns:

Same as obj but with objects converted to quantities.

Return type:

dict

panoptes.utils.serializers.from_json(msg: str) → dict[source]

Convert a JSON string into a Python object.

Astropy quanitites will be converted from a {"value": val, "unit": unit} format. Additionally, the following units will be converted if the value ends with the exact string:

  • deg

  • m

  • s

Time-like values are not parsed, however see example below.

Examples:

>>> from panoptes.utils.serializers import from_json
>>> config_str = '{"name":"Mauna Loa","elevation":{"value":3397.0,"unit":"m"}}'
>>> from_json(config_str)
{'name': 'Mauna Loa', 'elevation': <Quantity 3397. m>}

# Invalid values will be returned as is.
>>> from_json('{"horizon":{"value":42.0,"unit":"degr"}}')
{'horizon': {'value': 42.0, 'unit': 'degr'}}

# The following will convert if final string:
>>> from_json('{"horizon": "42.0 deg"}')
{'horizon': <Quantity 42. deg>}

>>> from_json('{"elevation": "1000 m"}')
{'elevation': <Quantity 1000. m>}

>>> from_json('{"readout_time": "10 s"}')
{'readout_time': <Quantity 10. s>}

# Be careful with short unit names in extended format!
>>> horizon = from_json('{"horizon":{"value":42.0,"unit":"d"}}')
>>> horizon['horizon']
<Quantity 42. d>
>>> horizon['horizon'].decompose()
<Quantity 3628800. s>

>>> from panoptes.utils.time import current_time
>>> time_str = to_json({"current_time": current_time().datetime})
>>> from_json(time_str)['current_time']
2019-04-08T06:43:28.232406

>>> from astropy.time import Time
>>> Time(from_json(time_str)['current_time'])
<Time object: scale='utc' format='isot' value=2019-04-08T06:43:28.232>
Parameters:

msg (str) – The JSON string representation of the object.

Returns:

The loaded object.

Return type:

dict

panoptes.utils.serializers.from_yaml(msg: str, parse: bool = True) → dict[source]

Convert a YAML string into a Python object.

This is a thin-wrapper around ruamel.YAML.load that also parses the results looking for astropy.units.Quantity objects.

Comments are preserved as long as the object remains YAML (lost on conversion to JSON, for example).

See from_json for examples of astropy unit parsing.

Examples

Note how comments in the YAML are preserved.

>>> config_str = '''name: Testing PANOPTES Unit
... pan_id: PAN000
...
... location:
...   latitude: 19.54 deg
...   longitude: -155.58 deg
...   name: Mauna Loa Observatory  # Can be anything
... '''

>>> config = from_yaml(config_str)
>>> config['location']['latitude']
<Quantity 19.54 deg>

>>> yaml_config = to_yaml(config)
>>> yaml_config
''' name: Testing PANOPTES Unit
... pan_id: PAN000  # CHANGE NAME
...
... location:
...   latitude: 19.54 deg
...   longitude: value: -155.58 deg
...   name: Mauna Loa Observatory  # Can be anything
... '''
>>> yaml_config == config_str
True
Parameters:

  • msg (str) – The YAML string representation of the object.

  • parse (bool) – If objects should be parsed via _parse_all_objects, default True.

Returns:

The ordered dict representing the YAML string, with appropriate

object deserialization.

Return type:

collections.OrderedDict

panoptes.utils.serializers.serialize_all_objects(obj)[source]

Iterate the obj items and serialize each value.

Note

See the to/from_json/yaml methods, which use this function.

Parameters:

obj (dict) – The dictionary object to be iterated.

Returns:

The same as obj but with the values serialized.

Return type:

dict

panoptes.utils.serializers.serialize_object(obj)[source]

Serialize the given object.

This is a custom serializer function used by to_json to serialize individual objects. Also called in a loop by serialize_all_objects.

>>> from panoptes.utils.serializers import serialize_object
>>> from dateutil.parser import parse as date_parse
>>> from astropy import units as u

>>> serialize_object(42 * u.meter)
'42.0 m'

>>> party_time = date_parse('1999-12-31 11:59:59')
>>> type(party_time)
 <class 'datetime.datetime'>
>>> serialize_object(party_time)
'1999-12-31T11:59:59.000'

Note

See the to/from_json/yaml methods, which use this function.

Parameters:

obj (any) – The object to be serialized.

Returns:

panoptes.utils.serializers.to_json(obj, filename: str | Path | TextIO | BinaryIO | None = None, append: bool = True, **kwargs) → str[source]

Convert a Python object to a JSON string.

Will handle datetime objects as well as astropy.unit.Quantity objects. Astropy quantities will be converted to a dict: {“value”: val, “unit”: unit}.

Examples:

>>> from panoptes.utils.serializers import to_json
>>> from astropy import units as u
>>> config = { "name": "Mauna Loa", "elevation": 3397 * u.meter }
>>> to_json(config)
'{"name": "Mauna Loa", "elevation": "3397.0 m"}'

>>> to_json({"numpy_array": np.arange(10)})
'{"numpy_array": [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]}'

>>> from panoptes.utils.time import current_time
>>> to_json({"current_time": current_time()})
'{"current_time": "2019-04-08 22:19:28.402198"}'
Parameters:

  • obj (object) – The object to be converted to JSON, usually a dict.

  • filename – Path to file for saving. Can be a string path, pathlib.Path object, or open filehandle.

  • append (bool, optional) – Append to filename, default True. Setting False will clobber the file.

  • **kwargs – Keyword arguments passed to json.dumps.

Returns:

The JSON string representation of the object.

Return type:

str

panoptes.utils.serializers.to_yaml(obj, **kwargs) → str[source]

Serialize a Python object to a YAML string.

This will properly serialize the following:

  • datetime.datetime

  • astropy.time.Time

  • astropy.units.Quantity

Examples

Also see the examples from_yaml.

>>> import os
>>> os.environ['POCSTIME'] = '1999-12-31 23:49:49'
>>> from panoptes.utils.time import current_time
>>> t0 = current_time()
>>> t0
<Time object: scale='utc' format='iso' value=1999-12-31 23:49:49.000>

>>> to_yaml({'astropy time -> astropy time': t0})
"astropy time -> astropy time: '1999-12-31T23:49:49.000'\n"

>>> to_yaml({'datetime -> astropy time': t0.datetime})
"datetime -> astropy time: '1999-12-31T23:49:49.000'\n"

>>> # Can pass a `stream` parameter to save to file
>>> with open('temp.yaml', 'w') as f:
...     to_yaml({'my_object': 42}, stream=f)
Parameters:

  • obj (dict) – The object to be converted to be serialized.

  • **kwargs – Arguments passed to ruamel.yaml.dump. See Examples.

Returns:

The YAML string representation of the object.

Return type:

str

panoptes.utils.time module

class panoptes.utils.time.CountdownTimer(duration: int | float, name: str = '')[source]

Bases: object

Simple timer object for tracking whether a time duration has elapsed.

A countdown timer that tracks elapsed time and provides methods to check if the timer has expired, how much time is left, and sleep until expiration.

expired() → bool[source]

Return a boolean, telling if the timeout has expired.

Returns:

If timer has expired.

Return type:

bool

restart() → None[source]

Restart the timed duration.

sleep(max_sleep: int | float | None = None, log_level: str = 'DEBUG') → bool[source]

Sleep until the timer expires, or for max_sleep, whichever is sooner.

Parameters:

  • max_sleep (int or None) – Number of seconds to wait for, or None.

  • log_level (str) – Log level for sleeping message, default DEBUG.

Returns:

True if slept for less than time_left(), False otherwise.

time_left() → float[source]

Return how many seconds are left until the timeout expires.

Returns:

Number of seconds remaining in timer, zero if is_non_blocking=True.

Return type:

int

panoptes.utils.time.current_time(flatten: bool = False, datetime: bool = False, pretty: bool = False) → Time | datetime | str[source]

Convenience method to return the “current” time according to the system.

Note

If the $POCSTIME environment variable is set then this will return the time given in the variable. This is used for setting specific times during testing. After checking the value of POCSTIME the environment variable will also be incremented by one second so that subsequent calls to this function will generate monotonically increasing times.

Operation of POCS from $POCS/bin/pocs_shell will clear the POCSTIME variable.

>>> os.environ['POCSTIME'] = '1999-12-31 23:59:59'
>>> party_time = current_time(pretty=True)
>>> party_time
'1999-12-31 23:59:59'

# Next call is one second later when using $POCSTIME.
>>> y2k = current_time(pretty=True)
>>> y2k
'2000-01-01 00:00:00'

Note

The time returned from this function is not timezone aware. All times are UTC.

>>> from panoptes.utils.time import current_time
>>> current_time()
<Time object: scale='utc' format='datetime' value=2018-10-07 22:29:03.009873>

>>> current_time(datetime=True)
datetime.datetime(2018, 10, 7, 22, 29, 26, 594368)

>>> current_time(pretty=True)
'2018-10-07 22:29:51'
Returns:

Object representing now.

Return type:

astropy.time.Time

panoptes.utils.time.flatten_time(t: Time) → str[source]

Given an astropy time, flatten to have no extra chars besides integers.

>>> from astropy.time import Time
>>> from panoptes.utils.time import flatten_time
>>> t0 = Time('1999-12-31 23:59:59')
>>> t0.isot
'1999-12-31T23:59:59.000'

>>> flatten_time(t0)
'19991231T235959'
Parameters:

t (astropy.time.Time) – The time to be flattened.

Returns:

The flattened string representation of the time.

Return type:

str

panoptes.utils.time.wait_for_events(events: ~threading.Event | list[~threading.Event], timeout: float | ~astropy.units.quantity.Quantity = 600, sleep_delay: float | ~astropy.units.quantity.Quantity = <Quantity 5. s>, callback: ~collections.abc.Callable | None = None) → bool[source]

Wait for event(s) to be set.

This method will wait for a maximum of timeout seconds for all the events to complete.

Checks every sleep_delay seconds for the events to be set.

If provided, the callback will be called every sleep_delay seconds. The callback should return True to continue waiting otherwise False to interrupt the loop and return from the function.

>>> import time
>>> import threading
>>> from panoptes.utils.time import wait_for_events
>>> # Create some events, normally something like taking an image.
>>> event0 = threading.Event()
>>> event1 = threading.Event()

>>> # Wait for 30 seconds but interrupt after 1 second by returning False from callback.
>>> def interrupt_cb(): time.sleep(1); return False
>>> # The function will return False if events are not set.
>>> wait_for_events([event0, event1], timeout=30, callback=interrupt_cb)
False

>>> # Timeout will raise an exception.
>>> wait_for_events([event0, event1], timeout=1)
Traceback (most recent call last):
  File "<input>", line 1, in <module>
  File ".../panoptes-utils/src/panoptes/utils/time.py", line 254, in wait_for_events
panoptes.utils.error.Timeout: Timeout: Timeout waiting for generic event

>>> # Set the events in another thread for normal usage.
>>> def set_events(): time.sleep(1); event0.set(); event1.set()
>>> threading.Thread(target=set_events).start()
>>> wait_for_events([event0, event1], timeout=30)
True
Parameters:

  • events (list(threading.Event)) – An Event or list of Events to wait on.

  • timeout (float|`astropy.units.Quantity`) – Timeout in seconds to wait for events, default 600 seconds.

  • sleep_delay (float, optional) – Time in seconds between event checks.

  • callback (callable) – A periodic callback that should return True to continue waiting or False to interrupt the loop. Can also be used for e.g. custom logging.

Returns:

True if events were set, False otherwise.

Return type:

bool

Raises:

error.Timeout – Raised if events have not all been set before timeout seconds.

panoptes.utils.utils module

panoptes.utils.utils.altaz_to_radec(alt: float | Quantity | None = None, az: float | Quantity | None = None, location: EarthLocation | None = None, obstime: Time | None = None, **kwargs) → SkyCoord[source]

Convert alt/az degrees to RA/Dec SkyCoord.

>>> from panoptes.utils.utils import altaz_to_radec
>>> from astropy.coordinates import EarthLocation
>>> from astropy import units as u
>>> keck = EarthLocation.of_site('Keck Observatory')
...

>>> altaz_to_radec(alt=75, az=180, location=keck, obstime='2020-02-02T20:20:02.02')
<SkyCoord (ICRS): (ra, dec) in deg
    (281.78..., 4.807...)>

>>> # Can use quantities or not.
>>> alt = 4500 * u.arcmin
>>> az = 180 * u.degree
>>> altaz_to_radec(alt=alt, az=az, location=keck, obstime='2020-02-02T20:20:02.02')
<SkyCoord (ICRS): (ra, dec) in deg
    (281.78..., 4.807...)>

>>> # Will use current time if none given.
>>> altaz_to_radec(alt=35, az=90, location=keck)
<SkyCoord (ICRS): (ra, dec) in deg
    (..., ...)>

>>> # Must pass a `location` instance.
>>> altaz_to_radec()
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  ...
    assert location is not None
AssertionError
Parameters:

  • alt (astropy.units.Quantity or scalar) – Altitude.

  • az (astropy.units.Quantity or scalar) – Azimuth.

  • location (astropy.coordinates.EarthLocation, required) – A valid location.

  • obstime (None, optional) – Time for object, defaults to current_time

Returns:

Coordinates corresponding to the AltAz.

Return type:

astropy.coordinates.SkyCoord

panoptes.utils.utils.get_free_space(directory: str | Path | None = None) → Quantity[source]

Return the amoung of freespace in gigabytes for given directory.

>>> from panoptes.utils.utils import get_free_space
>>> get_free_space()
<Quantity ... Gbyte>

>>> get_free_space(directory='/')
<Quantity ... Gbyte>
Parameters:

directory (str, optional) – Path to directory. If None defaults to root.

Returns:

The number of gigabytes avialable in folder.

Return type:

astropy.units.Quantity

panoptes.utils.utils.get_quantity_value(quantity: Quantity | float, unit: str | Unit | None = None) → float[source]

Thin-wrapper around the astropy.units.Quantity.to_value method.

If passed something other than a Quantity will simply return the original object.

>>> from astropy import units as u
>>> from panoptes.utils.utils import get_quantity_value

>>> get_quantity_value(60 * u.second)
60.0

>>> # Can convert between units.
>>> get_quantity_value(60 * u.minute, unit='second')
3600.0

>>> get_quantity_value(60 * u.minute, unit=u.second)
3600.0

>>> get_quantity_value(60)
60

>>> get_quantity_value('J2000.0', unit='jyear_str')
'J2000.0'

>>> get_quantity_value('J2000.0')
'J2000.0'
Parameters:

  • quantity (astropy.units.Quantity or scalar) – Quantity to extract numerical value from.

  • unit (astropy.units.Unit, optional) – unit to convert to.

Returns:

numerical value of the Quantity after conversion to the specified unit.

Return type:

float

panoptes.utils.utils.listify(obj) → list[source]

Given an object, return a list.

Always returns a list. If obj is None, returns empty list, if obj is list, just returns obj, otherwise returns list with obj as single member.

If a dict object is passed then this function will return a list of only the values.

>>> listify(42)
[42]
>>> listify('foo')
['foo']
>>> listify(None)
[]
>>> listify(['a'])
['a']

>>> my_dict = dict(a=42, b='foo')
>>> listify(my_dict)
[42, 'foo']
>>> listify(my_dict.values())
[42, 'foo']
>>> listify(my_dict.keys())
['a', 'b']
Returns:

You guessed it.

Return type:

list

panoptes.utils.utils.normalize_file_input(file_input: str | Path | TextIO | BinaryIO) → str[source]

Normalize file input to a string path.

This function accepts string paths, pathlib.Path objects, and open filehandles and returns a string path that can be used with functions that expect a filename.

Parameters:

file_input – A string path, pathlib.Path object, or open filehandle.

Returns:

The normalized file path as a string.

Return type:

str

Raises:

ValueError – If the input is an open filehandle without a name attribute, or if the input type is not supported.

Module contents