File size: 3,331 Bytes
38622c4
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
cc02af3
 
 
 
 
 
 
 
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
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
"""Завантаження відкритих довідкових БД (детерміноване джерело для L1/L2a/L2b).

Усі коди узгоджені між файлами: RxNorm (ліки), LOINC (аналізи), SNOMED (діагнози).

Loader-и кешуються (`lru_cache`) і повертають **незмінні** структури: заморожені
Pydantic-моделі (`refmodels.FrozenModel`) у tuple/MappingProxyType. Це водночас
прибирає перебудову об'єктів на кожен виклик і унеможливлює псування спільного
довідника випадковою мутацією (раніше від цього захищали свіжі копії).
"""
from __future__ import annotations

import json
from collections.abc import Mapping
from functools import lru_cache
from types import MappingProxyType
from typing import Any, Optional

from .config import DATA_DIR
from .refmodels import (
    CondTerm,
    DoseLimit,
    Interaction,
    LabTerm,
    MedTerm,
    RangeSpec,
    RefRange,
    Vocab,
)


@lru_cache(maxsize=None)
def _raw(name: str) -> Any:
    return json.loads((DATA_DIR / name).read_text(encoding="utf-8"))


@lru_cache(maxsize=None)
def load_interactions() -> tuple[Interaction, ...]:
    """Пари RxNorm-кодів із тяжкістю взаємодії."""
    return tuple(Interaction(**p) for p in _raw("interactions.json")["pairs"])


@lru_cache(maxsize=None)
def load_dose_limits() -> Mapping[str, DoseLimit]:
    """RxNorm → DoseLimit."""
    return MappingProxyType(
        {d["rxnorm"]: DoseLimit(**d) for d in _raw("dose_limits.json")["limits"]}
    )


@lru_cache(maxsize=None)
def load_range_specs() -> Mapping[str, RangeSpec]:
    """LOINC → RangeSpec (з опціональними варіантами за статтю)."""
    return MappingProxyType({
        loinc: RangeSpec(**spec)
        for loinc, spec in _raw("reference_ranges.json")["ranges"].items()
    })


def range_for(loinc: str, sex: Optional[str] = None) -> Optional[RefRange]:
    """Референсні межі показника, розвʼязані під стать пацієнта (або None)."""
    spec = load_range_specs().get(loinc)
    return spec.resolve(sex) if spec is not None else None


@lru_cache(maxsize=None)
def load_vocab() -> Vocab:
    """Словник для StubExtractor: labs / medications / conditions."""
    raw = _raw("vocab.json")
    return Vocab(
        labs={k: LabTerm(**v) for k, v in raw.get("labs", {}).items()},
        medications={k: MedTerm(**v) for k, v in raw.get("medications", {}).items()},
        conditions={k: CondTerm(**v) for k, v in raw.get("conditions", {}).items()},
    )


def load_sample_docs(lang: str = "uk") -> list[dict]:
    """Демо-документи мовою ``lang`` (``uk``|``en``). Щоразу свіжа копія.

    Набори дзеркальні: ті самі значення й ті самі призначення, тож сигнали
    безпеки збігаються — демо не залежить від того, якою мовою його дивляться.
    """
    key = "en" if lang == "en" else "uk"
    return json.loads(json.dumps(_raw("sample_docs.json")[key]))