use std::collections::HashMap;
use super::columns::{
read_table_deferring, Column, RawTable, TableWriter, TextColumn, Values, DOCUMENT_COLUMNS, EVENT_COLUMNS,
LINK_COLUMNS,
};
use super::model::{
Bounds, ClinicalRecords, Descriptor, Document, Event, Link, DESCRIPTOR, DOCUMENTS, EVENTS, GROUP, LINKS, SCHEMA,
};
use super::select::{Selection, SelectionPolicy};
use crate::h5::data;
use crate::h5::ops;
use crate::json::repr_str;
use crate::storage::codecs::CodecProfile;
use crate::{Error, Result};
#[derive(Clone, Copy)]
struct Cells<'a>(Option<&'a Column>);
impl<'a> Cells<'a> {
fn of(table: &'a RawTable, name: &str) -> Cells<'a> {
Cells(table.column(name))
}
fn text(self, i: usize) -> Option<String> {
self.0.and_then(|c| c.text(i))
}
fn i64(self, i: usize) -> Option<i64> {
self.0.and_then(|c| c.i64(i))
}
fn u64(self, i: usize) -> Option<u64> {
self.0.and_then(|c| c.u64(i))
}
fn f64(self, i: usize) -> Option<f64> {
self.0.and_then(|c| c.f64(i))
}
}
fn bounds(lo: Cells<'_>, hi: Cells<'_>, i: usize) -> Option<Bounds> {
match (lo.i64(i), hi.i64(i)) {
(Some(lo), Some(hi)) => Some(Bounds { lo, hi }),
_ => None,
}
}
pub fn events_of(table: &RawTable) -> Vec<Event> {
let c = |name: &str| Cells::of(table, name);
let (event_id, record_id, kind, temporal_type, status) =
(c("event_id"), c("record_id"), c("kind"), c("temporal_type"), c("status"));
let (start_lo, start_hi, end_lo, end_hi, available_lo, available_hi) = (
c("effective_start_lo_us"),
c("effective_start_hi_us"),
c("effective_end_lo_us"),
c("effective_end_hi_us"),
c("available_lo_us"),
c("available_hi_us"),
);
let (timepoint_id, encounter_id, code_system, code, code_version) =
(c("timepoint_id"), c("encounter_id"), c("code_system"), c("code"), c("code_version"));
let (value_num, value_comparator, unit, value_text, missing_reason, prov) =
(c("value_num"), c("value_comparator"), c("unit"), c("value_text"), c("missing_reason"), c("prov"));
(0..table.rows)
.map(|i| Event {
event_id: event_id.text(i).unwrap_or_default(),
record_id: record_id.text(i).unwrap_or_default(),
kind: kind.text(i).unwrap_or_default(),
temporal_type: temporal_type.text(i).unwrap_or_default(),
effective_start: bounds(start_lo, start_hi, i),
effective_end: bounds(end_lo, end_hi, i),
available: bounds(available_lo, available_hi, i),
status: status.text(i).unwrap_or_default(),
timepoint_id: timepoint_id.text(i),
encounter_id: encounter_id.text(i),
code_system: code_system.text(i),
code: code.text(i),
code_version: code_version.text(i),
value_num: value_num.f64(i),
value_comparator: value_comparator.text(i),
unit: unit.text(i),
value_text: value_text.text(i),
missing_reason: missing_reason.text(i),
prov: prov.text(i),
})
.collect()
}
pub fn documents_of(table: &RawTable) -> Vec<Document> {
let c = |name: &str| Cells::of(table, name);
let (document_id, media_type, text, language, source_type) =
(c("document_id"), c("media_type"), c("text"), c("language"), c("source_type"));
(0..table.rows)
.map(|i| Document {
document_id: document_id.text(i).unwrap_or_default(),
media_type: media_type.text(i).unwrap_or_default(),
text: text.text(i).unwrap_or_default(),
language: language.text(i),
source_type: source_type.text(i),
})
.collect()
}
pub fn links_of(table: &RawTable) -> Vec<Link> {
let c = |name: &str| Cells::of(table, name);
let (source_type, source_id, relation, target_type, target_id) =
(c("source_type"), c("source_id"), c("relation"), c("target_type"), c("target_id"));
let (start, end, target_annotation_id, asserted_by_event_id) =
(c("source_start"), c("source_end"), c("target_annotation_id"), c("asserted_by_event_id"));
(0..table.rows)
.map(|i| Link {
source_type: source_type.text(i).unwrap_or_default(),
source_id: source_id.text(i).unwrap_or_default(),
relation: relation.text(i).unwrap_or_default(),
target_type: target_type.text(i).unwrap_or_default(),
target_id: target_id.text(i).unwrap_or_default(),
source_span: match (start.u64(i), end.u64(i)) {
(Some(a), Some(b)) => Some((a, b)),
_ => None,
},
target_annotation_id: target_annotation_id.text(i),
asserted_by_event_id: asserted_by_event_id.text(i),
})
.collect()
}
pub fn storage_order(events: &[Event]) -> Vec<usize> {
let mut order: Vec<usize> = (0..events.len()).collect();
order.sort_by_key(|&i| {
let e = &events[i];
(e.effective_start.is_none(), e.effective_start.map(|s| (s.lo, s.hi)), e.event_id.as_str())
});
order
}
pub fn write(root: &hdf5::Group, records: &ClinicalRecords, profile: &CodecProfile) -> Result<()> {
if ops::exists(root, GROUP) {
return Err(Error::invalid(format!("{} already exists; it is replaced, never merged", repr_str(GROUP))));
}
let group = ops::new_group(root, GROUP)?;
data::create_scalar_string(&group, DESCRIPTOR, &records.descriptor.dumps())?;
write_events(&ops::new_group(&group, EVENTS)?, &records.events, profile)?;
if !records.documents.is_empty() {
write_documents(&ops::new_group(&group, DOCUMENTS)?, &records.documents, profile)?;
}
if !records.links.is_empty() {
write_links(&ops::new_group(&group, LINKS)?, &records.links, profile)?;
}
Ok(())
}
fn write_events(group: &hdf5::Group, events: &[Event], profile: &CodecProfile) -> Result<()> {
let rows: Vec<&Event> = storage_order(events).into_iter().map(|i| &events[i]).collect();
let mut w = TableWriter::new(group.clone(), profile, rows.len());
let text = |f: fn(&Event) -> Option<&str>| -> Vec<Option<&str>> { rows.iter().map(|e| f(e)).collect() };
let lo =
|f: fn(&Event) -> Option<Bounds>| -> Vec<Option<i64>> { rows.iter().map(|e| f(e).map(|b| b.lo)).collect() };
let hi =
|f: fn(&Event) -> Option<Bounds>| -> Vec<Option<i64>> { rows.iter().map(|e| f(e).map(|b| b.hi)).collect() };
for spec in &EVENT_COLUMNS {
match spec.name {
"event_id" => w.utf8(spec, &text(|e| Some(e.event_id.as_str())))?,
"record_id" => w.utf8(spec, &text(|e| Some(e.record_id.as_str())))?,
"kind" => w.utf8(spec, &text(|e| Some(e.kind.as_str())))?,
"temporal_type" => w.utf8(spec, &text(|e| Some(e.temporal_type.as_str())))?,
"effective_start_lo_us" => w.i64(spec, &lo(|e| e.effective_start))?,
"effective_start_hi_us" => w.i64(spec, &hi(|e| e.effective_start))?,
"effective_end_lo_us" => w.i64(spec, &lo(|e| e.effective_end))?,
"effective_end_hi_us" => w.i64(spec, &hi(|e| e.effective_end))?,
"available_lo_us" => w.i64(spec, &lo(|e| e.available))?,
"available_hi_us" => w.i64(spec, &hi(|e| e.available))?,
"status" => w.utf8(spec, &text(|e| Some(e.status.as_str())))?,
"timepoint_id" => w.utf8(spec, &text(|e| e.timepoint_id.as_deref()))?,
"encounter_id" => w.utf8(spec, &text(|e| e.encounter_id.as_deref()))?,
"code_system" => w.utf8(spec, &text(|e| e.code_system.as_deref()))?,
"code" => w.utf8(spec, &text(|e| e.code.as_deref()))?,
"code_version" => w.utf8(spec, &text(|e| e.code_version.as_deref()))?,
"value_num" => w.f64(spec, &rows.iter().map(|e| e.value_num).collect::<Vec<_>>())?,
"value_comparator" => w.utf8(spec, &text(|e| e.value_comparator.as_deref()))?,
"unit" => w.utf8(spec, &text(|e| e.unit.as_deref()))?,
"value_text" => w.utf8(spec, &text(|e| e.value_text.as_deref()))?,
"missing_reason" => w.utf8(spec, &text(|e| e.missing_reason.as_deref()))?,
"prov" => w.utf8(spec, &text(|e| e.prov.as_deref()))?,
other => unreachable!("event column {other} has no writer"),
}
}
Ok(())
}
fn write_documents(group: &hdf5::Group, documents: &[Document], profile: &CodecProfile) -> Result<()> {
let mut rows: Vec<&Document> = documents.iter().collect();
rows.sort_by(|a, b| a.document_id.cmp(&b.document_id));
let mut w = TableWriter::new(group.clone(), profile, rows.len());
let text = |f: fn(&Document) -> Option<&str>| -> Vec<Option<&str>> { rows.iter().map(|d| f(d)).collect() };
for spec in &DOCUMENT_COLUMNS {
match spec.name {
"document_id" => w.utf8(spec, &text(|d| Some(d.document_id.as_str())))?,
"media_type" => w.utf8(spec, &text(|d| Some(d.media_type.as_str())))?,
"text" => w.utf8(spec, &text(|d| Some(d.text.as_str())))?,
"language" => w.utf8(spec, &text(|d| d.language.as_deref()))?,
"source_type" => w.utf8(spec, &text(|d| d.source_type.as_deref()))?,
other => unreachable!("document column {other} has no writer"),
}
}
Ok(())
}
fn write_links(group: &hdf5::Group, links: &[Link], profile: &CodecProfile) -> Result<()> {
let mut rows: Vec<&Link> = links.iter().collect();
rows.sort();
let mut w = TableWriter::new(group.clone(), profile, rows.len());
let text = |f: fn(&Link) -> Option<&str>| -> Vec<Option<&str>> { rows.iter().map(|l| f(l)).collect() };
for spec in &LINK_COLUMNS {
match spec.name {
"source_type" => w.utf8(spec, &text(|l| Some(l.source_type.as_str())))?,
"source_id" => w.utf8(spec, &text(|l| Some(l.source_id.as_str())))?,
"relation" => w.utf8(spec, &text(|l| Some(l.relation.as_str())))?,
"target_type" => w.utf8(spec, &text(|l| Some(l.target_type.as_str())))?,
"target_id" => w.utf8(spec, &text(|l| Some(l.target_id.as_str())))?,
"source_start" => w.u64(spec, &rows.iter().map(|l| l.source_span.map(|s| s.0)).collect::<Vec<_>>())?,
"source_end" => w.u64(spec, &rows.iter().map(|l| l.source_span.map(|s| s.1)).collect::<Vec<_>>())?,
"target_annotation_id" => w.utf8(spec, &text(|l| l.target_annotation_id.as_deref()))?,
"asserted_by_event_id" => w.utf8(spec, &text(|l| l.asserted_by_event_id.as_deref()))?,
other => unreachable!("link column {other} has no writer"),
}
}
Ok(())
}
pub fn recognised(root: &hdf5::Group) -> bool {
let Some(group) = ops::child_group(root, GROUP) else { return false };
let Some(meta) = ops::child_dataset(&group, DESCRIPTOR) else { return false };
let Ok(text) = data::read_scalar_string(&meta) else { return false };
match crate::json::loads(&text) {
Ok(value) => value.get("schema").and_then(serde_json::Value::as_str) == Some(SCHEMA),
Err(_) => false,
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DocumentInfo {
pub document_id: String,
pub media_type: String,
pub language: Option<String>,
pub source_type: Option<String>,
pub n_bytes: u64,
row: usize,
}
#[derive(Debug, Default)]
pub struct Documents {
infos: Vec<DocumentInfo>,
by_id: HashMap<String, usize>,
text: Option<TextColumn>,
}
impl Documents {
pub fn open(root: &hdf5::Group, projection: bool) -> Result<Option<Documents>> {
let Some(group) = ops::child_group(root, GROUP) else { return Ok(None) };
let table = match ops::child_group(&group, DOCUMENTS) {
None => None,
Some(g) => {
let table = read_table_deferring(&g, DOCUMENTS, &DOCUMENT_COLUMNS, projection, &["text"])?;
refuse(&table)?;
Some(table)
}
};
Documents::from_table(root, table.as_ref()).map(Some)
}
fn from_table(root: &hdf5::Group, table: Option<&RawTable>) -> Result<Documents> {
let Some(table) = table else { return Ok(Documents::default()) };
refuse(table)?;
let column = table.column("text");
let text = match column.map(|c| &c.values) {
Some(Values::Deferred { offsets }) => {
let g = root.group(&format!("{GROUP}/{DOCUMENTS}"))?;
Some(TextColumn::open(&g, "text", offsets.clone(), "/clinical/documents/text")?)
}
Some(_) => return Err(Error::invalid("document text is read on demand: defer the `text` column")),
None => None,
};
let c = |name: &str| Cells::of(table, name);
let (document_id, media_type, language, source_type) =
(c("document_id"), c("media_type"), c("language"), c("source_type"));
let infos: Vec<DocumentInfo> = (0..table.rows)
.map(|i| DocumentInfo {
document_id: document_id.text(i).unwrap_or_default(),
media_type: media_type.text(i).unwrap_or_default(),
language: language.text(i),
source_type: source_type.text(i),
n_bytes: column.and_then(|c| c.span(i)).map_or(0, |(a, b)| b - a),
row: i,
})
.collect();
let by_id = infos.iter().enumerate().map(|(i, d)| (d.document_id.clone(), i)).collect();
Ok(Documents { infos, by_id, text })
}
pub fn infos(&self) -> &[DocumentInfo] {
&self.infos
}
pub fn len(&self) -> usize {
self.infos.len()
}
pub fn is_empty(&self) -> bool {
self.infos.is_empty()
}
pub fn info(&self, document_id: &str) -> Option<&DocumentInfo> {
self.by_id.get(document_id).map(|i| &self.infos[*i])
}
pub fn text(&self, document_id: &str) -> Result<String> {
let info = self.info(document_id).ok_or_else(|| {
crate::sample::reader::missing_key(
"document",
document_id,
self.infos.iter().map(|d| d.document_id.clone()),
)
})?;
match &self.text {
Some(column) => column.cell(info.row),
None => Ok(String::new()),
}
}
pub fn document(&self, document_id: &str) -> Result<Document> {
let text = self.text(document_id)?;
let info = self.info(document_id).expect("found above");
Ok(Document {
document_id: info.document_id.clone(),
media_type: info.media_type.clone(),
text,
language: info.language.clone(),
source_type: info.source_type.clone(),
})
}
pub fn all(&self) -> Result<Vec<Document>> {
let mut texts = match &self.text {
Some(column) => column.cells()?,
None => Vec::new(),
};
Ok(self
.infos
.iter()
.map(|d| Document {
document_id: d.document_id.clone(),
media_type: d.media_type.clone(),
text: texts.get_mut(d.row).map(std::mem::take).unwrap_or_default(),
language: d.language.clone(),
source_type: d.source_type.clone(),
})
.collect())
}
}
#[derive(Debug)]
pub struct Clinical {
pub descriptor: Descriptor,
pub events: Vec<Event>,
pub links: Vec<Link>,
pub projection: bool,
documents: Documents,
by_event: HashMap<String, usize>,
}
fn refuse(table: &RawTable) -> Result<()> {
match table.problems.iter().find(|p| !p.code.starts_with('W')) {
None => Ok(()),
Some(p) => Err(Error::coded(
p.code,
format!(
"{}: {} (the clinical tables are malformed; `medh5 validate` lists every problem)",
p.location, p.message
),
)),
}
}
impl Clinical {
pub fn open(root: &hdf5::Group, projection: bool) -> Result<Option<Clinical>> {
let Some(group) = ops::child_group(root, GROUP) else { return Ok(None) };
let meta =
ops::child_dataset(&group, DESCRIPTOR).ok_or_else(|| Error::coded("E801", "`clinical/meta` is absent"))?;
let descriptor = Descriptor::loads(&data::read_scalar_string(&meta)?)?;
let read = |name: &str, specs, deferred: &[&str]| -> Result<Option<RawTable>> {
match ops::child_group(&group, name) {
None => Ok(None),
Some(g) => {
let table = read_table_deferring(&g, name, specs, projection, deferred)?;
refuse(&table)?;
Ok(Some(table))
}
}
};
let events = read(EVENTS, &EVENT_COLUMNS[..], &[])?
.ok_or_else(|| Error::coded("E804", "`clinical/events` is absent"))?;
let links = read(LINKS, &LINK_COLUMNS[..], &[])?;
let documents = read(DOCUMENTS, &DOCUMENT_COLUMNS[..], &["text"])?;
Clinical::from_tables(root, descriptor, &events, documents.as_ref(), links.as_ref(), projection).map(Some)
}
pub fn from_tables(
root: &hdf5::Group,
descriptor: Descriptor,
events: &RawTable,
documents: Option<&RawTable>,
links: Option<&RawTable>,
projection: bool,
) -> Result<Clinical> {
for table in [Some(events), links].into_iter().flatten() {
refuse(table)?;
}
let documents = Documents::from_table(root, documents)?;
let events = events_of(events);
let links = links.map(links_of).unwrap_or_default();
let by_event = events.iter().enumerate().map(|(i, e)| (e.event_id.clone(), i)).collect();
Ok(Clinical { descriptor, events, links, projection, documents, by_event })
}
pub fn event(&self, event_id: &str) -> Option<&Event> {
self.by_event.get(event_id).map(|i| &self.events[*i])
}
pub fn documents(&self) -> &[DocumentInfo] {
self.documents.infos()
}
pub fn document_info(&self, document_id: &str) -> Option<&DocumentInfo> {
self.documents.info(document_id)
}
pub fn text(&self, document_id: &str) -> Result<String> {
self.documents.text(document_id)
}
pub fn document(&self, document_id: &str) -> Result<Document> {
self.documents.document(document_id)
}
pub fn records(&self) -> Result<ClinicalRecords> {
Ok(ClinicalRecords {
descriptor: self.descriptor.clone(),
events: self.events.clone(),
documents: self.documents.all()?,
links: self.links.clone(),
})
}
#[doc(hidden)]
pub fn select(&self, cutoff: i64, policy: &SelectionPolicy) -> Result<Selection> {
if self.projection {
return Err(Error::Version(format!(
"this clinical profile is a later minor version's, read as a projection: this engine implements MEDH5 \
up to {}, and a selection from it would claim what that version may define otherwise, so it is \
refused (1.1 §2.2)",
crate::version::LATEST_VERSION
)));
}
let links: Vec<(usize, &Link)> = self.links.iter().map(|l| (0, l)).collect();
super::select::select(&self.events, &links, cutoff, policy)
}
pub fn links_from<'a>(&'a self, source_type: &'a str, source_id: &'a str) -> impl Iterator<Item = &'a Link> + 'a {
self.links.iter().filter(move |l| l.source_type == source_type && l.source_id == source_id)
}
pub fn links_to<'a>(&'a self, target_type: &'a str, target_id: &'a str) -> impl Iterator<Item = &'a Link> + 'a {
self.links.iter().filter(move |l| l.target_type == target_type && l.target_id == target_id)
}
pub fn summary(&self) -> serde_json::Value {
let mut kinds: std::collections::BTreeMap<&str, usize> = Default::default();
for e in &self.events {
*kinds.entry(e.kind.as_str()).or_default() += 1;
}
let records: std::collections::BTreeSet<&str> = self.events.iter().map(|e| e.record_id.as_str()).collect();
serde_json::json!({
"clock": self.descriptor.clock.to_json(),
"events": self.events.len(),
"records": records.len(),
"kinds": kinds,
"documents": self.documents.len(),
"links": self.links.len(),
"unknown_availability": self.events.iter().filter(|e| e.available.is_none()).count(),
})
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn s4_events_are_stored_by_known_start_then_id() {
let at = |id: &str, start: Option<(i64, i64)>| Event {
event_id: id.into(),
effective_start: start.map(|(lo, hi)| Bounds::new(lo, hi)),
..Default::default()
};
let events = [
at("u2", None),
at("b", Some((5, 9))),
at("a", Some((5, 9))),
at("u1", None),
at("c", Some((5, 6))),
at("z", Some((-3, -3))),
];
let ids: Vec<&str> = storage_order(&events).into_iter().map(|i| events[i].event_id.as_str()).collect();
assert_eq!(ids, ["z", "c", "a", "b", "u1", "u2"]);
}
}