from pynetdicom import AE
from pynetdicom.sop_class import ModalityWorklistInformationFind
from pydicom.dataset import FileMetaDataset
from pydicom.sequence import Sequence
from hl7apy.core import Message, Segment
from hl7apy.parser import parse_message
import socket
import logging
logging.basicConfig(level=logging.DEBUG)
MWL_SERVER = {
'IP': '192.168.10.225',
'PORT': 11112,
'AE_TITLE': 'WORKLIST',
}
HL7_SERVER = {
'IP': '192.168.10.225',
'PORT': 2575,
}
AE_CLIENT = 'OPH4PY'
study_data = {
'PatientID': '12345',
'PatientName': 'Doe^John',
'PatientBirthDate': '19700101',
'PatientSex': 'M',
'StudyID': '1.2.3',
'AccessionNumber': '1234567',
'ReferringPhysicianName': 'Smith^John',
'StudyDescription': 'Eye examination',
'ScheduledProcedureStepStartDate': '20230818',
'ScheduledProcedureStepStartTime': '0900',
'Modality': 'OP',
'RequestedProcedureID': '1001',
'RequestedProcedureDescription': 'Opthalmology Procedure',
'ScheduledStationAETitle': 'PENTACAM',
'ScheduledPerformingPhysicianName': 'Kerry^John',
'ScheduledProcedureStepLocation': 'Room 1',
'PreMedication': 'None',
'SpecialNeeds': 'None',
}
def get_MWL(study_data=None):
pacs_ip = MWL_SERVER['IP']
pacs_port = MWL_SERVER['PORT']
remote_ae_title = MWL_SERVER['AE_TITLE']
ae = AE(ae_title=AE_CLIENT)
ae.add_requested_context(ModalityWorklistInformationFind)
ds = Dataset()
ds.PatientName = study_data.get('PatientName', '') if study_data else ''
ds.PatientID = study_data.get('PatientID', '') if study_data else ''
ds.PatientBirthDate = study_data.get('PatientBirthDate', '') if study_data else ''
ds.PatientSex = study_data.get('PatientSex', '') if study_data else ''
ds.StudyID = study_data.get('StudyID', '') if study_data else ''
ds.AccessionNumber = study_data.get('AccessionNumber', '') if study_data else ''
ds.ReferringPhysicianName = study_data.get('ReferringPhysicianName', '') if study_data else ''
ds.StudyDescription = study_data.get('StudyDescription', '') if study_data else ''
spss = Dataset()
spss.ScheduledProcedureStepStartDate = study_data.get('ScheduledProcedureStepStartDate', '') if study_data else ''
spss.Modality = study_data.get('Modality', '') if study_data else ''
spss.RequestedProcedureID = study_data.get('RequestedProcedureID', '') if study_data else ''
spss.RequestedProcedureDescription = study_data.get('RequestedProcedureDescription', '') if study_data else ''
spss.ScheduledStationAETitle = study_data.get('ScheduledStationAETitle', '') if study_data else ''
spss.ScheduledPerformingPhysicianName = study_data.get('ScheduledPerformingPhysicianName', '') if study_data else ''
spss.ScheduledProcedureStepLocation = study_data.get('ScheduledProcedureStepLocation', '') if study_data else ''
spss.PreMedication = study_data.get('PreMedication', '') if study_data else ''
spss.SpecialNeeds = study_data.get('SpecialNeeds', '') if study_data else ''
ds.ScheduledProcedureStepSequence = [spss]
assoc = ae.associate(pacs_ip, pacs_port, ae_title=remote_ae_title)
studies = []
if assoc.is_established:
responses = assoc.send_c_find(ds, ModalityWorklistInformationFind)
for (status, identifier) in responses:
if status:
print('C-FIND query status: 0x{0:04x}'.format(status.Status))
if status.Status == 0xFF00:
study_details = {
'PatientName': getattr(identifier, 'PatientName', ''),
'PatientID': getattr(identifier, 'PatientID', ''),
'PatientBirthDate': getattr(identifier, 'PatientBirthDate', ''),
'PatientSex': getattr(identifier, 'PatientSex', ''),
'StudyID': getattr(identifier, 'StudyID', ''),
'AccessionNumber': getattr(identifier, 'AccessionNumber', ''),
'ReferringPhysicianName': getattr(identifier, 'ReferringPhysicianName', ''),
'StudyDescription': getattr(identifier, 'StudyDescription', ''),
'ScheduledProcedureStepStartDate': getattr(identifier.ScheduledProcedureStepSequence[0], 'ScheduledProcedureStepStartDate', ''),
'Modality': getattr(identifier.ScheduledProcedureStepSequence[0], 'Modality', ''),
'RequestedProcedureID': getattr(identifier.ScheduledProcedureStepSequence[0], 'RequestedProcedureID', ''),
'RequestedProcedureDescription': getattr(identifier.ScheduledProcedureStepSequence[0], 'RequestedProcedureDescription', ''),
'ScheduledStationAETitle': getattr(identifier.ScheduledProcedureStepSequence[0], 'ScheduledStationAETitle', ''),
'ScheduledPerformingPhysicianName': getattr(identifier.ScheduledProcedureStepSequence[0], 'ScheduledPerformingPhysicianName', ''),
'ScheduledProcedureStepLocation': getattr(identifier.ScheduledProcedureStepSequence[0], 'ScheduledProcedureStepLocation', ''),
'PreMedication': getattr(identifier.ScheduledProcedureStepSequence[0], 'PreMedication', ''),
'SpecialNeeds': getattr(identifier.ScheduledProcedureStepSequence[0], 'SpecialNeeds', ''),
}
studies.append(study_details)
else:
print('Connection timed out, was aborted or received invalid response')
assoc.release()
return {'status': 'Success', 'studies': studies}
else:
print('Failed to associate with the PACS server.')
return {'status': 'Failed', 'studies': []}
def post_MWL(study_data):
"""
Function to add a study to the Modality Worklist (MWL).
:param study_data: Dictionary containing study details
:return: JSON response with status and message
"""
pacs_ip = MWL_SERVER['IP']
pacs_port = MWL_SERVER['PORT']
remote_ae_title = MWL_SERVER['AE_TITLE']
ae = AE(ae_title=AE_CLIENT)
ae.add_requested_context(ModalityWorklistInformationFind)
ds = Dataset()
ds.add_new(0x00100010, 'PN', study_data['PatientName']) # OK
ds.add_new(0x00100020, 'LO', study_data['PatientID']) # OK
ds.add_new(0x00100030, 'DA', study_data['PatientBirthDate']) # OK
ds.add_new(0x00100040, 'CS', study_data['PatientSex']) # OK
ds.add_new(0x0020000D, 'UI', study_data['StudyID']) # OK
ds.add_new(0x00080050, 'SH', study_data['AccessionNumber']) # OK
# ds.add_new(0x00400006, 'PN', study_data['ReferringPhysicianName']) # NOT OK
# ds.add_new(0x00080060, 'CS', study_data['Modality']) # OK
sps = Dataset()
sps.add_new(0x00400002, 'DA', study_data['ScheduledProcedureStepStartDate']) # OK
sps.add_new(0x00401001, 'SH', study_data['RequestedProcedureID']) # OK
#sps.add_new(0x00321060, 'LO', study_data['RequestedProcedureDescription']) # OK
sps.add_new(0x00400001, 'AE', study_data['ScheduledStationAETitle']) # OK
sps.add_new(0x00400006, 'SH', study_data['ScheduledPerformingPhysicianName']) # OK
sps.add_new(0x00400011, 'SH', study_data['ScheduledProcedureStepLocation']) # OK
sps.add_new(0x00400020, 'CS', 'SCHEDULED') # OK
# sps.add_new(0x00400007, 'LO', study_data['StudyDescription']) # OK
# sps.add_new(0x00400012, 'LO', study_data['PreMedication']) # OK
# sps.add_new(0x00380050, 'LO', study_data['SpecialNeeds']) # OK
ds.add_new(0x00400002, 'SQ', [sps]) # OK
# You'll need to modify this line based on the specific SOP Class UID for the MWL item
ds.SOPClassUID = '1.2.840.10008.5.1.4.31' # OK
# SOP Instance UID should be a unique identifier for the worklist item
ds.SOPInstanceUID = '1.2.3.4.77'
# Create file meta information and set Transfer Syntax UID
file_meta = FileMetaDataset()
#file_meta.MediaStorageSOPClassUID = '1.2.840.10008.5.1.4.31' # MWL SOP Class UID
#file_meta.MediaStorageSOPInstanceUID = '1.2.3.4.5' # Unique SOP Instance UID
file_meta.ImplementationClassUID = '1.2.40.0.13.1.3' # Implementation Class UID OK
file_meta.TransferSyntaxUID = '1.2.840.10008.1.2' # Explicit VR Little Endian OK
ds.file_meta = file_meta
# Additional attributes can be added here as required
print('_______________')
print(ds)
print('_______________')
assoc = ae.associate(pacs_ip, pacs_port, ae_title=remote_ae_title)
if assoc.is_established:
# Use N-CREATE service to add the MWL item
# 1.2.840.10008.5.1.4.31
response = assoc.send_n_create(ds, '1.2.840.10008.5.1.4.31')
#print("N-CREATE response:", response[0])
assoc.release()
if response[0] == 0x0000:
return {'status': 'Success', 'message': 'Study added to MWL successfully.', 'study': response[1]}
else:
return {'status': 'Failed', 'message': f'Error adding study to MWL, status code: {response[0]}'}
else:
return {'status': 'Failed', 'message': 'Failed to associate with the PACS server.'}
def post_HL7(study_data):
msg = Message(version='2.5')
# MSH Segment
msh_segment = Segment('MSH', version='2.5')
msh_segment.MSH_3 = 'SendingApp'
msh_segment.MSH_4 = 'SendingFacility'
msh_segment.MSH_5 = 'ReceivingApp'
msh_segment.MSH_6 = 'ReceivingFacility'
msh_segment.MSH_7 = study_data['ScheduledProcedureStepStartDate']
msh_segment.MSH_9 = 'ORM^O01'
msh_segment.MSH_10 = '123456'
msh_segment.MSH_11 = 'P'
msh_segment.MSH_12 = '2.5'
msg.add_segment(msh_segment)
# PID Segment
pid_segment = Segment('PID', version='2.5')
pid_segment.PID_3 = study_data['PatientID']
pid_segment.PID_5 = study_data['PatientName']
pid_segment.PID_7 = study_data['PatientBirthDate']
pid_segment.PID_8 = study_data['PatientSex']
msg.add_segment(pid_segment)
# PV1 Segment
pv1_segment = Segment('PV1', version='2.5')
pv1_segment.PV1_7 = study_data['ScheduledPerformingPhysicianName']
pv1_segment.PV1_3 = study_data['ScheduledProcedureStepLocation']
msg.add_segment(pv1_segment)
# OBR Segment
obr_segment = Segment('OBR', version='2.5')
obr_segment.OBR_2 = study_data['AccessionNumber']
obr_segment.OBR_4 = study_data['RequestedProcedureDescription']
obr_segment.OBR_18 = study_data['ScheduledStationAETitle']
obr_segment.OBR_19 = study_data['RequestedProcedureID']
obr_segment.OBR_24 = study_data['Modality']
msg.add_segment(obr_segment)
# ORC Segment
orc_segment = Segment('ORC', version='2.5')
orc_segment.ORC_12 = study_data['ReferringPhysicianName']
msg.add_segment(orc_segment)
# Convert HL7 Message to String
hl7_str = msg.to_er7()
print('hl7str', hl7_str)
# Envoyer le message HL7 au serveur
try:
with socket.create_connection((HL7_SERVER['IP'], HL7_SERVER['PORT'])) as sock:
sock.sendall(hl7_str.encode('utf-8'))
response = sock.recv(1024).decode('utf-8')
# Vous pouvez parser la réponse en utilisant la bibliothèque hl7apy, si nécessaire
response_msg = parse_message(response)
# Vérifier le succès de l'opération
if response_msg.MSA.MSA_1.acknowledgment_code == 'AA':
return {'status': 'Success', 'message': 'Study added to MWL successfully.'}
else:
return {'status': 'Failed', 'message': 'Error adding study to MWL.'}
except Exception as e:
print(f'Failed to send HL7 message: {e}')
return {'status': 'Failed', 'message': 'Failed to send HL7 message.'}
if __name__ == '__main__':
response=post_MWL(study_data)
print('Post',response)
#response = get_MWL()
#print('MWL:',response)
The function get_MWL works and returns the MWL.
