Skip to content

utils.py

This module contains general pipeline utility functions.

StopWatch

A simple stopwatch to simplify timing code.

Source code in vast_pipeline/utils/utils.py
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
class StopWatch():
    """
    A simple stopwatch to simplify timing code.
    """

    def __init__(self) -> None:
        """
        Initialise the StopWatch

        Returns:
            None.
        """
        self._init = datetime.now()
        self._last = self._init

    def reset(self) -> float:
        """
        Reset the stopwatch and return the time since last reset (seconds).

        Returns:
            The time in seconds since the last reset.
        """
        now = datetime.now()
        diff = (now - self._last).total_seconds()
        self._last = now

        return diff

    def reset_init(self) -> float:
        """
        Reset the stopwatch and return the total time since initialisation.

        Returns:
            The time in seconds since the initialisation.
        """
        now = datetime.now()
        diff = (now - self._init).total_seconds()
        self._last = self._init = now

        return diff

__init__(self)

Initialise the StopWatch

Returns:

Type Description
None

None.

Source code in vast_pipeline/utils/utils.py
26
27
28
29
30
31
32
33
34
def __init__(self) -> None:
    """
    Initialise the StopWatch

    Returns:
        None.
    """
    self._init = datetime.now()
    self._last = self._init

reset(self)

Reset the stopwatch and return the time since last reset (seconds).

Returns:

Type Description
float

The time in seconds since the last reset.

Source code in vast_pipeline/utils/utils.py
36
37
38
39
40
41
42
43
44
45
46
47
def reset(self) -> float:
    """
    Reset the stopwatch and return the time since last reset (seconds).

    Returns:
        The time in seconds since the last reset.
    """
    now = datetime.now()
    diff = (now - self._last).total_seconds()
    self._last = now

    return diff

reset_init(self)

Reset the stopwatch and return the total time since initialisation.

Returns:

Type Description
float

The time in seconds since the initialisation.

Source code in vast_pipeline/utils/utils.py
49
50
51
52
53
54
55
56
57
58
59
60
def reset_init(self) -> float:
    """
    Reset the stopwatch and return the total time since initialisation.

    Returns:
        The time in seconds since the initialisation.
    """
    now = datetime.now()
    diff = (now - self._init).total_seconds()
    self._last = self._init = now

    return diff

check_read_write_perm(path, perm='W')

Assess the file permission on a path.

Parameters:

Name Type Description Default
path str

The system path to assess.

required
perm str

The permission to check for.

'W'

Returns:

Type Description
None

None

Raises:

Type Description
IOError

The permission is not valid on the checked directory.

Source code in vast_pipeline/utils/utils.py
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def check_read_write_perm(path: str, perm: str='W') -> None:
    """
    Assess the file permission on a path.

    Args:
        path: The system path to assess.
        perm: The permission to check for.

    Returns:
        None

    Raises:
        IOError: The permission is not valid on the checked directory.
    """
    assert perm in ('R', 'W', 'X'), 'permission not supported'

    perm_map = {'R': os.R_OK, 'W': os.W_OK, 'X': os.X_OK}
    if not os.access(path, perm_map[perm]):
        msg = f'permission not valid on folder: {path}'
        logger.error(msg)
        raise IOError(msg)

    pass

deg2dms(deg, dms_format=False, precision=2)

Convert angle in degrees into a DMS formatted string. e.g.

Parameters:

Name Type Description Default
deg float

The angle to convert in degrees.

required
dms_format optional

If True, use "d", "m", and "s" as the coorindate separator, otherwise use ":". Defaults to False.

False
precision optional

Floating point precision of the arcseconds component. Can be 0 or a positive integer. Negative values will be interpreted as 0. Defaults to 2.

2

Returns:

Type Description
str

deg formatted as a DMS string.

Example

deg2dms(12.582438888888889) '+12:34:56.78' deg2dms(2.582438888888889, dms_format=True) '+02d34m56.78s' deg2dms(-12.582438888888889, precision=1) '-12:34:56.8'

Source code in vast_pipeline/utils/utils.py
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def deg2dms(deg: float, dms_format: bool = False, precision: int = 2) -> str:
    """Convert angle in degrees into a DMS formatted string. e.g.

    Args:
        deg: The angle to convert in degrees.
        dms_format (optional): If `True`, use "d", "m", and "s" as the coorindate
            separator, otherwise use ":". Defaults to False.
        precision (optional): Floating point precision of the arcseconds component.
            Can be 0 or a positive integer. Negative values will be interpreted as 0.
            Defaults to 2.

    Returns:
        `deg` formatted as a DMS string.

    Example:
        >>> deg2dms(12.582438888888889)
        '+12:34:56.78'
        >>> deg2dms(2.582438888888889, dms_format=True)
        '+02d34m56.78s'
        >>> deg2dms(-12.582438888888889, precision=1)
        '-12:34:56.8'
    """

    sign, sex = deg2sex(deg)
    signchar = "+" if sign == 1 else "-"
    precision = precision if precision >= 0 else 0
    sec_width = 3 + precision if precision > 0 else 2

    if dms_format:
        return f'{signchar}{sex[0]:02d}d{sex[1]:02d}m{sex[2]:0{sec_width}.{precision}f}s'

    return f'{signchar}{sex[0]:02d}:{sex[1]:02d}:{sex[2]:0{sec_width}.{precision}f}'

deg2hms(deg, hms_format=False, precision=2)

Convert angle in degrees into a HMS formatted string. e.g.

Parameters:

Name Type Description Default
deg float

The angle to convert in degrees.

required
hms_format optional

If True, use "h", "m", and "s" as the coorindate separator, otherwise use ":". Defaults to False.

False
precision optional

Floating point precision of the seconds component. Can be 0 or a positive integer. Negative values will be interpreted as 0. Defaults to 2.

2

Returns:

Type Description
str

deg formatted as an HMS string.

Example

deg2hms(188.73658333333333) '12:34:56.78' deg2hms(-188.73658333333333, hms_format=True) '12h34m56.78s' deg2hms(188.73658333333333, precision=1) '12:34:56.8'

Source code in vast_pipeline/utils/utils.py
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
def deg2hms(deg: float, hms_format: bool = False, precision: int = 2) -> str:
    """Convert angle in degrees into a HMS formatted string. e.g.

    Args:
        deg: The angle to convert in degrees.
        hms_format (optional): If `True`, use "h", "m", and "s" as the coorindate
            separator, otherwise use ":". Defaults to False.
        precision (optional): Floating point precision of the seconds component.
            Can be 0 or a positive integer. Negative values will be interpreted as 0.
            Defaults to 2.

    Returns:
        `deg` formatted as an HMS string.

    Example:
        >>> deg2hms(188.73658333333333)
        '12:34:56.78'
        >>> deg2hms(-188.73658333333333, hms_format=True)
        '12h34m56.78s'
        >>> deg2hms(188.73658333333333, precision=1)
        '12:34:56.8'
    """
    sign, sex = deg2sex(deg / 15.)
    precision = precision if precision >= 0 else 0
    sec_width = 3 + precision if precision > 0 else 2

    if hms_format:
        return f'{sex[0]:02d}h{sex[1]:02d}m{sex[2]:0{sec_width}.{precision}f}s'

    return f'{sex[0]:02d}:{sex[1]:02d}:{sex[2]:0{sec_width}.{precision}f}'

deg2sex(deg)

Converts an angle in degrees to a tuple containing its sexagesimal components.

Parameters:

Name Type Description Default
deg float

The angle to convert in degrees.

required

Returns:

Type Description
int

A nested tuple in the form (sign, (degrees, minutes, seconds)), where sign is

Tuple[float, float, float]

either -1 or 1.

Example

deg2sex(12.582438888888889) (12, 34, 56.78000000000182) deg2sex(-12.582438888888889) (-12, 34, 56.78000000000182)

Source code in vast_pipeline/utils/utils.py
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def deg2sex(deg: float) -> Tuple[int, Tuple[float, float, float]]:
    """
    Converts an angle in degrees to a tuple containing its sexagesimal components.

    Args:
        deg: The angle to convert in degrees.

    Returns:
        A nested tuple in the form (sign, (degrees, minutes, seconds)), where sign is
        either -1 or 1.

    Example:
        >>> deg2sex(12.582438888888889)
        (12, 34, 56.78000000000182)
        >>> deg2sex(-12.582438888888889)
        (-12, 34, 56.78000000000182)
    """

    sign = -1 if deg < 0 else 1
    adeg = abs(deg)
    degf = m.floor(adeg)
    mins = (adeg - degf) * 60.
    minsf = int(m.floor(mins))
    secs = (mins - minsf) * 60.

    return (sign, (degf, minsf, secs))

dict_merge(dct, merge_dct, add_keys=True)

Recursive dict merge. Inspired by dict.update(), instead of updating only top-level keys, dict_merge recurses down into dicts nested to an arbitrary depth, updating keys. The merge_dct is merged into dct.

This version will return a copy of the dictionary and leave the original arguments untouched.

The optional argument add_keys, determines whether keys which are present in merge_dict but not dct should be included in the new dict.

Parameters:

Name Type Description Default
dct dict

onto which the merge is executed

required
merge_dct dict

dct merged into dct

required
add_keys bool

whether to add new keys

True

Returns:

Name Type Description
dict Dict[Any, Any]

updated dict

Source code in vast_pipeline/utils/utils.py
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
def dict_merge(dct: Dict[Any, Any], merge_dct: Dict[Any, Any], add_keys=True) -> Dict[Any, Any]:
    """Recursive dict merge. Inspired by dict.update(), instead of
    updating only top-level keys, dict_merge recurses down into dicts nested
    to an arbitrary depth, updating keys. The `merge_dct` is merged into
    `dct`.

    This version will return a copy of the dictionary and leave the original
    arguments untouched.

    The optional argument `add_keys`, determines whether keys which are
    present in `merge_dict` but not `dct` should be included in the
    new dict.

    Args:
        dct (dict): onto which the merge is executed
        merge_dct (dict): dct merged into dct
        add_keys (bool): whether to add new keys

    Returns:
        dict: updated dict
    """
    dct = dct.copy()
    if not add_keys:
        merge_dct = {k: merge_dct[k] for k in set(dct).intersection(set(merge_dct))}

    for k, v in merge_dct.items():
        if (
            k in dct
            and isinstance(dct[k], dict)
            and isinstance(merge_dct[k], collections.Mapping)
        ):
            dct[k] = dict_merge(dct[k], merge_dct[k], add_keys=add_keys)
        else:
            dct[k] = merge_dct[k]

    return dct

eq_to_cart(ra, dec)

Find the cartesian co-ordinates on the unit sphere given the eq. co-ords. ra, dec should be in degrees.

Parameters:

Name Type Description Default
ra float

The right ascension coordinate, in degrees, to convert.

required
dec float

The declination coordinate, in degrees, to convert.

required

Returns:

Type Description
Tuple[float, float, float]

The cartesian coordinates.

Source code in vast_pipeline/utils/utils.py
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
def eq_to_cart(ra: float, dec: float) -> Tuple[float, float, float]:
    """
    Find the cartesian co-ordinates on the unit sphere given the eq.
    co-ords. ra, dec should be in degrees.

    Args:
        ra: The right ascension coordinate, in degrees, to convert.
        dec: The declination coordinate, in degrees, to convert.

    Returns:
        The cartesian coordinates.
    """
    # TODO: This part of the code can probably be removed along with the
    # storage of these coodinates on the image.
    return (
        m.cos(m.radians(dec)) * m.cos(m.radians(ra)),# Cartesian x
        m.cos(m.radians(dec)) * m.sin(m.radians(ra)),# Cartesian y
        m.sin(m.radians(dec))# Cartesian z
    )

equ2gal(ra, dec)

Convert equatorial coordinates to galactic

Parameters:

Name Type Description Default
ra float

Right ascension in units of degrees.

required
dec float

Declination in units of degrees.

required

Returns:

Name Type Description
Tuple float, float

Galactic longitude and latitude in degrees.

Source code in vast_pipeline/utils/utils.py
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
def equ2gal(ra: float, dec: float) -> Tuple[float, float]:
    """
    Convert equatorial coordinates to galactic

    Args:
        ra (float): Right ascension in units of degrees.
        dec (float): Declination in units of degrees.

    Returns:
        Tuple (float, float): Galactic longitude and latitude in degrees.
    """
    c = SkyCoord(np.float(ra), np.float(dec), unit=(u.deg, u.deg), frame='icrs')
    l = c.galactic.l.deg
    b = c.galactic.b.deg

    return l, b

gal2equ(l, b)

Convert galactic coordinates to equatorial.

Parameters:

Name Type Description Default
l float

Galactic longitude in degrees.

required
b float

Galactic latitude in degrees.

required

Returns:

Name Type Description
Tuple float, float

Right ascension and declination in units of degrees.

Source code in vast_pipeline/utils/utils.py
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
def gal2equ(l: float, b: float) -> Tuple[float, float]:
    """
    Convert galactic coordinates to equatorial.

    Args:
        l (float): Galactic longitude in degrees.
        b (float): Galactic latitude in degrees.

    Returns:
        Tuple (float, float): Right ascension and declination in units of degrees.
    """
    c = SkyCoord(l=np.float(l) * u.deg, b=np.float(b) * u.deg, frame='galactic')
    ra = c.icrs.ra.deg
    dec = c.icrs.dec.deg

    return ra, dec

optimize_floats(df)

Downcast float columns in a pd.DataFrame to the smallest data type without losing any information.

Credit to Robbert van der Gugten.

Parameters:

Name Type Description Default
df pd.DataFrame

input dataframe, no specific columns.

required

Returns:

Type Description
pd.DataFrame

The input dataframe with the float64 type

pd.DataFrame

columns downcasted.

Source code in vast_pipeline/utils/utils.py
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
def optimize_floats(df: pd.DataFrame) -> pd.DataFrame:
    """
    Downcast float columns in a pd.DataFrame to the smallest
    data type without losing any information.

    Credit to Robbert van der Gugten.

    Args:
        df:
            input dataframe, no specific columns.

    Returns:
        The input dataframe with the `float64` type
        columns downcasted.
    """
    floats = df.select_dtypes(include=['float64']).columns.tolist()
    df[floats] = df[floats].apply(pd.to_numeric, downcast='float')

    return df

optimize_ints(df)

Downcast integer columns in a pd.DataFrame to the smallest data type without losing any information.

Credit to Robbert van der Gugten.

Parameters:

Name Type Description Default
df pd.DataFrame

Input dataframe, no specific columns.

required

Returns:

Type Description
pd.DataFrame

The input dataframe with the int64 type

pd.DataFrame

columns downcasted.

Source code in vast_pipeline/utils/utils.py
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
def optimize_ints(df: pd.DataFrame) -> pd.DataFrame:
    """
    Downcast integer columns in a pd.DataFrame to the smallest
    data type without losing any information.

    Credit to Robbert van der Gugten.

    Args:
        df:
            Input dataframe, no specific columns.

    Returns:
        The input dataframe with the `int64` type
        columns downcasted.
    """
    ints = df.select_dtypes(include=['int64']).columns.tolist()
    df[ints] = df[ints].apply(pd.to_numeric, downcast='integer')

    return df

parse_coord(coord_string, coord_frame='icrs')

the following assumptions are made
  • if both coordinate components are decimals, they are assumed to be in degrees.
  • if a sexagesimal coordinate is given and the frame is galactic, both components are assumed to be in degrees. For any other frame, the first component is assumed to be in hourangles and the second in degrees.

Parse a coordinate string and return a SkyCoord. The units may be expressed within coord_string e.g. "21h52m03.1s -62d08m19.7s", "18.4d +43.1d". If no units are given, Will raise a ValueError if SkyCoord is unable to parse coord_string.

Parameters:

Name Type Description Default
coord_string str

The coordinate string to parse.

required
coord_frame str

The frame of coord_string. Defaults to "icrs".

'icrs'

Returns:

Type Description
SkyCoord

SkyCoord

Source code in vast_pipeline/utils/utils.py
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
def parse_coord(coord_string: str, coord_frame: str = "icrs") -> SkyCoord:
    """Parse a coordinate string and return a SkyCoord. The units may be expressed within
    `coord_string` e.g. "21h52m03.1s -62d08m19.7s", "18.4d +43.1d". If no units are given,
    the following assumptions are made:
        - if both coordinate components are decimals, they are assumed to be in degrees.
        - if a sexagesimal coordinate is given and the frame is galactic, both components
            are assumed to be in degrees. For any other frame, the first component is
            assumed to be in hourangles and the second in degrees.
    Will raise a ValueError if SkyCoord is unable to parse `coord_string`.

    Args:
        coord_string (str): The coordinate string to parse.
        coord_frame (str, optional): The frame of `coord_string`. Defaults to "icrs".

    Returns:
        SkyCoord
    """
    # if both coord components are decimals, assume they're in degrees, otherwise assume
    # hourangles and degrees. Note that the unit parameter is ignored if the units are
    # not ambiguous i.e. if coord_string contains the units (e.g. 18.4d, 5h35m, etc)
    try:
        _ = [float(x) for x in coord_string.split()]
        unit = "deg"
    except ValueError:
        if coord_frame == "galactic":
            unit = "deg"
        else:
            unit = "hourangle,deg"

    coord = SkyCoord(coord_string, unit=unit, frame=coord_frame)

    return coord

Last update: March 2, 2022
Created: March 2, 2022