<filename>examples/discriminative_reranking_nmt/huggingface_generation/common/utils.py
import itertools
import os
import pickle
from pathlib import Path
from typing import Any, List, Dict, Iterable, Union, Optional

from torch import Tensor

project_dir = Path(__file__).parent.parent.parent

LABELS_PAD = -100


def read_lines(path: Path) -> List[str]:
    content = path.read_text()
    lines = content.split('\n')
    if lines[-1] == '':
        lines = lines[:-1]
    return lines


def append_lines(lines: List[str], path: Path) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    lines = [line.replace('\n', ' ') for line in lines]
    content = '\n'.join(lines) + '\n'
    with open(str(path), 'a') as f:
        f.write(content)


def get_hf_cache_dir() -> str:
    master_cache_dir = os.environ.get("XDG_CACHE_HOME", "~/.cache")
    hf_cache_dir = Path(master_cache_dir) / "huggingface"
    return str(hf_cache_dir)


def dump_pickle(to_dump: Any, path: Union[str, Path]) -> None:
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    with open(path, 'wb') as f:
        pickle.dump(to_dump, f)


def load_pickle(path: Union[str, Path]) -> Any:
    with open(path, 'rb') as f:
        obj = pickle.load(f)
    return obj


def flatten(list_of_lists: Iterable[Iterable]) -> List:
    return list(itertools.chain.from_iterable(list_of_lists))


def dict_of_lists_to_list_of_dicts(dict_of_lists: Dict[Any, List]) -> List[Dict]:
    return [dict(zip(dict_of_lists.keys(), t)) for t in zip(*dict_of_lists.values())]


def list_of_dicts_to_dict_of_lists(list_of_dicts: List[Dict]) -> Dict[Any, List]:
    return {k: [dic[k] for dic in list_of_dicts] for k in list_of_dicts[0].keys()}


def convert_tensor_items_to_lists(d: Dict) -> Dict:
    d = dict(d)
    for key, value in d.items():
        if isinstance(value, Tensor):
            d[key] = value.cpu().tolist()
    return d


def attach_remote_debugger(ip: Optional[str], port: Optional[int]):
    if (ip, port) == (None, None):
        return
    elif None in (ip, port):
        raise ValueError("cfg.common.remote_debug: must provide both ip and port, or neither.")

    import pydevd_pycharm
    pydevd_pycharm.settrace(host=ip, port=port, stdoutToServer=True, stderrToServer=True)

<filename>examples/discriminative_reranking_nmt/huggingface_generation/common/utils.py
import itertools
import os
import pickle
from pathlib import Path
from typing import Any, List, Dict, Iterable, Union, Optional

from torch import Tensor

project_dir = Path(__file__).parent.parent.parent

LABELS_PAD = -100


def read_lines(path: Path) -> List[str]:
    content = path.read_text()
    lines = content.split('\n')
    if lines[-1] == '':
        lines = lines[:-1]
    return lines


def append_lines(lines: List[str], path: Path) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    lines = [line.replace('\n', ' ') for line in lines]
    content = '\n'.join(lines) + '\n'
    with open(str(path), 'a') as f:
        f.write(content)


def get_hf_cache_dir() -> str:
    master_cache_dir = os.environ.get("XDG_CACHE_HOME", "~/.cache")
    hf_cache_dir = Path(master_cache_dir) / "huggingface"
    return str(hf_cache_dir)


def dump_pickle(to_dump: Any, path: Union[str, Path]) -> None:
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    with open(path, 'wb') as f:
        pickle.dump(to_dump, f)


def load_pickle(path: Union[str, Path]) -> Any:
    with open(path, 'rb') as f:
        obj = pickle.load(f)
    return obj


def flatten(list_of_lists: Iterable[Iterable]) -> List:
    return list(itertools.chain.from_iterable(list_of_lists))


def dict_of_lists_to_list_of_dicts(dict_of_lists: Dict[Any, List]) -> List[Dict]:
    return [dict(zip(dict_of_lists.keys(), t)) for t in zip(*dict_of_lists.values())]


def list_of_dicts_to_dict_of_lists(list_of_dicts: List[Dict]) -> Dict[Any, List]:
    return {k: [dic[k] for dic in list_of_dicts] for k in list_of_dicts[0].keys()}


def convert_tensor_items_to_lists(d: Dict) -> Dict:
    d = dict(d)
    for key, value in d.items():
        if isinstance(value, Tensor):
            d[key] = value.cpu().tolist()
    return d


def attach_remote_debugger(ip: Optional[str], port: Optional[int]):
    if (ip, port) == (None, None):
        return
    elif None in (ip, port):
        raise ValueError("cfg.common.remote_debug: must provide both ip and port, or neither.")

    import pydevd_pycharm
    pydevd_pycharm.settrace(host=ip, port=port, stdoutToServer=True, stderrToServer=True)
