✅ lint, fix docs and tests

acq2sqlite.py

@@ -122,7 +122,15 @@ def __init__(self, sql=None):
         #  only add if not already in the DB
         acq_uniq_col = set(self.all_columns) - set(self.CONSTS) - set(["filename"])
         assert acq_uniq_col == set(
-            ["AcqTime", "AcqDate", "SeriesNumber", "SubID", "Operator", "Shims", "Station"]
+            [
+                "AcqTime",
+                "AcqDate",
+                "SeriesNumber",
+                "SubID",
+                "Operator",
+                "Shims",
+                "Station",
+            ]
         )
         # TODO: include station?
 

change_header.py

@@ -117,8 +117,8 @@ def gen_ids(new_id: str) -> List[pydicom.DataElement]:
     'example_name'
     >>> data_els[0].VR
     'PN'
-    >>> data_els[0].tag
-    (0010,0010)
+    >>> data_els[0].tag # doctest: +NORMALIZE_WHITESPACE
+    (0010, 0010)
     """
     return [
         pydicom.DataElement(value=new_id, VR="PN", tag=(0x0010, 0x0010)),

check_template.py

@@ -1,9 +1,13 @@
 #!/usr/bin/env python3
+"""
+quick cli tool to run template checker against tsv input
+likely form ``./dcmmeta2tsv.py $file``::
+    ./dcmmeta2tsv.py $file | ./check_template.py
 
-import os
+Added 2024-10-08. Consider removing 2024-11-20
+"""
 import sys
-
-from acq2sqlite import DBQuery, TagValue, have_pipe_data
+from acq2sqlite import DBQuery, have_pipe_data
 
 db = DBQuery()
 

compliance_check_html.py

@@ -1,17 +1,20 @@
-from template_checker import TemplateChecker, CheckResult
 from jinja2 import Template
 
+from template_checker import CheckResult, TemplateChecker
+
+
 def load_template(template_path: str) -> Template:
     """
     Load an HTML template from the template.html file
 
     :param template_path: Path to the HTML template file.
     :returns: A Jinja2 Template object.
     """
-    with open(template_path, 'r') as file:
+    with open(template_path, "r") as file:
         template_content = file.read()
     return Template(template_content)
 
+
 def generate_html_report(check_result: CheckResult, template_path: str) -> str:
     """
     Generate an HTML report of DICOM header comparison, highlighting mismatches.
@@ -31,12 +34,14 @@ def generate_html_report(check_result: CheckResult, template_path: str) -> str:
         expected_value = check_result["template"].get(header, "N/A")
         actual_value = check_result["input"].get(header, "N/A")
         class_name = "mismatch" if header in check_result["errors"] else "match"
-        rows.append({
-            "header": header, 
-            "expected_value": expected_value,
-            "actual_value": actual_value,
-            "class_name": class_name
-        })
+        rows.append(
+            {
+                "header": header,
+                "expected_value": expected_value,
+                "actual_value": actual_value,
+                "class_name": class_name,
+            }
+        )
 
     # Load the template from the file
     template = load_template(template_path)
@@ -45,5 +50,3 @@ def generate_html_report(check_result: CheckResult, template_path: str) -> str:
     html = template.render(rows=rows)
 
     return html
-
-

dcmmeta2tsv.py

@@ -7,7 +7,7 @@
 import re
 import sys
 import warnings
-from typing import Optional, TypedDict
+from typing import TypedDict, Optional
 
 import pydicom
 
@@ -18,14 +18,11 @@
 
 logging.basicConfig(level=os.environ.get("LOGLEVEL", logging.INFO))
 
-
 class NULLVAL:
     """Container to imitate ``pydicom.dcmread``.
     object that has ``obj.value`` for when a dicom tag does not exist.
     Using "null" to match AFNI's dicom_hinfo missing text"""
-
-    value: str = "null"
-
+    value : str = "null"
 
 TagTuple = tuple[int, int]
 
@@ -99,41 +96,37 @@ def csa_fetch(csa_tr: dict, itemname: str) -> str:
         val = NULLVAL.value
     return val
 
-
 def read_shims(csa_s: Optional[dict]) -> list:
     """
-    read current shim parameters from ASCCONV
-    from 0x0029,0x1020 CSA Series Header Info
+    :param: csa_s ``0x0029,0x1020`` CSA **Series** Header Info::
+        csa_s = dcmmeta2tsv.read_csa(dcm.get(())
 
-    csa_s = dcmmeta2tsv.read_csa(dcm.get(())
+    :return: list of shim values in order of CHM matlab code
 
     CHM maltab code concats
       sAdjData.uiAdjShimMode
       sGRADSPEC.asGPAData[0].lOffset{X,Y,Z}
       sGRADSPEC.alShimCurrent[0:4]
       sTXSPEC.asNucleusInfo[0].lFrequency
+
     >>> csa_s = pydicom.dcmread('example_dicoms/RewardedAnti_good.dcm').get((0x0029, 0x1020))
     >>> ",".join(read_shims(read_csa(csa_s)))
     '1174,-2475,4575,531,-20,59,54,-8,123160323,4'
-    >>> read_shims(None)
-    [None]*10
+    >>> read_shims(None)  # doctest: +ELLIPSIS, +NORMALIZE_WHITESPACE
+    ['null', ...'null']
     """
 
     if csa_s is None:
         csa_s = {}
     try:
-        asccov = csa_s["tags"]["MrPhoenixProtocol"]["items"][0]
+        asccov = csa_s['tags']['MrPhoenixProtocol']['items'][0]
     except KeyError:
-        return [NULLVAL.value] * 10
-
-    key = "|".join(
-        [
-            "sAdjData.uiAdjShimMode",
-            "sGRADSPEC.asGPAData\\[0\\].lOffset[XYZ]",
-            "sGRADSPEC.alShimCurrent\\[[0-4]\\]",
-            "sTXSPEC.asNucleusInfo\\[0\\].lFrequency",
-        ]
-    )
+        return [NULLVAL.value]*10
+
+    key = "|".join(['sAdjData.uiAdjShimMode',
+                    'sGRADSPEC.asGPAData\\[0\\].lOffset[XYZ]',
+                    'sGRADSPEC.alShimCurrent\\[[0-4]\\]',
+                    'sTXSPEC.asNucleusInfo\\[0\\].lFrequency'])
 
     # keys are like
     #   sGRADSPEC.asGPAData[0].lOffsetX\t = \t1174
@@ -168,12 +161,12 @@ def read_tags(dcm_path: os.PathLike, tags: TagDicts) -> TagValues:
 
     :param dcm_path: dicom file with headers to extract
     :param tags: ordered dictionary with 'tag' key as hex pair, see :py:func:`tagpair_to_hex`
-    :return: dict[tag,value] values in same order as ``tags`` \
+    :return: dict[tag,value] values in same order as ``tags``
 
     >>> tr = {'name': 'TR', 'tag': (0x0018,0x0080), 'loc': 'header'}
     >>> ipat = {'name': 'iPAT', 'tag': 'ImaPATModeText', 'loc': 'csa'}
     >>> list(read_tags('example_dicoms/RewardedAnti_good.dcm', [ipat, tr]).values())
-    ['p2', '1300', 'example_dicoms/RewardedAnti_good.dcm']
+    ['p2', '1300.0', 'example_dicoms/RewardedAnti_good.dcm']
 
     >>> list(read_tags('example_dicoms/DNE.dcm', [ipat,tr]).values())
     ['null', 'null', 'example_dicoms/DNE.dcm']
@@ -235,3 +228,4 @@ def read_dicom_tags(self, dcm_path: os.PathLike) -> TagValues:
     for dcm_path in sys.argv[1:]:
         all_tags = dtr.read_dicom_tags(dcm_path).values()
         print("\t".join([str(x) for x in all_tags]))
+

generated_report.html

@@ -1,4 +1,3 @@
-
 <html>
 <head>
     <style>
@@ -71,4 +70,4 @@ <h2>DICOM Header Compliance Report</h2>
 
     </table>
 </body>
-</html>
+</html>

mrqart.py

@@ -19,34 +19,38 @@
 
 Station = str
 Sequence = str
+
+
 class CurSeqStation:
     "Current Sequence settings at a MR Scanner station"
-    series_seqname : str
-    station : str
-    count : int
+    series_seqname: str
+    station: str
+    count: int
+
     def __init__(self, station: Station):
         "initialize new series"
         self.station = station
-        self.series_seqname=""
+        self.series_seqname = ""
         self.count = 0
 
     def update_isnew(self, series, seqname: Sequence) -> bool:
         """
         Maintain count of repeats seen
         :return:  True if is new
         """
-        serseq=f"{series}{seqname}"
+        serseq = f"{series}{seqname}"
         if self.series_seqname == serseq:
             self.count += 1
             return False
 
-        self.series_seqname=serseq
+        self.series_seqname = serseq
         self.count = 0
         return True
 
     def __repr__(self) -> str:
         return f"{self.station} {self.series_seqname} {self.count}"
 
+
 #: Websocket port used to send updates to browser
 WS_PORT = 5000
 #: HTTP port used to serve static/index.html
@@ -121,12 +125,13 @@ async def track_ws(websocket):
 def session_from_fname(dcm_fname: os.PathLike) -> Sequence:
     """
     We can use the file name to see if session name has changed.
-    Don't need to read the dicom header -- if we know the station name
-        extract
-         ls /data/dicomstream/20241016.MRQART_test.24.10.16_16_50_16_DST_1.3.12.2.1107.5.2.43.67078/|head
-    001_000001_000001.dcm
-        ...
-        001_000017_000066.dcm
+    Don't need to read the dicom header -- if we know the station name.
+    Can extract from ``001_sequencenum_seriesnum``::
+
+      ls /data/dicomstream/20241016.MRQART_test.24.10.16_16_50_16_DST_1.3.12.2.1107.5.2.43.67078/|head
+      001_000001_000001.dcm
+      ...
+      001_000017_000066.dcm
     """
     session = os.path.basename(dcm_fname)
     (proj, sequence, number) = session.split("_")
@@ -167,15 +172,19 @@ async def monitor_dirs(watcher, dcm_checker):
             # only send to browser if new
             if current_ses.update_isnew(hdr["SeriesNumber"], hdr["SequenceName"]):
                 logging.debug("first time seeing  %s", current_ses)
-                msg = {'station': hdr["Station"],
-                       'type': 'new',
-                       'content': dcm_checker.check_header(hdr)}
+                msg = {
+                    "station": hdr["Station"],
+                    "type": "new",
+                    "content": dcm_checker.check_header(hdr),
+                }
                 logging.debug(msg)
                 broadcast(WS_CONNECTIONS, json.dumps(msg, default=list))
             else:
-                msg = {'station': hdr["Station"],
-                       'type': 'update',
-                       'content': current_ses.count}
+                msg = {
+                    "station": hdr["Station"],
+                    "type": "update",
+                    "content": current_ses.count,
+                }
                 broadcast(WS_CONNECTIONS, json.dumps(msg, default=list))
                 logging.debug("already have %s", STATE[seq["Station"]])
 
@@ -184,8 +193,8 @@ async def monitor_dirs(watcher, dcm_checker):
 
         else:
             logging.warning("non dicom file %s", event.name)
-            # if we want to do this, we need msg formated
-            #broadcast(WS_CONNECTIONS, f"non-dicom file: {event}")
+            # if we want to do this, we need msg formatted
+            # broadcast(WS_CONNECTIONS, f"non-dicom file: {event}")
 
 
 async def main(path):
@@ -196,7 +205,8 @@ async def main(path):
     dcm_checker = TemplateChecker()
     watcher = aionotify.Watcher()
     watcher.watch(
-        path=path, flags=FOLLOW_FLAGS
+        path=path,
+        flags=FOLLOW_FLAGS,
         # NB. prev had just aionotify.Flags.CREATE but that triggers too early (partial file)
     )  # aionotify.Flags.MODIFY|aionotify.Flags.CREATE |aionotify.Flags.DELETE)
     asyncio.create_task(monitor_dirs(watcher, dcm_checker))
@@ -215,5 +225,7 @@ async def main(path):
 if __name__ == "__main__":
     # TODO: watch based on input argument
     # TODO: watch all sub directories?
-    watch_dir = os.path.join(FILEDIR, "/data/dicomstream/20241119.testMRQARAT.testMRQARAT/")
+    watch_dir = os.path.join(
+        FILEDIR, "/data/dicomstream/20241119.testMRQARAT.testMRQARAT/"
+    )
     asyncio.run(main(watch_dir))

readme.md

@@ -19,7 +19,7 @@ See
  * should look at n echos 0018,0086 and collapse across dicoms to make protocol mutliecho parameter
  * precision changes between realtime streaming and offline dicom writes? see `check_template.py`
 
- * inotify `CREATE` is catches files before they finish writting. Watch `CLOSE_WRITE` instead.
+ * inotify `CREATE` is catches files before they finish writing. Watch `CLOSE_WRITE` instead.
   Test slower write with `smbclient`
   ```
   smbclient -U mrqart //localhost/dicomstream/ -c 'put 001_000001_000002.dcm sim/y.dcm'

sphinx/index.rst

@@ -22,6 +22,8 @@ Code
    dcmmeta2tsv
    acq2sqlite
    change_header
+   template_checker
+   compliance_check_html
 
 
 Overview

template_checker.py

@@ -2,10 +2,10 @@
 check a header against best template
 """
 
-from typing import Optional, TypedDict
+from typing import TypedDict
 
 from acq2sqlite import DBQuery
-from dcmmeta2tsv import DicomTagReader, TagDicts, TagValues
+from dcmmeta2tsv import DicomTagReader, TagValues
 
 ErrorCompare = TypedDict("ErrorCompare", {"have": str, "expect": str})
 CheckResult = TypedDict(
@@ -93,13 +93,13 @@ def check_header(self, hdr) -> CheckResult:
             "errors": errors,
             "input": hdr,
             "template": dict(template),
-
         }
+
     def check_row(self, row: dict) -> CheckResult:
-        """ 
+        """
         Check a single SQL row against its template.
 
-        :parm row: Dictionary of header parameters (a row from SQL query)
+        :param row: Dictionary of header parameters (a row from SQL query)
         :returns: Conforming status, errors, and comparison information.
         """
 
@@ -111,8 +111,8 @@ def check_row(self, row: dict) -> CheckResult:
         errors = find_errors(template, row) if template else {}
 
         return {
-                "conforms": not errors,
-                "errors": errors,
-                "input": row, 
-                "template": template,
+            "conforms": not errors,
+            "errors": errors,
+            "input": row,
+            "template": template,
         }