Chapter 5: HEMS Resources and Servicing

How helicopters and cars are modelled, built from the rota files, scheduled for servicing and kept in a SimPy FilterStore.

Written for: Intermediate Coder Confident Coder Decision-makers Domain experts General public

In Calls and Patient Journeys we saw DES_HEMS turn each daily call into a Patient and ask for a resource, and we saw it trigger a “daily servicing check” at the start of every simulated day. This chapter looks at the other side of that request: what a resource is, where the list of resources comes from, and how the model keeps unavailable ones out of reach.

The use case: a helicopter that is due a service

Devon Air Ambulance operates two helicopters (registrations g-daas and g-daan) and three rapid-response cars. A helicopter cannot fly while it is being serviced, and a service takes about three weeks. If the simulation ignored this, a scenario with fewer aircraft would look better than it deserves.

So the model has to answer three questions before a single patient is created:

1. Which resources exist, and when are they on shift?
2. When will each one be in for servicing, over the whole simulated period?
3. On any given day, which resources can be handed to a patient, and which are parked in the workshop?

We will follow one helicopter, g-daas, through all three. The answers live in three small classes (Ambulance, HEMS and HEMSAvailability) and several CSV files in actual_data/, the folder that the app fills in (Streamlit App and Scenario Parameters).

HEMS resource: one helicopter or car, as an instance of HEMS. A single physical vehicle can appear as more than one HEMS resource (see below).

Key concepts

Ambulance and HEMS

Ambulance is the generic “something that responds to an incident”. It holds a type, a callsign and a handful of timestamp placeholders for the stages of a job.

air_ambulance_des/class_ambulance.py:16-28

    def __init__(self, ambulance_type="ambulance", callsign="AMBULANCE"):
        self.mobile = ""
        self.as_scene = ""
        self.leaving_scene = ""
        self.at_hospital = ""
        self.clear = ""
        self.stood_down = ""

        self.ambulance_type = ambulance_type
        self.callsign = callsign

    def what_am_i(self):
        print(f"I am {self.ambulance_type}")

It is deliberately thin. All the interesting behaviour lives in its subclass HEMS, which adds the attributes the allocation logic needs:

air_ambulance_des/class_hems.py:15-69

    def __init__(
        self,
        callsign: str,
        callsign_group: str,
        vehicle_type: str,
        category: str,
        registration: str,
        summer_start: str,
        winter_start: str,
        summer_end: str,
        winter_end: str,
        servicing_schedule: pd.DataFrame,
        resource_id: str = None,
        summer_season: list[int] = None,
    ):
        # Inherit all parent class functions
        super().__init__(ambulance_type="HEMS")

        self.utilityClass = Utils()

        self.callsign = callsign
        self.callsign_group = callsign_group
        self.available = 1
        self.being_serviced = False
        self.flying_time = 0
        self.vehicle_type = vehicle_type
        self.category = category
        self.registration = registration
        self.summer_start = summer_start
        self.winter_start = winter_start
        self.summer_end = summer_end
        self.winter_end = winter_end

        # Read in the summer months from a file
        summer_months = pd.read_csv("actual_data/rota_start_end_months.csv")
        summer_start_month = int(
            summer_months[summer_months["what"] == "summer_start_month"][
                "month"
            ].values[0]
        )
        summer_end_month = int(
            summer_months[summer_months["what"] == "summer_end_month"]["month"].values[
                0
            ]
        )

        self.summer_season = [
            x for x in range(summer_start_month, summer_end_month + 1)
        ]

        # Pre-determine the servicing schedule when the resource is created
        self.servicing_schedule = servicing_schedule

        self.in_use = False
        self.resource_id = resource_id

The attributes worth knowing:

callsign and callsign group

The callsign (for example H70 or CC72) is what the crew is called on the radio. The callsign group (an integer such as 70) ties together the vehicles that share one crew: the helicopter H70 and the car CC70 are staffed by the same people, so they must never be out at the same time. The allocation code in Resource Allocation relies on this.

registration

The identifier of the physical vehicle (g-daas, cc70). Servicing is a property of the registration, not the callsign.

vehicle type and category

helicopter or car, and a care category (CC critical care or EC enhanced care) describing the clinical capability the resource is being used for. Care categories were introduced in Calls and Patient Journeys.

summer and winter shift hours

Four hour-of-day values giving when the resource starts and stops work in each season. The months that count as summer are read from actual_data/rota_start_end_months.csv (the dimmed method in the excerpt above).

servicing schedule

A small DataFrame of service_start_date and service_end_date rows, calculated up front and handed to the resource when it is built.

in_use and being_serviced

Two flags that the allocation code reads. in_use is set while a patient has the resource; being_serviced is updated whenever the servicing schedule is checked.

Shifts: hems_resource_on_shift

The shift check is a pure function of the hour and the month.

air_ambulance_des/class_hems.py:107-120

    def hems_resource_on_shift(self, hour: int, month: int) -> bool:
        """
        Function to determine whether the HEMS resource is within
        its operational hours
        """

        # Assuming summer hours are quarters 2 and 3 i.e. April-September
        # Can be modified if required.
        # SR NOTE: If changing these, please also modify in
        # write_run_params() function in des_parallel_process
        start = self.summer_start if month in self.summer_season else self.winter_start
        end = self.summer_end if month in self.summer_season else self.winter_end

        return self.utilityClass.is_time_in_range(int(hour), int(start), int(end))

hems_resource_on_shift picks the summer or winter hours depending on the month (highlighted), then asks Utils.is_time_in_range whether the hour falls inside them. That helper copes with shifts that cross midnight: in utils.py it returns start <= current < end when start <= end, and current >= start or current < end otherwise. This matters because the H70 enhanced-care row in the rota runs from 19 to 2.

The rota: one row per role, not per vehicle

actual_data/HEMS_ROTA.csv has one row per callsign and care category:

Example: actual_data/HEMS_ROTA.csv (first rows)

callsign,category,vehicle_type,callsign_group,summer_start,winter_start,summer_end,winter_end
H70,CC,helicopter,70,7,7,19,19
H70,EC,helicopter,70,19,19,2,2
CC70,CC,car,70,7,7,19,19
CC70,EC,car,70,19,19,2,2

The shipped file also has rows for H71 (an enhanced care helicopter) and CC72 (an enhanced care car). The same aircraft therefore appears twice: as a critical-care helicopter by day and as an enhanced-care helicopter overnight. Each row becomes its own HEMS object, and both objects share the registration g-daas. That is why you will see the code identify a resource by category and registration together, and why the active registrations tracked during allocation matter: they stop the same airframe being sent in both roles at once.

Building the resources

HEMSAvailability is the class that owns all of this. It is created once per run by DES_HEMS and is where the two SimPy stores live.

NoteRefresher: SimPy FilterStore

A SimPy Store is a queue of Python objects: put(item) adds one and get() removes one, waiting (in simulated time) if none is available. A FilterStore adds a get(filter) variant that takes the first item for which the filter function returns true. Both put and get return events, so inside a process you write item = yield store.get(...).

air_ambulance_des/class_hems_availability.py:99-123

        self.HEMS_resources_list = []

        self.active_callsign_groups = (
            set()
        )  # Prevents same crew being used twice...hopefully...
        self.active_registrations = set()  # Prevents same vehicle being used twice
        self.active_callsigns = set()

        # Create a store for HEMS resources
        self.store = FilterStore(env)

        self.serviceStore = FilterStore(env)

        # Prepare HEMS resources for ingesting into store
        self.prep_HEMS_resources()

        # Populate the store with HEMS resources
        self.populate_store()

        # Daily servicing check (in case sim starts during a service)
        [dow, hod, weekday, month, qtr, current_dt] = (
            self.utilityClass.date_time_of_call(self.sim_start_date, self.env.now)
        )

        self.daily_servicing_check(current_dt, hod, month)

Two stores are created. store holds resources that are in the base and may be dispatched. serviceStore is the service store: a holding area for anything currently in the workshop. Moving a resource between them is how the model takes it out of circulation, with no extra “is it serviced?” test needed on the hot path of allocation (though being_serviced is still checked, as a second line of defence).

The constructor then calls prep_HEMS_resources() to build the objects and populate_store() to put them in store. Finally it runs a daily servicing check, in case the simulation starts mid-service (see the warning below).

Reading the rota and servicing files

prep_HEMS_resources starts by joining three small tables:

air_ambulance_des/class_hems_availability.py:229-241

        SERVICING_SCHEDULE = pd.read_csv("actual_data/service_schedules_by_model.csv")
        SERVICE_HISTORY = pd.read_csv("actual_data/service_history.csv", na_values=0)
        CALLSIGN_REGISTRATION = pd.read_csv(
            "actual_data/callsign_registration_lookup.csv"
        )

        SERVICING_SCHEDULE = SERVICING_SCHEDULE.merge(
            CALLSIGN_REGISTRATION, how="right", on="model"
        )

        SERVICING_SCHEDULE = SERVICING_SCHEDULE.merge(
            SERVICE_HISTORY, how="left", on="registration"
        )

• service_schedules_by_model.csv says how often each aircraft model is serviced (service_schedule_months) and for how many weeks (service_duration_weeks).
• callsign_registration_lookup.csv maps each registration to a callsign and a model.
• service_history.csv gives the date each registration was last serviced. Cars have a 0 there, which na_values=0 turns into a missing value, so they never get a service schedule.

The result is one row per registration with everything needed to project its future services.

Projecting service dates

For each registration with a known last service, the method walks forward in time until it passes the simulation end date, asking find_next_service_date for each next slot.

air_ambulance_des/class_hems_availability.py:245-288

        for index, row in SERVICING_SCHEDULE.iterrows():
            # self.debug(row)
            current_resource_service_dates = []
            # Check if service date provided
            if not pd.isna(row["last_service"]):
                # self.debug(f"Checking {row['registration']} with previous service date of {row['last_service']}")
                last_service = datetime.strptime(row["last_service"], "%Y-%m-%d")
                service_date = last_service

                while last_service < self.sim_end_date:
                    end_date = (
                        last_service
                        + timedelta(weeks=int(row["service_duration_weeks"]))
                        + timedelta(weeks=self.serviing_buffer_weeks)
                    )

                    service_date, end_date = self.find_next_service_date(
                        last_service,
                        row["service_schedule_months"],
                        service_dates,
                        row["service_duration_weeks"],
                    )

                    schedule.append((row["registration"], service_date, end_date))
                    # self.debug(service_date)
                    service_dates.append(
                        {
                            "service_start_date": service_date,
                            "service_end_date": end_date,
                        }
                    )

                    current_resource_service_dates.append(
                        {
                            "service_start_date": service_date,
                            "service_end_date": end_date,
                        }
                    )
                    # self.debug(service_dates)
                    # self.debug(current_resource_service_dates)
                    last_service = service_date
            else:
                schedule.append((row["registration"], None, None))
                # self.debug(schedule)

The highlighted call is the heart of it. Each result is appended three ways: to schedule (every registration, which becomes the DataFrame written to data/service_dates.csv), to service_dates (a list shared across all resources, so later aircraft can avoid earlier ones), and to current_resource_service_dates. The end_date computed just before the call (which uses the servicing buffer) is overwritten by the call’s return value.

The loop’s bound is self.sim_end_date, which the constructor sets to the simulation end plus an extra 365 days. The comment there explains why: a service date can be walked backwards into the previous year, so the schedule needs some slack.

Choosing a date: find_next_service_date

find_next_service_date starts from “last service plus the interval”, then prefers a configured month if that is possible, then nudges earlier until the slot is legal.

air_ambulance_des/class_hems_availability.py:749-779

    def find_next_service_date(
        self,
        last_service_date: datetime,
        interval_months: int,
        service_dates: list,
        service_duration: int,
    ) -> list[datetime]:
        """
        Function to determine the next service date for a resource. The date is determine by
        the servicing schedule for the resource, the preferred month of servicing, and to
        avoid dates that fall in either school holidays or when other resources are being serviced.
        """

        next_due_date = last_service_date + relativedelta(
            months=interval_months
        )  # Approximate month length
        end_date = next_due_date + timedelta(weeks=service_duration)

        preferred_date = datetime(next_due_date.year, self.servicing_preferred_month, 2)
        preferred_end_date = preferred_date + timedelta(weeks=service_duration)

        if next_due_date.month > preferred_date.month:
            preferred_date += relativedelta(years=1)

        # self.debug(f"Next due: {next_due_date} with end date {end_date} and preferred_date is {preferred_date} with pref end {preferred_end_date}")

        # If preferred date is valid, use it
        if preferred_date <= next_due_date and not self.is_during_school_holidays(
            preferred_date, preferred_end_date
        ):
            next_due_date = preferred_date

The preferred date is the 2nd of servicing_preferred_month (January by default) in the year the service falls due. If the service is due after that month, the preferred date rolls to the next year. The preferred date is only used if it is on or before the due date and does not clash with school holidays; it is never later than the due date, so an aircraft is never serviced late.

air_ambulance_des/class_hems_availability.py:781-793

        while True:
            if self.is_during_school_holidays(
                next_due_date, end_date
            ) or self.is_other_resource_being_serviced(
                next_due_date, end_date, service_dates
            ):
                next_due_date -= timedelta(days=1)
                end_date = next_due_date + timedelta(weeks=service_duration)
                continue
            else:
                break

        return [next_due_date, end_date]

The loop is simple and a little brutal: while the proposed window overlaps a school holiday (is_during_school_holidays) or another resource’s service (is_other_resource_being_serviced), move the start one day earlier and try again. Both helpers rely on do_ranges_overlap, which is the standard interval test max(start1, start2) <= min(end1, end2).

Tip

School holidays are loaded in the constructor with load_school_holidays("actual_data/school_holidays.csv") from school_holidays.py. The function accepts UK dates (dd/mm/yyyy, as in the shipped CSV) or ISO dates (as written by the app when you import a template), trying UK format first, and raises a ValueError if a holiday ends before it starts. Servicing is steered away from these periods by the search loop above.

Turning rows into HEMS objects

With the service dates in a DataFrame, the rota is read and merged with the servicing information, and each row becomes a resource:

air_ambulance_des/class_hems_availability.py:299-325

        HEMS_RESOURCES = (
            pd.read_csv("actual_data/HEMS_ROTA.csv")
            # Add model and servicing rules
            .merge(SERVICING_SCHEDULE, on=["callsign", "vehicle_type"], how="left")
        )

        for index, row in HEMS_RESOURCES.iterrows():
            s = service_df[service_df["registration"] == row["registration"]]
            # self.debug(s)

            # Create new HEMS resource and add to HEMS_resource_list
            # pd.DataFrame(columns=['year', 'service_start_date', 'service_end_date'])
            hems = HEMS(
                callsign=row["callsign"],
                callsign_group=row["callsign_group"],
                vehicle_type=row["vehicle_type"],
                category=row["category"],
                registration=row["registration"],
                summer_start=row["summer_start"],
                winter_start=row["winter_start"],
                summer_end=row["summer_end"],
                winter_end=row["winter_end"],
                servicing_schedule=s,
                resource_id=row["registration"],
            )

            self.HEMS_resources_list.append(hems)

The merge on callsign and vehicle_type is what attaches a registration and a model to each rota row. Each resource gets only its own registration’s slice of the servicing table (s). Finally populate_store simply calls self.store.put(h) for every object in HEMS_resources_list.

Figure 1 summarises the flow from CSV files to stores.

flowchart LR
    A[HEMS_ROTA.csv] --> P[prep_HEMS_resources]
    B[callsign_registration_lookup.csv] --> P
    C[service_schedules_by_model.csv] --> P
    D[service_history.csv] --> P
    E[school_holidays.csv] --> F[find_next_service_date]
    P --> F
    F --> P
    P --> L[HEMS objects with servicing schedules]
    L --> S[(store)]
    S <-->|daily_servicing_check| W[(serviceStore)]

Figure 1: From input files to the two stores

The daily servicing check

Schedules are fixed up front, but the stores only change when the check runs. DES_HEMS.generate_calls runs it once at the start of every simulated day:

air_ambulance_des/des_hems.py:245-248

                # Also run scripts to check HEMS resources to see whether they are starting/finishing service
                yield self.env.process(
                    self.hems_resources.daily_servicing_check(current_dt, hod, month)
                )

Because daily_servicing_check is a generator containing yield statements, it must be wrapped in env.process and yielded, which blocks the call generator until the shuffling is done. Figure 2 shows what happens inside.

sequenceDiagram
    participant D as DES_HEMS
    participant A as HEMSAvailability
    participant H as HEMS (g-daas)
    participant S as serviceStore
    participant M as store
    D->>A: daily_servicing_check(current_dt)
    A->>H: unavailable_due_to_service(current_dt)
    H-->>A: True or False (GDAAS_service)
    loop each resource in serviceStore
        A->>H: service_check(current_dt, GDAAS_service)
        H-->>A: still in service?
        A->>S: get finished resource
        A->>M: put it back
    end
    loop each resource in store
        A->>H: service_check(current_dt, GDAAS_service)
        H-->>A: due for service?
        A->>M: get resource
        A->>S: put it in the workshop
    end

Figure 2: One daily servicing check, simplified

The first half of the method decides who is coming back:

air_ambulance_des/class_hems_availability.py:139-173

        GDAAS_service = False

        all_resources = self.serviceStore.items + self.store.items
        for h in all_resources:
            if h.registration == "g-daas":
                GDAAS_service = h.unavailable_due_to_service(current_dt)
                break

        self.debug(f"GDAAS_service is {GDAAS_service}")

        # --- Return from serviceStore to store ---
        to_return = [
            (s.category, s.registration)
            for s in self.serviceStore.items
            if not s.service_check(current_dt, GDAAS_service)  # Note the NOT here!
        ]

        # Attempted fix for gap after return from H70 duties
        for h in self.store.items:
            if h.registration == "g-daan" and not GDAAS_service:
                h.callsign_group = 71
                h.callsign = "H71"

        if to_return:
            self.debug("Service store has items to return")

        for category, registration in to_return:
            s = yield self.serviceStore.get(
                lambda item: item.category == category
                and item.registration == registration
            )
            yield self.store.put(s)
            self.debug(
                f"Returned [{s.category} / {s.registration}] from service to store"
            )

It first looks for the resource registered as g-daas and asks whether it is in service today, storing the answer as GDAAS_service. Then it builds a list of (category, registration) pairs from serviceStore whose service_check is now false (note the not). It collects the pairs first and moves them afterwards, because changing a store while iterating over its items would be unsafe. Each move is a get with a filter on category and registration, followed by a put on the main store. Filtering on both is necessary because of the two-objects-per-airframe point made earlier.

The second half is the mirror image:

air_ambulance_des/class_hems_availability.py:175-198

        # --- Send from store to serviceStore ---
        to_service = [
            (h.category, h.registration)
            for h in self.store.items
            if h.service_check(current_dt, GDAAS_service)
        ]

        for category, registration in to_service:
            self.debug("****************")
            self.debug(
                f"HEMS [{category} / {registration}] being serviced, removing from store"
            )

            h = yield self.store.get(
                lambda item: item.category == category
                and item.registration == registration
            )

            self.debug(
                f"HEMS [{h.category} / {h.registration}] successfully removed from store"
            )
            yield self.serviceStore.put(h)
            self.debug(f"HEMS [{h.category} / {h.registration}] moved to service store")
            self.debug("***********")

Anything in store for which service_check returns true is taken out and placed in serviceStore. Further down, the method only logs status through current_store_status, which is a debugging helper.

The shared aircraft: H70 and H71

Devon has two helicopters, but the rota only knows two callsigns, H70 and H71. When the usual H70 aircraft (g-daas) is in the workshop, the other aircraft must fly as H70 so that the callsign-group-70 crew can still be dispatched. service_check (the second method in the HEMS excerpt above) implements that by rewriting the callsign and group on the g-daan resource:

Aircraft	g-daas in service	g-daas not in service
g-daas	in the workshop	H70
g-daan	H70 (callsign group 70)	H71

The code is at class_hems.py:71-89:

air_ambulance_des/class_hems.py:71-89

    def service_check(self, current_dt: pd.Timestamp, GDAAS_service: bool) -> bool:
        if self.registration == "g-daan":
            if GDAAS_service:
                self.callsign_group = 70
                self.callsign = "H70"

            else:
                self.callsign_group = 71
                self.callsign = "H71"
                self.in_use = (
                    False  # We might need to re-think this in 24/7 scenarios although
                )
                # presumably the callsign will not change during an incident, only after.

            # GDASS being serviced

            # print(f"reg {self.registration} now has callsign {self.callsign}")

        return self.unavailable_due_to_service(current_dt)

When g-daas is in for servicing (GDAAS_service true), g-daan takes over as H70; when g-daas returns, it reverts to H71. Either way, the method ends by returning unavailable_due_to_service, which scans the resource’s own service windows:

air_ambulance_des/class_hems.py:91-105

    def unavailable_due_to_service(self, current_dt: pd.Timestamp) -> bool:
        """
        Returns logical value denoting whether the HEMS resource is currently
        unavailable due to being serviced

        """

        for index, row in self.servicing_schedule.iterrows():
            # print(row)
            if row["service_start_date"] <= current_dt <= row["service_end_date"]:
                self.being_serviced = True
                return True

        self.being_serviced = False
        return False

That is the only place being_serviced is set: it is true if current_dt lies inside any [service_start_date, service_end_date] window. Because it is a linear scan of a short table once a day per resource, its cost is negligible.

current_dt itself comes from Utils.date_time_of_call, which converts simulation minutes into a calendar date; see Sampling Distributions and Random Streams.

WarningA call that does nothing in the constructor

HEMSAvailability.__init__ ends with self.daily_servicing_check(current_dt, hod, month), with the comment “in case sim starts during a service”. But daily_servicing_check is a generator function, and calling a generator function without iterating it (or wrapping it in env.process) runs none of its body. As written, that constructor call appears to be a no-op. In practice the first real check happens when generate_calls starts day one, at simulation time zero, so the effect is limited to anything that happens before that point. It is worth knowing if you ever construct HEMSAvailability outside DES_HEMS and expect the stores to be correct immediately.

Tracing g-daas through a run

Putting it together for our helicopter:

1. Setup. prep_HEMS_resources reads its last service date from service_history.csv (4 January 2022) and its model’s interval (8 months for the Airbus H145, 3 weeks duration, as the shipped CSVs stand). It projects service windows until a year past the simulation end, avoiding school holidays and the other aircraft’s windows. Two HEMS objects (CC and EC) are created with these windows and put in store.
2. Every morning. The check runs. Most days nothing moves.
3. Service starts. unavailable_due_to_service becomes true for g-daas. Both of its HEMS objects move to serviceStore, and g-daan takes the H70 callsign for the duration.
4. During the service. Resource Allocation iterates over store.items, so the grounded aircraft cannot be seen, let alone chosen.
5. Service ends. The next morning’s check sees the window has passed, moves both objects back and g-daan reverts to H71.

The shipped service_schedules_by_model.csv gives the H145 an 8-month interval and the EC135 a 12-month interval, with a 3-week duration for each; cars have 0 and never enter the workshop. The preferred month is a constructor argument of HEMSAvailability. A servicing_buffer_weeks argument (default 4) also exists, but the buffer is only used in a value that is overwritten straight away, so it does not currently lengthen the service windows (a service window is exactly the model’s service_duration_weeks).

Conclusion

A resource is a HEMS object (a subclass of Ambulance) with a callsign, callsign group, registration, vehicle type, care category, seasonal shift hours and a pre-computed servicing schedule. HEMSAvailability builds them from the rota and servicing CSVs, plans service windows that dodge school holidays and each other, keeps dispatchable resources in a FilterStore, and parks serviced ones in a second store through a daily check.

Next, in Resource Allocation, we will see how HEMSAvailability searches that store for the best resource for each patient, records the reason for its choice and releases the resource afterwards.