PM Interval Calculation in the CMMS Routing Pipeline

PM interval calculation is the scheduling-math stage of the CMMS Architecture & Maintenance Taxonomy domain — the component that converts asset telemetry, failure history, and operational calendars into the deterministic due dates that trigger work order release downstream.

When intervals are hardcoded, stale, or non-deterministic, the cost surfaces twice: maintenance engineers chase work orders that fire too early and burn labor on healthy assets, while facilities managers absorb unplanned downtime when intervals stretch past the point of failure. Treating the interval as recomputed state — derived deterministically on every cycle from runtime, cycle counts, and supplier-independent reliability parameters — closes that gap. This guide implements the calculation stage end to end: prerequisites, the input/output data contract, a step-by-step Python build, a configuration reference, validation checks, and the failure modes you will actually hit in production.

Prerequisites

This component runs as a worker inside the maintenance routing service. It consumes already-validated telemetry snapshots and emits scheduling directives; it never reads raw sensor frames or writes to dispatch tables directly. Before you deploy it, confirm the following are in place.

• Python 3.11+ with pydantic>=2.6 for schema enforcement on the telemetry contract and numpy>=1.26 for the mean-time-between-failures statistics. Standard-library datetime and zoneinfo handle calendar normalization; no third-party scheduler is required because the worker is driven by the broker described in async batch processing.
• CMMS REST API v1 with read access to the asset master (GET /api/v1/assets/{asset_id}) and write access to the PM schedule resource (PUT /api/v1/pm-schedules/{asset_id}). The API must honor an idempotency key and return 409 Conflict on stale sequence writes so an out-of-order recalculation cannot regress a due date.
• A resolved asset master. Interval math is only meaningful once equipment is anchored to its functional location and parent-child dependencies, which is established through asset hierarchy design. An orphaned asset has no criticality class, so its interval cannot be risk-weighted.
• Environment variables: CMMS_BASE_URL, CMMS_API_TOKEN, and PM_SAFETY_MARGIN (default 0.85). The token must carry the pm:write and assets:read scopes; a token missing pm:write fails closed at the publish step rather than silently dropping the schedule update.

Architecture and Data Contract

The component sits between validated telemetry snapshots and the routing engine. It runs four boundaries that keep the calculation honest and prevent routing logic from leaking upstream into the math:

• Ingestion boundary: validated telemetry enters a staging buffer; no calculation runs until the snapshot passes schema validation.
• Derivation boundary: deterministic formulas consume a snapshot and emit a base interval. Same input, same output, every time — the stage is stateless and idempotent.
• Normalization boundary: the base interval is aligned to facility shift windows and compliance deadlines, producing a concrete due timestamp.
• Publication boundary: the normalized schedule is pushed to the CMMS over an authenticated REST endpoint with an idempotency key, where it triggers routing, skill matching, and — when a job needs spares — parts availability checks. None of that routing is performed by this component.

The contract across the derivation boundary is explicit. The input is a TelemetrySnapshot; the output is a BaseInterval. Encoding both as pydantic models means a malformed snapshot is rejected at the boundary instead of producing a plausible-but-wrong interval that silently mis-schedules an asset.

from datetime import datetime
from typing import List, Literal
from pydantic import BaseModel, Field, field_validator


class TelemetrySnapshot(BaseModel):
    """Input contract: a validated reliability snapshot for one asset."""
    asset_id: str = Field(..., min_length=6, max_length=64)
    runtime_hours: float = Field(..., gt=0)
    cycle_count: int = Field(..., ge=0)
    failure_intervals_hrs: List[float] = Field(default_factory=list)
    manufacturer_recommended_interval_hrs: float = Field(..., gt=0)
    asset_criticality: Literal[
        "SAFETY", "ENVIRONMENTAL", "PRODUCTION_CRITICAL", "STANDARD"
    ]
    sequence_id: int = Field(..., ge=0)
    snapshot_ts: datetime

    @field_validator("failure_intervals_hrs")
    @classmethod
    def no_negative_intervals(cls, v: List[float]) -> List[float]:
        if any(x <= 0 for x in v):
            raise ValueError("failure_intervals_hrs must be positive")
        return v


class BaseInterval(BaseModel):
    """Output contract: a deterministic interval before calendar alignment."""
    asset_id: str
    value: float
    unit: Literal["hours", "cycles"]
    confidence_score: float = Field(..., ge=0.0, le=1.0)
    sequence_id: int

Step-by-Step Implementation

1. Define the canonical work order payload

An interval only matters because it eventually emits a work order. To stay consistent with the rest of the pipeline, the worker imports the same WorkOrderPayload used everywhere on this site rather than redefining its own shape. The SLA fields — priority, requested_completion, and escalation_tier — are what the routing engine reads when it decides how urgently a generated PM should dispatch. Schema details live in work order schema standards.

from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import Dict, List, Optional


class Priority(str, Enum):
    CRITICAL = "critical"
    HIGH = "high"
    STANDARD = "standard"
    PLANNED = "planned"


@dataclass
class WorkOrderPayload:
    """Canonical CMMS work order — SLA fields are mandatory site-wide."""
    work_order_id: str
    asset_id: str
    part_skus: List[str]
    required_quantities: Dict[str, int]
    priority: Priority = Priority.STANDARD
    requested_completion: Optional[datetime] = None
    escalation_tier: int = 0
    status: str = "open"
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))

2. Derive the base interval deterministically

The derivation stage stays stateless: it accepts a telemetry snapshot and configuration parameters, then returns a single base interval. When failure history is present it computes a mean time between failures and derates it by a safety margin; when history is absent it falls back to the manufacturer baseline. Cyclic assets — pumps, conveyors, presses — route to cycle-based intervals so the schedule tracks duty rather than wall-clock time. Sourcing the safety margin from configuration, never hardcoding it, lets reliability engineers tune cover without a code deploy. The statistical method behind the failure-curve conversion is covered in depth in defining PM intervals based on MTBF data.

import math
import numpy as np


def derive_base_interval(
    snapshot: TelemetrySnapshot,
    safety_margin: float = 0.85,
    load_factor: float = 1.0,
) -> BaseInterval:
    """Deterministic interval from MTBF or manufacturer spec, with derating."""
    if snapshot.failure_intervals_hrs:
        # Statistical cover: derate the observed mean time between failures.
        mtbf_hours = float(np.mean(snapshot.failure_intervals_hrs))
        base_hrs = mtbf_hours * safety_margin
        confidence = 0.95 if len(snapshot.failure_intervals_hrs) >= 5 else 0.80
    else:
        # No history: fall back to the manufacturer baseline.
        base_hrs = snapshot.manufacturer_recommended_interval_hrs * safety_margin
        confidence = 0.70

    # Derate further for harsh duty (load_factor > 1.0 shortens the interval).
    adjusted_hrs = base_hrs / max(load_factor, 0.1)

    # Cyclic assets schedule on cycles, not hours.
    if snapshot.cycle_count > 0:
        cycles_per_hour = snapshot.cycle_count / snapshot.runtime_hours
        interval_cycles = math.floor(adjusted_hrs * cycles_per_hour)
        return BaseInterval(
            asset_id=snapshot.asset_id,
            value=float(interval_cycles),
            unit="cycles",
            confidence_score=confidence,
            sequence_id=snapshot.sequence_id,
        )

    return BaseInterval(
        asset_id=snapshot.asset_id,
        value=round(adjusted_hrs, 2),
        unit="hours",
        confidence_score=confidence,
        sequence_id=snapshot.sequence_id,
    )

3. Normalize the interval to a maintenance window

A raw interval rarely lands on a moment the facility can actually act on. The normalization stage projects the interval forward from now, then walks ahead to the nearest permitted shift window without violating a compliance deadline or colliding with a planned production shutdown. Using zoneinfo keeps multi-site scheduling correct across daylight-saving transitions, so a 06:00 window in one plant does not drift by an hour twice a year.

from datetime import datetime, timedelta
from zoneinfo import ZoneInfo


def normalize_to_window(
    base_interval: BaseInterval,
    now: datetime,
    facility_tz: ZoneInfo,
    allowed_windows: list[tuple[int, int]],  # (start_hour, end_hour), local time
    cycles_to_hours: float = 0.5,
    max_deferral_days: int = 3,
) -> datetime:
    """Shift the derived interval to the nearest compliant maintenance window."""
    if base_interval.unit == "hours":
        delta = timedelta(hours=base_interval.value)
    else:
        # Approximate cycle duration only to place the job on the calendar.
        delta = timedelta(hours=base_interval.value * cycles_to_hours)

    raw_due = now.astimezone(facility_tz) + delta

    # Walk forward up to the deferral limit to find a permitted window.
    for day_offset in range(max_deferral_days + 1):
        candidate = raw_due + timedelta(days=day_offset)
        for start, end in allowed_windows:
            if start <= candidate.hour < end:
                # Clamp to window start for a deterministic due timestamp.
                return candidate.replace(
                    hour=start, minute=0, second=0, microsecond=0
                )

    # No window inside the deferral limit — return raw_due for manual review.
    return raw_due

4. Score routing priority from interval confidence

The routing engine reads a priority tier to decide queue order. Low-confidence intervals on production-critical assets escalate so a noisy reliability signal never quietly delays a high-value job, while safety and environmental assets and imminent compliance deadlines always pin to the top tier. The mapping is purely functional — no side effects, only deterministic tier resolution.

def score_priority(
    asset_criticality: str,
    interval_confidence: float,
    compliance_deadline_days: int,
) -> Priority:
    """Deterministic priority mapping for dispatch queue insertion."""
    if asset_criticality in ("SAFETY", "ENVIRONMENTAL") or compliance_deadline_days <= 1:
        return Priority.CRITICAL
    if interval_confidence < 0.70 and asset_criticality == "PRODUCTION_CRITICAL":
        return Priority.HIGH
    if compliance_deadline_days <= 7:
        return Priority.STANDARD
    return Priority.PLANNED

5. Publish the schedule idempotently

Telemetry recalculations arrive out of order and retries duplicate payloads. A monotonic sequence_id plus a checksum gives the CMMS everything it needs to reject stale or duplicate writes, so a network retry can never regress a due date or emit a phantom PM.

import hashlib
import logging
import requests

logger = logging.getLogger(__name__)


class PmSchedulePublisher:
    def __init__(self, cmms_base_url: str, api_token: str, timeout: int = 10):
        self.base_url = cmms_base_url.rstrip("/")
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_token}",
            "Content-Type": "application/json",
            "X-Client-Id": "cmms-pm-interval-worker-v1",
        })
        self.timeout = timeout

    def publish(
        self, interval: BaseInterval, due_ts: datetime, priority: Priority
    ) -> bool:
        """Idempotent PUT to the CMMS routing engine with sequence guarding."""
        payload = {
            "asset_id": interval.asset_id,
            "interval": {"value": interval.value, "unit": interval.unit},
            "confidence_score": interval.confidence_score,
            "due_ts": due_ts.isoformat(),
            "priority": priority.value,
            "sequence_id": interval.sequence_id,
            "calculated_at": datetime.now(timezone.utc).isoformat(),
            "checksum": hashlib.sha256(
                f"{interval.asset_id}:{interval.sequence_id}:{due_ts.isoformat()}".encode()
            ).hexdigest(),
        }
        try:
            resp = self.session.put(
                f"{self.base_url}/api/v1/pm-schedules/{interval.asset_id}",
                json=payload,
                timeout=self.timeout,
            )
            resp.raise_for_status()
            logger.info("scheduled %s | seq:%s | due:%s",
                        interval.asset_id, interval.sequence_id, due_ts.date())
            return True
        except requests.exceptions.HTTPError as e:
            if e.response is not None and e.response.status_code == 409:
                logger.warning(
                    "stale sequence for %s | seq:%s — skipping duplicate",
                    interval.asset_id, interval.sequence_id,
                )
                return False
            logger.error("HTTP error scheduling %s: %s", interval.asset_id, e)
            raise
        except requests.exceptions.RequestException as e:
            logger.error("network/timeout scheduling %s: %s", interval.asset_id, e)
            raise

Configuration Reference

Keep every tunable in a version-controlled configuration registry, not in the worker source. The defaults below are conservative starting points for general-purpose rotating and fixed plant equipment.

Validation and Testing

Interval math must be reproducible, so the highest-value test asserts that the same snapshot always yields the same interval. A single deterministic assertion catches accidental nondeterminism — unsorted inputs, clock-dependent math, set iteration order — before it reaches production and starts mis-scheduling assets.

def test_interval_is_deterministic():
    snap = TelemetrySnapshot(
        asset_id="PMP-CWP-0007",
        runtime_hours=8760.0,
        cycle_count=0,
        failure_intervals_hrs=[2100.0, 1900.0, 2200.0, 2000.0, 1800.0],
        manufacturer_recommended_interval_hrs=2500.0,
        asset_criticality="PRODUCTION_CRITICAL",
        sequence_id=42,
        snapshot_ts=datetime(2026, 6, 28, tzinfo=timezone.utc),
    )
    a = derive_base_interval(snap)
    b = derive_base_interval(snap)
    assert a == b
    # mean MTBF 2000 hrs * 0.85 safety margin = 1700.0
    assert a.value == 1700.0
    assert a.unit == "hours"
    assert a.confidence_score == 0.95  # 5+ samples

On a successful publish, the worker emits a single structured log line per asset — scheduled PMP-CWP-0007 | seq:42 | due:2026-09-15 — which is the canonical signal that the routing engine received the new schedule. A 409 write produces stale sequence ... skipping duplicate; seeing that line under retry storms confirms idempotency is holding rather than indicating a fault. Assert against both log lines in integration tests to verify the full derive-to-publish path.

Failure Modes and Troubleshooting

Expand each scenario for the root cause, the diagnostic log excerpt, and the fix. The checklist items render as interactive checkboxes — work through them in order.

PM work orders fire too early and burn labor on healthy assets

• Confirm safety_margin is not set so low that intervals collapse far inside the failure mean; 0.5 on a stable asset doubles PM frequency for no reliability gain.
• Check whether load_factor is mistakenly above 1.0 for a benign environment — log line derived interval <n>h (load_factor=<x>) reveals the derate that was applied.
• Root cause is almost always an over-conservative margin or an inherited harsh-duty load factor; recalibrate both against the asset’s real duty class.

Intervals swing between recalculation cycles

• Verify failure_intervals_hrs carries enough samples; a two-point history produces a high-variance mean that chases noise each cycle.
• Confirm the worker holds the manufacturer fallback below confidence_floor instead of publishing a sparse-data interval.
• Root cause is a small or unsorted failure sample feeding the MTBF mean; widen the history window and gate on confidence before publishing.

Due dates land outside any permitted shift window

• Diagnostic: normalize returned raw_due (no window in <n> days) — normalization exhausted max_deferral_days without a match.
• Check allowed_windows against the live crew roster; an empty or mistyped window list forces every job to fall through to manual review.
• Root cause is usually a timezone mismatch — the snapshot clock and facility_tz disagree — so candidate hours never satisfy the window test; pin both to the facility zone.

Schedule writes silently disappear under load

• Look for the absence of any scheduled or stale sequence line for affected assets — neither success nor conflict logged.
• Confirm the API token still carries pm:write; an expired or under-scoped token fails closed at publish.
• Check the dead-letter queue — payloads past max_retries land there with full context for replay rather than vanishing, and reconciliation can reprocess them after the upstream fix.

Frequently Asked Questions

Should intervals be recalculated on every telemetry snapshot or on a schedule?

Recalculate on every validated snapshot, but make the math idempotent so redundant cycles are cheap and safe. Event-driven recalculation keeps intervals aligned with real duty, and a nightly reconciliation pass over the asset fleet catches equipment whose telemetry was delayed or dropped. The same broker that drives async batch processing handles both modes.

How many failures do I need before trusting an MTBF-based interval?

Treat anything under five observed failure intervals as low confidence and keep the manufacturer baseline as the fallback until the sample grows. A sparse history produces a high-variance mean that swings the interval every cycle; gate on confidence_floor so the worker only switches to the statistical interval once the signal stabilizes.

Where should the safety margin live?

In a version-controlled configuration registry keyed per asset class, never in the worker source. That lets reliability engineers tighten cover for safety-critical assets without a deployment and keeps the margin auditable alongside the rest of the pipeline configuration.

How does a calculated interval actually become a work order?

The worker publishes a due timestamp and priority to the CMMS; the routing engine generates the work order when that timestamp arrives, matches technician skills, and — for jobs needing spares — runs parts availability checks before dispatch. This component never creates the work order itself, which keeps the math stage stateless and replayable.

Related

Feed risk-weighting from the equipment tree in asset hierarchy design, keep generated PMs schema-valid through work order schema standards, gate window access with security access boundaries, convert failure curves into intervals with defining PM intervals based on MTBF data, and confirm spares before dispatch via parts availability checks.

Part of: CMMS Architecture & Maintenance Taxonomy.