refactor 'get_delivery_data_items'
@@ -19,23 +19,41 @@ import pymedphys._base.delivery
from . import extract
-def get_delivery_data_items(single_icom_stream):
- shrunk_stream, mu = extract.extract(single_icom_stream, "Delivery MU")
+def get_delivery_data_items(single_icom_stream: bytes):
+ """Convert a single timestep of an iCom byte stream into a delivery timestep.
+
+ Parameters
+ ----------
+ single_icom_stream
+ The message provided by the Elekta iCom Vx stream for one timestep.
+
+ Returns
+ -------
+ meterset
+ The machine meterset (MU) value.
+ gantry
+ The gantry angle.
+ collimator
+ The collimator angle
+ mlc
+ The MLC positions adjusted to be in the ``pymedphys.Delivery``
+ coordinate system.
+ jaw
+ The Jaw positions adjusted to be in the ``pymedphys.Delivery``
+ coordinate system.
+ """
+
+ shrunk_stream, meterset = extract.extract(single_icom_stream, "Delivery MU")
shrunk_stream, gantry = extract.extract(shrunk_stream, "Gantry")
shrunk_stream, collimator = extract.extract(shrunk_stream, "Collimator")
- shrunk_stream, mlc = extract.extract_coll(shrunk_stream, b"MLCX", 160)
- mlc = np.array(mlc)
- mlc = mlc.reshape((80, 2))
- mlc = np.fliplr(np.flipud(mlc * 10))
- mlc[:, 1] = -mlc[:, 1]
- mlc = np.round(mlc, 10)
+ shrunk_stream, raw_mlc = extract.extract_coll(shrunk_stream, b"MLCX", 160)
+ mlc = _convert_icom_mlc_to_delivery_coords(raw_mlc)
- shrunk_stream, jaw = extract.extract_coll(shrunk_stream, b"ASYMY", 2)
- jaw = np.round(np.array(jaw) * 10, 10)
- jaw = np.flipud(jaw)
+ shrunk_stream, raw_jaw = extract.extract_coll(shrunk_stream, b"ASYMY", 2)
+ jaw = _convert_icom_jaw_to_delivery_coords(raw_jaw)
- return mu, gantry, collimator, mlc, jaw
+ return meterset, gantry, collimator, mlc, jaw
def delivery_from_icom_stream(icom_stream):
@@ -66,3 +84,20 @@ class DeliveryIcom(
return cls( # pylint: disable = protected-access
*delivery_from_icom_stream(icom_stream)
)._filter_cps()
+
+
+def _convert_icom_mlc_to_delivery_coords(raw_mlc):
+ mlc = np.array(raw_mlc)
+ mlc = mlc.reshape((80, 2))
+ mlc = np.fliplr(np.flipud(mlc * 10))
+ mlc[:, 1] = -mlc[:, 1]
+ mlc = np.round(mlc, 10)
+
+ return mlc
+
+
+def _convert_icom_jaw_to_delivery_coords(raw_jaw):
+ jaw = np.round(np.array(raw_jaw) * 10, 10)
+ jaw = np.flipud(jaw)
+
+ return jaw
