Source code for esrf_pathlib._schemas.definitions.esrf_utils

import datetime
import os
import re
from enum import Enum
from typing import Dict
from typing import Tuple
from typing import Type
from typing import Union

from .types import ConceptValueType

_BEAMLINE_NAME_TO_DIR = {
    "id23-1": "id23eh1",
    "id23-2": "id23eh2",
    "id30a-1": "id30a1",
    "id30a-2": "id30a2",
    "id30a-3": "id30a3",
}
_BEAMLINE_DIR_TO_NAME = {v: k for k, v in _BEAMLINE_NAME_TO_DIR.items()}

# Names of proposal, beamline, sample, dataset
_FORBIDDEN_CHARS_BASE = {os.sep, "{", "}", "%", "\x00"}
_FORBIDDEN_CHARS_WINDOWS = {"<", ">", ":", '"', "'", "/", "\\", "|", "?", "*"} | {
    chr(i) for i in range(32)
}
_FORBIDDEN_CHARS_URL = {"@", "#", ":", "?"}

_FORBIDDEN_CHARS = (
    _FORBIDDEN_CHARS_BASE | _FORBIDDEN_CHARS_WINDOWS | _FORBIDDEN_CHARS_URL
)
_BLISS_ALLOWED_CHARS = f"[^{re.escape(''.join(_FORBIDDEN_CHARS))}]"
# BLISS_NAME_PATTERN = f"{_BLISS_ALLOWED_CHARS}+"
BLISS_NAME_PATTERN = rf"{_BLISS_ALLOWED_CHARS}+(?<!\.h5)"

SESSION_DATE_PATTERN = r"\d{4}(?:0[1-9]|1[0-2])(?:0[1-9]|[12]\d|3[01])"  # YYYYMMDD




def deserialize_session_date(
    session_date: Union[str, datetime.date, datetime.datetime],
) -> datetime.date:
    if isinstance(session_date, datetime.datetime):
        return session_date.date()
    if isinstance(session_date, datetime.date):
        return session_date
    return datetime.datetime.strptime(session_date, "%Y%m%d").date()





def serialize_session_date(
    session_date: Union[str, datetime.date, datetime.datetime],
) -> str:
    if isinstance(session_date, str):
        return session_date
    if isinstance(session_date, datetime.datetime):
        session_date = session_date.date()
    return session_date.strftime("%Y%m%d")





def serialize_beamline(normalized_beamline: str) -> str:
    name = normalized_beamline.lower()
    return _BEAMLINE_NAME_TO_DIR.get(name, name)





def derive_normalized_beamline(values: Dict[str, ConceptValueType]) -> str:
    name = values["beamline"].lower()
    return _BEAMLINE_DIR_TO_NAME.get(name, name)





def deserializer_enum_type(
    data_type: Union[str, Enum], enum_type: Type[Enum], string_to_enum: Dict[str, Enum]
) -> Enum:
    if isinstance(data_type, str) and data_type in string_to_enum:
        return string_to_enum[data_type]
    return enum_type(data_type)





def serializer_enum_type(
    data_type: Union[str, Enum], enum_type: Type[Enum], string_to_enum: Dict[str, Enum]
) -> str:
    data_type = deserializer_enum_type(data_type, enum_type, string_to_enum)
    enum_to_string = {v: k for k, v in string_to_enum.items()}
    return enum_to_string[data_type]





def named_directory(name: str) -> Tuple[str, str, bool]:
    """Use for a named directory with optionally unknown content."""
    return ("{unknown_part}", name, False)