Coverage for nexusLIMS/harvesters/reservation_event.py: 100%
130 statements
« prev ^ index » next coverage.py v7.11.3, created at 2026-03-24 05:23 +0000
1"""
2A representation of calendar reservations.
4This module contains a class to represent calendar reservations and
5associated metadata harvest metadata from various calendar sources.
6The expectation is that submodules of this module will have a method named
7``res_event_from_session`` implemented to handle fetching a ReservationEvent
8object from a :py:class:`nexusLIMS.db.session_handler.Session` object.
9"""
11from dataclasses import dataclass
12from datetime import datetime
13from typing import List
15from lxml import etree
17from nexusLIMS.instruments import Instrument
20@dataclass
21class ReservationEvent:
22 """
23 A representation of a single calendar reservation.
25 The representation is independent of the type of calendar the reservation was
26 made with. ``ReservationEvent`` is a common interface that is used by the record
27 building code.
29 Any attribute can be None to indicate it was not present or no value was
30 provided. The :py:meth:`as_xml` method is used to serialize the information
31 contained within a ``ReservationEvent`` into an XML representation that is
32 compatible with the Nexus Facility ``Experiment`` schema.
34 Parameters
35 ----------
36 experiment_title
37 The title of the event
38 instrument
39 The instrument associated with this reservation
40 last_updated : datetime.datetime
41 The time this event was last updated
42 username
43 The username of the user indicated in this event
44 user_full_name
45 The full name of the user for this event
46 created_by
47 The username of the user that created this event
48 created_by_full_name
49 The full name of the user that created this event
50 start_time
51 The time this event was scheduled to start
52 end_time
53 The time this event was scheduled to end
54 reservation_type
55 The "type" or category of this event (such as User session, service,
56 etc.)
57 experiment_purpose
58 The user-entered purpose of this experiment
59 sample_details
60 A list of the user-entered sample details for this experiment. The
61 length of the list must match that given in ``sample_pid`` and
62 ``sample_name``.
63 sample_pid
64 A list of sample PIDs provided by the user. The
65 length of the list must match that given in ``sample_details`` and
66 ``sample_name``.
67 sample_name
68 A list of user-friendly sample names (not a PID). The
69 length of the list must match that given in ``sample_details`` and
70 ``sample_pid``.
71 project_name
72 A list of the user-entered project names for this experiment. The
73 length of the list must match that given in ``project_id`` and
74 ``project_ref``.
75 project_id
76 A list of the specific project IDs within a research group/division. The
77 length of the list must match that given in ``project_name`` and
78 ``project_ref``.
79 project_ref
80 A list of (optional) links to this project in another database. The
81 length of the list must match that given in ``project_name`` and
82 ``project_id``.
83 internal_id
84 The identifier assigned to this event (if any) by the calendaring system
85 division
86 An identifier of the division this experiment was performed for (i.e.
87 the user's division)
88 group
89 An identifier of the group this experiment was performed for (i.e.
90 the user's group)
91 url
92 A web-accessible link to a summary of this reservation
93 """
95 experiment_title: str | None = None
96 instrument: Instrument | None = None
97 last_updated: datetime | None = None
98 username: str | None = None
99 user_full_name: str | None = None
100 created_by: str | None = None
101 created_by_full_name: str | None = None
102 start_time: datetime | None = None
103 end_time: datetime | None = None
104 reservation_type: str | None = None
105 experiment_purpose: str | None = None
106 sample_details: List[str | None] | None = None
107 sample_pid: List[str | None] | None = None
108 sample_name: List[str | None] | None = None
109 sample_elements: List[List[str] | None] | None = None
110 project_name: List[str | None] | None = None
111 project_id: List[str | None] | None = None
112 project_ref: List[str | None] | None = None
113 internal_id: str | None = None
114 division: str | None = None
115 group: str | None = None
116 url: str | None = None
118 def __post_init__(self):
119 """Post-initialization to coerce arguments to lists and validate."""
120 # coerce sample arguments into lists
121 self.sample_details = (
122 self.sample_details
123 if isinstance(self.sample_details, list)
124 else [self.sample_details]
125 )
126 self.sample_pid = (
127 self.sample_pid if isinstance(self.sample_pid, list) else [self.sample_pid]
128 )
129 self.sample_name = (
130 self.sample_name
131 if isinstance(self.sample_name, list)
132 else [self.sample_name]
133 )
134 # sample elements should be a list of List[str] or None; we shouldn't really be
135 # doing the above coercion anyway, so we'll assume the caller knows what
136 # they're doing and used the right argument type
137 if self.sample_elements is None:
138 self.sample_elements = [None]
140 # coerce project arguments into lists
141 self.project_name = (
142 self.project_name
143 if isinstance(self.project_name, list)
144 else [self.project_name]
145 )
146 self.project_id = (
147 self.project_id if isinstance(self.project_id, list) else [self.project_id]
148 )
149 self.project_ref = (
150 self.project_ref
151 if isinstance(self.project_ref, list)
152 else [self.project_ref]
153 )
155 # raise error if all sample values are not none and have different
156 # lengths (this shouldn't happen):
157 self._check_arg_lists()
159 def _check_arg_lists(self):
160 for check_name, arg_names, lists in zip(
161 ["sample", "project"],
162 [
163 "[sample_details, sample_pid, sample_name]",
164 "[project_name, project_id, project_ref]",
165 ],
166 [
167 [self.sample_details, self.sample_pid, self.sample_name],
168 [self.project_name, self.project_id, self.project_ref],
169 ],
170 ):
171 if all(x is not None for x in lists):
172 length = len(lists[0])
173 if not all(len(lst) == length for lst in lists[1:]):
174 msg = (
175 f"Length of {check_name} arguments must be the same. The "
176 "lengths of the following arguments were "
177 f"{arg_names} : "
178 f"{[len(list_) for list_ in lists]}"
179 )
180 raise ValueError(msg)
182 def __repr__(self):
183 """Return custom representation of a ReservationEvent."""
184 if self.username and self.start_time and self.end_time:
185 return (
186 f"Event for {self.username} on {self.instrument.name} "
187 f"from "
188 f"{self.instrument.localize_datetime(self.start_time).isoformat()} "
189 f"to {self.instrument.localize_datetime(self.end_time).isoformat()}"
190 )
191 return "No matching calendar event" + (
192 f" for {self.instrument.name}" if self.instrument else ""
193 )
195 def as_xml(self) -> etree.Element:
196 """
197 Get an XML representation of this ReservationEvent.
199 Returns
200 -------
201 root : lxml.etree.Element
202 The reservation event serialized as XML that matches the
203 Nexus Experiment schema
204 """
205 root = etree.Element("root")
207 # top-level nodes
208 title_el = etree.SubElement(root, "title")
209 if self.experiment_title:
210 title_el.text = self.experiment_title
211 else:
212 title_el.text = f"Experiment on the {self.instrument.display_name}"
213 if self.start_time:
214 title_el.text += f" on {self.start_time.strftime('%A %b. %d, %Y')}"
215 if self.internal_id:
216 id_el = etree.SubElement(root, "id")
217 id_el.text = self.internal_id
219 # summary node
220 root = self._add_summary_node(root)
222 # sample nodes
223 root = self._add_sample_nodes(root)
225 # project nodes
226 return self._add_project_nodes(root)
228 def _add_summary_node(self, root):
229 summary_el = etree.SubElement(root, "summary")
230 if self.user_full_name:
231 experimenter_el = etree.SubElement(summary_el, "experimenter")
232 experimenter_el.text = self.user_full_name
233 elif self.username:
234 experimenter_el = etree.SubElement(summary_el, "experimenter")
235 experimenter_el.text = self.username
236 if self.instrument:
237 instr_el = etree.SubElement(summary_el, "instrument")
238 instr_el.text = self.instrument.display_name
239 pid = self.instrument.name
240 # temporary workaround for duplicate harvesters for some instruments
241 if self.instrument.harvester == "nemo" and self.instrument.name.endswith(
242 "_n",
243 ): # pragma: no cover
244 pid = self.instrument.name.strip("_n")
245 instr_el.set("pid", pid)
246 if self.start_time:
247 start_el = etree.SubElement(summary_el, "reservationStart")
248 if self.instrument is not None:
249 start_el.text = self.instrument.localize_datetime(
250 self.start_time,
251 ).isoformat()
252 else:
253 start_el.text = self.start_time.isoformat()
254 if self.end_time:
255 end_el = etree.SubElement(summary_el, "reservationEnd")
256 if self.instrument is not None:
257 end_el.text = self.instrument.localize_datetime(
258 self.end_time,
259 ).isoformat()
260 else:
261 end_el.text = self.end_time.isoformat()
262 if self.experiment_purpose:
263 motivation_el = etree.SubElement(summary_el, "motivation")
264 motivation_el.text = self.experiment_purpose
265 if self.url:
266 summary_el.set("ref", self.url)
268 return root
270 def _add_sample_nodes(self, root):
271 if self.sample_pid is not None:
272 # if any of the sample arguments are not none, they should be
273 # lists, so we should create a sample element for each one
274 for pid, name, details, elements in zip(
275 self.sample_pid,
276 self.sample_name,
277 self.sample_details,
278 self.sample_elements,
279 ):
280 # create one sample subelement for each sample in our lists
281 sample_el = etree.SubElement(root, "sample")
282 if pid is not None:
283 sample_el.set("ref", pid)
284 if name is not None:
285 sample_name_el = etree.SubElement(sample_el, "name")
286 sample_name_el.text = name
287 if details is not None:
288 sample_detail_el = etree.SubElement(sample_el, "description")
289 sample_detail_el.text = details
290 if elements is not None:
291 sample_elements_el = etree.SubElement(sample_el, "elements")
292 for element in elements:
293 etree.SubElement(sample_elements_el, element)
294 return root
296 def _add_project_nodes(self, root):
297 if self.project_name is not None:
298 for name, pid, ref in zip(
299 self.project_name,
300 self.project_id,
301 self.project_ref,
302 ):
303 project_el = etree.SubElement(root, "project")
304 if name is not None:
305 project_name_el = etree.SubElement(project_el, "name")
306 project_name_el.text = name
307 if self.division is not None:
308 division_el = etree.SubElement(project_el, "division")
309 division_el.text = self.division
310 if self.group is not None:
311 group_el = etree.SubElement(project_el, "group")
312 group_el.text = self.group
313 if pid is not None:
314 proj_id_el = etree.SubElement(project_el, "project_id")
315 proj_id_el.text = pid
316 if ref is not None:
317 proj_ref_el = etree.SubElement(project_el, "ref")
318 proj_ref_el.text = ref
320 return root