From 304c0c854bf6b5a8cc1324c1333d8898e09597c5 Mon Sep 17 00:00:00 2001 From: sjswerdloff Date: Sun, 5 May 2024 22:01:47 +1200 Subject: [PATCH] Reformat with black, cleanup based on flake8 (#38) * replaced hardcoded IP addresses and Ports with those loaded from a configuration json file (ApplicationEntities.json) * reformatted with black -l 127 reformatted with isort --profile black * cleaned out unused imports per flake8 using treatment_delivery_type in rtbdi_factory (real bug caught by flake8) instead of hardcoded TREATMENT * enforce more comprehensive flake8 now that the ui files from Qt Creator have per file excludes and the entire repo has been tidied --- .github/workflows/python-app.yml | 2 +- .../TDWII_PPVS_subscriber/basescp.py | 49 +--- .../TDWII_PPVS_subscriber/cstore_handler.py | 13 +- .../TDWII_PPVS_subscriber/echoscp.py | 55 ++-- .../TDWII_PPVS_subscriber/nevent_receiver.py | 51 ++-- .../ppvs_subscriber_widget.py | 74 +++-- .../TDWII_PPVS_subscriber/ppvsscp.py | 103 +++---- .../TDWII_PPVS_subscriber/storescp.py | 69 ++--- .../ui_tdwii_ppvs_subscriber.py | 179 +++++++----- .../TDWII_PPVS_subscriber/watchscu.py | 141 ++++----- tdwii_plus_examples/cmove_inputs.py | 8 +- tdwii_plus_examples/handlers.py | 68 ++--- tdwii_plus_examples/nactionscu.py | 34 +-- tdwii_plus_examples/ncreatescu.py | 22 +- tdwii_plus_examples/nevent_receiver.py | 27 +- tdwii_plus_examples/nevent_sender.py | 29 +- .../generate_course_sessions_for_plan.py | 11 +- .../rtbdi_creator/mainbdiwidget.py | 72 ++--- .../rtbdi_creator/rtbdi_factory.py | 275 +++++++++--------- tdwii_plus_examples/rtbdi_creator/ui_form.py | 183 +++++++----- tdwii_plus_examples/tdwii_config.py | 4 +- .../tests/test_nevent_sender.py | 33 +-- tdwii_plus_examples/upsdb.py | 42 +-- tdwii_plus_examples/upsscp.py | 19 +- tdwii_plus_examples/watchscu.py | 24 +- 25 files changed, 733 insertions(+), 854 deletions(-) diff --git a/.github/workflows/python-app.yml b/.github/workflows/python-app.yml index 01e39c6..51ecf97 100644 --- a/.github/workflows/python-app.yml +++ b/.github/workflows/python-app.yml @@ -62,7 +62,7 @@ jobs: # while flake8 complains (this warning is not PEP 8 compliant), If W503 pops up, that should be disabled also # poetry run flake8 tdwii_plus_examples --count --exit-zero --extend-ignore=E203 --max-line-length=127 --statistics # poetry run flake8 tdwii_plus_examples --count --exit-zero --max-complexity=10 --statistics - poetry run flake8 --count --exit-zero --extend-ignore=E203 --max-line-length=127 --statistics --per-file-ignores="tdwii_plus_examples/rtbdi_creator/ui_form.py:E266,F401,E501 tdwii_plus_examples/TDWII_PPVS_subscriber/ui_tdwii_ppvs_subscriber.py:E266,F401,E501" tdwii_plus_examples + poetry run flake8 --count --extend-ignore=E203 --max-line-length=127 --statistics --per-file-ignores="tdwii_plus_examples/rtbdi_creator/ui_form.py:E266,F401,E501 tdwii_plus_examples/TDWII_PPVS_subscriber/ui_tdwii_ppvs_subscriber.py:E266,F401,E501" tdwii_plus_examples - name: Test with pytest run: | poetry run pytest tdwii_plus_examples/tests diff --git a/tdwii_plus_examples/TDWII_PPVS_subscriber/basescp.py b/tdwii_plus_examples/TDWII_PPVS_subscriber/basescp.py index 28a7876..4a2b75f 100644 --- a/tdwii_plus_examples/TDWII_PPVS_subscriber/basescp.py +++ b/tdwii_plus_examples/TDWII_PPVS_subscriber/basescp.py @@ -1,40 +1,17 @@ -import logging -import os -import sys -from argparse import Namespace -from configparser import ConfigParser -from datetime import datetime -from typing import Tuple -import pydicom.config -from tdwii_plus_examples.TDWII_PPVS_subscriber.nevent_receiver_handlers import handle_echo -from pynetdicom import ( - AE, - ALL_TRANSFER_SYNTAXES, - UnifiedProcedurePresentationContexts, - _config, - _handlers, - evt, -) +from argparse import Namespace +from pynetdicom import AE from pynetdicom.apps.common import setup_logging -from pynetdicom.sop_class import Verification -from pynetdicom.utils import set_ae - class BaseSCP: - def __init__(self, - ae_title:str="BASE_SCP", - port:int=11112, - logger=None, - bind_address:str="" - ): - + def __init__(self, ae_title: str = "BASE_SCP", port: int = 11112, logger=None, bind_address: str = ""): + self.ae_title = ae_title self.port = port if logger is None: - logger_args = Namespace(log_type='d', log_level='debug') - self.logger = setup_logging(logger_args, "base_scp") + logger_args = Namespace(log_type="d", log_level="debug") + self.logger = setup_logging(logger_args, "base_scp") else: self.logger = logger self.bind_address = bind_address @@ -45,17 +22,13 @@ def __init__(self, self._add_handlers() def _add_contexts(self): - # self.ae.add_supported_context(Verification, ALL_TRANSFER_SYNTAXES) - pass # base class, do nothing, pure virtual - + # self.ae.add_supported_context(Verification, ALL_TRANSFER_SYNTAXES) + pass # base class, do nothing, pure virtual def _add_handlers(self): - # self.handlers.append((evt.EVT_C_ECHO, handle_echo, [None, self.logger])) - pass # base class, do nothing, pure virtual - + # self.handlers.append((evt.EVT_C_ECHO, handle_echo, [None, self.logger])) + pass # base class, do nothing, pure virtual def run(self): # Listen for incoming association requests - self.threaded_server = self.ae.start_server((self.bind_address, self.port), - evt_handlers=self.handlers, - block=False) + self.threaded_server = self.ae.start_server((self.bind_address, self.port), evt_handlers=self.handlers, block=False) diff --git a/tdwii_plus_examples/TDWII_PPVS_subscriber/cstore_handler.py b/tdwii_plus_examples/TDWII_PPVS_subscriber/cstore_handler.py index 248ac27..542a253 100644 --- a/tdwii_plus_examples/TDWII_PPVS_subscriber/cstore_handler.py +++ b/tdwii_plus_examples/TDWII_PPVS_subscriber/cstore_handler.py @@ -1,16 +1,14 @@ """Utility classes and functions for the apps.""" -import logging import os -from struct import pack -from pydicom import dcmread -from pydicom.datadict import tag_for_keyword, repeater_has_keyword, get_entry + +# from pydicom import dcmread +# from pydicom.datadict import get_entry, repeater_has_keyword, tag_for_keyword from pydicom.dataset import Dataset from pydicom.filewriter import write_file_meta_info -from pydicom.tag import Tag +# from pydicom.tag import Tag from pydicom.uid import DeflatedExplicitVRLittleEndian - from pynetdicom.dsutils import encode @@ -56,8 +54,7 @@ def handle_store(event, args, app_logger): sop_instance = ds.SOPInstanceUID except Exception as exc: app_logger.error( - "Unable to decode the received dataset or missing 'SOP Class " - "UID' and/or 'SOP Instance UID' elements" + "Unable to decode the received dataset or missing 'SOP Class " "UID' and/or 'SOP Instance UID' elements" ) app_logger.exception(exc) # Unable to decode dataset diff --git a/tdwii_plus_examples/TDWII_PPVS_subscriber/echoscp.py b/tdwii_plus_examples/TDWII_PPVS_subscriber/echoscp.py index 30fbd00..bb0d68d 100644 --- a/tdwii_plus_examples/TDWII_PPVS_subscriber/echoscp.py +++ b/tdwii_plus_examples/TDWII_PPVS_subscriber/echoscp.py @@ -1,60 +1,39 @@ -import logging -import os -import sys -from argparse import Namespace -from configparser import ConfigParser -from datetime import datetime -from typing import Tuple + from time import sleep -import pydicom.config -from tdwii_plus_examples.TDWII_PPVS_subscriber.nevent_receiver_handlers import handle_echo + from pynetdicom import ( - AE, ALL_TRANSFER_SYNTAXES, - UnifiedProcedurePresentationContexts, - _config, - _handlers, evt, ) -from pynetdicom.apps.common import setup_logging from pynetdicom.sop_class import Verification -from pynetdicom.utils import set_ae + from tdwii_plus_examples.TDWII_PPVS_subscriber.basescp import BaseSCP +from tdwii_plus_examples.TDWII_PPVS_subscriber.nevent_receiver_handlers import ( + handle_echo, +) + class EchoSCP: - def __init__(self, - ae_title:str="ECHO_SCP", - port:int=11112, - logger=None, - bind_address:str="" - ): - - BaseSCP.__init__(self, - ae_title=ae_title, - port=port, - logger=logger, - bind_address=bind_address) - + def __init__(self, ae_title: str = "ECHO_SCP", port: int = 11112, logger=None, bind_address: str = ""): + + BaseSCP.__init__(self, ae_title=ae_title, port=port, logger=logger, bind_address=bind_address) + def _add_contexts(self): BaseSCP._add_contexts(self) self.ae.add_supported_context(Verification, ALL_TRANSFER_SYNTAXES) - - def _add_handlers(self): - BaseSCP._add_handlers(self) - self.handlers.append((evt.EVT_C_ECHO, handle_echo, [None, self.logger])) - - - + BaseSCP._add_handlers(self) + self.handlers.append((evt.EVT_C_ECHO, handle_echo, [None, self.logger])) def run(self): # Listen for incoming association requests BaseSCP.run(self) -if __name__ == '__main__': + +if __name__ == "__main__": myecho_scp = EchoSCP() myecho_scp.run() - while True: sleep(100) # sleep forever - \ No newline at end of file + while True: + sleep(100) # sleep forever diff --git a/tdwii_plus_examples/TDWII_PPVS_subscriber/nevent_receiver.py b/tdwii_plus_examples/TDWII_PPVS_subscriber/nevent_receiver.py index 09e5023..ed9edc0 100644 --- a/tdwii_plus_examples/TDWII_PPVS_subscriber/nevent_receiver.py +++ b/tdwii_plus_examples/TDWII_PPVS_subscriber/nevent_receiver.py @@ -1,27 +1,17 @@ -import logging -import os -import sys -from argparse import Namespace -from configparser import ConfigParser -from datetime import datetime -from typing import Tuple + from time import sleep -import pydicom.config -from tdwii_plus_examples.TDWII_PPVS_subscriber.nevent_receiver_handlers import handle_echo, handle_nevent from pynetdicom import ( - AE, ALL_TRANSFER_SYNTAXES, UnifiedProcedurePresentationContexts, - _config, - _handlers, evt, ) -from pynetdicom.apps.common import setup_logging -from pynetdicom.sop_class import Verification -from pynetdicom.utils import set_ae from tdwii_plus_examples.TDWII_PPVS_subscriber.basescp import BaseSCP from tdwii_plus_examples.TDWII_PPVS_subscriber.echoscp import EchoSCP +from tdwii_plus_examples.TDWII_PPVS_subscriber.nevent_receiver_handlers import ( + handle_nevent, +) + def nevent_cb(**kwargs): logger = None @@ -31,9 +21,7 @@ def nevent_cb(**kwargs): logger.info("nevent_cb invoked") event_type_id = 0 # not a valid type ID if logger: - logger.info( - "TODO: Invoke application response appropriate to content of N-EVENT-REPORT-RQ" - ) + logger.info("TODO: Invoke application response appropriate to content of N-EVENT-REPORT-RQ") if "type_id" in kwargs.keys(): event_type_id = kwargs["type_id"] if logger: @@ -54,9 +42,7 @@ def nevent_cb(**kwargs): elif event_type_id == 3: if logger: logger.info("UPS Progress Report") - logger.info( - "Probably time to see if the Beam (number) changed, or if adaptation is taking or took place" - ) + logger.info("Probably time to see if the Beam (number) changed, or if adaptation is taking or took place") elif event_type_id == 4: if logger: logger.info("SCP Status Change") @@ -75,26 +61,21 @@ def nevent_cb(**kwargs): if logger: logger.warning(f"Unknown Event Type ID: {event_type_id}") + class NEventReceiver(EchoSCP): - def __init__(self, - nevent_callback=None, - ae_title:str="NEVENT_RECEIVER", - port:int=11115, - logger=None, - bind_address:str="" - ): + def __init__( + self, nevent_callback=None, ae_title: str = "NEVENT_RECEIVER", port: int = 11115, logger=None, bind_address: str = "" + ): if nevent_callback is None: self.nevent_callback = nevent_cb # fallback to something that just logs incoming events else: self.nevent_callback = nevent_callback - EchoSCP.__init__(self,ae_title=ae_title,port=port,logger=logger,bind_address=bind_address) + EchoSCP.__init__(self, ae_title=ae_title, port=port, logger=logger, bind_address=bind_address) def _add_contexts(self): EchoSCP._add_contexts(self) for cx in UnifiedProcedurePresentationContexts: - self.ae.add_supported_context( - cx.abstract_syntax, ALL_TRANSFER_SYNTAXES, scp_role=True, scu_role=False - ) + self.ae.add_supported_context(cx.abstract_syntax, ALL_TRANSFER_SYNTAXES, scp_role=True, scu_role=False) def _add_handlers(self): EchoSCP._add_handlers(self) @@ -103,7 +84,9 @@ def _add_handlers(self): def run(self): BaseSCP.run(self) -if __name__ == '__main__': + +if __name__ == "__main__": my_scp = NEventReceiver() my_scp.run() - while True: sleep(100) # sleep forever \ No newline at end of file + while True: + sleep(100) # sleep forever diff --git a/tdwii_plus_examples/TDWII_PPVS_subscriber/ppvs_subscriber_widget.py b/tdwii_plus_examples/TDWII_PPVS_subscriber/ppvs_subscriber_widget.py index e7a316f..e095052 100755 --- a/tdwii_plus_examples/TDWII_PPVS_subscriber/ppvs_subscriber_widget.py +++ b/tdwii_plus_examples/TDWII_PPVS_subscriber/ppvs_subscriber_widget.py @@ -3,12 +3,9 @@ import sys from pathlib import Path -from PySide6.QtWidgets import QApplication, QWidget, QFileDialog -from PySide6.QtCore import Qt, Slot, QDateTime # pylint: disable=no-name-in-module - - from ppvsscp import PPVS_SCP -from watchscu import WatchSCU +from PySide6.QtCore import QDateTime, Qt, Slot # pylint: disable=no-name-in-module +from PySide6.QtWidgets import QApplication, QFileDialog, QWidget # Important: # You need to run the following command to generate the ui_form.py file @@ -16,6 +13,10 @@ # pyside2-uic form.ui -o ui_form.py from ui_tdwii_ppvs_subscriber import Ui_MainPPVSSubscriberWidget +import tdwii_config +from watchscu import WatchSCU + + class PPVS_SubscriberWidget(QWidget): def __init__(self, parent=None): super().__init__(parent) @@ -32,7 +33,7 @@ def __init__(self, parent=None): @Slot() def _import_staging_dir_clicked(self): - dialog = QFileDialog(self,"Import Staging Dir") + dialog = QFileDialog(self, "Import Staging Dir") dialog.setAcceptMode(QFileDialog.AcceptMode.AcceptOpen) dialog.setFileMode(QFileDialog.Directory) dialog.setOption(QFileDialog.ShowDirsOnly, True) @@ -50,64 +51,62 @@ def _toggle_subscription(self): else: self._unsubscribe_from_ups() - @Slot() def _restart_scp(self): ppvs_scp_ae_title = self.ui.ppvs_ae_line_edit.text() staging_dir = self.ui.import_staging_dir_line_edit.text() print(f"PPVS AE Title: {ppvs_scp_ae_title} using {staging_dir} for caching data") # PPVS_SCP combines the NEVENT SCP and C-STORE SCP - self.ppvs_scp = PPVS_SCP(nevent_callback= self._nevent_callback, - ae_title = ppvs_scp_ae_title, - ) + self.ppvs_scp = PPVS_SCP( + nevent_callback=self._nevent_callback, + ae_title=ppvs_scp_ae_title, + ) self.ppvs_scp.run() self.watch_scu = WatchSCU(self.ui.ppvs_ae_line_edit.text()) - ip_addr = "127.0.0.1" - port = 11114 - self.watch_scu.set_subscription_ae(self.ui.ups_ae_line_edit.text(), ip_addr=ip_addr,port=port) - + upsscp_ae_title = self.ui.ups_ae_line_edit.text() + ip_addr = tdwii_config.known_ae_ipaddr[upsscp_ae_title] + port = tdwii_config.known_ae_port[upsscp_ae_title] + self.watch_scu.set_subscription_ae(upsscp_ae_title, ip_addr=ip_addr, port=port) @Slot() def _get_ups(self): # do C-FIND-RQ - + pass - def _subscribe_to_ups(self, match_on_step_state=False, match_on_beam_number=False)->bool: + def _subscribe_to_ups(self, match_on_step_state=False, match_on_beam_number=False) -> bool: if self.watch_scu is None: my_ae_title = self.ui.ppvs_ae_line_edit.text() watch_scu = WatchSCU(my_ae_title) - # hard code for the moment, deal with configuration of AE's soon - ip_addr = "127.0.0.1" - port = 11114 upsscp_ae_title = self.ui.ups_ae_line_edit.text() - watch_scu.set_subscription_ae(upsscp_ae_title, ip_addr=ip_addr,port=port) + ip_addr = tdwii_config.known_ae_ipaddr[upsscp_ae_title] + port = tdwii_config.known_ae_port[upsscp_ae_title] + watch_scu.set_subscription_ae(upsscp_ae_title, ip_addr=ip_addr, port=port) else: watch_scu = self.watch_scu - + matching_keys = None - if (match_on_beam_number or match_on_step_state): - matching_keys = watch_scu.create_data_set(match_on_beam_number=match_on_beam_number, - match_on_step_state=match_on_step_state) + if match_on_beam_number or match_on_step_state: + matching_keys = watch_scu.create_data_set( + match_on_beam_number=match_on_beam_number, match_on_step_state=match_on_step_state + ) success = watch_scu.subscribe(matching_keys=matching_keys) if success and self.watch_scu is None: self.watch_scu = watch_scu return success - - def _unsubscribe_from_ups(self): if self.watch_scu is None: my_ae_title = self.ui.ppvs_ae_line_edit.text() watch_scu = WatchSCU(my_ae_title) - # hard code for the moment, deal with configuration of AE's soon - ip_addr = "127.0.0.1" - port = 11114 upsscp_ae_title = self.ui.ups_ae_line_edit.text() - watch_scu.set_subscription_ae(upsscp_ae_title, ip_addr=ip_addr,port=port) + ip_addr = tdwii_config.known_ae_ipaddr[upsscp_ae_title] + port = tdwii_config.known_ae_port[upsscp_ae_title] + + watch_scu.set_subscription_ae(upsscp_ae_title, ip_addr=ip_addr, port=port) else: watch_scu = self.watch_scu - + matching_keys = None success = watch_scu.unsubscribe(matching_keys=matching_keys) if success and self.watch_scu is None: @@ -122,9 +121,7 @@ def _nevent_callback(self, **kwargs): logger.info("nevent_cb invoked") event_type_id = 0 # not a valid type ID if logger: - logger.info( - "TODO: Invoke application response appropriate to content of N-EVENT-REPORT-RQ" - ) + logger.info("TODO: Invoke application response appropriate to content of N-EVENT-REPORT-RQ") if "type_id" in kwargs.keys(): event_type_id = kwargs["type_id"] if logger: @@ -147,9 +144,7 @@ def _nevent_callback(self, **kwargs): elif event_type_id == 3: if logger: logger.info("UPS Progress Report") - logger.info( - "Probably time to see if the Beam (number) changed, or if adaptation is taking or took place" - ) + logger.info("Probably time to see if the Beam (number) changed, or if adaptation is taking or took place") self._get_ups() elif event_type_id == 4: if logger: @@ -170,7 +165,7 @@ def _nevent_callback(self, **kwargs): logger.warning(f"Unknown Event Type ID: {event_type_id}") -def restart_ppvs_scp(ae_title:str, output_dir:Path=None) -> str: +def restart_ppvs_scp(ae_title: str, output_dir: Path = None) -> str: """_summary_ Args: @@ -183,11 +178,10 @@ def restart_ppvs_scp(ae_title:str, output_dir:Path=None) -> str: str: error string, empty if startup was successful """ - - if __name__ == "__main__": app = QApplication(sys.argv) + tdwii_config.load_ae_config() widget = PPVS_SubscriberWidget() widget.show() sys.exit(app.exec()) diff --git a/tdwii_plus_examples/TDWII_PPVS_subscriber/ppvsscp.py b/tdwii_plus_examples/TDWII_PPVS_subscriber/ppvsscp.py index 0d3c73c..af83b4c 100644 --- a/tdwii_plus_examples/TDWII_PPVS_subscriber/ppvsscp.py +++ b/tdwii_plus_examples/TDWII_PPVS_subscriber/ppvsscp.py @@ -1,82 +1,71 @@ -import logging + import os -import sys -from argparse import Namespace -from configparser import ConfigParser -from datetime import datetime -from typing import Tuple from time import sleep -import pydicom.config -from pynetdicom.apps.common import handle_store + +from basescp import BaseSCP from pynetdicom import ( - AE, ALL_TRANSFER_SYNTAXES, AllStoragePresentationContexts, - _config, - _handlers, - evt, + ) -from pynetdicom.apps.common import setup_logging -from pynetdicom.sop_class import Verification -from pynetdicom.utils import set_ae -from basescp import BaseSCP -from echoscp import EchoSCP -from tdwii_plus_examples.TDWII_PPVS_subscriber.nevent_receiver import NEventReceiver from storescp import StoreSCP -class PPVS_SCP(NEventReceiver,StoreSCP): - def __init__(self, - ae_title:str="PPVS_SCP", - port:int=11115, - logger=None, - bind_address:str="", - storage_presentation_contexts=AllStoragePresentationContexts, - transfer_syntaxes=ALL_TRANSFER_SYNTAXES, - custom_cstore_handler = None, - nevent_callback = None, - store_directory=os.path.curdir - ): +from tdwii_plus_examples.TDWII_PPVS_subscriber.nevent_receiver import NEventReceiver + + +class PPVS_SCP(NEventReceiver, StoreSCP): + def __init__( + self, + ae_title: str = "PPVS_SCP", + port: int = 11115, + logger=None, + bind_address: str = "", + storage_presentation_contexts=AllStoragePresentationContexts, + transfer_syntaxes=ALL_TRANSFER_SYNTAXES, + custom_cstore_handler=None, + nevent_callback=None, + store_directory=os.path.curdir, + ): # self.storage_presentation_contexts = storage_presentation_contexts # self.transfer_syntaxes = transfer_syntaxes self.nevent_callback = nevent_callback - StoreSCP.__init__(self, - ae_title=ae_title, - port=port, - logger=logger, - bind_address=bind_address, - custom_handler=custom_cstore_handler, - storage_presentation_contexts=storage_presentation_contexts, - transfer_syntaxes=transfer_syntaxes, - store_directory=store_directory) - NEventReceiver.__init__(self, - nevent_callback=nevent_callback, - ae_title=ae_title, - port=port, - logger=logger, - bind_address=bind_address, - ) - - + StoreSCP.__init__( + self, + ae_title=ae_title, + port=port, + logger=logger, + bind_address=bind_address, + custom_handler=custom_cstore_handler, + storage_presentation_contexts=storage_presentation_contexts, + transfer_syntaxes=transfer_syntaxes, + store_directory=store_directory, + ) + NEventReceiver.__init__( + self, + nevent_callback=nevent_callback, + ae_title=ae_title, + port=port, + logger=logger, + bind_address=bind_address, + ) + def _add_contexts(self): StoreSCP._add_contexts(self) NEventReceiver._add_contexts(self) - - - - def _add_handlers(self): - StoreSCP._add_handlers(self) - NEventReceiver._add_handlers(self) + StoreSCP._add_handlers(self) + NEventReceiver._add_handlers(self) def run(self): # Listen for incoming association requests BaseSCP.run(self) -if __name__ == '__main__': + +if __name__ == "__main__": my_scp = PPVS_SCP() my_scp.run() - while True: sleep(100) # sleep forever - \ No newline at end of file + while True: + sleep(100) # sleep forever diff --git a/tdwii_plus_examples/TDWII_PPVS_subscriber/storescp.py b/tdwii_plus_examples/TDWII_PPVS_subscriber/storescp.py index 4bf86da..3d66120 100644 --- a/tdwii_plus_examples/TDWII_PPVS_subscriber/storescp.py +++ b/tdwii_plus_examples/TDWII_PPVS_subscriber/storescp.py @@ -1,41 +1,29 @@ -import logging import os -import sys from argparse import Namespace -from configparser import ConfigParser -from datetime import datetime -from pathlib import Path -from typing import Tuple from time import sleep -import pydicom.config +from basescp import BaseSCP +from cstore_handler import handle_store +from echoscp import EchoSCP from pynetdicom import ( - AE, ALL_TRANSFER_SYNTAXES, AllStoragePresentationContexts, - _config, - _handlers, evt, ) -from pynetdicom.apps.common import setup_logging -from pynetdicom.sop_class import Verification -from pynetdicom.utils import set_ae -from basescp import BaseSCP -from echoscp import EchoSCP -from cstore_handler import handle_store class StoreSCP(EchoSCP): - def __init__(self, - ae_title:str="STORE_SCP", - port:int=11112, - logger=None, - bind_address:str="", - storage_presentation_contexts=AllStoragePresentationContexts, - transfer_syntaxes=ALL_TRANSFER_SYNTAXES, - custom_handler = None, - store_directory = os.path.curdir, - ): + def __init__( + self, + ae_title: str = "STORE_SCP", + port: int = 11112, + logger=None, + bind_address: str = "", + storage_presentation_contexts=AllStoragePresentationContexts, + transfer_syntaxes=ALL_TRANSFER_SYNTAXES, + custom_handler=None, + store_directory=os.path.curdir, + ): self.storage_presentation_contexts = storage_presentation_contexts self.transfer_syntaxes = transfer_syntaxes if custom_handler is None: @@ -43,35 +31,26 @@ def __init__(self, else: self.handle_cstore = custom_handler self.store_directory = store_directory - EchoSCP.__init__(self, - ae_title=ae_title, - port=port, - logger=logger, - bind_address=bind_address) - + EchoSCP.__init__(self, ae_title=ae_title, port=port, logger=logger, bind_address=bind_address) + def _add_contexts(self): EchoSCP._add_contexts(self) for context in self.storage_presentation_contexts: - self.ae.add_supported_context(context.abstract_syntax, self.transfer_syntaxes ) - - - + self.ae.add_supported_context(context.abstract_syntax, self.transfer_syntaxes) def _add_handlers(self): - EchoSCP._add_handlers(self) - args = Namespace(ignore=False, output_directory = self.store_directory) - - self.handlers.append((evt.EVT_C_STORE, self.handle_cstore, [args, self.logger])) - - + EchoSCP._add_handlers(self) + args = Namespace(ignore=False, output_directory=self.store_directory) + self.handlers.append((evt.EVT_C_STORE, self.handle_cstore, [args, self.logger])) def run(self): # Listen for incoming association requests BaseSCP.run(self) -if __name__ == '__main__': + +if __name__ == "__main__": my_scp = StoreSCP() my_scp.run() - while True: sleep(100) # sleep forever - \ No newline at end of file + while True: + sleep(100) # sleep forever diff --git a/tdwii_plus_examples/TDWII_PPVS_subscriber/ui_tdwii_ppvs_subscriber.py b/tdwii_plus_examples/TDWII_PPVS_subscriber/ui_tdwii_ppvs_subscriber.py index d88d19c..e8a27a2 100644 --- a/tdwii_plus_examples/TDWII_PPVS_subscriber/ui_tdwii_ppvs_subscriber.py +++ b/tdwii_plus_examples/TDWII_PPVS_subscriber/ui_tdwii_ppvs_subscriber.py @@ -8,191 +8,240 @@ ## WARNING! All changes made in this file will be lost when recompiling UI file! ################################################################################ -from PySide6.QtCore import (QCoreApplication, QDate, QDateTime, QLocale, - QMetaObject, QObject, QPoint, QRect, - QSize, QTime, QUrl, Qt) -from PySide6.QtGui import (QBrush, QColor, QConicalGradient, QCursor, - QFont, QFontDatabase, QGradient, QIcon, - QImage, QKeySequence, QLinearGradient, QPainter, - QPalette, QPixmap, QRadialGradient, QTransform) -from PySide6.QtWidgets import (QApplication, QCheckBox, QComboBox, QDateTimeEdit, - QGridLayout, QGroupBox, QHeaderView, QLabel, - QLineEdit, QPushButton, QSizePolicy, QTreeWidget, - QTreeWidgetItem, QWidget) +from PySide6.QtCore import ( + QCoreApplication, + QDate, + QDateTime, + QLocale, + QMetaObject, + QObject, + QPoint, + QRect, + QSize, + Qt, + QTime, + QUrl, +) +from PySide6.QtGui import ( + QBrush, + QColor, + QConicalGradient, + QCursor, + QFont, + QFontDatabase, + QGradient, + QIcon, + QImage, + QKeySequence, + QLinearGradient, + QPainter, + QPalette, + QPixmap, + QRadialGradient, + QTransform, +) +from PySide6.QtWidgets import ( + QApplication, + QCheckBox, + QComboBox, + QDateTimeEdit, + QGridLayout, + QGroupBox, + QHeaderView, + QLabel, + QLineEdit, + QPushButton, + QSizePolicy, + QTreeWidget, + QTreeWidgetItem, + QWidget, +) + class Ui_MainPPVSSubscriberWidget(object): def setupUi(self, MainPPVSSubscriberWidget): if not MainPPVSSubscriberWidget.objectName(): - MainPPVSSubscriberWidget.setObjectName(u"MainPPVSSubscriberWidget") + MainPPVSSubscriberWidget.setObjectName("MainPPVSSubscriberWidget") MainPPVSSubscriberWidget.resize(800, 659) self.group_box_aes_and_machine = QGroupBox(MainPPVSSubscriberWidget) - self.group_box_aes_and_machine.setObjectName(u"group_box_aes_and_machine") + self.group_box_aes_and_machine.setObjectName("group_box_aes_and_machine") self.group_box_aes_and_machine.setGeometry(QRect(30, 30, 622, 70)) self.gridLayout = QGridLayout(self.group_box_aes_and_machine) - self.gridLayout.setObjectName(u"gridLayout") + self.gridLayout.setObjectName("gridLayout") self.ups_ae_label = QLabel(self.group_box_aes_and_machine) - self.ups_ae_label.setObjectName(u"ups_ae_label") + self.ups_ae_label.setObjectName("ups_ae_label") self.gridLayout.addWidget(self.ups_ae_label, 0, 0, 1, 1) self.ups_ae_line_edit = QLineEdit(self.group_box_aes_and_machine) - self.ups_ae_line_edit.setObjectName(u"ups_ae_line_edit") + self.ups_ae_line_edit.setObjectName("ups_ae_line_edit") self.gridLayout.addWidget(self.ups_ae_line_edit, 0, 1, 1, 1) self.qr_ae_label = QLabel(self.group_box_aes_and_machine) - self.qr_ae_label.setObjectName(u"qr_ae_label") + self.qr_ae_label.setObjectName("qr_ae_label") self.gridLayout.addWidget(self.qr_ae_label, 0, 2, 1, 1) self.qr_ae_line_edit = QLineEdit(self.group_box_aes_and_machine) - self.qr_ae_line_edit.setObjectName(u"qr_ae_line_edit") + self.qr_ae_line_edit.setObjectName("qr_ae_line_edit") self.gridLayout.addWidget(self.qr_ae_line_edit, 0, 3, 1, 1) self.machine_name_label = QLabel(self.group_box_aes_and_machine) - self.machine_name_label.setObjectName(u"machine_name_label") + self.machine_name_label.setObjectName("machine_name_label") self.gridLayout.addWidget(self.machine_name_label, 0, 4, 1, 1) self.machine_name_line_edit = QLineEdit(self.group_box_aes_and_machine) - self.machine_name_line_edit.setObjectName(u"machine_name_line_edit") + self.machine_name_line_edit.setObjectName("machine_name_line_edit") self.gridLayout.addWidget(self.machine_name_line_edit, 0, 5, 1, 1) self.group_box_ppvs_scp = QGroupBox(MainPPVSSubscriberWidget) - self.group_box_ppvs_scp.setObjectName(u"group_box_ppvs_scp") + self.group_box_ppvs_scp.setObjectName("group_box_ppvs_scp") self.group_box_ppvs_scp.setGeometry(QRect(20, 120, 496, 171)) self.gridLayout_2 = QGridLayout(self.group_box_ppvs_scp) - self.gridLayout_2.setObjectName(u"gridLayout_2") + self.gridLayout_2.setObjectName("gridLayout_2") self.label = QLabel(self.group_box_ppvs_scp) - self.label.setObjectName(u"label") + self.label.setObjectName("label") self.gridLayout_2.addWidget(self.label, 1, 0, 1, 1) self.ppvs_ae_line_edit = QLineEdit(self.group_box_ppvs_scp) - self.ppvs_ae_line_edit.setObjectName(u"ppvs_ae_line_edit") + self.ppvs_ae_line_edit.setObjectName("ppvs_ae_line_edit") self.gridLayout_2.addWidget(self.ppvs_ae_line_edit, 0, 2, 1, 1) self.import_staging_dir_line_edit = QLineEdit(self.group_box_ppvs_scp) - self.import_staging_dir_line_edit.setObjectName(u"import_staging_dir_line_edit") + self.import_staging_dir_line_edit.setObjectName("import_staging_dir_line_edit") self.gridLayout_2.addWidget(self.import_staging_dir_line_edit, 2, 0, 1, 4) self.ppvs_ae_label = QLabel(self.group_box_ppvs_scp) - self.ppvs_ae_label.setObjectName(u"ppvs_ae_label") + self.ppvs_ae_label.setObjectName("ppvs_ae_label") self.gridLayout_2.addWidget(self.ppvs_ae_label, 0, 0, 1, 1) self.ppvs_restart_push_button = QPushButton(self.group_box_ppvs_scp) - self.ppvs_restart_push_button.setObjectName(u"ppvs_restart_push_button") + self.ppvs_restart_push_button.setObjectName("ppvs_restart_push_button") self.gridLayout_2.addWidget(self.ppvs_restart_push_button, 3, 3, 1, 1) self.import_staging_directory_push_button = QPushButton(self.group_box_ppvs_scp) - self.import_staging_directory_push_button.setObjectName(u"import_staging_directory_push_button") + self.import_staging_directory_push_button.setObjectName("import_staging_directory_push_button") self.gridLayout_2.addWidget(self.import_staging_directory_push_button, 1, 2, 1, 1) self.subscribe_ups_checkbox = QCheckBox(self.group_box_ppvs_scp) - self.subscribe_ups_checkbox.setObjectName(u"subscribe_ups_checkbox") + self.subscribe_ups_checkbox.setObjectName("subscribe_ups_checkbox") self.gridLayout_2.addWidget(self.subscribe_ups_checkbox, 3, 0, 1, 1) self.auto_download_checkbox = QCheckBox(self.group_box_ppvs_scp) - self.auto_download_checkbox.setObjectName(u"auto_download_checkbox") + self.auto_download_checkbox.setObjectName("auto_download_checkbox") self.gridLayout_2.addWidget(self.auto_download_checkbox, 3, 1, 1, 2) self.group_box_cfind_request_and_response = QGroupBox(MainPPVSSubscriberWidget) - self.group_box_cfind_request_and_response.setObjectName(u"group_box_cfind_request_and_response") + self.group_box_cfind_request_and_response.setObjectName("group_box_cfind_request_and_response") self.group_box_cfind_request_and_response.setGeometry(QRect(20, 300, 751, 243)) self.gridLayout_3 = QGridLayout(self.group_box_cfind_request_and_response) - self.gridLayout_3.setObjectName(u"gridLayout_3") + self.gridLayout_3.setObjectName("gridLayout_3") self.scheduled_date_label = QLabel(self.group_box_cfind_request_and_response) - self.scheduled_date_label.setObjectName(u"scheduled_date_label") + self.scheduled_date_label.setObjectName("scheduled_date_label") self.gridLayout_3.addWidget(self.scheduled_date_label, 0, 0, 1, 1) self.ups_response_tree_widget = QTreeWidget(self.group_box_cfind_request_and_response) __qtreewidgetitem = QTreeWidgetItem() - __qtreewidgetitem.setText(0, u"1"); + __qtreewidgetitem.setText(0, "1") self.ups_response_tree_widget.setHeaderItem(__qtreewidgetitem) - self.ups_response_tree_widget.setObjectName(u"ups_response_tree_widget") + self.ups_response_tree_widget.setObjectName("ups_response_tree_widget") self.gridLayout_3.addWidget(self.ups_response_tree_widget, 0, 1, 7, 1) self.soonest_date_time_edit = QDateTimeEdit(self.group_box_cfind_request_and_response) - self.soonest_date_time_edit.setObjectName(u"soonest_date_time_edit") + self.soonest_date_time_edit.setObjectName("soonest_date_time_edit") self.soonest_date_time_edit.setCalendarPopup(True) self.gridLayout_3.addWidget(self.soonest_date_time_edit, 1, 0, 1, 1) self.schedule_to_label = QLabel(self.group_box_cfind_request_and_response) - self.schedule_to_label.setObjectName(u"schedule_to_label") + self.schedule_to_label.setObjectName("schedule_to_label") self.gridLayout_3.addWidget(self.schedule_to_label, 2, 0, 1, 1) self.latest_date_time_edit = QDateTimeEdit(self.group_box_cfind_request_and_response) - self.latest_date_time_edit.setObjectName(u"latest_date_time_edit") + self.latest_date_time_edit.setObjectName("latest_date_time_edit") self.latest_date_time_edit.setEnabled(True) self.latest_date_time_edit.setCalendarPopup(True) self.gridLayout_3.addWidget(self.latest_date_time_edit, 3, 0, 1, 1) self.step_status_label = QLabel(self.group_box_cfind_request_and_response) - self.step_status_label.setObjectName(u"step_status_label") + self.step_status_label.setObjectName("step_status_label") self.gridLayout_3.addWidget(self.step_status_label, 4, 0, 1, 1) self.step_status_combo_box = QComboBox(self.group_box_cfind_request_and_response) - self.step_status_combo_box.setObjectName(u"step_status_combo_box") + self.step_status_combo_box.setObjectName("step_status_combo_box") self.step_status_combo_box.setMaxVisibleItems(6) self.step_status_combo_box.setMaxCount(6) self.gridLayout_3.addWidget(self.step_status_combo_box, 5, 0, 1, 1) self.push_button_get_ups = QPushButton(self.group_box_cfind_request_and_response) - self.push_button_get_ups.setObjectName(u"push_button_get_ups") + self.push_button_get_ups.setObjectName("push_button_get_ups") self.gridLayout_3.addWidget(self.push_button_get_ups, 6, 0, 1, 1) self.groupBox = QGroupBox(MainPPVSSubscriberWidget) - self.groupBox.setObjectName(u"groupBox") + self.groupBox.setObjectName("groupBox") self.groupBox.setGeometry(QRect(20, 550, 361, 80)) self.get_listed_inputs_push_button = QPushButton(self.groupBox) - self.get_listed_inputs_push_button.setObjectName(u"get_listed_inputs_push_button") + self.get_listed_inputs_push_button.setObjectName("get_listed_inputs_push_button") self.get_listed_inputs_push_button.setGeometry(QRect(30, 30, 131, 32)) self.get_rtss_and_ct_push_button = QPushButton(self.groupBox) - self.get_rtss_and_ct_push_button.setObjectName(u"get_rtss_and_ct_push_button") + self.get_rtss_and_ct_push_button.setObjectName("get_rtss_and_ct_push_button") self.get_rtss_and_ct_push_button.setGeometry(QRect(170, 30, 161, 32)) self.retranslateUi(MainPPVSSubscriberWidget) QMetaObject.connectSlotsByName(MainPPVSSubscriberWidget) + # setupUi def retranslateUi(self, MainPPVSSubscriberWidget): - MainPPVSSubscriberWidget.setWindowTitle(QCoreApplication.translate("MainPPVSSubscriberWidget", u"Widget", None)) - self.group_box_aes_and_machine.setTitle(QCoreApplication.translate("MainPPVSSubscriberWidget", u"group_box_aes_and_machine_name", None)) - self.ups_ae_label.setText(QCoreApplication.translate("MainPPVSSubscriberWidget", u"UPS AE", None)) - self.qr_ae_label.setText(QCoreApplication.translate("MainPPVSSubscriberWidget", u"QR AE", None)) - self.machine_name_label.setText(QCoreApplication.translate("MainPPVSSubscriberWidget", u"Machine Name", None)) - self.group_box_ppvs_scp.setTitle(QCoreApplication.translate("MainPPVSSubscriberWidget", u"ppvs_scp_group_box", None)) - self.label.setText(QCoreApplication.translate("MainPPVSSubscriberWidget", u"Import Staging Directory:", None)) - self.ppvs_ae_label.setText(QCoreApplication.translate("MainPPVSSubscriberWidget", u"Event and Store SCP AE", None)) - self.ppvs_restart_push_button.setText(QCoreApplication.translate("MainPPVSSubscriberWidget", u"Restart SCP", None)) - self.import_staging_directory_push_button.setText(QCoreApplication.translate("MainPPVSSubscriberWidget", u"Choose Staging Directory", None)) - self.subscribe_ups_checkbox.setText(QCoreApplication.translate("MainPPVSSubscriberWidget", u"Subscribe to UPS", None)) - self.auto_download_checkbox.setText(QCoreApplication.translate("MainPPVSSubscriberWidget", u"Auto Download", None)) - self.group_box_cfind_request_and_response.setTitle(QCoreApplication.translate("MainPPVSSubscriberWidget", u"group_box_cfind_request_and_response", None)) - self.scheduled_date_label.setText(QCoreApplication.translate("MainPPVSSubscriberWidget", u"scheduled date range", None)) - self.schedule_to_label.setText(QCoreApplication.translate("MainPPVSSubscriberWidget", u"to", None)) - self.step_status_label.setText(QCoreApplication.translate("MainPPVSSubscriberWidget", u"Procedure Step Status", None)) - self.push_button_get_ups.setText(QCoreApplication.translate("MainPPVSSubscriberWidget", u"Get UPS", None)) - self.groupBox.setTitle(QCoreApplication.translate("MainPPVSSubscriberWidget", u"group_box_get_reference_data", None)) - self.get_listed_inputs_push_button.setText(QCoreApplication.translate("MainPPVSSubscriberWidget", u"Get Listed Inputs", None)) - self.get_rtss_and_ct_push_button.setText(QCoreApplication.translate("MainPPVSSubscriberWidget", u"Get RTSS and CT", None)) - # retranslateUi + MainPPVSSubscriberWidget.setWindowTitle(QCoreApplication.translate("MainPPVSSubscriberWidget", "Widget", None)) + self.group_box_aes_and_machine.setTitle( + QCoreApplication.translate("MainPPVSSubscriberWidget", "group_box_aes_and_machine_name", None) + ) + self.ups_ae_label.setText(QCoreApplication.translate("MainPPVSSubscriberWidget", "UPS AE", None)) + self.qr_ae_label.setText(QCoreApplication.translate("MainPPVSSubscriberWidget", "QR AE", None)) + self.machine_name_label.setText(QCoreApplication.translate("MainPPVSSubscriberWidget", "Machine Name", None)) + self.group_box_ppvs_scp.setTitle(QCoreApplication.translate("MainPPVSSubscriberWidget", "ppvs_scp_group_box", None)) + self.label.setText(QCoreApplication.translate("MainPPVSSubscriberWidget", "Import Staging Directory:", None)) + self.ppvs_ae_label.setText(QCoreApplication.translate("MainPPVSSubscriberWidget", "Event and Store SCP AE", None)) + self.ppvs_restart_push_button.setText(QCoreApplication.translate("MainPPVSSubscriberWidget", "Restart SCP", None)) + self.import_staging_directory_push_button.setText( + QCoreApplication.translate("MainPPVSSubscriberWidget", "Choose Staging Directory", None) + ) + self.subscribe_ups_checkbox.setText(QCoreApplication.translate("MainPPVSSubscriberWidget", "Subscribe to UPS", None)) + self.auto_download_checkbox.setText(QCoreApplication.translate("MainPPVSSubscriberWidget", "Auto Download", None)) + self.group_box_cfind_request_and_response.setTitle( + QCoreApplication.translate("MainPPVSSubscriberWidget", "group_box_cfind_request_and_response", None) + ) + self.scheduled_date_label.setText(QCoreApplication.translate("MainPPVSSubscriberWidget", "scheduled date range", None)) + self.schedule_to_label.setText(QCoreApplication.translate("MainPPVSSubscriberWidget", "to", None)) + self.step_status_label.setText(QCoreApplication.translate("MainPPVSSubscriberWidget", "Procedure Step Status", None)) + self.push_button_get_ups.setText(QCoreApplication.translate("MainPPVSSubscriberWidget", "Get UPS", None)) + self.groupBox.setTitle(QCoreApplication.translate("MainPPVSSubscriberWidget", "group_box_get_reference_data", None)) + self.get_listed_inputs_push_button.setText( + QCoreApplication.translate("MainPPVSSubscriberWidget", "Get Listed Inputs", None) + ) + self.get_rtss_and_ct_push_button.setText( + QCoreApplication.translate("MainPPVSSubscriberWidget", "Get RTSS and CT", None) + ) + # retranslateUi diff --git a/tdwii_plus_examples/TDWII_PPVS_subscriber/watchscu.py b/tdwii_plus_examples/TDWII_PPVS_subscriber/watchscu.py index 0b70e7d..4a7fe03 100644 --- a/tdwii_plus_examples/TDWII_PPVS_subscriber/watchscu.py +++ b/tdwii_plus_examples/TDWII_PPVS_subscriber/watchscu.py @@ -2,36 +2,36 @@ import logging from typing import Optional, Tuple -from pydicom import Dataset, dcmread +from pydicom import Dataset from pydicom.errors import InvalidDicomError from pydicom.uid import UID from pynetdicom import AE, Association, UnifiedProcedurePresentationContexts -from pynetdicom._globals import DEFAULT_MAX_LENGTH -from pynetdicom.apps.common import setup_logging from pynetdicom.sop_class import UnifiedProcedureStepPush, UPSGlobalSubscriptionInstance -class WatchSCU(): - def __init__(self, calling_ae_title:str, - receiving_ae_title:str=None, - ): +class WatchSCU: + def __init__( + self, + calling_ae_title: str, + receiving_ae_title: str = None, + ): self.calling_ae_title = calling_ae_title if receiving_ae_title is None: self.receiving_ae_title = self.calling_ae_title else: self.receiving_ae_title = receiving_ae_title - def create_action_info(self, match_on_beam_number:bool=False, - match_on_step_state:bool=False)->Dataset: - return create_action_info(match_on_beam_number=match_on_beam_number, - match_on_step_state=match_on_step_state) - - def set_subscription_ae(self, watched_ae_title:str, ip_addr:str=None, port:int=None): + def create_action_info(self, match_on_beam_number: bool = False, match_on_step_state: bool = False) -> Dataset: + return create_action_info(match_on_beam_number=match_on_beam_number, match_on_step_state=match_on_step_state) + + def set_subscription_ae(self, watched_ae_title: str, ip_addr: str = None, port: int = None): self.watched_ae_title = watched_ae_title self.ip_addr = ip_addr self.port = port - def subscribe(self, watched_ae_title:str=None, ip_addr:str=None, port:int=None, matching_keys:Dataset=None)->bool: + def subscribe( + self, watched_ae_title: str = None, ip_addr: str = None, port: int = None, matching_keys: Dataset = None + ) -> bool: success = True watched_ip_addr = ip_addr if watched_ip_addr is None: @@ -51,9 +51,12 @@ def subscribe(self, watched_ae_title:str=None, ip_addr:str=None, port:int=None, ) if assoc.is_established: try: - status, response = send_global_watch_registration(calling_ae_title=self.calling_ae_title, - receiving_ae_title=self.receiving_ae_title, - assoc=assoc,action_info=matching_keys) + status, response = send_global_watch_registration( + calling_ae_title=self.calling_ae_title, + receiving_ae_title=self.receiving_ae_title, + assoc=assoc, + action_info=matching_keys, + ) dbg_message = f"Status: {status} response {response}" logging.getLogger().debug(dbg_message) except InvalidDicomError: @@ -70,49 +73,53 @@ def subscribe(self, watched_ae_title:str=None, ip_addr:str=None, port:int=None, success = False return success - - def unsubscribe(self, watched_ae_title:str=None, ip_addr:str=None, port:int=None, matching_keys:Dataset=None)->bool: - success = True - watched_ip_addr = ip_addr - if watched_ip_addr is None: - watched_ip_addr = self.ip_addr - watched_port = port - if watched_port is None: - watched_port = self.port - if watched_ae_title is None: - watched_ae_title = self.watched_ae_title - - - ae = AE(ae_title=self.calling_ae_title) - assoc = ae.associate( - watched_ip_addr, - watched_port, - contexts=UnifiedProcedurePresentationContexts, - ae_title=watched_ae_title, - ) - if assoc.is_established: - try: - status, response = send_global_watch_delete_registration(calling_ae_title=self.calling_ae_title, - receiving_ae_title=self.receiving_ae_title, - assoc=assoc,action_info=matching_keys) - dbg_message = f"Status: {status} response {response}" - logging.getLogger().debug(dbg_message) - except InvalidDicomError: - logging.getLogger().error("Bad DICOM: ") - success = False - except Exception as exc: - logging.getLogger().error("Watch Delete Registration (N-ACTION-RQ) failed") - logging.getLogger().exception(exc) - success = False - - assoc.release() - else: - logging.getLogger().error("Watch Delete Registration failed to make Association") + + def unsubscribe( + self, watched_ae_title: str = None, ip_addr: str = None, port: int = None, matching_keys: Dataset = None + ) -> bool: + success = True + watched_ip_addr = ip_addr + if watched_ip_addr is None: + watched_ip_addr = self.ip_addr + watched_port = port + if watched_port is None: + watched_port = self.port + if watched_ae_title is None: + watched_ae_title = self.watched_ae_title + + ae = AE(ae_title=self.calling_ae_title) + assoc = ae.associate( + watched_ip_addr, + watched_port, + contexts=UnifiedProcedurePresentationContexts, + ae_title=watched_ae_title, + ) + if assoc.is_established: + try: + status, response = send_global_watch_delete_registration( + calling_ae_title=self.calling_ae_title, + receiving_ae_title=self.receiving_ae_title, + assoc=assoc, + action_info=matching_keys, + ) + dbg_message = f"Status: {status} response {response}" + logging.getLogger().debug(dbg_message) + except InvalidDicomError: + logging.getLogger().error("Bad DICOM: ") success = False + except Exception as exc: + logging.getLogger().error("Watch Delete Registration (N-ACTION-RQ) failed") + logging.getLogger().exception(exc) + success = False + + assoc.release() + else: + logging.getLogger().error("Watch Delete Registration failed to make Association") + success = False + + return success + - return success - - def send_action( assoc: Association, class_uid: UID, @@ -138,8 +145,9 @@ def send_action( return assoc.send_n_action(action_info, action_type, class_uid, instance_uid) -def send_global_watch_registration( calling_ae_title:str, receiving_ae_title:str, - assoc: Association, action_info: Dataset = None): +def send_global_watch_registration( + calling_ae_title: str, receiving_ae_title: str, assoc: Association, action_info: Dataset = None +): """_summary_ Args: @@ -164,13 +172,14 @@ def send_global_watch_registration( calling_ae_title:str, receiving_ae_title:str assoc=assoc, class_uid=ds.RequestedSOPClassUID, instance_uid=ds.RequestedSOPInstanceUID, - action_type=3, # subscribe + action_type=3, # subscribe action_info=ds, ) -def send_global_watch_delete_registration( calling_ae_title:str, receiving_ae_title:str, - assoc: Association, action_info: Dataset = None): +def send_global_watch_delete_registration( + calling_ae_title: str, receiving_ae_title: str, assoc: Association, action_info: Dataset = None +): """_summary_ Args: @@ -195,13 +204,12 @@ def send_global_watch_delete_registration( calling_ae_title:str, receiving_ae_ti assoc=assoc, class_uid=ds.RequestedSOPClassUID, instance_uid=ds.RequestedSOPInstanceUID, - action_type=4, # unsubscribe + action_type=4, # unsubscribe action_info=ds, ) -def create_action_info(match_on_beam_number:bool=False, - match_on_step_state:bool=False)->Dataset: +def create_action_info(match_on_beam_number: bool = False, match_on_step_state: bool = False) -> Dataset: if not (match_on_beam_number or match_on_step_state): return None else: @@ -209,4 +217,3 @@ def create_action_info(match_on_beam_number:bool=False, # populate with the sequence and elements logging.getLogger().error("Matching Keys for WatchSCU Not Implemented") return ds - diff --git a/tdwii_plus_examples/cmove_inputs.py b/tdwii_plus_examples/cmove_inputs.py index 7d1ec48..8bff089 100755 --- a/tdwii_plus_examples/cmove_inputs.py +++ b/tdwii_plus_examples/cmove_inputs.py @@ -30,9 +30,7 @@ def get_port_for_ae_title(retrieve_ae_title): return known_ae_port[retrieve_ae_title] -def cmove_specific_input( - retrieve_ae_title, dest_ae_title, patient_id, study_uid, series_uid, instance_uid -): +def cmove_specific_input(retrieve_ae_title, dest_ae_title, patient_id, study_uid, series_uid, instance_uid): """C-MOVE request for a specific instance Args: @@ -75,9 +73,7 @@ def cmove_specific_input( if assoc.is_established: # Use the C-MOVE service to send the identifier - responses = assoc.send_c_move( - ds, dest_ae_title, StudyRootQueryRetrieveInformationModelMove - ) + responses = assoc.send_c_move(ds, dest_ae_title, StudyRootQueryRetrieveInformationModelMove) for status, identifier in responses: if status: print("C-MOVE query status: 0x{0:04x}".format(status.Status)) diff --git a/tdwii_plus_examples/handlers.py b/tdwii_plus_examples/handlers.py index 107409b..168a844 100644 --- a/tdwii_plus_examples/handlers.py +++ b/tdwii_plus_examples/handlers.py @@ -8,6 +8,7 @@ # from pydicom.dataset import FileMetaDataset from pydicom.errors import InvalidDicomError +from pynetdicom import AE, UnifiedProcedurePresentationContexts # from pynetdicom.dimse_primitives import N_ACTION # from pynetdicom.dsutils import encode @@ -17,13 +18,13 @@ UPSFilteredGlobalSubscriptionInstance, UPSGlobalSubscriptionInstance, ) -from pynetdicom import AE, Association, UnifiedProcedurePresentationContexts # from recursive_print_ds import print_ds from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker -from upsdb import Instance, InvalidIdentifier, add_instance, search + import tdwii_config +from upsdb import Instance, InvalidIdentifier, add_instance, search _SERVICE_STATUS = { "SCHEDULED": { @@ -59,6 +60,7 @@ REMOTE_AE_CONFIG_FILE = "ApplicationEntities.json" tdwii_config.load_ae_config(REMOTE_AE_CONFIG_FILE) + def _add_global_subscriber(subscriber_ae_title: str, deletion_lock: bool = False, logger=None): if subscriber_ae_title not in _global_subscribers.keys(): _global_subscribers[subscriber_ae_title] = deletion_lock @@ -727,15 +729,15 @@ def handle_ncreate(event, storage_dir, db_path, cli_config, logger): return 0xC210 # Check attributes satisfy CC.2.5.1.3 UPS Attribute Service Requirements - if not (('ProcedureStepState' in ds) and ('InputReadinessState' in ds)): + if not (("ProcedureStepState" in ds) and ("InputReadinessState" in ds)): logger.error("UPS is missing required attributes") - return 0x0120, None # Missing Attribute - elif not ds.ProcedureStepState == 'SCHEDULED': + return 0x0120, None # Missing Attribute + elif not ds.ProcedureStepState == "SCHEDULED": logger.error("UPS State not SCHEDULED") - return 0xC309, None # The provided value of UPS State was not "SCHEDULED" - elif not ds.InputReadinessState in ('INCOMPLETE', 'UNAVAILABLE', 'READY'): + return 0xC309, None # The provided value of UPS State was not "SCHEDULED" + elif ds.InputReadinessState not in ("INCOMPLETE", "UNAVAILABLE", "READY"): logger.error("Input Readiness State not valid") - return 0x0106, None # Invalid Attribute Value + return 0x0106, None # Invalid Attribute Value # More requirements need check from Table CC.2.5-3 and TDW-II # Transaction UID present and empty # Scheduled Procedure Step Priority present and not empty @@ -795,39 +797,41 @@ def handle_ncreate(event, storage_dir, db_path, cli_config, logger): # Database successfully updated, notify any globally subscribed AE # Get AET of UPS Event SCP (which is the AET of the UPS Watch SCP) - acceptor = event.assoc.acceptor - + acceptor = event.assoc.acceptor + # Set event information and type event_info = Dataset() event_info.ProcedureStepState = ds.ProcedureStepState event_info.InputReadinessState = ds.InputReadinessState - + # UPS Assigned when Scheduled Station Name or Scheduled Human Performers is defined # Only Scheduled Station Name is relevant for assignment to TDD in TDW-II - # As the SCP may choose to not send duplicate messages to an AE, only UPS State Report events + # As the SCP may choose to not send duplicate messages to an AE, only UPS State Report events # could maybe be sent and properly documented in conformance statement - if ('ScheduledStationNameCodeSequence' in ds): + if "ScheduledStationNameCodeSequence" in ds: event_type = 5 event_info.ScheduledStationNameCodeSequence = ds.ScheduledStationNameCodeSequence - if ('ScheduledHumanPerformersSequence' in ds): + if "ScheduledHumanPerformersSequence" in ds: event_info.ScheduledHumanPerformersSequence = ds.ScheduledHumanPerformersSequence - if ('HumanPerformerOrganization' in ds): + if "HumanPerformerOrganization" in ds: event_info.HumanPerformerOrganization = ds.HumanPerformerOrganization - logger.info(f"Send UPS Assigned event from {acceptor.ae_title} to subscribed AEs " - f"(assigned to {ds.ScheduledStationNameCodeSequence[0].CodeValue})") + logger.info( + f"Send UPS Assigned event from {acceptor.ae_title} to subscribed AEs " + f"(assigned to {ds.ScheduledStationNameCodeSequence[0].CodeValue})" + ) else: - # UPS State Report otherwise + # UPS State Report otherwise event_type = 1 logger.info(f"Send UPS State Report event from {acceptor.ae_title} to subscribed AEs") for globalsubscriber in _global_subscribers: # Request association with subscriber ae = AE(ae_title=acceptor.ae_title) - if (not globalsubscriber in tdwii_config.known_ae_ipaddr): + if globalsubscriber not in tdwii_config.known_ae_ipaddr: logger.error(f"{globalsubscriber} missing IP Address configuration in {REMOTE_AE_CONFIG_FILE}") continue - if (not globalsubscriber in tdwii_config.known_ae_port): + if globalsubscriber not in tdwii_config.known_ae_port: logger.error(f"{globalsubscriber} missing Port configuration in {REMOTE_AE_CONFIG_FILE}") continue @@ -840,37 +844,35 @@ def handle_ncreate(event, storage_dir, db_path, cli_config, logger): ae_title=globalsubscriber, max_pdu=16382, ) - + if assoc.is_established: - message_id=0 + message_id = 0 try: - if (event_type==1): - logger.info(f"Sending UPS State Report: {ds.SOPInstanceUID}, {ds.ProcedureStepState}") - message_id +=1 + if event_type == 1: + logger.info(f"Sending UPS State Report: {ds.SOPInstanceUID}, {ds.ProcedureStepState}") + message_id += 1 assoc.send_n_event_report(event_info, event_type, UnifiedProcedureStepPush, ds.SOPInstanceUID, message_id) - elif (event_type==5): # The assignment took place at the time of creation - # notify of creation first, i.e. event type == 1 + elif event_type == 5: # The assignment took place at the time of creation + # notify of creation first, i.e. event type == 1 logger.info(f"Sending UPS State Report: {ds.SOPInstanceUID}, {ds.ProcedureStepState}") - message_id +=1 + message_id += 1 assoc.send_n_event_report(event_info, 1, UnifiedProcedureStepPush, ds.SOPInstanceUID, message_id) # if the assignment happened after the creation (e.g. via N-SET or internal change in a TMS) # then *only* send an N-EVENT-REPORT regarding the UPS Assignment logger.info(f"Sending UPS Assignment: {ds.ScheduledStationNameCodeSequence}") - message_id +=1 + message_id += 1 assoc.send_n_event_report(event_info, event_type, UnifiedProcedureStepPush, ds.SOPInstanceUID, message_id) logger.info(f"Notified global subscriber: {globalsubscriber}") except InvalidDicomError: logger.error("Bad DICOM: ") except Exception as exc: - logger.error( - "UPS State Report as Event Notification (N-EVENT-REPORT-RQ) failed" - ) + logger.error("UPS State Report as Event Notification (N-EVENT-REPORT-RQ) failed") logger.exception(exc) assoc.release() else: - logger.error(f"Failed to establish association with subscriber: {globalsubscriber}") + logger.error(f"Failed to establish association with subscriber: {globalsubscriber}") return 0x0000, ds diff --git a/tdwii_plus_examples/nactionscu.py b/tdwii_plus_examples/nactionscu.py index d248069..8c65eb3 100755 --- a/tdwii_plus_examples/nactionscu.py +++ b/tdwii_plus_examples/nactionscu.py @@ -50,9 +50,7 @@ def send_action( return assoc.send_n_action(action_info, action_type, class_uid, instance_uid) -def send_procedure_step_state_change( - assoc: Association, new_state: str, ups_uid: UID, transaction_uid: UID -): +def send_procedure_step_state_change(assoc: Association, new_state: str, ups_uid: UID, transaction_uid: UID): ds = Dataset() ds.RequestedSOPInstanceUID = ups_uid # .ds.RequestingAE = calling_ae @@ -69,9 +67,7 @@ def send_procedure_step_state_change( ) -def send_global_watch_registration( - args: argparse.Namespace, assoc: Association, action_info: Dataset = None -): +def send_global_watch_registration(args: argparse.Namespace, assoc: Association, action_info: Dataset = None): """_summary_ Args: @@ -105,26 +101,19 @@ def _setup_argparser(): """Setup the command line arguments""" # Description parser = argparse.ArgumentParser( - description=( - "The nactionscu application implements a Service Class User " - "(SCU) for the UPS Push SOP Class. " - ), + description=("The nactionscu application implements a Service Class User " "(SCU) for the UPS Push SOP Class. "), usage="nactionscu [options] addr port", ) # Parameters req_opts = parser.add_argument_group("Parameters") - req_opts.add_argument( - "addr", help="TCP/IP address or hostname of DICOM peer", type=str - ) + req_opts.add_argument("addr", help="TCP/IP address or hostname of DICOM peer", type=str) req_opts.add_argument("port", help="TCP/IP port number of peer", type=int) req_opts.add_argument("ups_uid", help="SOP Instance UID of the UPS", type=str) # General Options gen_opts = parser.add_argument_group("General Options") - gen_opts.add_argument( - "--version", help="print version information and exit", action="store_true" - ) + gen_opts.add_argument("--version", help="print version information and exit", action="store_true") output = gen_opts.add_mutually_exclusive_group() output.add_argument( "-q", @@ -242,10 +231,7 @@ def _setup_argparser(): "-pdu", "--max-pdu", metavar="[n]umber of bytes", - help=( - f"set max receive pdu to n bytes (0 for unlimited, " - f"default: {DEFAULT_MAX_LENGTH})" - ), + help=(f"set max receive pdu to n bytes (0 for unlimited, " f"default: {DEFAULT_MAX_LENGTH})"), type=int, default=DEFAULT_MAX_LENGTH, ) @@ -381,9 +367,7 @@ def main(args=None): else: requested_state = args.requested_procedure_step_state - status_dataset, response = send_procedure_step_state_change( - assoc, requested_state, args.ups_uid, transaction_uid - ) + status_dataset, response = send_procedure_step_state_change(assoc, requested_state, args.ups_uid, transaction_uid) print(f"Status Code: 0x{status_dataset.Status:X}") print("Status Dataset:") @@ -391,9 +375,7 @@ def main(args=None): except InvalidDicomError: APP_LOGGER.error("Bad DICOM: ") except Exception as exc: - APP_LOGGER.error( - "Request to change UPS Procedure Step State (N-ACTION-RQ) failed" - ) + APP_LOGGER.error("Request to change UPS Procedure Step State (N-ACTION-RQ) failed") APP_LOGGER.exception(exc) assoc.release() diff --git a/tdwii_plus_examples/ncreatescu.py b/tdwii_plus_examples/ncreatescu.py index c477eb2..0018618 100755 --- a/tdwii_plus_examples/ncreatescu.py +++ b/tdwii_plus_examples/ncreatescu.py @@ -40,9 +40,7 @@ def _setup_argparser(): # Parameters req_opts = parser.add_argument_group("Parameters") - req_opts.add_argument( - "addr", help="TCP/IP address or hostname of DICOM peer", type=str - ) + req_opts.add_argument("addr", help="TCP/IP address or hostname of DICOM peer", type=str) req_opts.add_argument("port", help="TCP/IP port number of peer", type=int) req_opts.add_argument( "path", @@ -54,9 +52,7 @@ def _setup_argparser(): # General Options gen_opts = parser.add_argument_group("General Options") - gen_opts.add_argument( - "--version", help="print version information and exit", action="store_true" - ) + gen_opts.add_argument("--version", help="print version information and exit", action="store_true") output = gen_opts.add_mutually_exclusive_group() output.add_argument( "-q", @@ -146,10 +142,7 @@ def _setup_argparser(): "-pdu", "--max-pdu", metavar="[n]umber of bytes", - help=( - f"set max receive pdu to n bytes (0 for unlimited, " - f"default: {DEFAULT_MAX_LENGTH})" - ), + help=(f"set max receive pdu to n bytes (0 for unlimited, " f"default: {DEFAULT_MAX_LENGTH})"), type=int, default=DEFAULT_MAX_LENGTH, ) @@ -181,10 +174,7 @@ def _setup_argparser(): misc_opts.add_argument( "-cx", "--required-contexts", - help=( - "only request the presentation contexts required for the " - "input DICOM file(s)" - ), + help=("only request the presentation contexts required for the " "input DICOM file(s)"), action="store_true", ) @@ -297,9 +287,7 @@ def main(args=None): sys.exit() # Request association with remote - assoc = ae.associate( - args.addr, args.port, ae_title=args.called_aet, max_pdu=args.max_pdu - ) + assoc = ae.associate(args.addr, args.port, ae_title=args.called_aet, max_pdu=args.max_pdu) if assoc.is_established: ii = 1 for fpath in lfiles: diff --git a/tdwii_plus_examples/nevent_receiver.py b/tdwii_plus_examples/nevent_receiver.py index 0807c22..305ab81 100755 --- a/tdwii_plus_examples/nevent_receiver.py +++ b/tdwii_plus_examples/nevent_receiver.py @@ -7,7 +7,6 @@ from configparser import ConfigParser import pydicom.config -from nevent_receiver_handlers import handle_echo, handle_nevent from pynetdicom import ( AE, ALL_TRANSFER_SYNTAXES, @@ -20,6 +19,7 @@ from pynetdicom.sop_class import Verification from pynetdicom.utils import set_ae +from nevent_receiver_handlers import handle_echo, handle_nevent # Use `None` for empty values pydicom.config.use_none_as_empty_text_VR_value = True @@ -88,9 +88,7 @@ def _setup_argparser(): # General Options gen_opts = parser.add_argument_group("General Options") - gen_opts.add_argument( - "--version", help="print version information and exit", action="store_true" - ) + gen_opts.add_argument("--version", help="print version information and exit", action="store_true") output = gen_opts.add_mutually_exclusive_group() output.add_argument( "-q", @@ -191,10 +189,7 @@ def _setup_argparser(): ) db_opts.add_argument( "--clean", - help=( - "remove all entries from the database and delete the " - "corresponding stored instances" - ), + help=("remove all entries from the database and delete the " "corresponding stored instances"), action="store_true", ) @@ -209,9 +204,7 @@ def nevent_cb(**kwargs): logger.info("nevent_cb invoked") event_type_id = 0 # not a valid type ID if logger: - logger.info( - "TODO: Invoke application response appropriate to content of N-EVENT-REPORT-RQ" - ) + logger.info("TODO: Invoke application response appropriate to content of N-EVENT-REPORT-RQ") if "type_id" in kwargs.keys(): event_type_id = kwargs["type_id"] if logger: @@ -232,9 +225,7 @@ def nevent_cb(**kwargs): elif event_type_id == 3: if logger: logger.info("UPS Progress Report") - logger.info( - "Probably time to see if the Beam (number) changed, or if adaptation is taking or took place" - ) + logger.info("Probably time to see if the Beam (number) changed, or if adaptation is taking or took place") elif event_type_id == 4: if logger: logger.info("SCP Status Change") @@ -353,9 +344,7 @@ def main(args=None): # Unified Procedure Step SCP for cx in UnifiedProcedurePresentationContexts: - ae.add_supported_context( - cx.abstract_syntax, ALL_TRANSFER_SYNTAXES, scp_role=True, scu_role=False - ) + ae.add_supported_context(cx.abstract_syntax, ALL_TRANSFER_SYNTAXES, scp_role=True, scu_role=False) APP_LOGGER.info(f"Configured for instance_dir = {instance_dir}") # Set our handler bindings @@ -372,9 +361,7 @@ def main(args=None): ] # Listen for incoming association requests - ae.start_server( - (app_config["bind_address"], app_config.getint("port")), evt_handlers=handlers - ) + ae.start_server((app_config["bind_address"], app_config.getint("port")), evt_handlers=handlers) if __name__ == "__main__": diff --git a/tdwii_plus_examples/nevent_sender.py b/tdwii_plus_examples/nevent_sender.py index 6b8903f..c3b27f3 100755 --- a/tdwii_plus_examples/nevent_sender.py +++ b/tdwii_plus_examples/nevent_sender.py @@ -70,9 +70,7 @@ def send_ups_state_report( if reason_for_cancellation: event_info.ReasonForCancellation = reason_for_cancellation if discontinuation_reason_code_seq: - event_info.ProcedureStepDiscontinuationReasonCodeSequence = ( - discontinuation_reason_code_seq - ) + event_info.ProcedureStepDiscontinuationReasonCodeSequence = discontinuation_reason_code_seq return send_nevent( assoc, @@ -114,16 +112,12 @@ def _setup_argparser(): # Parameters req_opts = parser.add_argument_group("Parameters") - req_opts.add_argument( - "addr", help="TCP/IP address or hostname of DICOM peer", type=str - ) + req_opts.add_argument("addr", help="TCP/IP address or hostname of DICOM peer", type=str) req_opts.add_argument("port", help="TCP/IP port number of peer", type=int) # General Options gen_opts = parser.add_argument_group("General Options") - gen_opts.add_argument( - "--version", help="print version information and exit", action="store_true" - ) + gen_opts.add_argument("--version", help="print version information and exit", action="store_true") output = gen_opts.add_mutually_exclusive_group() output.add_argument( "-q", @@ -223,10 +217,7 @@ def _setup_argparser(): "-pdu", "--max-pdu", metavar="[n]umber of bytes", - help=( - f"set max receive pdu to n bytes (0 for unlimited, " - f"default: {DEFAULT_MAX_LENGTH})" - ), + help=(f"set max receive pdu to n bytes (0 for unlimited, " f"default: {DEFAULT_MAX_LENGTH})"), type=int, default=DEFAULT_MAX_LENGTH, ) @@ -356,22 +347,18 @@ def main(args=None): status, response = send_ups_state_report(assoc, UID("1.2.3.4"), "SCHEDULED") APP_LOGGER.info(f"Status: {os.linesep}{status}") APP_LOGGER.info(f"Response: {os.linesep}{response}") - - status, response = send_ups_state_report( - assoc, UID("1.2.3.4"), "IN PROGRESS" - ) + + status, response = send_ups_state_report(assoc, UID("1.2.3.4"), "IN PROGRESS") APP_LOGGER.info(f"Status: {os.linesep}{status}") APP_LOGGER.info(f"Response: {os.linesep}{response}") - + status, response = send_ups_state_report(assoc, UID("1.2.3.4"), "COMPLETED") APP_LOGGER.info(f"Status: {os.linesep}{status}") APP_LOGGER.info(f"Response: {os.linesep}{response}") except InvalidDicomError: APP_LOGGER.error("Bad DICOM: ") except Exception as exc: - APP_LOGGER.error( - "UPS State Report as Event Notification (N-EVENT-REPORT-RQ) failed" - ) + APP_LOGGER.error("UPS State Report as Event Notification (N-EVENT-REPORT-RQ) failed") APP_LOGGER.exception(exc) assoc.release() diff --git a/tdwii_plus_examples/rtbdi_creator/generate_course_sessions_for_plan.py b/tdwii_plus_examples/rtbdi_creator/generate_course_sessions_for_plan.py index 1421f35..97b370f 100644 --- a/tdwii_plus_examples/rtbdi_creator/generate_course_sessions_for_plan.py +++ b/tdwii_plus_examples/rtbdi_creator/generate_course_sessions_for_plan.py @@ -1,8 +1,9 @@ #! /usr/bin/env python -from sys import argv from datetime import datetime, timedelta -from time import strftime +from sys import argv + + from rtbdi_factory import gen_one_session, load_plan if __name__ == "__main__": @@ -12,7 +13,7 @@ planned_fractions = plan_ds.FractionGroupSequence[0].NumberOfFractionsPlanned start_date = datetime.now() date = start_date - + for i in range(planned_fractions): - gen_one_session(plan_ds, i+1, date, retrieve_ae_title) - date += timedelta(days=1) \ No newline at end of file + gen_one_session(plan_ds, i + 1, date, retrieve_ae_title) + date += timedelta(days=1) diff --git a/tdwii_plus_examples/rtbdi_creator/mainbdiwidget.py b/tdwii_plus_examples/rtbdi_creator/mainbdiwidget.py index ef92d9f..1fc7776 100755 --- a/tdwii_plus_examples/rtbdi_creator/mainbdiwidget.py +++ b/tdwii_plus_examples/rtbdi_creator/mainbdiwidget.py @@ -1,14 +1,26 @@ #!/usr/bin/env python # This Python file uses the following encoding: utf-8 +import sys from datetime import datetime from pathlib import Path from typing import List -import sys -from PySide6.QtWidgets import QApplication, QWidget, QFileDialog, QMessageBox # pylint: disable=no-name-in-module -from PySide6.QtCore import Qt, Slot, QDateTime # pylint: disable=no-name-in-module -from rtbdi_factory import (create_rtbdi_from_rtion_plan, create_ups_from_plan_and_bdi, load_plan, write_rtbdi, - is_tx_record_for_bdi, is_tx_record_for_plan, load_treatment_records, write_ups +from PySide6.QtCore import QDateTime, Qt, Slot # pylint: disable=no-name-in-module +from PySide6.QtWidgets import ( # pylint: disable=no-name-in-module + QApplication, + QFileDialog, + QMessageBox, + QWidget, +) +from rtbdi_factory import ( + create_rtbdi_from_rtion_plan, + create_ups_from_plan_and_bdi, + is_tx_record_for_bdi, + is_tx_record_for_plan, + load_plan, + load_treatment_records, + write_rtbdi, + write_ups, ) # Important: @@ -17,6 +29,7 @@ # pyside2-uic form.ui -o ui_form.py from ui_form import Ui_MainBDIWidget + class MainBDIWidget(QWidget): """Main UI for Creating an RT BDI based on an RT (Ion) Plan The UI provides a File Dialog to locate the RT (Ion) Plan @@ -28,6 +41,7 @@ class MainBDIWidget(QWidget): Args: QWidget (_type_): description """ + def __init__(self, parent=None): super().__init__(parent) self.ui = Ui_MainBDIWidget() @@ -40,24 +54,22 @@ def __init__(self, parent=None): self.ui.push_button_export_ups.clicked.connect(self._export_ups_button_clicked) self.plan = None self.rtbdi = None - self.export_path = Path("~/") # home for a default isn't the worst choice + self.export_path = Path("~/") # home for a default isn't the worst choice self.fraction_number = 1 self.retrieve_ae_title = "" self.scheduled_datetime = datetime.now @Slot() def _plan_button_clicked(self): - file_name, ok = QFileDialog.getOpenFileName(self, - "Open Plan", "~/", "Image Files (*.dcm)") + file_name, ok = QFileDialog.getOpenFileName(self, "Open Plan", "~/", "Image Files (*.dcm)") if file_name: path = Path(file_name) self.ui.lineedit_plan_selector.insert(str(path)) # print("Plan Button Clicked") - @Slot() def _bdidir_button_clicked(self): - dialog = QFileDialog(self,"BDI Export Dir") + dialog = QFileDialog(self, "BDI Export Dir") dialog.setAcceptMode(QFileDialog.AcceptMode.AcceptOpen) dialog.setFileMode(QFileDialog.Directory) dialog.setOption(QFileDialog.ShowDirsOnly, True) @@ -67,51 +79,42 @@ def _bdidir_button_clicked(self): if file_name: path = Path(file_name) self.ui.lineedit_bdidir_selector.insert(str(path)) -# fraction_number = round(self.ui.double_spin_box_fraction_number.value()) + + # fraction_number = round(self.ui.double_spin_box_fraction_number.value()) @Slot() def _bdi_export_button_clicked(self): fraction_number = round(self.ui.double_spin_box_fraction_number.value()) scheduled_date = self.ui.datetime_edit_scheduled_datetime.date() scheduled_time = self.ui.datetime_edit_scheduled_datetime.time() - print(f"Fraction Number: {fraction_number} "\ - f"Scheduled Date: {scheduled_date} "\ - f"Scheduled Time: {scheduled_time}") + print(f"Fraction Number: {fraction_number} " f"Scheduled Date: {scheduled_date} " f"Scheduled Time: {scheduled_time}") plan = load_plan(self.ui.lineedit_plan_selector.text()) self.plan = plan fraction_number = round(self.ui.double_spin_box_fraction_number.value()) self.fraction_number = fraction_number treatment_record_list = self._get_treatment_record_paths() - rtbdi = create_rtbdi_from_rtion_plan(plan, - fraction_number=fraction_number - ) + rtbdi = create_rtbdi_from_rtion_plan(plan, fraction_number=fraction_number, + treatment_record_list=treatment_record_list) self.rtbdi = rtbdi - bdi_path = Path(self.ui.lineedit_bdidir_selector.text(), - self.ui.line_edit_bdi_filename.text() - ) + bdi_path = Path(self.ui.lineedit_bdidir_selector.text(), self.ui.line_edit_bdi_filename.text()) write_rtbdi(rtbdi, bdi_path) self.export_path = bdi_path - @Slot() def _export_ups_button_clicked(self): self.retrieve_ae_title = self.ui.line_edit_move_scp_ae_title.text() self.scheduled_datetime = self.ui.datetime_edit_scheduled_datetime.dateTime().toPython() - + treatment_record_paths = self._get_treatment_record_paths() treatment_record_ds_list = load_treatment_records(treatment_record_paths) self._validate_treatment_records(treatment_record_ds_list) if self.rtbdi is None: self._bdi_export_button_clicked() - - ups = create_ups_from_plan_and_bdi(self.plan, - self.rtbdi, - self.retrieve_ae_title, - self.scheduled_datetime, - treatment_record_ds_list - ) - write_ups(ups,Path(self.ui.lineedit_bdidir_selector.text())) + ups = create_ups_from_plan_and_bdi( + self.plan, self.rtbdi, self.retrieve_ae_title, self.scheduled_datetime, treatment_record_ds_list + ) + write_ups(ups, Path(self.ui.lineedit_bdidir_selector.text())) def _validate_treatment_records(self, treatment_record_ds_list): """Confirms that the treatment records reference the plan @@ -132,16 +135,15 @@ def _validate_treatment_records(self, treatment_record_ds_list): if not is_tx_record_for_bdi(tx_record_ds, self.rtbdi): treatment_record_ds_list.remove(tx_record_ds) mismatched_records.append(tx_record_ds.SOPInstanceUID) - + if len(mismatched_records) != 0: - QMessageBox.show('
'.join(mismatched_records)) + QMessageBox.show("
".join(mismatched_records)) return - - def _get_treatment_record_paths(self)-> List[Path]: + def _get_treatment_record_paths(self) -> List[Path]: treatment_record_paths = [] model = self.ui.list_view_treatment_records.model() - if (model is not None): + if model is not None: for row in range(model.rowCount()): index = model.index(row, 0) item = model.data(index, Qt.ItemDataRole.DisplayRole) diff --git a/tdwii_plus_examples/rtbdi_creator/rtbdi_factory.py b/tdwii_plus_examples/rtbdi_creator/rtbdi_factory.py index 9116527..68641d8 100644 --- a/tdwii_plus_examples/rtbdi_creator/rtbdi_factory.py +++ b/tdwii_plus_examples/rtbdi_creator/rtbdi_factory.py @@ -1,16 +1,16 @@ """ RT Beams Delivery Instruction factory """ + from copy import deepcopy -import datetime from datetime import datetime -from time import strftime, strptime, mktime +from functools import lru_cache from pathlib import Path +from sys import argv +from time import mktime, strptime from typing import List -from functools import lru_cache -from sys import argv import pydicom -from pydicom import Dataset, DataElement, uid +from pydicom import DataElement, Dataset, uid from pydicom.filereader import dcmread from pydicom.filewriter import dcmwrite @@ -24,71 +24,71 @@ def load_plan(path: Path) -> Dataset: Returns: Dataset: a pydicom Dataset containing the rt plan information """ - plan = dcmread(path,force=True) + plan = dcmread(path, force=True) return plan -def load_treatment_records(treatment_record_paths: List[Path])->List[Dataset]: - tx_rec_ds_list = [ dcmread(x,force=True) for x in treatment_record_paths] +def load_treatment_records(treatment_record_paths: List[Path]) -> List[Dataset]: + tx_rec_ds_list = [dcmread(x, force=True) for x in treatment_record_paths] return tx_rec_ds_list -def is_tx_record_for_plan(tx_rec_ds:Dataset, plan:Dataset)->bool: +def is_tx_record_for_plan(tx_rec_ds: Dataset, plan: Dataset) -> bool: is_ion = tx_rec_ds.SOPClassUID == uid.RTIonBeamsTreatmentRecordStorage if tx_rec_ds.ReferencedRTPlanSequence[0].ReferencedSOPInstanceUID != plan.SOPInstanceUID: return False else: if is_ion: - planned_beam_numbers = [ x.BeamNumber for x in plan.IonBeamSequence] + planned_beam_numbers = [x.BeamNumber for x in plan.IonBeamSequence] for tx_session in tx_rec_ds.TreatmentSessionIonBeamSequence: referenced_beam_number = tx_session.ReferencedBeamNumber - if not referenced_beam_number in planned_beam_numbers: + if referenced_beam_number not in planned_beam_numbers: return False else: - planned_beam_numbers = [ x.BeamNumber for x in plan.BeamSequence] + planned_beam_numbers = [x.BeamNumber for x in plan.BeamSequence] for tx_session in tx_rec_ds.TreatmentSessionBeamSequence: referenced_beam_number = tx_session.ReferencedBeamNumber - if not referenced_beam_number in planned_beam_numbers: + if referenced_beam_number not in planned_beam_numbers: return False return True -def is_tx_record_for_bdi(tx_rec_ds:Dataset, bdi:Dataset)->bool: + +def is_tx_record_for_bdi(tx_rec_ds: Dataset, bdi: Dataset) -> bool: is_ion = tx_rec_ds.SOPClassUID == uid.RTIonBeamsTreatmentRecordStorage - bdi_current_fraction_list = [ x.CurrentFractionNumber for x in bdi.BeamTaskSequence] + bdi_current_fraction_list = [x.CurrentFractionNumber for x in bdi.BeamTaskSequence] if is_ion: for tx_session in tx_rec_ds.TreatmentSessionIonBeamSequence: current_fraction_number = tx_session.CurrentFractionNumber - if not current_fraction_number in bdi_current_fraction_list: + if current_fraction_number not in bdi_current_fraction_list: return False else: for tx_session in tx_rec_ds.TreatmentSessionBeamSequence: current_fraction_number = tx_session.CurrentFractionNumber - if not current_fraction_number in bdi_current_fraction_list: + if current_fraction_number not in bdi_current_fraction_list: return False return True -def write_rtbdi(bdi: Dataset, filename:Path): +def write_rtbdi(bdi: Dataset, filename: Path): if filename.is_dir(): - filename=filename.joinpath(f"RB_{str(bdi.SOPInstanceUID)}.dcm") + filename = filename.joinpath(f"RB_{str(bdi.SOPInstanceUID)}.dcm") _write_ds(bdi, filename) -def write_ups(ups:Dataset, filename:Path): +def write_ups(ups: Dataset, filename: Path): if filename.is_dir(): - filename=filename.joinpath(f"UPS_{str(ups.SOPInstanceUID)}.dcm") + filename = filename.joinpath(f"UPS_{str(ups.SOPInstanceUID)}.dcm") _write_ds(ups, filename) -def _write_ds(ds: Dataset, filename:Path): +def _write_ds(ds: Dataset, filename: Path): ds.fix_meta_info() - dcmwrite(filename,ds,write_like_original=False) + dcmwrite(filename, ds, write_like_original=False) -def create_rtbdi_from_rtion_plan(plan: Dataset, - fraction_number:int=1, - treatment_record_list:List[Dataset]=None, - bdi_uid_prefix:str=None) -> Dataset: +def create_rtbdi_from_rtion_plan( + plan: Dataset, fraction_number: int = 1, treatment_record_list: List[Dataset] = None, bdi_uid_prefix: str = None +) -> Dataset: """_summary_ Args: @@ -105,10 +105,10 @@ def create_rtbdi_from_rtion_plan(plan: Dataset, if elem.tag < 0x0020FFFF: bdi_ds[elem.tag] = deepcopy(elem) _delete_excess_plan_elements_from_bdi(bdi_ds) - now_date = datetime.now().strftime("%Y%m%d") # ,.date) - now_time = datetime.now().strftime("%H%M") # datetime.now().time, - bdi_ds['SOPClassUID'] = DataElement('SOPClassUID', "UI",uid.RTBeamsDeliveryInstructionStorage) - bdi_ds['SOPInstanceUID'] = DataElement('SOPInstanceUID', "UI",uid.generate_uid(bdi_uid_prefix)) + now_date = datetime.now().strftime("%Y%m%d") # ,.date) + now_time = datetime.now().strftime("%H%M") # datetime.now().time, + bdi_ds["SOPClassUID"] = DataElement("SOPClassUID", "UI", uid.RTBeamsDeliveryInstructionStorage) + bdi_ds["SOPInstanceUID"] = DataElement("SOPInstanceUID", "UI", uid.generate_uid(bdi_uid_prefix)) bdi_ds.InstanceCreationDate = now_date bdi_ds.InstanceCreationTime = now_time bdi_ds.SeriesDate = now_date @@ -117,8 +117,8 @@ def create_rtbdi_from_rtion_plan(plan: Dataset, bdi_ds.Manufacturer = "SJS Targeted Solutions" bdi_ds.ManufacturerModelName = "RTBDI Creator" bdi_ds.SoftwareVersions = "0.1" - - ref_plan_seq=Dataset() + + ref_plan_seq = Dataset() ref_plan_seq.ReferencedSOPClassUID = plan.SOPClassUID ref_plan_seq.ReferencedSOPInstanceUID = plan.SOPInstanceUID bdi_ds.ReferencedRTPlanSequence = pydicom.Sequence([ref_plan_seq]) @@ -126,26 +126,26 @@ def create_rtbdi_from_rtion_plan(plan: Dataset, list_of_beam_tasks = [] list_of_omitted_beam_tasks = [] if plan.SOPClassUID == uid.RTIonPlanStorage: - list_of_beams = [ x for x in plan.IonBeamSequence] + list_of_beams = [x for x in plan.IonBeamSequence] else: - list_of_beams = [ x for x in plan.BeamSequence] + list_of_beams = [x for x in plan.BeamSequence] for beam in list_of_beams: if is_completion_session: - if (_beam_already_treated_to_completion(beam.BeamNumber, treatment_record_list)): + if _beam_already_treated_to_completion(beam.BeamNumber, treatment_record_list): omitted_beam_task_seq = Dataset() omitted_beam_task_seq.ReferencedBeamNumber = beam.BeamNumber - omitted_beam_task_seq.ReasonForOmission = 'ALREADY_TREATED' + omitted_beam_task_seq.ReasonForOmission = "ALREADY_TREATED" list_of_omitted_beam_tasks.append(omitted_beam_task_seq) - else: + else: beam_task_seq = Dataset() _populate_default_beam_task_elements(beam_task_seq) beam_task_seq.ReferencedBeamNumber = beam.BeamNumber beam_task_seq.CurrentFractionNumber = fraction_number beam_task_seq.BeamTaskType = "TREAT" - treatment_delivery_type = 'TREATMENT' + treatment_delivery_type = "TREATMENT" if is_completion_session: - if _beam_already_partially_treated(beam.BeamNumber,treatment_record_list): - treatment_delivery_type = 'CONTINUATION' + if _beam_already_partially_treated(beam.BeamNumber, treatment_record_list): + treatment_delivery_type = "CONTINUATION" beam_task_seq.TreatmentDeliveryType = treatment_delivery_type list_of_beam_tasks.append(beam_task_seq) bdi_ds.BeamTaskSequence = pydicom.Sequence(list_of_beam_tasks) @@ -154,35 +154,38 @@ def create_rtbdi_from_rtion_plan(plan: Dataset, bdi_ds.is_little_endian = True return bdi_ds -def _beam_already_treated_to_completion(beam_number:int, treatment_record_list:List[Dataset])->bool: + +def _beam_already_treated_to_completion(beam_number: int, treatment_record_list: List[Dataset]) -> bool: for tx_record in treatment_record_list: if tx_record.SOPClassUID == uid.RTIonBeamsTreatmentRecordStorage: - tx_session_list = [ x for x in tx_record.TreatmentSessionIonBeamSequence] + tx_session_list = [x for x in tx_record.TreatmentSessionIonBeamSequence] else: - tx_session_list = [ x for x in tx_record.TreatmentSessionBeamSequence] + tx_session_list = [x for x in tx_record.TreatmentSessionBeamSequence] for tx_session in tx_session_list: - if (tx_session.ReferencedBeamNumber == beam_number) and (tx_session.TreatmentTerminationStatus == 'NORMAL'): + if (tx_session.ReferencedBeamNumber == beam_number) and (tx_session.TreatmentTerminationStatus == "NORMAL"): return True return False -def _beam_already_partially_treated(beam_number:int, treatment_record_list:List[Dataset])->bool: + +def _beam_already_partially_treated(beam_number: int, treatment_record_list: List[Dataset]) -> bool: for tx_record in treatment_record_list: if tx_record.SOPClassUID == uid.RTIonBeamsTreatmentRecordStorage: - tx_session_list = [ x for x in tx_record.TreatmentSessionIonBeamSequence] + tx_session_list = [x for x in tx_record.TreatmentSessionIonBeamSequence] else: - tx_session_list = [ x for x in tx_record.TreatmentSessionBeamSequence] + tx_session_list = [x for x in tx_record.TreatmentSessionBeamSequence] for tx_session in tx_session_list: - if (tx_session.ReferencedBeamNumber == beam_number): + if tx_session.ReferencedBeamNumber == beam_number: return True return False -def _populate_default_beam_task_elements(beam_task:Dataset): + +def _populate_default_beam_task_elements(beam_task: Dataset): beam_task.TableTopVerticalAdjustedPosition = 0 beam_task.TableTopLongitudinalAdjustedPosition = 0 beam_task.TableTopLateralAdjustedPosition = 0 @@ -194,26 +197,24 @@ def _populate_default_beam_task_elements(beam_task:Dataset): beam_task.TableTopLongitudinalSetupDisplacement = 0 beam_task.TableTopLateralSetupDisplacement = 0 -def _delete_excess_plan_elements_from_bdi(bdi_ds:Dataset): - excess_elements = [ "PositionReferenceIndicator", "FrameOfReferenceUID"] + +def _delete_excess_plan_elements_from_bdi(bdi_ds: Dataset): + excess_elements = ["PositionReferenceIndicator", "FrameOfReferenceUID"] for excess in excess_elements: if excess in bdi_ds: del bdi_ds[excess] -def _delete_excess_plan_elements_from_ups(ups_ds:Dataset): - excess_elements = [ "PositionReferenceIndicator", - "FrameOfReferenceUID", - "Modality"] + +def _delete_excess_plan_elements_from_ups(ups_ds: Dataset): + excess_elements = ["PositionReferenceIndicator", "FrameOfReferenceUID", "Modality"] for excess in excess_elements: if excess in ups_ds: del ups_ds[excess] -def create_ups_from_plan_and_bdi(plan:Dataset, - bdi:Dataset, - retrieve_ae_title:str, - scheduled_datetime: datetime, - treatment_records:List[Dataset]) -> Dataset: +def create_ups_from_plan_and_bdi( + plan: Dataset, bdi: Dataset, retrieve_ae_title: str, scheduled_datetime: datetime, treatment_records: List[Dataset] +) -> Dataset: """Build up the UPS Args: @@ -227,31 +228,31 @@ def create_ups_from_plan_and_bdi(plan:Dataset, if elem.tag < 0x0010FFFF: ups_ds[elem.tag] = deepcopy(elem) # _delete_excess_plan_elements_from_ups(ups_ds) # probably should rename the function - ups_ds.SOPClassUID = '1.2.840.10008.5.1.4.34.6.1' # UPS Push SOP Class + ups_ds.SOPClassUID = "1.2.840.10008.5.1.4.34.6.1" # UPS Push SOP Class ups_ds.SOPInstanceUID = uid.generate_uid() ups_ds.Manufacturer = "SJS Targeted Solutions" ups_ds.ManufacturerModelName = "RTBDI Creator" ups_ds.SoftwareVersions = "0.1" - + work_item = pydicom.Dataset() work_item.CodeValue = "121726" work_item.CodingSchemeDesignator = "DCM" - work_item.CodeMeaning = 'RT Treatment with Internal Verification' - scheduled_work_item_code_sequence = pydicom.Sequence( [ work_item ] ) - ups_ds.ScheduledWorkitemCodeSequence= scheduled_work_item_code_sequence + work_item.CodeMeaning = "RT Treatment with Internal Verification" + scheduled_work_item_code_sequence = pydicom.Sequence([work_item]) + ups_ds.ScheduledWorkitemCodeSequence = scheduled_work_item_code_sequence plan_reference_item = _create_referenced_instances_and_access_item(plan, retrieve_ae_title) bdi_reference_item = _create_referenced_instances_and_access_item(bdi, retrieve_ae_title) treatment_record_reference_items = [] for treatment_rec in treatment_records: treatment_rec_ref_item = _create_referenced_instances_and_access_item(treatment_rec, retrieve_ae_title) treatment_record_reference_items.append(treatment_rec_ref_item) - list_of_reference_items = [ ] + list_of_reference_items = [] list_of_reference_items.append(plan_reference_item) list_of_reference_items.append(bdi_reference_item) - treatment_delivery_type = 'TREATMENT' + treatment_delivery_type = "TREATMENT" if len(treatment_record_reference_items) > 0: list_of_reference_items += treatment_record_reference_items - treatment_delivery_type = 'CONTINUATION' + treatment_delivery_type = "CONTINUATION" ups_ds.InputInformationSequence = pydicom.Sequence(list_of_reference_items) scheduled_station_name_code_sequence_item = _create_scheduled_station_name_code_sequence_item(plan) @@ -259,67 +260,50 @@ def create_ups_from_plan_and_bdi(plan:Dataset, ups_ds.InputReadinessState = "READY" ups_ds.ProcedureStepState = "SCHEDULED" ups_ds.ScheduledProcedureStepPriority = "MEDIUM" - ups_ds.WorklistLabel = 'Worklist label for ' + plan.RTPlanLabel - ups_ds.ProcedureStepLabel = 'Treatment Step for ' + plan.RTPlanName + ups_ds.WorklistLabel = "Worklist label for " + plan.RTPlanLabel + ups_ds.ProcedureStepLabel = "Treatment Step for " + plan.RTPlanName ups_ds.ScheduledProcedureStepStartDateTime = scheduled_datetime scheduled_processing_parameters_list = [] - treatment_delivery_concept = _create_code_seq_item('121740', - 'DCM', - 'Treatment Delivery Type' - ) - treatment_param_item = _create_ups_content_item('TEXT', - 'TREATMENT', - treatment_delivery_concept - ) - + treatment_delivery_concept = _create_code_seq_item("121740", "DCM", "Treatment Delivery Type") + # maybe should be using treatment_delivery_type variable instead of "TREATMENT" hardcoded constant + treatment_param_item = _create_ups_content_item("TEXT", treatment_delivery_type, treatment_delivery_concept) + scheduled_processing_parameters_list.append(treatment_param_item) - plan_label_concept = _create_code_seq_item('2018001', - '99IHERO2018', - 'Plan Label') - plan_label_value = 'No Plan Label' - if 'RTPlanLabel' in plan: + plan_label_concept = _create_code_seq_item("2018001", "99IHERO2018", "Plan Label") + plan_label_value = "No Plan Label" + if "RTPlanLabel" in plan: plan_label_value = plan.RTPlanLabel - elif 'RTPlanName' in plan: + elif "RTPlanName" in plan: plan_label_value = plan.RTPlanName - plan_label_item = _create_ups_content_item('TEXT', - plan_label_value, - plan_label_concept) - + plan_label_item = _create_ups_content_item("TEXT", plan_label_value, plan_label_concept) + scheduled_processing_parameters_list.append(plan_label_item) - current_fraction_concept = _create_code_seq_item('2018002', - '99IHERO2018', - 'Current Fraction Number' - ) - - - current_fraction_item = _create_ups_content_item('NUMERIC', - int(bdi.BeamTaskSequence[0].CurrentFractionNumber), - current_fraction_concept - ) - + current_fraction_concept = _create_code_seq_item("2018002", "99IHERO2018", "Current Fraction Number") + + current_fraction_item = _create_ups_content_item( + "NUMERIC", int(bdi.BeamTaskSequence[0].CurrentFractionNumber), current_fraction_concept + ) + scheduled_processing_parameters_list.append(current_fraction_item) - fractions_planned_concept = _create_code_seq_item('2018003', - '99IHERO2018', - 'Number of Fractions Planned' - ) - fractions_planned_item = _create_ups_content_item('NUMERIC', - int(plan.FractionGroupSequence[0].NumberOfFractionsPlanned), - fractions_planned_concept - ) + fractions_planned_concept = _create_code_seq_item("2018003", "99IHERO2018", "Number of Fractions Planned") + fractions_planned_item = _create_ups_content_item( + "NUMERIC", int(plan.FractionGroupSequence[0].NumberOfFractionsPlanned), fractions_planned_concept + ) scheduled_processing_parameters_list.append(fractions_planned_item) ups_ds.ScheduledProcessingParametersSequence = pydicom.Sequence(scheduled_processing_parameters_list) return ups_ds + @lru_cache -def _measurement_units_code_seq_item_no_units()->Dataset: - return _create_code_seq_item('1','UCUM','no units') +def _measurement_units_code_seq_item_no_units() -> Dataset: + return _create_code_seq_item("1", "UCUM", "no units") -def _create_code_seq_item(value:str|int, designator:str, meaning:str)->Dataset: +def _create_code_seq_item(value: str | int, designator: str, meaning: str) -> Dataset: code_seq_item = Dataset() code_seq_item.CodeValue = value code_seq_item.CodingSchemeDesignator = designator @@ -327,13 +311,13 @@ def _create_code_seq_item(value:str|int, designator:str, meaning:str)->Dataset: return code_seq_item -def _create_ups_content_item(value_type: str, value:any, code_seq_item:Dataset)->Dataset: +def _create_ups_content_item(value_type: str, value: any, code_seq_item: Dataset) -> Dataset: content_item = Dataset() content_item.ValueType = value_type - if value_type == 'TEXT': + if value_type == "TEXT": content_item.TextValue = str(value) - elif value_type == 'NUMERIC': - content_item.MeasurementUnitsCodeSequence = pydicom.Sequence([ _measurement_units_code_seq_item_no_units() ]) + elif value_type == "NUMERIC": + content_item.MeasurementUnitsCodeSequence = pydicom.Sequence([_measurement_units_code_seq_item_no_units()]) content_item.ConceptNameCodeSequence = pydicom.Sequence([code_seq_item]) content_item.NumericValue = value else: @@ -341,17 +325,17 @@ def _create_ups_content_item(value_type: str, value:any, code_seq_item:Dataset)- return content_item -def _datetime_to_dicom_date(dt:datetime)-> str: +def _datetime_to_dicom_date(dt: datetime) -> str: dicom_date = dt.strftime("%Y%m%d") return dicom_date -def _datetime_to_dicom_time(dt:datetime)-> str: +def _datetime_to_dicom_time(dt: datetime) -> str: dicom_time = dt.strftime("%H%m.%s") return dicom_time -def _create_referenced_instances_and_access_item(input_ds:Dataset, retrieve_ae_title:str)->Dataset: +def _create_referenced_instances_and_access_item(input_ds: Dataset, retrieve_ae_title: str) -> Dataset: ref_instance_seq_item = pydicom.Dataset() ref_instance_seq_item.TypeOfInstances = "DICOM" ref_instance_seq_item.StudyInstanceUID = input_ds.StudyInstanceUID @@ -366,7 +350,7 @@ def _create_referenced_instances_and_access_item(input_ds:Dataset, retrieve_ae_t return ref_instance_seq_item -def _create_scheduled_station_name_code_sequence_item(plan:Dataset)->Dataset: +def _create_scheduled_station_name_code_sequence_item(plan: Dataset) -> Dataset: machine_name = "" if plan.SOPClassUID == uid.RTIonPlanStorage: while len(machine_name) == 0: @@ -396,24 +380,24 @@ def main(args): tx_record_list = [] fraction_number = 1 scheduled_time = datetime.now() - retrieve_ae_title = 'TDWII_MOVE_SCP' - if (len(args)>2): + retrieve_ae_title = "TDWII_MOVE_SCP" + if len(args) > 2: fraction_number = int(args[2]) - if (len(args)>3): + if len(args) > 3: retrieve_ae_title = args[3] - if (len(args) > 4): - scheduled_time = datetime.fromtimestamp(mktime(strptime(argv[4],"%Y%m%d%H%M"))) - if (len(args)>5): + if len(args) > 4: + scheduled_time = datetime.fromtimestamp(mktime(strptime(argv[4], "%Y%m%d%H%M"))) + if len(args) > 5: treatment_record_paths = args[5:] tx_record_list = load_treatment_records(treatment_record_paths) - - bdi = create_rtbdi_from_rtion_plan(plan,fraction_number=fraction_number,treatment_record_list=tx_record_list) - write_rtbdi(bdi, Path.cwd()) - ups = create_ups_from_plan_and_bdi(plan, bdi,retrieve_ae_title, scheduled_time, tx_record_list) - write_ups(ups,Path.cwd()) + bdi = create_rtbdi_from_rtion_plan(plan, fraction_number=fraction_number, treatment_record_list=tx_record_list) + write_rtbdi(bdi, Path.cwd()) + ups = create_ups_from_plan_and_bdi(plan, bdi, retrieve_ae_title, scheduled_time, tx_record_list) + write_ups(ups, Path.cwd()) -def gen_one_session(plan:Dataset,fraction_number:int, scheduled_date_time:datetime, retrieve_ae_title:str): + +def gen_one_session(plan: Dataset, fraction_number: int, scheduled_date_time: datetime, retrieve_ae_title: str): """wrapper to simplify generation of an entire course worth of sessions by invoking this multiple times with incrementing fraction number and schedule datetime @@ -423,19 +407,24 @@ def gen_one_session(plan:Dataset,fraction_number:int, scheduled_date_time:dateti scheduled_date_time (datetime): _description_ retrieve_ae_title (str): _description_ """ - tx_record_list = [] - scheduled_time = scheduled_date_time - bdi = create_rtbdi_from_rtion_plan(plan,fraction_number,treatment_record_list=tx_record_list) - write_rtbdi(bdi, Path.cwd()) - ups = create_ups_from_plan_and_bdi(plan, bdi,retrieve_ae_title, scheduled_time, tx_record_list) - write_ups(ups,Path.cwd()) + tx_record_list = [] + scheduled_time = scheduled_date_time + bdi = create_rtbdi_from_rtion_plan(plan, fraction_number, treatment_record_list=tx_record_list) + write_rtbdi(bdi, Path.cwd()) + ups = create_ups_from_plan_and_bdi(plan, bdi, retrieve_ae_title, scheduled_time, tx_record_list) + write_ups(ups, Path.cwd()) if __name__ == "__main__": - if len(argv) < 2 : - print("Usage: python {argv[0]} plan_file ") - print("Defaults to fraction_number=1, retrieve_ae_title=TDWII_MOVE_SCP," - " scheduled_date_time=now, no interruptions/continuations" ) + if len(argv) < 2: + print( + 'Usage: python {argv[0]} plan_file ' + '' + ) + print( + "Defaults to fraction_number=1, retrieve_ae_title=TDWII_MOVE_SCP," + " scheduled_date_time=now, no interruptions/continuations" + ) else: - print(' '.join(argv[1:]) ) + print(" ".join(argv[1:])) main(argv) diff --git a/tdwii_plus_examples/rtbdi_creator/ui_form.py b/tdwii_plus_examples/rtbdi_creator/ui_form.py index 24f22ce..3411308 100644 --- a/tdwii_plus_examples/rtbdi_creator/ui_form.py +++ b/tdwii_plus_examples/rtbdi_creator/ui_form.py @@ -8,169 +8,222 @@ ## WARNING! All changes made in this file will be lost when recompiling UI file! ################################################################################ -from PySide6.QtCore import (QCoreApplication, QDate, QDateTime, QLocale, - QMetaObject, QObject, QPoint, QRect, - QSize, QTime, QUrl, Qt) -from PySide6.QtGui import (QBrush, QColor, QConicalGradient, QCursor, - QFont, QFontDatabase, QGradient, QIcon, - QImage, QKeySequence, QLinearGradient, QPainter, - QPalette, QPixmap, QRadialGradient, QTransform) -from PySide6.QtWidgets import (QApplication, QCheckBox, QDateTimeEdit, QDoubleSpinBox, - QGridLayout, QGroupBox, QLabel, QLineEdit, - QListView, QPushButton, QSizePolicy, QWidget) +from PySide6.QtCore import ( + QCoreApplication, + QDate, + QDateTime, + QLocale, + QMetaObject, + QObject, + QPoint, + QRect, + QSize, + Qt, + QTime, + QUrl, +) +from PySide6.QtGui import ( + QBrush, + QColor, + QConicalGradient, + QCursor, + QFont, + QFontDatabase, + QGradient, + QIcon, + QImage, + QKeySequence, + QLinearGradient, + QPainter, + QPalette, + QPixmap, + QRadialGradient, + QTransform, +) +from PySide6.QtWidgets import ( + QApplication, + QCheckBox, + QDateTimeEdit, + QDoubleSpinBox, + QGridLayout, + QGroupBox, + QLabel, + QLineEdit, + QListView, + QPushButton, + QSizePolicy, + QWidget, +) + class Ui_MainBDIWidget(object): def setupUi(self, MainBDIWidget): if not MainBDIWidget.objectName(): - MainBDIWidget.setObjectName(u"MainBDIWidget") + MainBDIWidget.setObjectName("MainBDIWidget") MainBDIWidget.resize(800, 600) self.group_box_bdi_variables = QGroupBox(MainBDIWidget) - self.group_box_bdi_variables.setObjectName(u"group_box_bdi_variables") + self.group_box_bdi_variables.setObjectName("group_box_bdi_variables") self.group_box_bdi_variables.setGeometry(QRect(40, 110, 491, 121)) self.gridLayout_3 = QGridLayout(self.group_box_bdi_variables) - self.gridLayout_3.setObjectName(u"gridLayout_3") + self.gridLayout_3.setObjectName("gridLayout_3") self.double_spin_box_fraction_number = QDoubleSpinBox(self.group_box_bdi_variables) - self.double_spin_box_fraction_number.setObjectName(u"double_spin_box_fraction_number") - self.double_spin_box_fraction_number.setAlignment(Qt.AlignRight|Qt.AlignTrailing|Qt.AlignVCenter) + self.double_spin_box_fraction_number.setObjectName("double_spin_box_fraction_number") + self.double_spin_box_fraction_number.setAlignment(Qt.AlignRight | Qt.AlignTrailing | Qt.AlignVCenter) self.double_spin_box_fraction_number.setDecimals(0) self.gridLayout_3.addWidget(self.double_spin_box_fraction_number, 0, 1, 1, 1) self.label_fraction_number = QLabel(self.group_box_bdi_variables) - self.label_fraction_number.setObjectName(u"label_fraction_number") + self.label_fraction_number.setObjectName("label_fraction_number") self.gridLayout_3.addWidget(self.label_fraction_number, 0, 0, 1, 1) self.list_view_treatment_records = QListView(self.group_box_bdi_variables) - self.list_view_treatment_records.setObjectName(u"list_view_treatment_records") + self.list_view_treatment_records.setObjectName("list_view_treatment_records") self.gridLayout_3.addWidget(self.list_view_treatment_records, 1, 0, 1, 1) self.push_button_load_treatment_records = QPushButton(self.group_box_bdi_variables) - self.push_button_load_treatment_records.setObjectName(u"push_button_load_treatment_records") + self.push_button_load_treatment_records.setObjectName("push_button_load_treatment_records") self.gridLayout_3.addWidget(self.push_button_load_treatment_records, 1, 1, 1, 1) self.group_box_plan_selection = QGroupBox(MainBDIWidget) - self.group_box_plan_selection.setObjectName(u"group_box_plan_selection") + self.group_box_plan_selection.setObjectName("group_box_plan_selection") self.group_box_plan_selection.setGeometry(QRect(40, 20, 581, 70)) self.gridLayout_4 = QGridLayout(self.group_box_plan_selection) - self.gridLayout_4.setObjectName(u"gridLayout_4") + self.gridLayout_4.setObjectName("gridLayout_4") self.label_plan_selector = QLabel(self.group_box_plan_selection) - self.label_plan_selector.setObjectName(u"label_plan_selector") + self.label_plan_selector.setObjectName("label_plan_selector") self.gridLayout_4.addWidget(self.label_plan_selector, 0, 0, 1, 1) self.lineedit_plan_selector = QLineEdit(self.group_box_plan_selection) - self.lineedit_plan_selector.setObjectName(u"lineedit_plan_selector") + self.lineedit_plan_selector.setObjectName("lineedit_plan_selector") self.gridLayout_4.addWidget(self.lineedit_plan_selector, 0, 1, 1, 1) self.push_button_plan_finder = QPushButton(self.group_box_plan_selection) - self.push_button_plan_finder.setObjectName(u"push_button_plan_finder") + self.push_button_plan_finder.setObjectName("push_button_plan_finder") self.gridLayout_4.addWidget(self.push_button_plan_finder, 0, 2, 1, 1) self.group_box_bdi_output = QGroupBox(MainBDIWidget) - self.group_box_bdi_output.setObjectName(u"group_box_bdi_output") + self.group_box_bdi_output.setObjectName("group_box_bdi_output") self.group_box_bdi_output.setGeometry(QRect(40, 230, 591, 193)) self.gridLayout_2 = QGridLayout(self.group_box_bdi_output) - self.gridLayout_2.setObjectName(u"gridLayout_2") + self.gridLayout_2.setObjectName("gridLayout_2") self.label_bdi_output_dir = QLabel(self.group_box_bdi_output) - self.label_bdi_output_dir.setObjectName(u"label_bdi_output_dir") + self.label_bdi_output_dir.setObjectName("label_bdi_output_dir") self.gridLayout_2.addWidget(self.label_bdi_output_dir, 0, 0, 1, 1) self.lineedit_bdidir_selector = QLineEdit(self.group_box_bdi_output) - self.lineedit_bdidir_selector.setObjectName(u"lineedit_bdidir_selector") + self.lineedit_bdidir_selector.setObjectName("lineedit_bdidir_selector") self.gridLayout_2.addWidget(self.lineedit_bdidir_selector, 0, 1, 1, 2) self.push_button_bdi_dir_finder = QPushButton(self.group_box_bdi_output) - self.push_button_bdi_dir_finder.setObjectName(u"push_button_bdi_dir_finder") + self.push_button_bdi_dir_finder.setObjectName("push_button_bdi_dir_finder") self.gridLayout_2.addWidget(self.push_button_bdi_dir_finder, 0, 3, 1, 1) self.checkbox_custom_bdi_filename = QCheckBox(self.group_box_bdi_output) - self.checkbox_custom_bdi_filename.setObjectName(u"checkbox_custom_bdi_filename") + self.checkbox_custom_bdi_filename.setObjectName("checkbox_custom_bdi_filename") self.gridLayout_2.addWidget(self.checkbox_custom_bdi_filename, 1, 0, 1, 2) self.line_edit_bdi_filename = QLineEdit(self.group_box_bdi_output) - self.line_edit_bdi_filename.setObjectName(u"line_edit_bdi_filename") + self.line_edit_bdi_filename.setObjectName("line_edit_bdi_filename") self.gridLayout_2.addWidget(self.line_edit_bdi_filename, 1, 2, 1, 1) self.push_button_export_bdi = QPushButton(self.group_box_bdi_output) - self.push_button_export_bdi.setObjectName(u"push_button_export_bdi") + self.push_button_export_bdi.setObjectName("push_button_export_bdi") self.gridLayout_2.addWidget(self.push_button_export_bdi, 3, 3, 1, 1) self.group_box_ups = QGroupBox(MainBDIWidget) - self.group_box_ups.setObjectName(u"group_box_ups") + self.group_box_ups.setObjectName("group_box_ups") self.group_box_ups.setGeometry(QRect(40, 430, 359, 151)) self.gridLayout = QGridLayout(self.group_box_ups) - self.gridLayout.setObjectName(u"gridLayout") + self.gridLayout.setObjectName("gridLayout") self.label_start_datetime = QLabel(self.group_box_ups) - self.label_start_datetime.setObjectName(u"label_start_datetime") + self.label_start_datetime.setObjectName("label_start_datetime") self.gridLayout.addWidget(self.label_start_datetime, 1, 0, 1, 1) self.datetime_edit_scheduled_datetime = QDateTimeEdit(self.group_box_ups) - self.datetime_edit_scheduled_datetime.setObjectName(u"datetime_edit_scheduled_datetime") + self.datetime_edit_scheduled_datetime.setObjectName("datetime_edit_scheduled_datetime") self.datetime_edit_scheduled_datetime.setDateTime(QDateTime(QDate(2023, 9, 1), QTime(0, 0, 0))) self.datetime_edit_scheduled_datetime.setCalendarPopup(True) self.gridLayout.addWidget(self.datetime_edit_scheduled_datetime, 1, 1, 1, 1) self.label_move_ae_title = QLabel(self.group_box_ups) - self.label_move_ae_title.setObjectName(u"label_move_ae_title") + self.label_move_ae_title.setObjectName("label_move_ae_title") self.gridLayout.addWidget(self.label_move_ae_title, 0, 0, 1, 1) self.line_edit_move_scp_ae_title = QLineEdit(self.group_box_ups) - self.line_edit_move_scp_ae_title.setObjectName(u"line_edit_move_scp_ae_title") + self.line_edit_move_scp_ae_title.setObjectName("line_edit_move_scp_ae_title") self.line_edit_move_scp_ae_title.setClearButtonEnabled(False) self.gridLayout.addWidget(self.line_edit_move_scp_ae_title, 0, 1, 1, 1) self.push_button_export_ups = QPushButton(self.group_box_ups) - self.push_button_export_ups.setObjectName(u"push_button_export_ups") + self.push_button_export_ups.setObjectName("push_button_export_ups") self.gridLayout.addWidget(self.push_button_export_ups, 2, 1, 1, 1) - self.retranslateUi(MainBDIWidget) QMetaObject.connectSlotsByName(MainBDIWidget) + # setupUi def retranslateUi(self, MainBDIWidget): - MainBDIWidget.setWindowTitle(QCoreApplication.translate("MainBDIWidget", u"RT Beams Delivery Instruction and UPS Creator", None)) -#if QT_CONFIG(tooltip) + MainBDIWidget.setWindowTitle( + QCoreApplication.translate("MainBDIWidget", "RT Beams Delivery Instruction and UPS Creator", None) + ) + # if QT_CONFIG(tooltip) MainBDIWidget.setToolTip("") -#endif // QT_CONFIG(tooltip) - self.group_box_bdi_variables.setTitle(QCoreApplication.translate("MainBDIWidget", u"BDI Customization", None)) - self.label_fraction_number.setText(QCoreApplication.translate("MainBDIWidget", u"Fraction #", None)) -#if QT_CONFIG(tooltip) - self.push_button_load_treatment_records.setToolTip(QCoreApplication.translate("MainBDIWidget", u"

Simulate a partially treated session.

The treatment records selected must reference the RT (Ion) Plan selected, and the Current Fraction Number in the RT (Ion) Beams Session records must match the Fraction # specified

", None)) -#endif // QT_CONFIG(tooltip) - self.push_button_load_treatment_records.setText(QCoreApplication.translate("MainBDIWidget", u"Treatment Records", None)) - self.group_box_plan_selection.setTitle(QCoreApplication.translate("MainBDIWidget", u"Plan Selection", None)) - self.label_plan_selector.setText(QCoreApplication.translate("MainBDIWidget", u"RT (Ion) Plan", None)) - self.push_button_plan_finder.setText(QCoreApplication.translate("MainBDIWidget", u"Find Plan", None)) - self.group_box_bdi_output.setTitle(QCoreApplication.translate("MainBDIWidget", u"BDI Output", None)) - self.label_bdi_output_dir.setText(QCoreApplication.translate("MainBDIWidget", u"BDI Output Dir", None)) - self.push_button_bdi_dir_finder.setText(QCoreApplication.translate("MainBDIWidget", u"Choose BDI Dir", None)) -#if QT_CONFIG(tooltip) - self.checkbox_custom_bdi_filename.setToolTip(QCoreApplication.translate("MainBDIWidget", u"

When checked, the value entered will be used for the RT Beams Delivery Instruction file name.

If left unchecked, the file name for the RT Beams Delivery Instruction will be of the format RB_<SOP Instance UID>.dcm

", None)) -#endif // QT_CONFIG(tooltip) - self.checkbox_custom_bdi_filename.setText(QCoreApplication.translate("MainBDIWidget", u"Custom BDI filename", None)) - self.push_button_export_bdi.setText(QCoreApplication.translate("MainBDIWidget", u"Export BDI", None)) - self.group_box_ups.setTitle(QCoreApplication.translate("MainBDIWidget", u"UPS Customization", None)) - self.label_start_datetime.setText(QCoreApplication.translate("MainBDIWidget", u"Scheduled DateTime", None)) - self.datetime_edit_scheduled_datetime.setDisplayFormat(QCoreApplication.translate("MainBDIWidget", u"dd/MM/yyyy h:mm AP", None)) - self.label_move_ae_title.setText(QCoreApplication.translate("MainBDIWidget", u"Move/Retrieve AE Title", None)) - self.push_button_export_ups.setText(QCoreApplication.translate("MainBDIWidget", u"Export UPS", None)) - # retranslateUi + # endif // QT_CONFIG(tooltip) + self.group_box_bdi_variables.setTitle(QCoreApplication.translate("MainBDIWidget", "BDI Customization", None)) + self.label_fraction_number.setText(QCoreApplication.translate("MainBDIWidget", "Fraction #", None)) + # if QT_CONFIG(tooltip) + self.push_button_load_treatment_records.setToolTip( + QCoreApplication.translate( + "MainBDIWidget", + "

Simulate a partially treated session.

The treatment records selected must reference the RT (Ion) Plan selected, and the Current Fraction Number in the RT (Ion) Beams Session records must match the Fraction # specified

", + None, + ) + ) + # endif // QT_CONFIG(tooltip) + self.push_button_load_treatment_records.setText(QCoreApplication.translate("MainBDIWidget", "Treatment Records", None)) + self.group_box_plan_selection.setTitle(QCoreApplication.translate("MainBDIWidget", "Plan Selection", None)) + self.label_plan_selector.setText(QCoreApplication.translate("MainBDIWidget", "RT (Ion) Plan", None)) + self.push_button_plan_finder.setText(QCoreApplication.translate("MainBDIWidget", "Find Plan", None)) + self.group_box_bdi_output.setTitle(QCoreApplication.translate("MainBDIWidget", "BDI Output", None)) + self.label_bdi_output_dir.setText(QCoreApplication.translate("MainBDIWidget", "BDI Output Dir", None)) + self.push_button_bdi_dir_finder.setText(QCoreApplication.translate("MainBDIWidget", "Choose BDI Dir", None)) + # if QT_CONFIG(tooltip) + self.checkbox_custom_bdi_filename.setToolTip( + QCoreApplication.translate( + "MainBDIWidget", + "

When checked, the value entered will be used for the RT Beams Delivery Instruction file name.

If left unchecked, the file name for the RT Beams Delivery Instruction will be of the format RB_<SOP Instance UID>.dcm

", + None, + ) + ) + # endif // QT_CONFIG(tooltip) + self.checkbox_custom_bdi_filename.setText(QCoreApplication.translate("MainBDIWidget", "Custom BDI filename", None)) + self.push_button_export_bdi.setText(QCoreApplication.translate("MainBDIWidget", "Export BDI", None)) + self.group_box_ups.setTitle(QCoreApplication.translate("MainBDIWidget", "UPS Customization", None)) + self.label_start_datetime.setText(QCoreApplication.translate("MainBDIWidget", "Scheduled DateTime", None)) + self.datetime_edit_scheduled_datetime.setDisplayFormat( + QCoreApplication.translate("MainBDIWidget", "dd/MM/yyyy h:mm AP", None) + ) + self.label_move_ae_title.setText(QCoreApplication.translate("MainBDIWidget", "Move/Retrieve AE Title", None)) + self.push_button_export_ups.setText(QCoreApplication.translate("MainBDIWidget", "Export UPS", None)) + # retranslateUi diff --git a/tdwii_plus_examples/tdwii_config.py b/tdwii_plus_examples/tdwii_config.py index 0bdb6a4..4cd90b8 100644 --- a/tdwii_plus_examples/tdwii_config.py +++ b/tdwii_plus_examples/tdwii_config.py @@ -45,6 +45,4 @@ def load_machine_map(path_to_machine_map=None): print(machine_ae_map) for key in machine_ae_map: value = machine_ae_map[key] - print( - f"Machine Name:{key} AE Title:{value} IPAddr:{known_ae_ipaddr[value]} Port:{known_ae_port[value]}" - ) + print(f"Machine Name:{key} AE Title:{value} IPAddr:{known_ae_ipaddr[value]} Port:{known_ae_port[value]}") diff --git a/tdwii_plus_examples/tests/test_nevent_sender.py b/tdwii_plus_examples/tests/test_nevent_sender.py index ff11d5a..bf5fe23 100644 --- a/tdwii_plus_examples/tests/test_nevent_sender.py +++ b/tdwii_plus_examples/tests/test_nevent_sender.py @@ -1,31 +1,15 @@ """Unit tests for nevent_sender.py""" -from time import sleep import os import subprocess import sys -import time -import pytest + from pydicom import Dataset, dcmread -from pydicom.uid import ( - DeflatedExplicitVRLittleEndian, - ExplicitVRBigEndian, - ExplicitVRLittleEndian, - ImplicitVRLittleEndian, -) -from pynetdicom import ( - AE, - ALL_TRANSFER_SYNTAXES, - UnifiedProcedurePresentationContexts, - evt, -) -from pynetdicom.sop_class import UnifiedProcedureStepPush, Verification - -# from nevent_receiver_handlers import handle_nevent -from tdwii_plus_examples.TDWII_PPVS_subscriber.nevent_receiver import NEventReceiver - -# debug_logger() + +from pynetdicom import AE, evt + +from pynetdicom.sop_class import UnifiedProcedureStepPush APP_DIR = os.path.join(os.path.dirname(__file__), "../") @@ -95,16 +79,13 @@ def handle_release(event): ae.network_timeout = 5 ae.add_supported_context(UnifiedProcedureStepPush) scp = ae.start_server(("localhost", 11115), block=False, evt_handlers=handlers) - p = self.func(["127.0.0.1", "11115"]) p.wait() assert p.returncode == 0 # sleep(1.0) scp.shutdown() - - assert events[0].event == evt.EVT_N_EVENT_REPORT current_event = events[0] nevent_primitive = current_event.request @@ -136,7 +117,7 @@ def handle_release(event): assert nevent_information.ProcedureStepState == "COMPLETED" assert events[3].event == evt.EVT_RELEASED - + # def test_no_peer(self, capfd): # """Test trying to connect to non-existent host.""" # p = self.func([DATASET_FILE]) @@ -146,7 +127,6 @@ def handle_release(event): # assert "Association request failed: unable to connect to remote" in err # assert "TCP Initialisation Error" in err - # def test_bad_input(self, capfd): # """Test being unable to read the input file.""" # p = self.func(["no-such-file.dcm", "-d"]) @@ -157,7 +137,6 @@ def handle_release(event): # assert "No suitable DICOM files found" in err # assert "Cannot access path: no-such-file.dcm" in err - class Testneventsender(neventsenderBase): """Tests for nevent_sender.py""" diff --git a/tdwii_plus_examples/upsdb.py b/tdwii_plus_examples/upsdb.py index 90afc24..3df17d4 100644 --- a/tdwii_plus_examples/upsdb.py +++ b/tdwii_plus_examples/upsdb.py @@ -193,11 +193,7 @@ def add_instance(ds, session, fpath=None): to the database file. """ # Check if instance is already in the database - result = ( - session.query(Instance) - .filter(Instance.sop_instance_uid == ds.SOPInstanceUID) - .all() - ) + result = session.query(Instance).filter(Instance.sop_instance_uid == ds.SOPInstanceUID).all() if result: instance = result[0] else: @@ -246,9 +242,7 @@ def add_instance(ds, session, fpath=None): if value is None: value = "SCHEDULED" elif value != "SCHEDULED": - raise InvalidIdentifier( - f"ProcedureStepState via N-CREATE must be SCHEDULED, received {value}" - ) + raise InvalidIdentifier(f"ProcedureStepState via N-CREATE must be SCHEDULED, received {value}") if value is not None: # All supported attributes have VM 1 @@ -377,9 +371,7 @@ def _check_identifier(identifier, model): # Part 4, C.4.1.1.3.1, C.4.2.1.4 and C.4.3.1.3.1: # (0008,0052) Query Retrieve Level is required in the Identifier if "QueryRetrieveLevel" not in identifier: - raise InvalidIdentifier( - "The Identifier contains no Query Retrieve Level element" - ) + raise InvalidIdentifier("The Identifier contains no Query Retrieve Level element") if model in _PATIENT_ROOT: attr = _PATIENT_ROOT[model] @@ -388,9 +380,7 @@ def _check_identifier(identifier, model): levels = list(attr.keys()) if identifier.QueryRetrieveLevel not in levels: - raise InvalidIdentifier( - "The Identifier's Query Retrieve Level value is invalid" - ) + raise InvalidIdentifier("The Identifier's Query Retrieve Level value is invalid") if len(identifier) == 1: raise InvalidIdentifier("The Identifier contains no keys") @@ -401,8 +391,7 @@ def _check_identifier(identifier, model): for sublevel in levels[ii + 1 :]: if any([kw in identifier for kw in attr[sublevel]]): raise InvalidIdentifier( - "The Identifier contains keys below the level " - "specified by the Query Retrieve Level" + "The Identifier contains keys below the level " "specified by the Query Retrieve Level" ) # The level is the same as that in the identifier so we're OK @@ -411,9 +400,7 @@ def _check_identifier(identifier, model): # The level is above that in the identifier so make sure the unique # keyword is present if attr[level][0] not in identifier: - raise InvalidIdentifier( - f"The Identifier is missing a unique key for " f"the '{level}' level" - ) + raise InvalidIdentifier(f"The Identifier is missing a unique key for " f"the '{level}' level") def clear(session): @@ -459,9 +446,7 @@ def remove_instance(instance_uid, session): session : sqlalchemy.orm.session.Session The session to use when querying the database for the instance. """ - matches = ( - session.query(Instance).filter(Instance.sop_instance_uid == instance_uid).all() - ) + matches = session.query(Instance).filter(Instance.sop_instance_uid == instance_uid).all() if matches: session.delete(matches[0]) session.commit() @@ -832,12 +817,8 @@ class Instance(Base): transaction_uid = Column(String(64)) station_name = Column(String(64)) - work_item_code_value = Column( - String(64) - ) # needs to be matched in conjunction with the scheme designator - work_item_scheme_designator = Column( - String(64) - ) # typically DCM, but could be 99IHERO2008 or other "private" schema + work_item_code_value = Column(String(64)) # needs to be matched in conjunction with the scheme designator + work_item_scheme_designator = Column(String(64)) # typically DCM, but could be 99IHERO2008 or other "private" schema work_item_meaning = Column(String(64)) start_date_time = Column(String(64)) procedure_step_state = Column(String(16)) @@ -915,10 +896,7 @@ def context(self): available for the Instance. """ if None in [self.sop_class_uid, self.transfer_syntax_uid]: - raise ValueError( - "Cannot determine which presentation context is required for " - "for the SOP Instance" - ) + raise ValueError("Cannot determine which presentation context is required for " "for the SOP Instance") return build_context(self.sop_class_uid, self.transfer_syntax_uid) diff --git a/tdwii_plus_examples/upsscp.py b/tdwii_plus_examples/upsscp.py index b956c4b..1c50d31 100755 --- a/tdwii_plus_examples/upsscp.py +++ b/tdwii_plus_examples/upsscp.py @@ -7,16 +7,6 @@ from configparser import ConfigParser import pydicom.config - -import upsdb -from handlers import ( - handle_echo, - handle_find, - handle_naction, - handle_ncreate, - handle_nget, - handle_nset, -) from pynetdicom import ( AE, ALL_TRANSFER_SYNTAXES, @@ -29,6 +19,15 @@ from pynetdicom.sop_class import Verification from pynetdicom.utils import set_ae +import upsdb +from handlers import ( + handle_echo, + handle_find, + handle_naction, + handle_ncreate, + handle_nget, + handle_nset, +) # Use `None` for empty values pydicom.config.use_none_as_empty_text_VR_value = True diff --git a/tdwii_plus_examples/watchscu.py b/tdwii_plus_examples/watchscu.py index e349521..5b2e85b 100755 --- a/tdwii_plus_examples/watchscu.py +++ b/tdwii_plus_examples/watchscu.py @@ -48,9 +48,7 @@ def send_action( return assoc.send_n_action(action_info, action_type, class_uid, instance_uid) -def send_global_watch_registration( - args: argparse.Namespace, assoc: Association, action_info: Dataset = None -): +def send_global_watch_registration(args: argparse.Namespace, assoc: Association, action_info: Dataset = None): """_summary_ Args: @@ -84,25 +82,18 @@ def _setup_argparser(): """Setup the command line arguments""" # Description parser = argparse.ArgumentParser( - description=( - "The watchscu application implements a Service Class User " - "(SCU) for the UPS Watch Class. " - ), + description=("The watchscu application implements a Service Class User " "(SCU) for the UPS Watch Class. "), usage="watchscu [options] addr port", ) # Parameters req_opts = parser.add_argument_group("Parameters") - req_opts.add_argument( - "addr", help="TCP/IP address or hostname of DICOM peer", type=str - ) + req_opts.add_argument("addr", help="TCP/IP address or hostname of DICOM peer", type=str) req_opts.add_argument("port", help="TCP/IP port number of peer", type=int) # General Options gen_opts = parser.add_argument_group("General Options") - gen_opts.add_argument( - "--version", help="print version information and exit", action="store_true" - ) + gen_opts.add_argument("--version", help="print version information and exit", action="store_true") output = gen_opts.add_mutually_exclusive_group() output.add_argument( "-q", @@ -172,7 +163,7 @@ def _setup_argparser(): metavar="[a]etitle", help="set receiver AE title of peer (default: NEVENT_RECEIVER)", type=str, - default="NEVENT_RECEIVER" + default="NEVENT_RECEIVER", ) net_opts.add_argument( "-ta", @@ -202,10 +193,7 @@ def _setup_argparser(): "-pdu", "--max-pdu", metavar="[n]umber of bytes", - help=( - f"set max receive pdu to n bytes (0 for unlimited, " - f"default: {DEFAULT_MAX_LENGTH})" - ), + help=(f"set max receive pdu to n bytes (0 for unlimited, " f"default: {DEFAULT_MAX_LENGTH})"), type=int, default=DEFAULT_MAX_LENGTH, )