Machine Maintenance
Schedule preventive maintenance across time slots to minimize cost while respecting crew-hour capacity and machine conflicts.
Schedule preventive maintenance across time slots to minimize cost while respecting crew-hour capacity and machine conflicts.
A multi-reasoner template that chains querying, graph analysis, rules-based classification, and prescriptive optimization to schedule preventive maintenance, surface hidden operational risk, and recommend cross-training to eliminate concentration vulnerabilities.
Browse files
Browse files
Browse files
What this template is for
Preventive maintenance reduces unplanned downtime, but scheduling it can be tricky. You typically have limited crew capacity per shift/day, some equipment can’t be serviced at the same time (shared tooling, access constraints, or specialist technicians), and some maintenance windows are more expensive (overtime/weekends).
This template shows how to build a small maintenance scheduling optimizer with RelationalAI. It schedules each machine into exactly one time slot, respects crew-hour limits, avoids conflicting machine pairs in the same slot, and minimizes total expected maintenance cost.
Who this is for
- You want an end-to-end example of prescriptive reasoning (optimization) using RelationalAI Semantics.
- You’re comfortable with basic Python and the idea of decision variables, constraints, and an objective.
What you’ll build
- A semantic model for machines, time slots, and conflicts (concepts + properties).
- A MILP scheduling model with one binary assignment variable per machine–slot pair.
- Constraints for exactly-once scheduling, crew-hour capacity, and conflict exclusions.
- A cost-minimizing solve using the HiGHS backend and a readable printed schedule.
What’s included
- Model + solve script:
machine_maintenance.py - Sample data:
data/machines.csv,data/time_slots.csv,data/conflicts.csv - Outputs: solver status + objective value + a machine-to-day schedule printed to stdout
Prerequisites
Access
- A Snowflake account that has the RAI Native App installed.
- A Snowflake user with permissions to access the RAI Native App.
Tools
- Python >= 3.10
Quickstart
Follow these steps to run the template with the included sample data.
-
Download the ZIP file for this template and extract it:
Terminal window curl -O https://private.relational.ai/templates/zips/v0.13/machine_maintenance.zipunzip machine_maintenance.zipcd machine_maintenance -
Create and activate a virtual environment
Terminal window python -m venv .venvsource .venv/bin/activatepython -m pip install -U pip -
Install dependencies
From this folder:
Terminal window python -m pip install . -
Configure Snowflake connection and RAI profile
Terminal window rai init -
Run the template
Terminal window python machine_maintenance.py -
Expected output
Status: OPTIMALTotal maintenance cost: $19500.00Maintenance schedule:machine dayCNC_Mill TuesdayDrill TuesdayLathe MondayPress MondayWelder Thursday
Template structure
.├─ README.md├─ pyproject.toml├─ machine_maintenance.py # main runner / entrypoint└─ data/ # sample input data ├─ machines.csv ├─ time_slots.csv └─ conflicts.csvStart here: machine_maintenance.py
Sample data
Data files are in data/.
machines.csv
Defines the machines to schedule, the effort required, and the relative cost of delaying maintenance.
| Column | Meaning |
|---|---|
id | Unique machine identifier |
name | Machine name |
maintenance_hours | Crew-hours required to complete maintenance |
failure_cost | Weight/cost used in the objective (higher means “more expensive to schedule into costly slots”) |
importance | Priority level (loaded for convenience; not used in the objective in this template) |
time_slots.csv
Defines available maintenance time slots.
| Column | Meaning |
|---|---|
id | Unique slot identifier |
day | Slot label used in output (e.g., Monday, Tuesday) |
crew_hours | Total crew-hours available in the slot |
cost_multiplier | Cost multiplier for scheduling into the slot (e.g., overtime) |
conflicts.csv
Defines machine pairs that cannot be maintained in the same time slot.
| Column | Meaning |
|---|---|
machine1_id | First machine in a conflicting pair |
machine2_id | Second machine in a conflicting pair |
Model overview
The semantic model for this template is built around four concepts.
Machine
A machine that requires preventive maintenance.
| Property | Type | Identifying? | Notes |
|---|---|---|---|
id | int | Yes | Loaded as the key from data/machines.csv |
name | string | No | Used for output labeling |
maintenance_hours | int | No | Consumed against slot capacity |
failure_cost | float | No | Scales assignment cost in the objective |
importance | int | No | Included as an extension hook |
TimeSlot
A maintenance window with limited crew capacity and a relative cost.
| Property | Type | Identifying? | Notes |
|---|---|---|---|
id | int | Yes | Loaded as the key from data/time_slots.csv |
day | string | No | Used for output labeling |
crew_hours | int | No | Capacity per slot |
cost_multiplier | float | No | Increases cost for expensive slots |
Conflict
A pair of machines that cannot be scheduled in the same slot.
| Property | Type | Identifying? | Notes |
|---|---|---|---|
machine1 | Machine | Part of compound key | Joined from conflicts.csv.machine1_id |
machine2 | Machine | Part of compound key | Joined from conflicts.csv.machine2_id |
Schedule (decision concept)
A decision entity that assigns a machine to a slot.
| Property | Type | Identifying? | Notes |
|---|---|---|---|
machine | Machine | Part of compound key | One dimension of the assignment |
slot | TimeSlot | Part of compound key | One dimension of the assignment |
assigned | float | No | Binary decision variable (0/1) |
How it works
This section walks through the highlights in machine_maintenance.py.
Import libraries and configure inputs
First, the script imports the Semantics APIs, sets DATA_DIR, and creates a Semantics Model container:
from pathlib import Path
import pandasfrom pandas import read_csv
from relationalai.semantics import Model, data, define, require, select, sum, wherefrom relationalai.semantics.reasoners.optimization import Solver, SolverModel
# --------------------------------------------------# Configure inputs# --------------------------------------------------
DATA_DIR = Path(__file__).parent / "data"
# Disable pandas inference of string types. This ensures that string columns# in the CSVs are loaded as object dtype. This is only required when using# relationalai versions prior to v1.0.pandas.options.future.infer_string = False
# --------------------------------------------------# Define semantic model & load data# --------------------------------------------------
# Create a Semantics model container.model = Model("machine_maintenance", config=globals().get("config", None), use_lqp=False)Define concepts and load CSV data
Next, it defines Machine and TimeSlot concepts and loads the corresponding CSVs using data(...).into(...):
# Machine concept: represents a machine with maintenance hours, failure cost, and importance level.Machine = model.Concept("Machine")Machine.id = model.Property("{Machine} has {id:int}")Machine.name = model.Property("{Machine} has {name:string}")Machine.maintenance_hours = model.Property("{Machine} has {maintenance_hours:int}")Machine.failure_cost = model.Property("{Machine} has {failure_cost:float}")Machine.importance = model.Property("{Machine} has {importance:int}")
# Load machine data from CSV.data(read_csv(DATA_DIR / "machines.csv")).into(Machine, keys=["id"])
# TimeSlot concept: represents a maintenance time slot with crew hour capacity and cost multiplier.TimeSlot = model.Concept("TimeSlot")TimeSlot.id = model.Property("{TimeSlot} has {id:int}")TimeSlot.day = model.Property("{TimeSlot} on {day:string}")TimeSlot.crew_hours = model.Property("{TimeSlot} has {crew_hours:int}")TimeSlot.cost_multiplier = model.Property("{TimeSlot} has {cost_multiplier:float}")
# Load time slot data from CSV.data(read_csv(DATA_DIR / "time_slots.csv")).into(TimeSlot, keys=["id"])Then, it loads conflict pairs from conflicts.csv and resolves machine IDs into Machine instances with where(...).define(...):
# Conflict concept: represents conflicts between machines that cannot be maintained at the same timeConflict = model.Concept("Conflict")Conflict.machine1 = model.Property("{Conflict} between {machine1:Machine}")Conflict.machine2 = model.Property("{Conflict} and {machine2:Machine}")
# Load machine conflict pairs from CSV.conflicts_data = data(read_csv(DATA_DIR / "conflicts.csv"))M2 = Machine.ref()where( Machine.id == conflicts_data.machine1_id, M2.id == conflicts_data.machine2_id).define( Conflict.new(machine1=Machine, machine2=M2))Define decision variables, constraints, and objective
With the input concepts in place, the script creates a Schedule decision concept and uses solve_for(..., type="bin") to create one binary assignment variable per machine–slot pair:
# Schedule decision concept: represents the assignment of machines to time slots.# The "assigned" property is a binary variable indicating whether a machine is scheduled in a slot.Schedule = model.Concept("Schedule")Schedule.machine = model.Property("{Schedule} for {machine:Machine}")Schedule.slot = model.Property("{Schedule} in {slot:TimeSlot}")Schedule.x_assigned = model.Property("{Schedule} is {assigned:float}")define(Schedule.new(machine=Machine, slot=TimeSlot))
Sch = Schedule.ref()Sch1 = Schedule.ref()Sch2 = Schedule.ref()
s = SolverModel(model, "cont")
# Variable: binary assignments.solve_for(Schedule.x_assigned, type="bin", name=["x", Schedule.machine.name, Schedule.slot.day])Next, it adds three families of constraints using require(...) and s.satisfy(...): schedule each machine exactly once, enforce per-slot crew-hour capacity, and prevent conflicting pairs from being scheduled in the same slot:
# Constraint: each machine scheduled exactly oncemachine_scheduled = sum(Sch.assigned).where(Sch.machine == Machine).per(Machine)exactly_once = require(machine_scheduled == 1)s.satisfy(exactly_once)
# Constraint: crew hours per slot not exceededslot_hours = sum(Sch.assigned * Sch.machine.maintenance_hours).where(Sch.slot == TimeSlot).per(TimeSlot)crew_limit = require(slot_hours <= TimeSlot.crew_hours)s.satisfy(crew_limit)
# Constraint: conflicting machines cannot be scheduled in same slotno_conflicts = require(Sch1.assigned + Sch2.assigned <= 1).where( Sch1.machine == Conflict.machine1, Sch2.machine == Conflict.machine2, Sch1.slot == Sch2.slot)s.satisfy(no_conflicts)Then, it minimizes total cost, computed as failure_cost * cost_multiplier for the chosen assignments:
# Objective: minimize total maintenance cost (base cost * slot multiplier)total_cost = sum(Schedule.x_assigned * Schedule.machine.failure_cost * Schedule.slot.cost_multiplier)s.minimize(total_cost)Solve and print results
Finally, it solves using the HiGHS backend (with a 60s time limit), prints the status/objective, and selects only assignments with Schedule.x_assigned > 0.5 for the output table:
solver = Solver("highs")s.solve(solver, time_limit_sec=60)
print(f"Status: {s.termination_status}")print(f"Total maintenance cost: ${s.objective_value:.2f}")
schedule = select( Schedule.machine.name.alias("machine"), Schedule.slot.day.alias("day")).where(Schedule.x_assigned > 0.5).to_df()
print("\nMaintenance schedule:")print(schedule.to_string(index=False))Customize this template
Use your own data
- Replace the CSVs under
data/with your own data, keeping the same headers (or update the loading logic inmachine_maintenance.py). - Ensure
conflicts.csvonly references valid machine IDs frommachines.csv. - Make sure total crew capacity across slots is sufficient for the total required maintenance hours (and that conflicts don’t force impossible packings).
Tune parameters
- To make certain days more expensive, edit
time_slots.csv.cost_multiplier. - To reflect staffing changes, edit
time_slots.csv.crew_hours. - To change the solve time budget, edit the call
s.solve(solver, time_limit_sec=60).
Extend the model
- Use
Machine.importancein the objective (for example, multiply it into the cost) if you want a second, coarse priority signal. - Add “must-schedule-by” deadlines by restricting which slots certain machines are allowed to use.
- Add technician skills or machine-type requirements by introducing additional concepts and capacity constraints.
Scale up and productionize
- Replace CSV ingestion with Snowflake sources.
- Write the chosen schedule back to Snowflake after solving.
Troubleshooting
Why does authentication/configuration fail?
- Run
rai initto create/updateraiconfig.toml. - If you have multiple profiles, set
RAI_PROFILEor switch profiles in your config.
Why does the script fail to connect to the RAI Native App?
- Verify the Snowflake account/role/warehouse and
rai_app_nameare correct inraiconfig.toml. - Ensure the RAI Native App is installed and you have access.
Why do I get ModuleNotFoundError?
- Confirm your virtual environment is activated.
- Install dependencies from this folder with
python -m pip install ..
Why does CSV loading fail (missing file or column)?
- Confirm the CSVs exist under
data/and the filenames match. - Ensure the headers match the expected schema:
machines.csv:id,name,maintenance_hours,failure_cost,importancetime_slots.csv:id,day,crew_hours,cost_multiplierconflicts.csv:machine1_id,machine2_id
Why do I get Status: INFEASIBLE?
- Check that total required hours can fit into the available
crew_hoursacross slots. - Conflicts can force additional separation: ensure conflicting machines have enough remaining capacity across distinct slots.
- Confirm every machine has at least one feasible slot (in this template, all machines can use all slots; if you extend the model with eligibility rules, this becomes a common cause).
Why is the printed schedule empty?
- The output filters on
Schedule.x_assigned > 0.5. If the solve did not succeed, no assignments may satisfy that. - Print
s.termination_statusand inspect whether the solve completed successfully.
What this template is for
Preventive maintenance reduces unplanned downtime, but scheduling it can be tricky. You typically have limited crew capacity per shift/day, some equipment can’t be serviced at the same time (shared tooling, access constraints, or specialist technicians), and some maintenance windows are more expensive (overtime/weekends).
This template shows how to build a small maintenance scheduling optimizer with RelationalAI. It schedules each machine into exactly one time slot, respects crew-hour limits, avoids conflicting machine pairs in the same slot, and minimizes total expected maintenance cost.
Who this is for
- You want an end-to-end example of prescriptive reasoning (optimization) using RelationalAI Semantics.
- You’re comfortable with basic Python and the idea of decision variables, constraints, and an objective.
What you’ll build
- A semantic model for machines, time slots, and conflicts (concepts + properties).
- A MILP scheduling model with one binary assignment variable per machine–slot pair.
- Constraints for exactly-once scheduling, crew-hour capacity, and conflict exclusions.
- A cost-minimizing solve using the HiGHS backend and a readable printed schedule.
What’s included
- Model + solve script:
machine_maintenance.py - Sample data:
data/machines.csv,data/time_slots.csv,data/conflicts.csv - Outputs: solver status + objective value + a machine-to-day schedule printed to stdout
Prerequisites
Access
- A Snowflake account that has the RAI Native App installed.
- A Snowflake user with permissions to access the RAI Native App.
Tools
- Python >= 3.10
Quickstart
Follow these steps to run the template with the included sample data.
-
Download the ZIP file for this template and extract it:
Terminal window curl -O https://private.relational.ai/templates/zips/v0.14/machine_maintenance.zipunzip machine_maintenance.zipcd machine_maintenance -
Create and activate a virtual environment
Terminal window python -m venv .venvsource .venv/bin/activatepython -m pip install -U pip -
Install dependencies
From this folder:
Terminal window python -m pip install . -
Configure Snowflake connection and RAI profile
Terminal window rai init -
Run the template
Terminal window python machine_maintenance.py -
Expected output
Status: OPTIMALTotal maintenance cost: $19500.00Maintenance schedule:machine dayCNC_Mill TuesdayDrill TuesdayLathe MondayPress MondayWelder Thursday
Template structure
.├─ README.md├─ pyproject.toml├─ machine_maintenance.py # main runner / entrypoint└─ data/ # sample input data ├─ machines.csv ├─ time_slots.csv └─ conflicts.csvStart here: machine_maintenance.py
Sample data
Data files are in data/.
machines.csv
Defines the machines to schedule, the effort required, and the relative cost of delaying maintenance.
| Column | Meaning |
|---|---|
id | Unique machine identifier |
name | Machine name |
maintenance_hours | Crew-hours required to complete maintenance |
failure_cost | Weight/cost used in the objective (higher means “more expensive to schedule into costly slots”) |
importance | Priority level (loaded for convenience; not used in the objective in this template) |
time_slots.csv
Defines available maintenance time slots.
| Column | Meaning |
|---|---|
id | Unique slot identifier |
day | Slot label used in output (e.g., Monday, Tuesday) |
crew_hours | Total crew-hours available in the slot |
cost_multiplier | Cost multiplier for scheduling into the slot (e.g., overtime) |
conflicts.csv
Defines machine pairs that cannot be maintained in the same time slot.
| Column | Meaning |
|---|---|
machine1_id | First machine in a conflicting pair |
machine2_id | Second machine in a conflicting pair |
Model overview
The semantic model for this template is built around four concepts.
Machine
A machine that requires preventive maintenance.
| Property | Type | Identifying? | Notes |
|---|---|---|---|
id | int | Yes | Loaded as the key from data/machines.csv |
name | string | No | Used for output labeling |
maintenance_hours | int | No | Consumed against slot capacity |
failure_cost | float | No | Scales assignment cost in the objective |
importance | int | No | Included as an extension hook |
TimeSlot
A maintenance window with limited crew capacity and a relative cost.
| Property | Type | Identifying? | Notes |
|---|---|---|---|
id | int | Yes | Loaded as the key from data/time_slots.csv |
day | string | No | Used for output labeling |
crew_hours | int | No | Capacity per slot |
cost_multiplier | float | No | Increases cost for expensive slots |
Conflict
A pair of machines that cannot be scheduled in the same slot.
| Property | Type | Identifying? | Notes |
|---|---|---|---|
machine1 | Machine | Part of compound key | Joined from conflicts.csv.machine1_id |
machine2 | Machine | Part of compound key | Joined from conflicts.csv.machine2_id |
Schedule (decision concept)
A decision entity that assigns a machine to a slot.
| Property | Type | Identifying? | Notes |
|---|---|---|---|
machine | Machine | Part of compound key | One dimension of the assignment |
slot | TimeSlot | Part of compound key | One dimension of the assignment |
assigned | float | No | Binary decision variable (0/1) |
How it works
This section walks through the highlights in machine_maintenance.py.
Import libraries and configure inputs
First, the script imports the Semantics APIs, sets DATA_DIR, and creates a Semantics Model container:
from pathlib import Path
import pandasfrom pandas import read_csv
from relationalai.semantics import Model, Relationship, data, define, require, select, sum, wherefrom relationalai.semantics.reasoners.optimization import Solver, SolverModel
# --------------------------------------------------# Configure inputs# --------------------------------------------------
DATA_DIR = Path(__file__).parent / "data"
# Disable pandas inference of string types. This ensures that string columns# in the CSVs are loaded as object dtype. This is only required when using# relationalai versions prior to v1.0.pandas.options.future.infer_string = False
# --------------------------------------------------# Define semantic model & load data# --------------------------------------------------
# Create a Semantics model container.model = Model("machine_maintenance", config=globals().get("config", None))Define concepts and load CSV data
Next, it defines Machine and TimeSlot concepts and loads the corresponding CSVs using data(...).into(...):
# Machine concept: represents a machine with maintenance hours, failure cost, and importance level.Machine = model.Concept("Machine")Machine.id = model.Property("{Machine} has {id:int}")Machine.name = model.Property("{Machine} has {name:string}")Machine.maintenance_hours = model.Property("{Machine} has {maintenance_hours:int}")Machine.failure_cost = model.Property("{Machine} has {failure_cost:float}")Machine.importance = model.Property("{Machine} has {importance:int}")
# Load machine data from CSV.data(read_csv(DATA_DIR / "machines.csv")).into(Machine, keys=["id"])
# TimeSlot concept: represents a maintenance time slot with crew hour capacity and cost multiplier.TimeSlot = model.Concept("TimeSlot")TimeSlot.id = model.Property("{TimeSlot} has {id:int}")TimeSlot.day = model.Property("{TimeSlot} on {day:string}")TimeSlot.crew_hours = model.Property("{TimeSlot} has {crew_hours:int}")TimeSlot.cost_multiplier = model.Property("{TimeSlot} has {cost_multiplier:float}")
# Load time slot data from CSV.data(read_csv(DATA_DIR / "time_slots.csv")).into(TimeSlot, keys=["id"])Then, it loads conflict pairs from conflicts.csv and resolves machine IDs into Machine instances with where(...).define(...):
# Conflict concept: represents conflicts between machines that cannot be maintained at the same timeConflict = model.Concept("Conflict")Conflict.machine1 = model.Relationship("{Conflict} between {machine1:Machine}")Conflict.machine2 = model.Relationship("{Conflict} and {machine2:Machine}")
# Load machine conflict pairs from CSV.conflicts_data = data(read_csv(DATA_DIR / "conflicts.csv"))OtherMachine = Machine.ref()where( Machine.id == conflicts_data.machine1_id, OtherMachine.id == conflicts_data.machine2_id).define( Conflict.new(machine1=Machine, machine2=OtherMachine))Define decision variables, constraints, and objective
With the input concepts in place, the script creates a Schedule decision concept and uses solve_for(..., type="bin") to create one binary assignment variable per machine–slot pair:
# Schedule decision concept: represents the assignment of machines to time slots.# The "assigned" property is a binary variable indicating whether a machine is scheduled in a slot.Schedule = model.Concept("Schedule")Schedule.machine = model.Relationship("{Schedule} for {machine:Machine}")Schedule.slot = model.Relationship("{Schedule} in {slot:TimeSlot}")Schedule.x_assigned = model.Property("{Schedule} is {assigned:float}")define(Schedule.new(machine=Machine, slot=TimeSlot))
ScheduleRef = Schedule.ref()ScheduleA = Schedule.ref()ScheduleB = Schedule.ref()
s = SolverModel(model, "cont")
# Variable: binary assignments.solve_for(Schedule.x_assigned, type="bin", name=["x", Schedule.machine.name, Schedule.slot.day])Next, it adds three families of constraints using require(...) and s.satisfy(...): schedule each machine exactly once, enforce per-slot crew-hour capacity, and prevent conflicting pairs from being scheduled in the same slot:
# Constraint: each machine scheduled exactly oncemachine_scheduled = sum(ScheduleRef.x_assigned).where(ScheduleRef.machine == Machine).per(Machine)exactly_once = require(machine_scheduled == 1)s.satisfy(exactly_once)
# Constraint: crew hours per slot not exceededslot_hours = sum(ScheduleRef.x_assigned * ScheduleRef.machine.maintenance_hours).where(ScheduleRef.slot == TimeSlot).per(TimeSlot)crew_limit = require(slot_hours <= TimeSlot.crew_hours)s.satisfy(crew_limit)
# Constraint: conflicting machines cannot be scheduled in same slotno_conflicts = require(ScheduleA.x_assigned + ScheduleB.x_assigned <= 1).where( ScheduleA.machine == Conflict.machine1, ScheduleB.machine == Conflict.machine2, ScheduleA.slot == ScheduleB.slot)s.satisfy(no_conflicts)Then, it minimizes total cost, computed as failure_cost * cost_multiplier for the chosen assignments:
# Objective: minimize total maintenance cost (base cost * slot multiplier)total_cost = sum(Schedule.x_assigned * Schedule.machine.failure_cost * Schedule.slot.cost_multiplier)s.minimize(total_cost)Solve and print results
Finally, it solves using the HiGHS backend (with a 60s time limit), prints the status/objective, and selects only assignments with Schedule.x_assigned > 0.5 for the output table:
solver = Solver("highs")s.solve(solver, time_limit_sec=60)
print(f"Status: {s.termination_status}")print(f"Total maintenance cost: ${s.objective_value:.2f}")
schedule = select( Schedule.machine.name.alias("machine"), Schedule.slot.day.alias("day")).where(Schedule.x_assigned > 0.5).to_df()
print("\nMaintenance schedule:")print(schedule.to_string(index=False))Customize this template
Use your own data
- Replace the CSVs under
data/with your own data, keeping the same headers (or update the loading logic inmachine_maintenance.py). - Ensure
conflicts.csvonly references valid machine IDs frommachines.csv. - Make sure total crew capacity across slots is sufficient for the total required maintenance hours (and that conflicts don’t force impossible packings).
Tune parameters
- To make certain days more expensive, edit
time_slots.csv.cost_multiplier. - To reflect staffing changes, edit
time_slots.csv.crew_hours. - To change the solve time budget, edit the call
s.solve(solver, time_limit_sec=60).
Extend the model
- Use
Machine.importancein the objective (for example, multiply it into the cost) if you want a second, coarse priority signal. - Add “must-schedule-by” deadlines by restricting which slots certain machines are allowed to use.
- Add technician skills or machine-type requirements by introducing additional concepts and capacity constraints.
Scale up and productionize
- Replace CSV ingestion with Snowflake sources.
- Write the chosen schedule back to Snowflake after solving.
Troubleshooting
Why does authentication/configuration fail?
- Run
rai initto create/updateraiconfig.toml. - If you have multiple profiles, set
RAI_PROFILEor switch profiles in your config.
Why does the script fail to connect to the RAI Native App?
- Verify the Snowflake account/role/warehouse and
rai_app_nameare correct inraiconfig.toml. - Ensure the RAI Native App is installed and you have access.
Why do I get ModuleNotFoundError?
- Confirm your virtual environment is activated.
- Install dependencies from this folder with
python -m pip install ..
Why does CSV loading fail (missing file or column)?
- Confirm the CSVs exist under
data/and the filenames match. - Ensure the headers match the expected schema:
machines.csv:id,name,maintenance_hours,failure_cost,importancetime_slots.csv:id,day,crew_hours,cost_multiplierconflicts.csv:machine1_id,machine2_id
Why do I get Status: INFEASIBLE?
- Check that total required hours can fit into the available
crew_hoursacross slots. - Conflicts can force additional separation: ensure conflicting machines have enough remaining capacity across distinct slots.
- Confirm every machine has at least one feasible slot (in this template, all machines can use all slots; if you extend the model with eligibility rules, this becomes a common cause).
Why is the printed schedule empty?
- The output filters on
Schedule.x_assigned > 0.5. If the solve did not succeed, no assignments may satisfy that. - Print
s.termination_statusand inspect whether the solve completed successfully.
What this template is for
Manufacturing facilities must schedule preventive maintenance for machines with ML-predicted failure probabilities. The challenge is that surface-level metrics (like OEE) can mask structural vulnerabilities — a plant that looks mid-tier on performance may actually carry the highest concentration risk, discoverable only by chaining multiple analytical layers.
This template uses RelationalAI’s querying, graph analysis, rules-based classification, and prescriptive reasoning (optimization) capabilities in a five-stage multi-reasoner workflow:
- Querying computes OEE by facility, surfaces sensor anomalies, and identifies machines with the steepest failure degradation trajectories. Plant_B looks worst at 61.4% OEE — but Plant_A, at 68.2%, has 7 of 9 sensor anomalies and the 3 steepest degradation curves.
- Graph analysis builds a machine dependency graph from shared-technician qualifications. All 30 machines form a single connected cluster, and Pump-type machines score highest on betweenness centrality (24.0) as the most constrained scheduling bottlenecks.
- Rules derive seven compliance flags and chain three of them (chronic downtime, high-risk, overdue) into a composite risk tier. M013 (Pump, Plant_A) is the only Critical-tier machine — it triggers all three flags.
- Prescriptive optimization schedules 20 maintenance jobs across 4 periods at $605K total cost, assigning qualified technicians. The optimizer consumes per-period failure predictions from Stage 0, betweenness centrality from Stage 1, and overdue-maintenance flags from Stage 2.
- Resilience analysis reveals that all 3 Turbine-qualified technicians are in Houston_TX, forcing 67% of scheduled Turbine jobs to require travel. Cross-training T006 (Senior, Chicago_IL) for $3,200 over 5 weeks eliminates this geographic concentration risk.
Each stage enriches the shared ontology, and downstream stages consume those enrichments — this is the accretive ontology enrichment pattern. No Python dicts carry state between stages; the ontology is the single source of truth:
- Stage 0 writes
Machine.performance_ratio,Machine.quality_ratio,Machine.anomaly_count,MachinePeriod.predicted_fp— consumed by Stage 2’s rules AND Stage 3’s objective. Both downstream reasoners see the same derived signals. - Stage 1 writes
Machine.betweenness(normalized centrality) — consumed by Stage 3’s failure cost term. Bottleneck machines are more expensive to leave vulnerable. - Stage 2 writes
Machine.is_overdue_maintenance,Machine.is_high_risk,Machine.is_chronic_downtime,Machine.risk_tier— the overdue flag feeds a hard scheduling constraint in Stage 3 (overdue machines must be maintained by period 2). - Stage 3 writes
x_maintain,x_vulnerable,x_assigned(decision variables) — parsed in Stage 4 to analyze technician utilization and concentration risk.
Reasoner overview
| Stage | Reasoner | Reads from ontology | Writes to ontology | Role |
|---|---|---|---|---|
| 0 | Querying | ProductionRun, SensorReading, FailurePrediction | Machine.performance_ratio, Machine.quality_ratio, Machine.anomaly_count, MachinePeriod.predicted_fp | Plant_C leads at 79.8% OEE; Plant_A mid at 68.2% but has 7 of 9 sensor anomalies and the 3 steepest failure trajectories (M001 +0.230, M013 +0.228, M016 +0.219). |
| 1 | Graph | Qualification, Machine (as node_concept) | Machine.betweenness (normalized centrality) | All 30 machines form 1 connected cluster. Pump-type machines are the top bottlenecks (betweenness=24.0). Centrality scores feed the failure cost multiplier in Stage 3. |
| 2 | Rules | Machine (all derived properties from Stages 0-1) | Machine.is_overdue_maintenance, Machine.is_high_risk, Machine.is_chronic_downtime, Machine.risk_tier | 6 overdue, 1 high-risk, 3 chronic downtime. Composite tier: M013 is Critical (all 3 flags), M016 is Elevated (2 of 3). Overdue flag becomes a hard constraint in Stage 3. |
| 3 | Prescriptive | MachinePeriod.predicted_fp, Machine.betweenness, Machine.is_overdue_maintenance | x_maintain, x_vulnerable, x_assigned (decision variables) | 20 jobs across 4 periods at $605K total cost. Per-period failure predictions (not static probability) weight the objective. Overdue machines scheduled by period 2. |
| 4 | Analysis | Solution variables, Qualification, TrainingOption | (terminal — prints recommendations) | All 3 Turbine techs in Houston_TX — 67% of Turbine jobs require travel. Best cross-training: T006 (Chicago_IL, Senior) at $3,200 / 5 weeks. |
Why this problem matters
OEE dashboards and failure-probability rankings are how most plants prioritize maintenance today. But these metrics evaluate machines in isolation — they miss structural dependencies between machines, technicians, and locations that create cascading risk. A plant where all Turbine-qualified technicians happen to work from the same office looks fine on every individual metric. The concentration risk is invisible until someone leaves, a certification expires, or a weather event disrupts the location — at which point multiple machines lose coverage simultaneously.
The multi-reasoner approach is necessary because no single analytical technique surfaces this risk. Querying reveals sensor anomalies that OEE masks. Graph analysis exposes which machines share technician pools. Rules chain individual flags into composite risk tiers. Optimization produces a schedule, and resilience analysis stress-tests that schedule against the qualification structure. Each layer reveals something the previous one missed.
Key design patterns demonstrated
- Accretive ontology enrichment — each stage writes derived properties (betweenness, risk_tier, predicted_fp) that downstream stages consume, building a progressively richer model
- Rules chaining — three boolean flags (is_chronic_downtime, is_high_risk, is_overdue_maintenance) are composed into a single risk_tier property using exhaustive enumeration with
model.not_() - Graph directly on domain concept — the Graph reasoner uses
Machinedirectly asnode_concept, so centrality scores are stored as Machine properties without a mirror concept - Per-period failure predictions — the optimization objective uses
MachinePeriod.predicted_fp(period-specific) rather than staticMachine.failure_probability, giving the solver time-varying cost information - Post-solve resilience analysis — Stage 4 inspects the solution and qualification structure to identify concentration risk, producing actionable cross-training recommendations without re-solving
Who this is for
- Manufacturing and plant managers scheduling preventive maintenance
- Operations researchers exploring multi-reasoner pipelines in RelationalAI
- Developers learning how to chain querying, graph, rules, and optimization in a single model
What you’ll build
- Machine-level production aggregates, OEE components, and anomaly counts as derived properties
- A machine dependency graph with cluster detection and centrality scoring
- Seven compliance rules as derived Relationships and Properties, including a composite risk tier that chains three boolean flags
- Binary decision variables for maintenance timing, vulnerability tracking, and technician assignment
- Cumulative coverage, capacity, and overdue-deadline constraints
- A cost minimization objective that incorporates per-period failure predictions and graph centrality
- Geographic concentration risk analysis with cross-training recommendations
What’s included
machine_maintenance.py— Main script with five chained reasoning stagesdata/machines.csv— 30 machines with failure probability, criticality (1-5), duration, and parts costdata/technicians.csv— 10 technicians with skill levels, certifications, hourly rates, and locationsdata/availability.csv— Technician availability across the 4-period planning horizondata/qualifications.csv— Mapping of which technicians can service which machine typesdata/parts_inventory.csv— Spare parts stock levels at each facilitydata/certification_expiry.csv— Days remaining on technician certifications per machine typedata/sensors.csv— 60 sensors (2 per machine) with warning and critical thresholdsdata/sensor_readings.csv— 240 periodic sensor measurements with anomaly flagsdata/failure_predictions.csv— 120 per-period failure probability trajectories with predicted failure modesdata/downtime_events.csv— 129 downtime events with fault categories and durationsdata/production_runs.csv— 120 production runs with planned, actual, and good quantitiesdata/training_options.csv— 13 cross-training options with costs and durationspyproject.toml— Python project configuration with dependencies
Prerequisites
Access
- A Snowflake account that has the RAI Native App installed.
- A Snowflake user with permissions to access the RAI Native App.
Tools
- Python >= 3.10
- RelationalAI Python SDK (
relationalai) >= 1.0.13
Quickstart
-
Download the ZIP file for this template and extract it:
Terminal window curl -O https://private.relational.ai/templates/zips/v1/machine_maintenance.zipunzip machine_maintenance.zipcd machine_maintenance -
Create a virtual environment and activate it:
Terminal window python -m venv .venvsource .venv/bin/activatepython -m pip install --upgrade pip -
Install dependencies:
Terminal window python -m pip install . -
Configure your RAI connection:
Terminal window rai init -
Run the template:
Terminal window python machine_maintenance.py -
Expected output:
======================================================================STAGE 0: Querying -- Operational Intelligence======================================================================OEE proxy by facility (Performance x Quality):Plant_C: Perf=81.3%, Qual=98.1%, OEE=79.8%Plant_A: Perf=69.8%, Qual=97.8%, OEE=68.2%Plant_B: Perf=62.6%, Qual=98.1%, OEE=61.4%Sensor anomalies (9 readings across 5 machines):M013 (Pump, Plant_A): 3 anomaliesM001 (Turbine, Plant_A): 2 anomaliesM016 (Turbine, Plant_A): 2 anomaliesM002 (Compressor, Plant_B): 1 anomaliesM006 (Turbine, Plant_C): 1 anomaliesBy facility: {'Plant_A': 7, 'Plant_B': 1, 'Plant_C': 1}Steepest failure trajectories (period 1 -> 4):M001 (Turbine, Plant_A): 0.102 -> 0.332 (+0.230) [bearing_wear]M013 (Pump, Plant_A): 0.435 -> 0.663 (+0.228) [impeller_erosion]M016 (Turbine, Plant_A): 0.263 -> 0.482 (+0.219) [bearing_wear]...======================================================================STAGE 1: Graph Analysis -- Dependency Clusters & Centrality======================================================================Dependency clusters found: 1Top bottleneck machines (betweenness centrality):M003 (Pump, Plant_C): betweenness=24.0000, failure_prob=0.089M008 (Pump, Plant_B): betweenness=24.0000, failure_prob=0.076M013 (Pump, Plant_A): betweenness=24.0000, failure_prob=0.435...======================================================================STAGE 2: Rules -- Compliance Flags & Composite Risk Tier======================================================================Overdue maintenance (6 machines):M002 (Compressor_Beta_1): RUL=3.7h < duration=6hM006 (Turbine_Alpha_2): RUL=3.4h < duration=8hM013 (Pump_Gamma_3): RUL=2.3h < duration=4h...High-risk machines (1):M013 (Pump_Gamma_3): prob=0.435, crit=4Anomalous machines (5):M013 (Pump_Gamma_3, Plant_A): 3 anomaliesM001 (Turbine_Alpha_1, Plant_A): 2 anomaliesM016 (Turbine_Alpha_4, Plant_A): 2 anomalies...Chronic downtime machines (>8 events, 3 machines):M001 (Turbine_Alpha_1, Plant_A): 12 events, 1635 min total downtimeM016 (Turbine_Alpha_4, Plant_A): 11 events, 1314 min total downtimeM013 (Pump_Gamma_3, Plant_A): 10 events, 1272 min total downtimeComposite risk tier:Critical (1): M013Elevated (1): M016Standard (28): M001, M002, ...Parts needing reorder (4):P001 (Spindle Bearings, Plant_A): stock=25 <= min_order=50...Expiring certifications (5):T001 (Alice_Johnson): Compressor -- 22 days remainingT004 (Diana_Chen): Pump -- 8 days remaining...======================================================================STAGE 3: Prescriptive -- Maintenance Scheduling======================================================================Status: OPTIMALObjective value: 605240.61Maintenance schedule (20 jobs):Period 1:M002 (Compressor, Plant_B, crit=5)M006 (Turbine, Plant_C, crit=5)M013 (Pump, Plant_A, crit=4)M016 (Turbine, Plant_A, crit=3)...Period 2: ...Period 3: ...Period 4: ...Technician assignments (20):Period 1:M002: T003 (6h x $65/h = $390) [TRAVEL]M013: T006 (4h x $88/h = $352) [TRAVEL]...======================================================================STAGE 4: Resilience -- Concentration Risk Analysis======================================================================Technician utilization in optimal schedule:T003 (Charlie_Brown, Junior, Houston_TX): 5 assignments (25%)T004 (Diana_Chen, Junior, Chicago_IL): 5 assignments (25%)...Qualification coverage by machine type:Compressor: 3 techs in Chicago_IL, Houston_TX -- gaps at Phoenix_AZGenerator: 3 techs in Chicago_IL, Phoenix_AZ -- gaps at Houston_TXMotor: 4 techs in Chicago_IL, Phoenix_AZ -- gaps at Houston_TXPump: 3 techs in Chicago_IL, Phoenix_AZ -- gaps at Houston_TXTurbine: 3 techs in Houston_TX -- CONCENTRATEDConcentration risk detail:Turbine: 6 machines across 3 facilities, all 3 qualified techs in Houston_TXLocal machines (Houston_TX): 2Remote machines (require travel): 4Scheduled Turbine jobs: 3, of which 2 require travel (67%)Qualified techs (all Houston_TX):T001 (Alice_Johnson, Senior)T002 (Bob_Martinez, Senior)T003 (Charlie_Brown, Junior)======================================================================RECOMMENDATION: Cross-Training to Eliminate Concentration Risk======================================================================Turbine -- add coverage outside Houston_TX:Best candidate: T006 (Fiona_Garcia, Senior, Chicago_IL)Cost: $3,200, Duration: 5 weeksAll candidates:T006 (Fiona_Garcia, Chicago_IL): $3,200, 5 weeksT005 (Edward_Smith, Chicago_IL): $3,500, 6 weeksT008 (Hannah_Wilson, Phoenix_AZ): $3,800, 6 weeksT009 (Ian_Taylor, Phoenix_AZ): $4,200, 8 weeksT004 (Diana_Chen, Chicago_IL): $5,500, 10 weeks
Template structure
.├── README.md├── pyproject.toml├── machine_maintenance.py└── data/ ├── machines.csv ├── technicians.csv ├── availability.csv ├── qualifications.csv ├── parts_inventory.csv ├── certification_expiry.csv ├── sensors.csv ├── sensor_readings.csv ├── failure_predictions.csv ├── downtime_events.csv ├── production_runs.csv └── training_options.csvHow it works
This section walks through the highlights in machine_maintenance.py.
Define concepts and load CSV data
The model defines concepts for machines (with ML-predicted failure probability and numeric criticality), technicians (with skills and hourly rates), qualifications linking technicians to machine types, parts inventory, certification expiry, sensors, sensor readings, failure predictions, downtime events, and production runs. All data is loaded from CSV files:
Machine = model.Concept("Machine", identify_by={"machine_id": String})Machine.failure_probability = model.Property( f"{Machine} has failure probability {Float:failure_probability}")Machine.criticality = model.Property(f"{Machine} has criticality {Integer:criticality}")
Technician = model.Concept("Technician", identify_by={"technician_id": String})Qualification = model.Concept( "Qualification", identify_by={"technician_id": String, "machine_type": String})
Sensor = model.Concept("Sensor", identify_by={"sensor_id": String})SensorReading = model.Concept( "SensorReading", identify_by={"sensor_id": String, "machine_id": String, "pid": Integer})FailurePrediction = model.Concept( "FailurePrediction", identify_by={"prediction_id": String})DowntimeEvent = model.Concept("DowntimeEvent", identify_by={"event_id": String})ProductionRun = model.Concept("ProductionRun", identify_by={"run_id": String})Machine-level derived aggregates are computed from the loaded data using aggs.sum and aggs.count, providing production ratios, downtime counts, and anomaly counts as derived properties:
Machine.total_planned_qty = model.Property( f"{Machine} has total planned qty {Float:total_planned_qty}")model.define(Machine.total_planned_qty( aggs.sum(ProductionRun.planned_quantity).per(Machine) .where(ProductionRun.machine(Machine)) | 0))
model.where(Machine.total_planned_qty > 0).define( Machine.performance_ratio( floats.float(Machine.total_actual_qty) / floats.float(Machine.total_planned_qty) ))Cross-product concepts define the scheduling decision space. MachinePeriod pairs each machine with each planning period and stores per-period failure predictions. TechnicianMachinePeriod is restricted to qualified pairs — technicians can only be assigned to machine types they are certified for:
MachinePeriod.predicted_fp = model.Property( f"{MachinePeriod} has predicted failure probability {Float:predicted_fp}")FPJoin = FailurePrediction.ref()model.where( MachinePeriod.machine_id == FPJoin.machine_id_str, MachinePeriod.pid == FPJoin.period_int,).define(MachinePeriod.predicted_fp(FPJoin.failure_probability))Stage 0: Querying — operational intelligence
The querying stage computes OEE proxy (Performance x Quality) by facility, surfaces machines with above-threshold sensor readings, and identifies the steepest failure degradation trajectories. All queries use model.select with derived properties:
oee_df = ( model.select( Machine.machine_id.alias("machine_id"), Machine.facility.alias("facility"), Machine.performance_ratio.alias("performance"), Machine.quality_ratio.alias("quality"), ) .to_df())Stage 1: Graph — dependency clusters and centrality
The Graph reasoner uses Machine directly as node_concept — no mirror concept needed. Edges connect machines when at least one technician is qualified for both machine types:
dep_graph = Graph( model, directed=False, weighted=False, node_concept=Machine, aggregator="sum")Weakly connected components identify dependency clusters (groups of machines that compete for the same technicians). Betweenness centrality scores bottleneck machines — those whose maintenance blocks the most scheduling options. The scores are normalized and stored directly on Machine:
Machine.betweenness_raw = model.Property( f"{Machine} has raw betweenness centrality {Float:betweenness_raw}")m_btwn = Machine.ref("m_btwn")model.define(m_btwn.betweenness_raw(btwn_score)).where(betweenness(m_btwn, btwn_score))max_betweenness = max(Machine.betweenness_raw)Machine.betweenness = model.Property( f"{Machine} has betweenness centrality {Float:betweenness}")m_norm = Machine.ref("m_norm")model.where(max_betweenness == 0).define(m_norm.betweenness(0.0))model.where(max_betweenness > 0).define( m_norm.betweenness(m_norm.betweenness_raw / max_betweenness))Stage 2: Rules — compliance flags and composite risk tier
Seven derived Relationships and Properties flag compliance issues. Each rule is a pure logic derivation using model.where(...).define(...):
Machine.is_overdue_maintenance = model.Relationship( f"{Machine} is overdue maintenance")model.where( Machine.remaining_useful_life < floats.float(Machine.maintenance_duration_hours)).define(Machine.is_overdue_maintenance())
Machine.is_chronic_downtime = model.Relationship(f"{Machine} has chronic downtime")model.where( Machine.downtime_event_count > CHRONIC_DOWNTIME_THRESHOLD).define(Machine.is_chronic_downtime())Individual flags are chained into a composite risk tier using model.not_() for negation. This exhaustively enumerates all eight combinations of three boolean flags:
Machine.risk_tier = model.Property(f"{Machine} has risk tier {String:risk_tier}")
# Critical: all 3 flags.model.where( Machine.is_chronic_downtime(), Machine.is_high_risk(), Machine.is_overdue_maintenance(),).define(Machine.risk_tier("Critical"))
# Elevated: exactly 2 of 3 flags.model.where( Machine.is_chronic_downtime(), Machine.is_high_risk(), model.not_(Machine.is_overdue_maintenance()),).define(Machine.risk_tier("Elevated"))Stage 3: Define decision variables, constraints, and objective
Three binary decision variables control the schedule: whether to maintain a machine in a period, whether it remains vulnerable, and whether a technician is assigned. The formulation includes four standard constraints (cumulative coverage, assignment linkage, technician capacity, parts/bay capacity) plus a hard constraint from Stage 2’s overdue flag:
maintained_by_deadline = ( sum(MachinePeriod_overdue.x_maintain) .where( MachinePeriod_overdue.machine(Machine_overdue), MachinePeriod_overdue.period(Period_overdue), Period_overdue.pid <= OVERDUE_DEADLINE, ) .per(Machine_overdue))problem.satisfy( model.require(maintained_by_deadline >= 1).where( Machine_overdue.is_overdue_maintenance() ))The objective minimizes expected total cost with three components. The failure cost term incorporates per-period failure predictions from Stage 0 and betweenness centrality from Stage 1, making it more expensive to leave bottleneck machines vulnerable in periods where their predicted failure probability is highest:
failure_cost = sum( MachinePeriod_outer.x_vulnerable * MachinePeriod_outer.predicted_fp * Machine_obj.estimated_parts_cost * Machine_obj.criticality * (1 + CENTRALITY_WEIGHT * Machine_obj.betweenness)).where( MachinePeriod_outer.machine(Machine_obj), MachinePeriod_outer.period(Period_outer))Solve and extract results
The model is solved using the HiGHS solver with a two-minute time limit. Assignment decisions are parsed from the solution to build the maintenance schedule:
problem.solve("highs", time_limit_sec=120)si = problem.solve_info()assert si.termination_status == "OPTIMAL"Stage 4: Resilience analysis and cross-training
After solving, the script analyzes qualification coverage by machine type and location. For each machine type, it checks whether all qualified technicians are concentrated in a single location — a geographic single-point-of-failure invisible to the optimizer:
for mtype in machine_types: qual_techs = qualifications_df[ qualifications_df["machine_type"] == mtype ]["technician_id"].tolist() tech_info = technicians_df[technicians_df["technician_id"].isin(qual_techs)] locations = tech_info["base_location"].unique().tolist() if len(locations) == 1: concentrated_types.append((mtype, locations[0], len(qual_techs)))For concentrated types, the script queries training_options.csv to recommend the cheapest candidate at a different location, producing a specific, costed action item (e.g., “Cross-train T006 for Turbine at $3,200 / 5 weeks”).
Customize this template
- Adjust centrality weight via
CENTRALITY_WEIGHTto control how strongly graph bottleneck scores influence scheduling priority. - Change the overdue deadline via
OVERDUE_DEADLINEto give more or fewer periods for overdue machines. - Extend the planning horizon by adding more periods to the availability and failure prediction data and increasing
PERIOD_HORIZON. - Adjust capacity limits via
PARTS_CAPACITY_PER_PERIODto see how tighter constraints shift scheduling priorities. - Tune travel cost via
TRAVEL_COST_PER_HOURto control preference for local vs. cross-facility assignments. - Add rule thresholds — adjust
failure_probability > 0.3orcriticality >= 4in the high-risk rule to match your risk tolerance. - Change chronic downtime threshold via
CHRONIC_DOWNTIME_THRESHOLDto control which machines are flagged. - Add sensor types — extend
sensors.csvwith new sensor types and adjustsensor_readings.csvwith corresponding measurements. - Add training options — extend
training_options.csvto explore different cross-training strategies.
Troubleshooting
Status: INFEASIBLE
- The overdue-maintenance constraint requires certain machines to be scheduled in early periods. If technician capacity is too tight, this can cause infeasibility.
- Try increasing
OVERDUE_DEADLINEfrom 2 to 3, or increasePARTS_CAPACITY_PER_PERIOD. - Check that technician hours capacity across all periods can accommodate all machines.
All machines maintained in period 1
- The solver minimizes total cost. If capacity allows, it may schedule all maintenance early to avoid vulnerability costs.
- Tighten
PARTS_CAPACITY_PER_PERIODto spread maintenance across periods.
Graph shows 0 edges
- This means no two machines share a qualified technician. Check that
qualifications.csvhas overlapping machine types across technicians. - The graph edge construction uses type-based joins: two machines connect if any technician is qualified for both their
machine_typevalues.
input definition is too large
- This occurs with large cross-products. The qualification-filtered assignment space avoids this issue for the default 30-machine dataset.
- If you scale up significantly, consider reducing data size or querying solver
results via
Variable.values(...)instead of broadmodel.select(...)patterns.
ModuleNotFoundError
- Make sure you activated the virtual environment and ran
python -m pip install .from the template directory. - The
pyproject.tomldeclares the required dependencies.
Connection or authentication errors
- Run
rai initto configure your Snowflake connection. - Verify that the RAI Native App is installed and your user has the required permissions.
No concentration risk detected in Stage 4
- This means all machine types have qualified technicians in multiple locations. The resilience analysis examines geographic diversity of the qualification pool, not individual assignment redundancy.
- Try modifying
qualifications.csvto concentrate a machine type’s technicians in one location to see how the analysis surfaces this risk.