15.2.123. camcops_server.cc_modules.cc_hl7


Copyright (C) 2012, University of Cambridge, Department of Psychiatry. Created by Rudolf Cardinal (rnc1001@cam.ac.uk).

This file is part of CamCOPS.

CamCOPS is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version.

CamCOPS is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.

You should have received a copy of the GNU General Public License along with CamCOPS. If not, see <https://www.gnu.org/licenses/>.

Core HL7 functions, e.g. to build HL7 v2 messages.

General HL7 sources:

To consider

Basic HL7 message structure:

We can then add an option for structure (XML), HTML, PDF export.

class camcops_server.cc_modules.cc_hl7.MLLPTimeoutClient(host: str, port: int, timeout_ms: Optional[int] = None)[source]

Class for MLLP TCP/IP transmission that implements timeouts.

__init__(host: str, port: int, timeout_ms: Optional[int] = None) None[source]

Creates MLLP client and opens socket.


Release the socket connection.

send(data: bytes) Tuple[bool, Optional[str]][source]

Low-level, direct access to the socket.send function (data must be already wrapped in an MLLP container). Blocks until the server returns.

Returns success, ack_msg.

send_message(message: Union[str, hl7.containers.Message]) Tuple[bool, Optional[str]][source]

Wraps a string or hl7.Message in a MLLP container and sends the message to the server.

Returns success, ack_msg.

camcops_server.cc_modules.cc_hl7.escape_hl7_text(s: str) str[source]

Escapes HL7 special characters.

camcops_server.cc_modules.cc_hl7.get_mod11_checkdigit(strnum: str) str[source]

Input: string containing integer. Output: MOD11 check digit (string).


camcops_server.cc_modules.cc_hl7.make_dg1_segment(set_id: int, diagnosis_datetime: pendulum.datetime.DateTime, coding_system: str, diagnosis_identifier: str, diagnosis_text: str, alternate_coding_system: str = '', alternate_diagnosis_identifier: str = '', alternate_diagnosis_text: str = '', diagnosis_type: str = 'F', diagnosis_classification: str = 'D', confidential_indicator: str = 'N', clinician_id_number: Optional[Union[str, int]] = None, clinician_surname: str = '', clinician_forename: str = '', clinician_middle_name_or_initial: str = '', clinician_suffix: str = '', clinician_prefix: str = '', clinician_degree: str = '', clinician_source_table: str = '', clinician_assigning_authority: str = '', clinician_name_type_code: str = '', clinician_identifier_type_code: str = '', clinician_assigning_facility: str = '', attestation_datetime: Optional[pendulum.datetime.DateTime] = None) hl7.containers.Segment[source]

Creates an HL7 diagnosis (DG1) segment.


set_id: Diagnosis sequence number, starting with 1 (use higher numbers
    for >1 diagnosis).
diagnosis_datetime: Date/time diagnosis was made.

coding_system: E.g. "I9C" for ICD9-CM; "I10" for ICD10.
diagnosis_identifier: Code.
diagnosis_text: Text.

alternate_coding_system: Optional alternate coding system.
alternate_diagnosis_identifier: Optional alternate code.
alternate_diagnosis_text: Optional alternate text.

diagnosis_type: A admitting, W working, F final.
diagnosis_classification: C consultation, D diagnosis, M medication,
    O other, R radiological scheduling, S sign and symptom,
    T tissue diagnosis, I invasive procedure not classified elsewhere.
confidential_indicator: Y yes, N no

clinician_id_number:              } Diagnosing clinician.
clinician_surname:                }
clinician_forename:               }
clinician_middle_name_or_initial: }
clinician_suffix:                 }
clinician_prefix:                 }
clinician_degree:                 }
clinician_source_table:           }
clinician_assigning_authority:    }
clinician_name_type_code:         }
clinician_identifier_type_code:   }
clinician_assigning_facility:     }

attestation_datetime: Date/time the diagnosis was attested.
camcops_server.cc_modules.cc_hl7.make_msh_segment(message_datetime: pendulum.datetime.DateTime, message_control_id: str) hl7.containers.Segment[source]

Creates an HL7 message header (MSH) segment.

camcops_server.cc_modules.cc_hl7.make_obr_segment(task: Task) hl7.containers.Segment[source]

Creates an HL7 observation request (OBR) segment.

camcops_server.cc_modules.cc_hl7.make_obx_segment(req: CamcopsRequest, task: Task, task_format: str, observation_identifier: str, observation_datetime: pendulum.datetime.DateTime, responsible_observer: str, export_options: TaskExportOptions) hl7.containers.Segment[source]

Creates an HL7 observation result (OBX) segment.

camcops_server.cc_modules.cc_hl7.make_pid_segment(forename: str, surname: str, dob: pendulum.date.Date, sex: str, address: str, patient_id_list: Optional[List[camcops_server.cc_modules.cc_simpleobjects.HL7PatientIdentifier]] = None) hl7.containers.Segment[source]

Creates an HL7 patient identification (PID) segment.

camcops_server.cc_modules.cc_hl7.msg_is_successful_ack(msg: hl7.containers.Message) Tuple[bool, Optional[str]][source]

Checks whether msg represents a successful acknowledgement message.