about summary refs log tree commit diff
path: root/uploader/datautils.py
blob: d132c4283905e8ecff0fa9778bf9d94d306f5d0c (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
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
"""Generic data utilities: Rename module."""
import math
import json
import base64
from functools import reduce
from typing import Union, Sequence

def enumerate_sequence(seq: Sequence[dict], start:int = 1) -> Sequence[dict]:
    """Enumerate sequence beginning at 1"""
    return tuple({**item, "sequence_number": seqno}
                 for seqno, item in enumerate(seq, start=start))


def order_by_family(items: tuple[dict, ...],
                    family_key: str = "Family",
                    order_key: str = "FamilyOrderId") -> list:
    """Order the populations by their families."""
    def __family_order__(item):
        orderval = item[order_key]
        return math.inf if orderval is None else orderval

    def __order__(ordered, current):
        _key = (__family_order__(current), current[family_key])
        return {
            **ordered,
            _key: ordered.get(_key, tuple()) + (current,)
        }

    return sorted(tuple(reduce(__order__, items, {}).items()),
                  key=lambda item: item[0][0])


def safe_int(val: Union[str, int, float]) -> int:
    """
    Convert val into an integer: if val cannot be converted, return a zero.
    """
    try:
        return int(val)
    except ValueError:
        return 0


def base64_encode_dict(dct: dict, **kwargs) -> bytes:
    """Base64 encode a dictionary. Takes the same keywords as `json.dumps` function."""
    return base64.urlsafe_b64encode(json.dumps(dct, **kwargs).encode("utf-8"))


def base64_decode_to_dict(value: str, **kwargs) -> dict:
    """Base64 encode a dictionary. Takes the same keywords as `json.loads` function."""
    return json.loads(base64.urlsafe_b64decode(value), **kwargs)