NIOSH Lifting Equation calculator for manual material handling risk assessment.
You are niosh-lifting-calculator - a specialized skill for assessing manual lifting tasks using the NIOSH Lifting Equation.
This skill enables AI-powered lifting risk assessment including:
from dataclasses import dataclass
from typing import Optional
import math
@dataclass
class LiftingTaskParameters:
"""
Input parameters for NIOSH Lifting Equation
"""
# Load characteristics
load_weight_lbs: float # Actual weight being lifted
# Origin parameters
horizontal_origin: float # H: Horizontal distance from midpoint between ankles (inches)
vertical_origin: float # V: Vertical height at origin (inches)
# Destination parameters
horizontal_dest: float # H at destination
vertical_dest: float # V at destination
# Task parameters
vertical_travel: float # D: Vertical travel distance (inches)
asymmetry_angle: float # A: Angle of asymmetry (degrees)
frequency: float # F: Lifts per minute
duration: float # Duration category: 1 (≤1hr), 2 (1-2hr), 8 (2-8hr)
coupling: str # "good", "fair", "poor"
def calculate_rwl(params: LiftingTaskParameters, at_origin: bool = True):
"""
Calculate Recommended Weight Limit using NIOSH equation
RWL = LC x HM x VM x DM x AM x FM x CM
LC = Load Constant = 51 lbs
"""
LC = 51 # Load Constant in lbs
# Select origin or destination for location-specific RWL
H = params.horizontal_origin if at_origin else params.horizontal_dest
V = params.vertical_origin if at_origin else params.vertical_dest
# Horizontal Multiplier (HM)
# HM = 10/H, where H is between 10-25 inches
H = max(10, min(H, 25)) # Clamp to valid range
HM = 10 / H
# Vertical Multiplier (VM)
# VM = 1 - 0.0075|V - 30|
VM = 1 - 0.0075 * abs(V - 30)
VM = max(0, VM) # Cannot be negative
# Distance Multiplier (DM)
# DM = 0.82 + 1.8/D
D = max(10, params.vertical_travel) # Minimum 10 inches
DM = 0.82 + (1.8 / D)
DM = min(1, DM) # Cannot exceed 1
# Asymmetric Multiplier (AM)
# AM = 1 - 0.0032A
A = min(135, params.asymmetry_angle) # Max 135 degrees
AM = 1 - (0.0032 * A)
# Frequency Multiplier (FM)
FM = get_frequency_multiplier(params.frequency, params.duration, V)
# Coupling Multiplier (CM)
CM = get_coupling_multiplier(params.coupling, V)
# Calculate RWL
RWL = LC * HM * VM * DM * AM * FM * CM
return {
"RWL": round(RWL, 1),
"multipliers": {
"LC": LC,
"HM": round(HM, 3),
"VM": round(VM, 3),
"DM": round(DM, 3),
"AM": round(AM, 3),
"FM": round(FM, 3),
"CM": round(CM, 3)
},
"location": "origin" if at_origin else "destination"
}
def get_frequency_multiplier(frequency, duration, vertical):
"""
Frequency Multiplier lookup table
"""
# Simplified FM table
FM_TABLE = {
# (frequency, duration, V>=30): FM value
(0.2, 1, True): 1.00, (0.2, 1, False): 1.00,
(0.5, 1, True): 0.97, (0.5, 1, False): 0.97,
(1, 1, True): 0.94, (1, 1, False): 0.94,
(2, 1, True): 0.91, (2, 1, False): 0.91,
(3, 1, True): 0.88, (3, 1, False): 0.88,
(4, 1, True): 0.84, (4, 1, False): 0.84,
(5, 1, True): 0.80, (5, 1, False): 0.80,
# Add more as needed
}
# Find closest match or interpolate
v_category = vertical >= 30
key = (min(15, frequency), int(duration), v_category)
# Default approximation
if frequency <= 0.2:
return 1.0
elif frequency >= 15:
return 0.0
else:
# Linear approximation
return max(0, 1 - 0.05 * frequency)
def get_coupling_multiplier(coupling, vertical):
"""
Coupling Multiplier based on handle quality
"""
CM_TABLE = {
("good", True): 1.00,
("good", False): 1.00,
("fair", True): 0.95,
("fair", False): 1.00,
("poor", True): 0.90,
("poor", False): 0.90
}
v_category = vertical >= 30
return CM_TABLE.get((coupling.lower(), v_category), 0.90)
def calculate_lifting_index(params: LiftingTaskParameters):
"""
Calculate Lifting Index
LI = Load Weight / RWL
LI interpretation:
- LI ≤ 1.0: Acceptable for most workers
- 1.0 < LI ≤ 3.0: Increased risk, some workers may be at risk
- LI > 3.0: Unacceptable for most workers
"""
# Calculate RWL at both origin and destination
rwl_origin = calculate_rwl(params, at_origin=True)
rwl_dest = calculate_rwl(params, at_origin=False)
# Use the more restrictive (lower) RWL
rwl = min(rwl_origin['RWL'], rwl_dest['RWL'])
limiting_location = "origin" if rwl_origin['RWL'] < rwl_dest['RWL'] else "destination"
# Calculate Lifting Index
li = params.load_weight_lbs / rwl if rwl > 0 else float('inf')
# Risk classification
if li <= 1.0:
risk_level = "LOW"
risk_description = "Task acceptable for most healthy workers"
elif li <= 2.0:
risk_level = "MODERATE"
risk_description = "Increased risk - consider job modifications"
elif li <= 3.0:
risk_level = "HIGH"
risk_description = "High risk - job redesign recommended"
else:
risk_level = "VERY HIGH"
risk_description = "Unacceptable risk - immediate redesign required"
return {
"lifting_index": round(li, 2),
"rwl": rwl,
"rwl_origin": rwl_origin['RWL'],
"rwl_dest": rwl_dest['RWL'],
"limiting_location": limiting_location,
"actual_weight": params.load_weight_lbs,
"risk_level": risk_level,
"risk_description": risk_description,
"multipliers_origin": rwl_origin['multipliers'],
"multipliers_dest": rwl_dest['multipliers']
}
def multi_task_lifting_index(tasks: list):
"""
Calculate Composite Lifting Index for multiple tasks
CLI = LI_max + sum of (LI_adjusted for remaining tasks)
"""
if not tasks:
return None
# Calculate individual LIs
task_results = []
for task in tasks:
result = calculate_lifting_index(task['params'])
result['task_name'] = task.get('name', 'Unnamed')
result['frequency'] = task['params'].frequency
task_results.append(result)
# Sort by LI descending
task_results.sort(key=lambda x: x['lifting_index'], reverse=True)
# Calculate CLI
cli = task_results[0]['lifting_index']
for i in range(1, len(task_results)):
# Frequency adjustment for additional tasks
# Simplified: add fraction of each additional LI
freq_factor = sum(t['frequency'] for t in task_results[:i+1]) / \
sum(t['frequency'] for t in task_results[:i])
cli += task_results[i]['lifting_index'] * (freq_factor - 1) / freq_factor
return {
"composite_lifting_index": round(cli, 2),
"individual_tasks": task_results,
"most_stressful_task": task_results[0]['task_name'],
"risk_level": get_risk_level(cli)
}
def get_risk_level(li):
if li <= 1.0:
return "LOW"
elif li <= 2.0:
return "MODERATE"
elif li <= 3.0:
return "HIGH"
else:
return "VERY HIGH"
def analyze_multipliers(result: dict):
"""
Identify which factors are limiting and provide recommendations
"""
multipliers = result.get('multipliers_origin', {})
limiting_factors = []
# Identify factors below threshold
thresholds = {
"HM": 0.7, # Horizontal distance issue
"VM": 0.8, # Vertical height issue
"DM": 0.8, # Travel distance issue
"AM": 0.8, # Asymmetry issue
"FM": 0.7, # Frequency issue
"CM": 0.9 # Coupling issue
}
recommendations = []
for factor, threshold in thresholds.items():
if factor in multipliers and multipliers[factor] < threshold:
limiting_factors.append(factor)
if factor == "HM":
recommendations.append({
"factor": "Horizontal Distance",
"issue": f"HM = {multipliers[factor]:.2f} - load too far from body",
"recommendations": [
"Move load closer to worker",
"Use conveyors or slides",
"Eliminate obstacles between worker and load",
"Reduce container width"
]
})
elif factor == "VM":
recommendations.append({
"factor": "Vertical Location",
"issue": f"VM = {multipliers[factor]:.2f} - lift height not optimal",
"recommendations": [
"Raise or lower origin/destination",
"Use lift tables or platforms",
"Eliminate floor-level or overhead lifts"
]
})
elif factor == "DM":
recommendations.append({
"factor": "Travel Distance",
"issue": f"DM = {multipliers[factor]:.2f} - excessive vertical travel",
"recommendations": [
"Reduce vertical travel distance",
"Use mechanical assists for long lifts",
"Reposition origin and destination"
]
})
elif factor == "AM":
recommendations.append({
"factor": "Asymmetry",
"issue": f"AM = {multipliers[factor]:.2f} - twisting required",
"recommendations": [
"Reposition load or destination to eliminate twist",
"Use turntables",
"Improve workstation layout"
]
})
elif factor == "FM":
recommendations.append({
"factor": "Frequency",
"issue": f"FM = {multipliers[factor]:.2f} - high lifting frequency",
"recommendations": [
"Reduce lifting frequency",
"Add job rotation",
"Use mechanical assists",
"Add rest breaks"
]
})
elif factor == "CM":
recommendations.append({
"factor": "Coupling",
"issue": f"CM = {multipliers[factor]:.2f} - poor grip/handles",
"recommendations": [
"Add handles to containers",
"Use containers with hand-holds",
"Improve grip surface"
]
})
return {
"limiting_factors": limiting_factors,
"recommendations": recommendations,
"priority": limiting_factors[0] if limiting_factors else None
}
def compare_modifications(baseline: LiftingTaskParameters, modifications: list):
"""
Compare baseline to proposed modifications
"""
baseline_result = calculate_lifting_index(baseline)
comparisons = [{
"scenario": "Baseline",
"changes": None,
"lifting_index": baseline_result['lifting_index'],
"rwl": baseline_result['rwl'],
"risk_level": baseline_result['risk_level'],
"improvement": 0
}]
for mod in modifications:
mod_result = calculate_lifting_index(mod['params'])
improvement = (baseline_result['lifting_index'] - mod_result['lifting_index']) / \
baseline_result['lifting_index'] * 100
comparisons.append({
"scenario": mod['name'],
"changes": mod.get('description', ''),
"lifting_index": mod_result['lifting_index'],
"rwl": mod_result['rwl'],
"risk_level": mod_result['risk_level'],
"improvement": round(improvement, 1)
})
# Sort by improvement
comparisons.sort(key=lambda x: x['lifting_index'])
return {
"comparisons": comparisons,
"best_scenario": comparisons[0]['scenario'],
"best_improvement": comparisons[0]['improvement']
}
This skill integrates with the following processes:
ergonomic-risk-assessment.jsworkstation-design-optimization.js{
"lifting_index": 1.8,
"rwl": 28.3,
"actual_weight": 51,
"risk_level": "MODERATE",
"limiting_factors": ["HM", "AM"],
"recommendations": [
{
"factor": "Horizontal Distance",
"recommendations": ["Move load closer to worker"]
}
],
"improvement_options": [
{
"scenario": "Add lift table",
"improvement": 45
}
]
}