Skip to content

Commit ba1e501

Browse files
bingli621bingli621
committed
implementing nxdict class
1 parent 9143b19 commit ba1e501

17 files changed

+489
-321
lines changed

‎.DS_Store

0 Bytes
Binary file not shown.

‎src/tavi/data/nxdict.py

+121-184
Original file line numberDiff line numberDiff line change
@@ -1,196 +1,133 @@
1-
from datetime import datetime
21
from typing import Optional
32

3+
import numpy as np
4+
5+
6+
class NexusDict(object):
7+
"""Read in dictionaries from DAS logs, instrument configuration json or sample json,
8+
format into NeXus style: nxentry_dict = {"attrs":{"attr1":attr1, ...}, "dataset":dataset}
9+
10+
Attributes:
11+
daslogs_dict (dict)
12+
instrument_dict (dict)
13+
sample_dict (dict)
14+
nxentry_dict (dict)
15+
16+
17+
Methods:
18+
set_attrs()
19+
get_attrs()
20+
get_dataset()
21+
set_dataset()
22+
23+
"""
24+
25+
def __init__(
26+
self,
27+
daslogs_dict: Optional[dict] = None,
28+
instrument_dict: Optional[dict] = None,
29+
sample_dict: Optional[dict] = None,
30+
) -> None:
31+
self.daslogs_dict = daslogs_dict
32+
self.instrument_dict = instrument_dict
33+
self.sample_dict = sample_dict
34+
self.nxentry_dict: dict = {"attrs": {}}
35+
36+
def set_attrs(self, **kwargs):
37+
for k, v in kwargs.items():
38+
self.nxentry_dict["attrs"].update({k: v})
39+
40+
def get_attrs(self, key: str, default=None):
41+
val = self.nxentry_dict["attrs"].get(key)
42+
return val if val is not None else default
43+
44+
def get_dataset(self, key: str, default=None):
45+
val = self.nxentry_dict.get(key)
46+
return val["dataset"] if val is not None else default
47+
48+
def set_dataset(self, key: str, dataset: str | dict | np.ndarray, **kwargs):
49+
"""Set dataset with proper format if dataset is a string or an array. Take directly if dictionary."""
450

5-
def _add_dataset_entry(
6-
entry_dict: dict,
7-
key: str,
8-
daslogs: dict,
9-
daslogs_key: Optional[str] = None,
10-
**kwargs,
11-
):
12-
key = key if daslogs_key is None else daslogs_key
13-
try:
14-
val = daslogs[key]
15-
entry_dict.update({key: {"dataset": val}})
16-
if not kwargs:
17-
return entry_dict
1851
attr_dict = {}
1952
for k, v in kwargs.items():
2053
attr_dict.update({k: v})
21-
entry_dict[key].update({"attrs": attr_dict})
22-
return entry_dict
23-
24-
except KeyError:
25-
print(f"Variable {key} cannot be found in DAS logs.")
26-
return entry_dict
27-
28-
29-
def nxsource(spicelogs, instrument_config_params):
30-
source = {
31-
"attrs": {"NX_class": "NXsource", "EX_required": "true"},
32-
"name": {
33-
"attrs": {"type": "NX_CHAR", "EX_required": "true"},
34-
"dataset": "HFIR",
35-
},
36-
"probe": {
37-
"attrs": {"type": "NX_CHAR", "EX_required": "true"},
38-
"dataset": "neutron",
39-
},
40-
}
54+
if "type" in kwargs.keys():
55+
match kwargs["type"]:
56+
case "NX_CHAR":
57+
dataset = str(dataset)
58+
case "NX_INT":
59+
dataset = dataset
60+
case "NX_FLOAT":
61+
dataset = dataset
62+
63+
self.nxentry_dict.update({key: {"dataset": dataset}})
64+
self.nxentry_dict[key].update({"attrs": attr_dict})
65+
66+
def set_dataset_from(
67+
self,
68+
key: str,
69+
source: Literal["DAS_DATA", "DAS_METADATA", "INSTRU", "SAMPLE"] = "DAS_DATA",
70+
daslogs_key: Optional[str] = None,
71+
**kwargs,
72+
):
73+
match source:
74+
case "DAS_DATA":
75+
if self.daslogs_dict is None:
76+
raise ValueError("Cannot find DAS logs.")
77+
case "DAS_METADATA":
78+
if self.daslogs_dict is None:
79+
raise ValueError("Cannot find DAS logs.")
80+
case "INSTRU":
81+
if self.instrument_dict is None:
82+
raise ValueError("Cannot find instrument configuration dict.")
83+
case "SMAPLE":
84+
if self.sample_config_params is None:
85+
raise ValueError("Cannot find sample configuration dict.")
86+
case _:
87+
raise ValueError(f"Unrecogonized source {source}.")
88+
89+
match source:
90+
case "DAS_DATA":
91+
try:
92+
val = self.daslogs_dict[key] if daslogs_key is None else self.daslogs_dict[daslogs_key]
93+
except KeyError:
94+
print(f"Variable {key} cannot be found in DAS logs.")
95+
return self.nxentry_dict
96+
97+
self.nxentry_dict.update({key: val})
98+
if not kwargs:
99+
return self.nxentry_dict
100+
101+
attr_dict = {}
102+
for k, v in kwargs.items():
103+
attr_dict.update({k: v})
104+
self.nxentry_dict[key].update({"attrs": attr_dict})
105+
return self.nxentry_dict
106+
case "DAS_METADATA":
107+
pass
108+
case "INSTRU":
109+
pass
110+
case "SMAPLE":
111+
pass
112+
113+
114+
def spicelogs_to_nested_dict(
115+
spicelogs: dict,
116+
instrument_dict: Optional[dict],
117+
sample_dict: Optional[dict],
118+
) -> dict:
119+
120+
nxsource = NexusDict()
121+
nxsource.set_attrs(NX_class="NXsource", EX_required="true")
122+
nxsource.set_dataset(key="name", dataset="HFIR", type="NX_CHAR", EX_required="true")
123+
nxsource.set_dataset(key="probe", dataset="neutron", type="NX_CHAR", EX_required="true")
124+
41125
# Effective distance from sample Distance as seen by radiation from sample.
42126
# This number should be negative to signify that it is upstream of the sample.
43127
# nxsource.attrs["distance"] = -0.0
44-
return source
45-
46-
47-
def nxmono(spicelogs, instrument_config_params):
48-
49-
mono = {"attrs": {"NX_class": "NXcrystal", "EX_required": "true"}}
50-
mono = _add_dataset_entry(mono, key="ei", daslogs=spicelogs, type="NX_FLOAT", EX_required="true", unit="meV")
51-
mono.update({"type": {"dataset": spicelogs["attrs"]["monochromator"], "attrs": {"type": "NX_CHAR"}}})
52-
mono.update({"sense": {"dataset": spicelogs["attrs"]["sense"][0], "attrs": {"type": "NX_CHAR"}}})
53-
mono = _add_dataset_entry(mono, key="m1", daslogs=spicelogs, type="NX_FLOAT", unit="degrees")
54-
mono = _add_dataset_entry(mono, key="m2", daslogs=spicelogs, type="NX_FLOAT", unit="degrees")
55-
mono = _add_dataset_entry(mono, key="mfocus", daslogs=spicelogs, type="NX_FLOAT")
56-
mono = _add_dataset_entry(mono, key="marc", daslogs=spicelogs, type="NX_FLOAT")
57-
mono = _add_dataset_entry(mono, key="mtrans", daslogs=spicelogs, type="NX_FLOAT")
58-
mono = _add_dataset_entry(mono, key="focal_length", daslogs=spicelogs, type="NX_FLOAT")
59-
60-
return mono
61-
62-
63-
def nxcoll(spicelogs, instrument_config_params):
64-
pass
65-
66-
67-
def nxana(spicelogs, instrument_config_params):
68-
ana = {"attrs": {"NX_class": "NXcrystal", "EX_required": "true"}}
69-
ana = _add_dataset_entry(ana, key="ef", daslogs=spicelogs, type="NX_FLOAT", EX_required="true", unit="meV")
70-
ana.update({"type": {"dataset": spicelogs["attrs"]["analyzer"], "attrs": {"type": "NX_CHAR"}}})
71-
ana.update({"sense": {"dataset": spicelogs["attrs"]["sense"][2], "attrs": {"type": "NX_CHAR"}}})
72-
ana = _add_dataset_entry(ana, key="a1", daslogs=spicelogs, type="NX_FLOAT", unit="degrees")
73-
ana = _add_dataset_entry(ana, key="a2", daslogs=spicelogs, type="NX_FLOAT", unit="degrees")
74-
ana = _add_dataset_entry(ana, key="afocus", daslogs=spicelogs, type="NX_FLOAT")
75-
for i in range(8):
76-
ana = _add_dataset_entry(ana, key=f"qm{i+1}", daslogs=spicelogs, type="NX_FLOAT")
77-
ana = _add_dataset_entry(ana, key=f"xm{i+1}", daslogs=spicelogs, type="NX_FLOAT")
78-
return ana
79-
80-
81-
def nxdet(spicelogs, instrument_config_params):
82-
det = {"attrs": {"NX_class": "NXdetector", "EX_required": "true"}}
83-
det = _add_dataset_entry(det, key="detector", daslogs=spicelogs, type="NX_INT", EX_required="true", unit="counts")
84-
# polar_angle
85-
return det
86-
87-
88-
def nxmonitor(spicelogs, instrument_config_params):
89-
monitor = {"attrs": {"NX_class": "NXmonitor", "EX_required": "true"}}
90-
preset_type = spicelogs["attrs"]["preset_type"]
91-
match preset_type:
92-
case "countfile": # polarization data
93-
print("Countfile preset type is not supported.")
94-
95-
case "normal":
96-
preset_channel = spicelogs["attrs"]["preset_channel"]
97-
monitor.update({"mode": {"dataset": preset_channel, "attrs": {"type": "NX_CHAR", "EX_required": "true"}}})
98-
monitor.update(
99-
{
100-
"preset": {
101-
"dataset": spicelogs["attrs"]["preset_value"],
102-
"attrs": {"type": "NX_FLOAT", "EX_required": "true"},
103-
}
104-
}
105-
)
106-
monitor = _add_dataset_entry(monitor, key="time", daslogs=spicelogs, type="NX_FLOAT", units="seconds")
107-
monitor = _add_dataset_entry(monitor, key="monitor", daslogs=spicelogs, type="NX_INT", units="counts")
108-
monitor = _add_dataset_entry(monitor, key="mcu", daslogs=spicelogs, type="NX_FLOAT")
109-
monitor_data_dict = monitor[preset_channel]
110-
monitor.update({"data": monitor_data_dict})
111-
112-
case _:
113-
print(f"Unrecogonized preset type {preset_type}.")
114-
115-
return monitor
116-
117-
118-
def nxsample(spicelogs, sample_config_params):
119-
return {}
120128

129+
nxmono = NexusDict(daslogs_dict=spicelogs)
130+
nxmono.set_attrs(NX_class="NXcrystal", EX_required="true")
131+
# nxmono.set_dataset_from(source="DAS_DATA", key="ei", type="NX_FLOAT", EX_required="true", unit="meV")
121132

122-
def nxslit(spicelogs, instrument_config_params):
123133
return {}
124-
125-
126-
def nxflipper(spicelogs, instrument_config_params):
127-
return {}
128-
129-
130-
def _spicelogs_to_nexus(spicelogs, instrument_config_params, sample_config_params):
131-
metadata = spicelogs["attrs"]
132-
# TODO timezone
133-
start_date_time = "{} {}".format(metadata["date"], metadata["time"])
134-
start_time = datetime.strptime(start_date_time, "%m/%d/%Y %I:%M:%S %p").isoformat()
135-
# if "end_time" in das_logs.attrs: # last scan never finished
136-
end_date_time = metadata["end_time"]
137-
end_time = datetime.strptime(end_date_time, "%m/%d/%Y %I:%M:%S %p").isoformat()
138-
139-
scan_dict = {
140-
"attrs": {
141-
"NX_class": "NXentry",
142-
"EX_required": "true",
143-
},
144-
"SPICElogs": spicelogs,
145-
"definition": {
146-
"attrs": {"EX_required": "true", "type": "NX_CHAR"},
147-
"dataset": "NXtas",
148-
},
149-
"title": {
150-
"attrs": {"EX_required": "true", "type": "NX_CHAR"},
151-
"dataset": metadata["scan_title"],
152-
},
153-
"start_time": {
154-
"attrs": {"EX_required": "true", "type": "NX_DATE_TIME"},
155-
"dataset": start_time,
156-
},
157-
"end_time": {
158-
"attrs": {"type": "NX_DATE_TIME"},
159-
"dataset": end_time,
160-
},
161-
"instrument": {
162-
"attrs": {"EX_required": "true", "NX_class": "NXinstrument"},
163-
"name": {"attrs": {"type": "NX_CHAR"}, "dataset": metadata["instrument"]},
164-
"source": nxsource(spicelogs, instrument_config_params),
165-
"collimator": nxcoll(spicelogs, instrument_config_params),
166-
"monochromator": nxmono(spicelogs, instrument_config_params),
167-
"analyser": nxana(spicelogs, instrument_config_params),
168-
"detector": nxdet(spicelogs, instrument_config_params),
169-
"slit": nxslit(spicelogs, instrument_config_params),
170-
"flipper": nxflipper(spicelogs, instrument_config_params),
171-
},
172-
"monitor": nxmonitor(spicelogs, instrument_config_params),
173-
"sample": nxsample(spicelogs, sample_config_params),
174-
"data": {},
175-
}
176-
return scan_dict
177-
178-
179-
def daslogs_to_nexus_dict(
180-
daslogs: dict,
181-
instrument_config_params: Optional[str],
182-
sample_config_params: Optional[str],
183-
) -> dict:
184-
"""Format DASlogs dict into NeXus dict"""
185-
186-
match (das_key := tuple(daslogs.keys())[0]):
187-
case "SPICElogs":
188-
scan_dict = _spicelogs_to_nexus(
189-
daslogs["SPICElogs"],
190-
instrument_config_params,
191-
sample_config_params,
192-
)
193-
case _:
194-
raise KeyError(f"Unrecogonized DASlogs key {das_key}.")
195-
196-
return scan_dict

0 commit comments

Comments
 (0)