Hospital Management

  1  from __future__ import annotations
  2  from abc import ABC, abstractmethod
  3  from dataclasses import dataclass, field
  4  from datetime import datetime, date, timedelta
  5  from enum import Enum
  6  from threading import RLock
  7  from sortedcontainers import SortedDict  # pip install sortedcontainers
  8  import uuid
  9  
 10  
 11  class Specialty(Enum):
 12      GENERAL = "general"
 13      CARDIOLOGY = "cardiology"
 14      PEDIATRICS = "pediatrics"
 15      ORTHOPEDICS = "orthopedics"
 16  
 17  
 18  class AppointmentStatus(Enum):
 19      REQUESTED = "requested"
 20      CONFIRMED = "confirmed"
 21      CHECKED_IN = "checked_in"
 22      COMPLETED = "completed"
 23      CANCELLED = "cancelled"
 24      NO_SHOW = "no_show"
 25  
 26  
 27  class EntryType(Enum):
 28      VISIT_NOTE = "visit_note"
 29      DIAGNOSIS = "diagnosis"
 30      PRESCRIPTION = "prescription"
 31      LAB_RESULT = "lab_result"
 32      CORRECTION = "correction"
 33  
 34  
 35  @dataclass(frozen=True)
 36  class Contact:
 37      phone: str
 38      email: str
 39  
 40  
 41  @dataclass(frozen=True)
 42  class InsurancePolicy:
 43      provider: str
 44      member_id: str
 45      group_number: str
 46  
 47  
 48  @dataclass
 49  class RecordEntry:
 50      id: str
 51      type: EntryType
 52      author_id: str
 53      timestamp: datetime
 54      text: str
 55      corrected_by: str | None = None       # points to a later entry if this was superseded
 56  
 57  
 58  class MedicalRecord:
 59      """Append-only. Corrections add new entries referencing the original."""
 60  
 61      def __init__(self, patient_id: str):
 62          self.patient_id = patient_id
 63          self._entries: list[RecordEntry] = []
 64          self._lock = RLock()
 65  
 66      def append(self, entry: RecordEntry) -> None:
 67          with self._lock:
 68              self._entries.append(entry)
 69  
 70      def correct(self, original_id: str, author_id: str, new_text: str) -> RecordEntry:
 71          with self._lock:
 72              original = next((e for e in self._entries if e.id == original_id), None)
 73              if original is None:
 74                  raise LookupError("original entry not found")
 75              correction = RecordEntry(
 76                  id=str(uuid.uuid4())[:8],
 77                  type=EntryType.CORRECTION,
 78                  author_id=author_id,
 79                  timestamp=datetime.utcnow(),
 80                  text=new_text,
 81              )
 82              self._entries.append(correction)
 83              # Point the original at its successor — history preserved both ways.
 84              original.corrected_by = correction.id
 85              return correction
 86  
 87      def entries(self, entry_type: EntryType | None = None) -> list[RecordEntry]:
 88          with self._lock:
 89              if entry_type is None:
 90                  return list(self._entries)
 91              return [e for e in self._entries if e.type == entry_type]
 92  
 93  
 94  @dataclass
 95  class Patient:
 96      id: str
 97      name: str
 98      dob: date
 99      contact: Contact
100      insurance: InsurancePolicy | None
101      record: MedicalRecord = field(init=False)
102  
103      def __post_init__(self):
104          self.record = MedicalRecord(patient_id=self.id)
105  
106  
107  @dataclass(frozen=True)
108  class TimeSlot:
109      start: datetime
110      end: datetime
111  
112      def overlaps(self, other: "TimeSlot") -> bool:
113          return self.start < other.end and other.start < self.end
114  
115  
116  class Doctor:
117      def __init__(self, doctor_id: str, name: str, specialty: Specialty):
118          self.id = doctor_id
119          self.name = name
120          self.specialty = specialty
121          self._bookings: SortedDict[datetime, TimeSlot] = SortedDict()
122          self._lock = RLock()
123  
124      def is_free(self, slot: TimeSlot) -> bool:
125          with self._lock:
126              idx = self._bookings.bisect_right(slot.start) - 1
127              if idx >= 0:
128                  prior = self._bookings.peekitem(idx)[1]
129                  if prior.end > slot.start:
130                      return False
131              idx_next = self._bookings.bisect_left(slot.start)
132              if idx_next < len(self._bookings):
133                  nxt = self._bookings.peekitem(idx_next)[1]
134                  if nxt.start < slot.end:
135                      return False
136              return True
137  
138      def reserve(self, slot: TimeSlot) -> None:
139          with self._lock:
140              if not self.is_free(slot):
141                  raise RuntimeError("slot unavailable")
142              self._bookings[slot.start] = slot
143  
144      def release(self, slot: TimeSlot) -> None:
145          with self._lock:
146              self._bookings.pop(slot.start, None)
147  
148  
149  @dataclass
150  class Prescription:
151      id: str
152      patient_id: str
153      prescriber_id: str
154      medication: str
155      dosage: str
156      duration: str
157      issued_at: datetime
158  
159  
160  @dataclass
161  class Encounter:
162      id: str
163      appointment_id: str
164      diagnoses: list[str] = field(default_factory=list)
165      prescriptions: list[Prescription] = field(default_factory=list)
166      notes: str = ""
167  
168  
169  @dataclass
170  class Appointment:
171      id: str
172      doctor_id: str
173      patient_id: str
174      slot: TimeSlot
175      status: AppointmentStatus = AppointmentStatus.REQUESTED
176      encounter_id: str | None = None
177      audit: list[tuple[datetime, str, AppointmentStatus]] = field(default_factory=list)
178  
179      def transition(self, by: str, new_status: AppointmentStatus) -> None:
180          self.audit.append((datetime.utcnow(), by, new_status))
181          self.status = new_status
182  
183  
184  class HospitalService:
185      def __init__(self):
186          self._patients: dict[str, Patient] = {}
187          self._doctors: dict[str, Doctor] = {}
188          self._appointments: dict[str, Appointment] = {}
189          self._encounters: dict[str, Encounter] = {}
190  
191      def register_patient(self, name: str, dob: date, contact: Contact,
192                            insurance: InsurancePolicy | None = None) -> Patient:
193          p = Patient(
194              id=str(uuid.uuid4())[:8], name=name, dob=dob,
195              contact=contact, insurance=insurance,
196          )
197          self._patients[p.id] = p
198          return p
199  
200      def onboard_doctor(self, name: str, specialty: Specialty) -> Doctor:
201          d = Doctor(doctor_id=str(uuid.uuid4())[:8], name=name, specialty=specialty)
202          self._doctors[d.id] = d
203          return d
204  
205      def book_appointment(self, doctor: Doctor, patient: Patient,
206                           slot: TimeSlot, by: str) -> Appointment:
207          doctor.reserve(slot)       # fails if unavailable
208          appt = Appointment(
209              id=str(uuid.uuid4())[:8],
210              doctor_id=doctor.id, patient_id=patient.id, slot=slot,
211          )
212          appt.transition(by, AppointmentStatus.CONFIRMED)
213          self._appointments[appt.id] = appt
214          return appt
215  
216      def check_in(self, appointment_id: str, by: str) -> None:
217          appt = self._appointments[appointment_id]
218          if appt.status != AppointmentStatus.CONFIRMED:
219              raise RuntimeError(f"cannot check in a {appt.status.value} appointment")
220          appt.transition(by, AppointmentStatus.CHECKED_IN)
221  
222      def complete_encounter(self, appointment_id: str, doctor_id: str,
223                             diagnoses: list[str], notes: str = "") -> Encounter:
224          appt = self._appointments[appointment_id]
225          if appt.status != AppointmentStatus.CHECKED_IN:
226              raise RuntimeError("can only close a checked-in appointment")
227          encounter = Encounter(
228              id=str(uuid.uuid4())[:8],
229              appointment_id=appointment_id,
230              diagnoses=diagnoses, notes=notes,
231          )
232          self._encounters[encounter.id] = encounter
233          appt.encounter_id = encounter.id
234          appt.transition(doctor_id, AppointmentStatus.COMPLETED)
235  
236          patient = self._patients[appt.patient_id]
237          for diagnosis in diagnoses:
238              patient.record.append(RecordEntry(
239                  id=str(uuid.uuid4())[:8],
240                  type=EntryType.DIAGNOSIS,
241                  author_id=doctor_id,
242                  timestamp=datetime.utcnow(),
243                  text=diagnosis,
244              ))
245          if notes:
246              patient.record.append(RecordEntry(
247                  id=str(uuid.uuid4())[:8],
248                  type=EntryType.VISIT_NOTE,
249                  author_id=doctor_id,
250                  timestamp=datetime.utcnow(),
251                  text=notes,
252              ))
253          return encounter
254  
255      def issue_prescription(self, encounter_id: str, doctor_id: str,
256                             patient_id: str, medication: str,
257                             dosage: str, duration: str) -> Prescription:
258          encounter = self._encounters[encounter_id]
259          rx = Prescription(
260              id=str(uuid.uuid4())[:8],
261              patient_id=patient_id,
262              prescriber_id=doctor_id,
263              medication=medication,
264              dosage=dosage,
265              duration=duration,
266              issued_at=datetime.utcnow(),
267          )
268          encounter.prescriptions.append(rx)
269          patient = self._patients[patient_id]
270          patient.record.append(RecordEntry(
271              id=rx.id,
272              type=EntryType.PRESCRIPTION,
273              author_id=doctor_id,
274              timestamp=rx.issued_at,
275              text=f"{medication} {dosage} for {duration}",
276          ))
277          return rx
278  
279      def cancel_appointment(self, appointment_id: str, by: str) -> None:
280          appt = self._appointments[appointment_id]
281          if appt.status in (AppointmentStatus.COMPLETED, AppointmentStatus.CANCELLED):
282              return
283          doctor = self._doctors[appt.doctor_id]
284          doctor.release(appt.slot)
285          appt.transition(by, AppointmentStatus.CANCELLED)
286  
287      def mark_no_shows(self, now: datetime | None = None,
288                        grace: timedelta = timedelta(minutes=15)) -> int:
289          """Any CONFIRMED appointment whose slot started more than `grace` ago without
290          a check-in transitions to NO_SHOW. Run this on a short cadence (e.g. every minute).
291          Returns the count flagged."""
292          check_at = now or datetime.utcnow()
293          count = 0
294          for appt in self._appointments.values():
295              if appt.status != AppointmentStatus.CONFIRMED:
296                  continue
297              if check_at >= appt.slot.start + grace:
298                  appt.transition("system", AppointmentStatus.NO_SHOW)
299                  count += 1
300          return count
301  
302  
303  if __name__ == "__main__":
304      svc = HospitalService()
305      alice = svc.register_patient("Alice Smith", date(1990, 5, 12),
306                                   Contact("555-0100", "alice@example.com"))
307      dr_kumar = svc.onboard_doctor("Dr. Kumar", Specialty.GENERAL)
308  
309      slot = TimeSlot(datetime(2026, 4, 15, 9, 0), datetime(2026, 4, 15, 9, 30))
310      appt = svc.book_appointment(dr_kumar, alice, slot, by="reception")
311      print(f"Booked {appt.id}, status {appt.status.value}")
312  
313      # Conflicting slot rejected.
314      try:
315          svc.book_appointment(dr_kumar, alice,
316                               TimeSlot(datetime(2026, 4, 15, 9, 15), datetime(2026, 4, 15, 9, 45)),
317                               by="reception")
318      except RuntimeError as e:
319          print(f"Overlapping booking rejected: {e}")
320  
321      svc.check_in(appt.id, by="nurse")
322      encounter = svc.complete_encounter(appt.id, dr_kumar.id,
323                                          diagnoses=["viral upper respiratory infection"],
324                                          notes="Rest, fluids, ibuprofen as needed")
325      rx = svc.issue_prescription(encounter.id, dr_kumar.id, alice.id,
326                                   medication="Ibuprofen 400mg", dosage="q8h", duration="5 days")
327      print(f"Encounter {encounter.id} with Rx {rx.id}")
328  
329      print(f"\nAlice's record ({len(alice.record.entries())} entries):")
330      for entry in alice.record.entries():
331          print(f"  [{entry.type.value}] {entry.text}")
332  
333      # A correction is appended, original preserved.
334      original_id = alice.record.entries(EntryType.DIAGNOSIS)[0].id
335      alice.record.correct(original_id, dr_kumar.id, "corrected: mild pharyngitis, viral")
336      corrected = next(e for e in alice.record.entries() if e.id == original_id)
337      print(f"\nOriginal diagnosis now links to correction {corrected.corrected_by}")
338  
339      # No-show sweep: a second patient never shows up, grace period passes.
340      bob = svc.register_patient("Bob Jones", date(1985, 2, 3),
341                                 Contact("555-0200", "bob@example.com"))
342      no_show_slot = TimeSlot(datetime(2026, 4, 15, 10, 0), datetime(2026, 4, 15, 10, 30))
343      skipped = svc.book_appointment(dr_kumar, bob, no_show_slot, by="reception")
344      # Simulate "now" = 20 minutes into the slot. Grace is 15 min → should flip to NO_SHOW.
345      simulated_now = no_show_slot.start + timedelta(minutes=20)
346      count = svc.mark_no_shows(now=simulated_now)
347      print(f"mark_no_shows flagged {count} appointment(s); bob's status = {skipped.status.value}")
348      assert skipped.status == AppointmentStatus.NO_SHOW