SafePath Training — Phase 2A: ChemIQ Integration
Overview
This document covers the HazCom Extension layer that activates when a company has both SafePath and ChemIQ (or HazCom Plan) modules enabled. It adds:
- Auto-generated HazCom courses from approved HazCom plan + SDS data
- Chemical-change retraining (new chemical added → triggers retraining)
- Plan-version retraining (new plan version published → triggers retraining)
- SDS-linked training content (references specific SDS sections and GHS data)
- GHS pictogram training (auto-generated lessons from inventory classifications)
Prerequisites:
- Phase 1 (1A–1F) complete — all core SafePath tables, services, and UI exist
- ChemIQ module with
chemiq_inventory,company_product_catalog,chemiq_sds_documentstables - HazCom Plan module with
hazcom_plans,hazcom_plan_sectionstables
Tier gating: Standard + Pro plans only (via SafePathCapability entitlements).
Files to create/modify:
| Action | File |
|---|---|
| Create | app/services/safepath/hazcom_integration_service.py |
| Create | app/schemas/safepath/hazcom_integration.py |
| Create | app/api/v1/safepath/hazcom_integration.py |
| Modify | app/services/hazcom/hazcom_plan_service.py (add hook in publish_plan) |
| Modify | app/services/chemiq/chemiq_inventory_service.py (add hook on chemical add) |
| Create | src/pages/safepath/courses/hazcom-generate.tsx |
| Modify | src/pages/safepath/courses/index.tsx (add "Generate from HazCom Plan" button) |
| Modify | src/services/api/safepath.api.ts (add API methods) |
| Modify | src/types/safepath.ts (add TypeScript interfaces) |
1. Data Flow & Architecture
1.1 How Modules Connect
┌─────────────────────────────────────────────────────────────────┐
│ HazCom Plan Module │
│ │
│ HazComPlan (approved) ──publish_plan()──→ HazComPlan (active) │
│ │ │ │
│ │ has sections (7 OSHA) │ HOOK │
│ ▼ ▼ │
│ HazComPlanSection SafePath Integration │
│ (training section content) (generate course outline) │
└─────────────────────────────────────────────────────────────────┘
│
│ references chemicals from
▼
┌─────────────────────────────────────────────────────────────────┐
│ ChemIQ Module │
│ │
│ ChemIQInventory ──→ CompanyProductCatalog ──→ SDSDocument │
│ (site chemicals) (company products) (global SDS) │
│ │ │ │
│ │ hazard_categories │ sections │
│ │ signal_word │ │
│ ▼ ▼ │
│ GHS pictograms SDSHazardInfo │
│ SDSComposition │
│ SDSSection (4-16) │
└─────────────────────────────────────────────────────────────────┘
│
│ feeds into
▼
┌─────────────────────────────────────────────────────────────────┐
│ SafePath Training Module │
│ │
│ SafePathCourse (plan_version_id ──→ HazComPlan) │
│ │ │
│ ├── SafePathLesson (auto-generated from SDS sections) │
│ │ - GHS overview lesson │
│ │ - PPE requirements lesson (SDS Section 8) │
│ │ - Emergency procedures lesson (SDS Section 6) │
│ │ - First aid measures lesson (SDS Section 4) │
│ │ │
│ └── SafePathQuiz (auto-generated from plan content) │
│ - Hazard identification questions │
│ - PPE selection questions │
│ - Emergency procedure questions │
└─────────────────────────────────────────────────────────────────┘
1.2 Trigger Points
| Trigger | Source Module | Action in SafePath |
|---|---|---|
| HazCom Plan published | HazCom Plan | Generate draft course outline from plan + SDS data |
| New chemical added to site | ChemIQ | Create retraining assignment for affected site employees |
| New plan version published | HazCom Plan | Create retraining assignment with updated content |
| SDS updated for existing chemical | ChemIQ | Flag course content for review (advisory, not auto-assign) |
1.3 Existing Model Support
The SafePath data model already has fields prepared for this integration:
SafePathCourse.plan_version_id— nullable UUID linking course to a HazCom plan versionSafePathAutoAssignmentRule.trigger_type = 'chemical_added'— already in the trigger_type check constraintSafePathAssignment.auto_rule_id— nullable FK to track which auto-rule created an assignment
2. Pydantic Schemas
File: app/schemas/safepath/hazcom_integration.py
"""SafePath HazCom Integration Schemas
Schemas for auto-generating courses from HazCom plans and SDS data,
and for chemical-change retraining triggers.
"""
from datetime import datetime
from typing import Optional, List
from uuid import UUID
from pydantic import BaseModel, Field, ConfigDict
# ============================================================================
# Course Generation Request/Response
# ============================================================================
class HazComCourseGenerateRequest(BaseModel):
"""Request to generate a draft course from an approved/active HazCom plan."""
plan_id: UUID = Field(..., description="ID of the HazCom plan to generate course from")
site_id: Optional[UUID] = Field(None, description="Limit chemical scope to a specific site")
include_ghs_overview: bool = Field(default=True, description="Include GHS classification overview lesson")
include_ppe_lesson: bool = Field(default=True, description="Include PPE requirements lesson (SDS Section 8)")
include_emergency_lesson: bool = Field(default=True, description="Include emergency procedures lesson (SDS Section 6)")
include_first_aid_lesson: bool = Field(default=True, description="Include first aid measures lesson (SDS Section 4)")
include_quiz: bool = Field(default=True, description="Auto-generate quiz from plan content")
quiz_question_count: int = Field(default=10, ge=5, le=30, description="Number of quiz questions to generate")
title_override: Optional[str] = Field(None, max_length=255, description="Custom course title (default: auto-generated)")
class GeneratedLessonPreview(BaseModel):
"""Preview of a lesson that will be generated."""
title: str
lesson_type: str # text, pdf, slides
content_summary: str # Brief description of what will be in the lesson
sds_sections_used: List[str] # e.g., ["Section 4", "Section 8"]
chemical_count: int # Number of chemicals referenced
class GeneratedQuizPreview(BaseModel):
"""Preview of quiz questions that will be generated."""
question_count: int
question_types: List[str] # e.g., ["mcq_single", "true_false"]
topics_covered: List[str] # e.g., ["GHS pictograms", "PPE requirements"]
class HazComCoursePreviewResponse(BaseModel):
"""Preview of what will be generated — shown to user before confirming."""
plan_id: UUID
plan_name: str
plan_version: int
site_name: Optional[str] = None
suggested_title: str
chemical_count: int
hazard_categories_found: List[str]
ghs_pictograms_found: List[str]
lessons: List[GeneratedLessonPreview]
quiz: Optional[GeneratedQuizPreview] = None
estimated_duration_minutes: int
class HazComCourseGenerateResponse(BaseModel):
"""Response after course generation — returns the created draft course."""
course_id: UUID
title: str
status: str # Always "draft"
lesson_count: int
quiz_question_count: int
plan_version_id: UUID
message: str
# ============================================================================
# Chemical Change Retraining
# ============================================================================
class ChemicalChangeEvent(BaseModel):
"""Event payload when a chemical is added/modified in ChemIQ."""
event_type: str = Field(..., pattern="^(chemical_added|chemical_removed|sds_updated)$")
company_id: UUID
site_id: UUID
product_catalog_id: UUID
product_name: str
hazard_categories: List[str] = []
ghs_pictograms: List[str] = []
sds_id: Optional[UUID] = None
class RetrainingTriggerResponse(BaseModel):
"""Response after processing a chemical change event."""
assignments_created: int
users_affected: List[str] # user full_names
course_id: Optional[UUID] = None
course_title: Optional[str] = None
message: str
# ============================================================================
# Plan Version Retraining
# ============================================================================
class PlanVersionRetrainingRequest(BaseModel):
"""Request to create retraining assignments after a new plan version is published."""
plan_id: UUID
previous_plan_id: Optional[UUID] = None # For delta comparison
auto_assign: bool = Field(default=False, description="Immediately assign or just flag for review")
due_date_offset_days: int = Field(default=30, ge=7, le=180)
class PlanVersionRetrainingResponse(BaseModel):
"""Response after processing plan version retraining."""
new_course_version_id: Optional[UUID] = None
assignments_created: int
message: str
# ============================================================================
# SDS-Linked Content Helpers
# ============================================================================
class SDSContentBlock(BaseModel):
"""A block of training content derived from a specific SDS section."""
sds_id: UUID
product_name: str
section_number: int
section_title: str
content_html: str # Rendered HTML for lesson display
hazard_pictograms: List[str] = [] # GHS pictogram codes
class GHSPictogramTrainingItem(BaseModel):
"""Training item for a specific GHS pictogram found in inventory."""
pictogram_code: str # e.g., "GHS01", "GHS02"
pictogram_name: str # e.g., "Exploding Bomb", "Flame"
hazard_description: str
chemicals_with_pictogram: List[str] # Product names
chemical_count: int
precautionary_statements: List[str] # P-codes from SDS
class SiteHazardSummary(BaseModel):
"""Summary of hazards at a site — used for course generation."""
site_id: UUID
site_name: str
total_chemicals: int
hazard_categories: List[str]
ghs_pictograms: List[str]
signal_words: List[str] # DANGER, WARNING
chemicals_by_pictogram: dict # { "GHS01": ["Chemical A", "Chemical B"] }
3. HazCom Integration Service
File: app/services/safepath/hazcom_integration_service.py
"""SafePath HazCom Integration Service
Generates training courses from HazCom plan data and SDS information.
Handles chemical-change retraining triggers and plan-version retraining.
"""
import uuid
from datetime import datetime, date, timedelta
from typing import Optional, List, Tuple
from uuid import UUID
from sqlalchemy import and_, func
from sqlalchemy.orm import Session, joinedload
from app.db.models.safepath import (
SafePathCourse, SafePathLesson, SafePathQuiz, SafePathQuizQuestion,
SafePathAssignment, SafePathAutoAssignmentRule, SafePathCourseCategory,
SafePathAuditLog,
)
from app.db.models.hazcom_plan import HazComPlan, HazComPlanSection
from app.db.models.chemiq_inventory import ChemIQInventory
from app.db.models.chemiq_product_catalog import CompanyProductCatalog
from app.db.models.chemiq_sds import SDSDocument, SDSHazardInfo, SDSComposition, SDSSection
from app.db.models.company import CompanySite
from app.db.models.user import User, CompanyUser
from app.schemas.safepath.hazcom_integration import (
HazComCourseGenerateRequest,
HazComCoursePreviewResponse,
HazComCourseGenerateResponse,
GeneratedLessonPreview,
GeneratedQuizPreview,
ChemicalChangeEvent,
RetrainingTriggerResponse,
PlanVersionRetrainingRequest,
PlanVersionRetrainingResponse,
SDSContentBlock,
GHSPictogramTrainingItem,
SiteHazardSummary,
)
# GHS pictogram lookup
GHS_PICTOGRAMS = {
"GHS01": {"name": "Exploding Bomb", "hazard": "Explosives, self-reactives, organic peroxides"},
"GHS02": {"name": "Flame", "hazard": "Flammable gases, aerosols, liquids, solids"},
"GHS03": {"name": "Flame Over Circle", "hazard": "Oxidizers"},
"GHS04": {"name": "Gas Cylinder", "hazard": "Compressed, liquefied, dissolved gases"},
"GHS05": {"name": "Corrosion", "hazard": "Corrosive to metals, skin, eyes"},
"GHS06": {"name": "Skull and Crossbones", "hazard": "Acute toxicity (fatal or toxic)"},
"GHS07": {"name": "Exclamation Mark", "hazard": "Irritant, sensitizer, narcotic effects"},
"GHS08": {"name": "Health Hazard", "hazard": "Carcinogen, mutagen, reproductive toxicity, organ toxicity"},
"GHS09": {"name": "Environment", "hazard": "Aquatic toxicity"},
}
class HazComIntegrationService:
"""Service for generating SafePath courses from HazCom plan and SDS data."""
def __init__(self, db: Session):
self.db = db
# ========================================================================
# Course Generation
# ========================================================================
def preview_course_generation(
self,
request: HazComCourseGenerateRequest,
company_id: UUID,
) -> Optional[HazComCoursePreviewResponse]:
"""Preview what a generated course would look like before creating it.
Reads the HazCom plan, collects SDS data for chemicals in scope,
and returns a preview of lessons, quiz, and hazard summary.
"""
# 1. Load the HazCom plan
plan = self.db.query(HazComPlan).filter(
HazComPlan.plan_id == request.plan_id,
HazComPlan.company_id == company_id,
HazComPlan.version_status.in_(["approved", "active"]),
).first()
if not plan:
return None
# 2. Get site info
site_name = None
site_id = request.site_id or plan.site_id
if site_id:
site = self.db.query(CompanySite).filter(
CompanySite.site_id == site_id
).first()
site_name = site.site_name if site else None
# 3. Collect chemicals and SDS data for the site
hazard_summary = self._get_site_hazard_summary(company_id, site_id, site_name or "All Sites")
# 4. Build lesson previews
lessons = self._build_lesson_previews(request, hazard_summary)
# 5. Build quiz preview
quiz = None
if request.include_quiz:
quiz = GeneratedQuizPreview(
question_count=request.quiz_question_count,
question_types=["mcq_single", "true_false"],
topics_covered=self._get_quiz_topics(request, hazard_summary),
)
# 6. Estimate duration
estimated_minutes = len(lessons) * 10 + (15 if quiz else 0)
return HazComCoursePreviewResponse(
plan_id=plan.plan_id,
plan_name=plan.plan_name,
plan_version=plan.version_number,
site_name=site_name,
suggested_title=request.title_override or f"HazCom Training — {site_name or 'Company-Wide'} (v{plan.version_number})",
chemical_count=hazard_summary.total_chemicals,
hazard_categories_found=hazard_summary.hazard_categories,
ghs_pictograms_found=hazard_summary.ghs_pictograms,
lessons=lessons,
quiz=quiz,
estimated_duration_minutes=estimated_minutes,
)
def generate_course(
self,
request: HazComCourseGenerateRequest,
company_id: UUID,
user_id: UUID,
) -> Optional[HazComCourseGenerateResponse]:
"""Generate a draft course from an approved/active HazCom plan.
Process:
1. Load plan + SDS data for chemicals in scope
2. Create SafePathCourse with plan_version_id link
3. Generate lessons from SDS sections (GHS overview, PPE, emergency, first aid)
4. Generate quiz questions from plan content
5. Return course in Draft status for trainer review
"""
# 1. Load the plan
plan = self.db.query(HazComPlan).filter(
HazComPlan.plan_id == request.plan_id,
HazComPlan.company_id == company_id,
HazComPlan.version_status.in_(["approved", "active"]),
).first()
if not plan:
return None
site_id = request.site_id or plan.site_id
# 2. Get HazCom category
hazcom_category = self.db.query(SafePathCourseCategory).filter(
SafePathCourseCategory.name == "HazCom / GHS",
SafePathCourseCategory.is_system == True,
).first()
# 3. Collect site hazard data
hazard_summary = self._get_site_hazard_summary(company_id, site_id, "")
sds_content_blocks = self._get_sds_content_blocks(company_id, site_id)
# 4. Create draft course
site_name = self._get_site_name(site_id) if site_id else "Company-Wide"
course = SafePathCourse(
course_id=uuid.uuid4(),
company_id=company_id,
title=request.title_override or f"HazCom Training — {site_name} (v{plan.version_number})",
description=f"Auto-generated HazCom training course from plan '{plan.plan_name}' version {plan.version_number}. "
f"Covers {hazard_summary.total_chemicals} chemicals, "
f"{len(hazard_summary.ghs_pictograms)} GHS pictogram categories.",
category_id=hazcom_category.category_id if hazcom_category else None,
estimated_duration_minutes=0, # Updated after lessons
passing_score_percent=80,
max_retakes=3,
status="draft",
version_number=1,
plan_version_id=plan.plan_id,
locale="en",
created_by=user_id,
)
self.db.add(course)
# 5. Generate lessons
lesson_count = 0
sort_order = 0
if request.include_ghs_overview:
sort_order += 1
lesson_count += 1
self._create_ghs_overview_lesson(course.course_id, sort_order, hazard_summary)
if request.include_ppe_lesson:
sort_order += 1
lesson_count += 1
self._create_sds_section_lesson(
course.course_id, sort_order,
"Personal Protective Equipment (PPE) Requirements",
section_number=8,
sds_content_blocks=sds_content_blocks,
)
if request.include_emergency_lesson:
sort_order += 1
lesson_count += 1
self._create_sds_section_lesson(
course.course_id, sort_order,
"Emergency Procedures & Spill Response",
section_number=6,
sds_content_blocks=sds_content_blocks,
)
if request.include_first_aid_lesson:
sort_order += 1
lesson_count += 1
self._create_sds_section_lesson(
course.course_id, sort_order,
"First Aid Measures",
section_number=4,
sds_content_blocks=sds_content_blocks,
)
# 6. Generate quiz
quiz_question_count = 0
if request.include_quiz:
quiz_question_count = self._create_hazcom_quiz(
course.course_id, hazard_summary, request.quiz_question_count
)
# 7. Update estimated duration
course.estimated_duration_minutes = lesson_count * 10 + (15 if quiz_question_count > 0 else 0)
# 8. Audit log
self._log_event(
company_id, user_id, "course_generated_from_hazcom",
"course", course.course_id,
{"plan_id": str(plan.plan_id), "plan_version": plan.version_number,
"lesson_count": lesson_count, "quiz_questions": quiz_question_count},
)
self.db.commit()
self.db.refresh(course)
return HazComCourseGenerateResponse(
course_id=course.course_id,
title=course.title,
status="draft",
lesson_count=lesson_count,
quiz_question_count=quiz_question_count,
plan_version_id=plan.plan_id,
message=f"Draft course created with {lesson_count} lessons and {quiz_question_count} quiz questions. "
f"Review and publish when ready.",
)
# ========================================================================
# Chemical-Change Retraining
# ========================================================================
def handle_chemical_change(
self,
event: ChemicalChangeEvent,
) -> RetrainingTriggerResponse:
"""Process a chemical-change event from ChemIQ.
Called when a new chemical is added to a site. Checks for active
auto-assignment rules with trigger_type='chemical_added' and creates
assignments for affected site employees.
"""
# 1. Find active auto-assignment rules for this company with chemical_added trigger
rules = self.db.query(SafePathAutoAssignmentRule).filter(
SafePathAutoAssignmentRule.company_id == event.company_id,
SafePathAutoAssignmentRule.trigger_type == "chemical_added",
SafePathAutoAssignmentRule.is_active == True,
).options(
joinedload(SafePathAutoAssignmentRule.course)
).all()
if not rules:
return RetrainingTriggerResponse(
assignments_created=0,
users_affected=[],
message="No active chemical_added auto-assignment rules found.",
)
# 2. Get employees at the affected site
site_users = self.db.query(User).join(
CompanyUser, CompanyUser.user_id == User.user_id
).filter(
CompanyUser.company_id == event.company_id,
CompanyUser.status == "active",
).all()
# Filter by site if rule config specifies site matching
assignments_created = 0
users_affected = []
course_used = None
for rule in rules:
# Check trigger_config for site filtering
config = rule.trigger_config or {}
target_site_ids = config.get("site_ids", [])
# If rule has site filter, only assign if matching
if target_site_ids and str(event.site_id) not in target_site_ids:
continue
course = rule.course
if not course or course.status != "published":
continue
course_used = course
due_date = date.today() + timedelta(days=rule.due_date_offset_days)
for user in site_users:
# Check if user already has an active assignment for this course
existing = self.db.query(SafePathAssignment).filter(
SafePathAssignment.course_id == course.course_id,
SafePathAssignment.assigned_to == user.user_id,
SafePathAssignment.status.in_(["assigned", "in_progress"]),
).first()
if existing:
continue # Skip — already assigned
assignment = SafePathAssignment(
assignment_id=uuid.uuid4(),
company_id=event.company_id,
course_id=course.course_id,
assigned_to=user.user_id,
due_date=due_date,
priority="normal",
status="assigned",
auto_rule_id=rule.rule_id,
notes=f"Auto-assigned: New chemical '{event.product_name}' added to site.",
)
self.db.add(assignment)
assignments_created += 1
users_affected.append(user.full_name)
self.db.commit()
return RetrainingTriggerResponse(
assignments_created=assignments_created,
users_affected=users_affected,
course_id=course_used.course_id if course_used else None,
course_title=course_used.title if course_used else None,
message=f"Created {assignments_created} retraining assignments for chemical '{event.product_name}'.",
)
# ========================================================================
# Plan-Version Retraining
# ========================================================================
def handle_plan_version_published(
self,
request: PlanVersionRetrainingRequest,
company_id: UUID,
user_id: UUID,
) -> PlanVersionRetrainingResponse:
"""Handle a new HazCom plan version being published.
Options:
1. Create a new version of the existing HazCom course (linked to old plan)
2. Auto-assign the new version to all employees who completed the old version
"""
plan = self.db.query(HazComPlan).filter(
HazComPlan.plan_id == request.plan_id,
HazComPlan.company_id == company_id,
HazComPlan.version_status == "active",
).first()
if not plan:
return PlanVersionRetrainingResponse(
assignments_created=0,
message="Plan not found or not in active status.",
)
# Find existing course linked to a previous version of this plan
existing_course = None
if request.previous_plan_id:
existing_course = self.db.query(SafePathCourse).filter(
SafePathCourse.company_id == company_id,
SafePathCourse.plan_version_id == request.previous_plan_id,
SafePathCourse.status == "published",
).first()
new_course_version_id = None
assignments_created = 0
if existing_course and request.auto_assign:
# Create new version of the course
from app.services.safepath.course_service import CourseService
course_service = CourseService(self.db)
new_course = course_service.create_new_version(
existing_course.course_id, company_id, user_id
)
if new_course:
new_course.plan_version_id = plan.plan_id
new_course_version_id = new_course.course_id
# Find users who completed the old version
completed_users = self.db.query(SafePathAssignment.assigned_to).filter(
SafePathAssignment.course_id == existing_course.course_id,
SafePathAssignment.status == "completed",
).distinct().all()
due_date = date.today() + timedelta(days=request.due_date_offset_days)
for (uid,) in completed_users:
assignment = SafePathAssignment(
assignment_id=uuid.uuid4(),
company_id=company_id,
course_id=new_course.course_id,
assigned_to=uid,
due_date=due_date,
priority="normal",
status="assigned",
notes=f"Retraining: HazCom plan updated to version {plan.version_number}.",
)
self.db.add(assignment)
assignments_created += 1
self._log_event(
company_id, user_id, "plan_version_retraining",
"course", new_course.course_id,
{"plan_id": str(plan.plan_id), "plan_version": plan.version_number,
"assignments_created": assignments_created},
)
self.db.commit()
return PlanVersionRetrainingResponse(
new_course_version_id=new_course_version_id,
assignments_created=assignments_created,
message=f"Plan version retraining processed. {assignments_created} assignments created."
if assignments_created > 0
else "Plan version retraining flagged for review. No auto-assignments created.",
)
# ========================================================================
# SDS Content Helpers
# ========================================================================
def get_site_hazard_summary(
self,
company_id: UUID,
site_id: UUID,
) -> Optional[SiteHazardSummary]:
"""Public method to get a hazard summary for a site."""
site = self.db.query(CompanySite).filter(
CompanySite.site_id == site_id,
CompanySite.company_id == company_id,
).first()
if not site:
return None
return self._get_site_hazard_summary(company_id, site_id, site.site_name)
def get_ghs_pictogram_training_items(
self,
company_id: UUID,
site_id: Optional[UUID] = None,
) -> List[GHSPictogramTrainingItem]:
"""Get GHS pictogram training items based on chemicals at a site.
Returns a list of GHS pictograms found in the site's chemical inventory,
along with which chemicals have each pictogram and precautionary statements.
"""
# Get all products with SDS hazard info
query = self.db.query(CompanyProductCatalog).filter(
CompanyProductCatalog.company_id == company_id,
)
if site_id:
product_ids = self.db.query(ChemIQInventory.company_product_id).filter(
ChemIQInventory.company_id == company_id,
ChemIQInventory.site_id == site_id,
).distinct()
query = query.filter(CompanyProductCatalog.company_product_id.in_(product_ids))
products = query.options(
joinedload(CompanyProductCatalog.current_sds)
).all()
# Build pictogram → chemicals mapping
pictogram_map = {} # { "GHS01": { "chemicals": [], "p_codes": set() } }
for product in products:
if not product.current_sds_id:
continue
hazard_info = self.db.query(SDSHazardInfo).filter(
SDSHazardInfo.sds_id == product.current_sds_id
).first()
if not hazard_info or not hazard_info.ghs_pictograms:
continue
pictograms = hazard_info.ghs_pictograms # JSONB list
p_codes = hazard_info.precautionary_codes or []
for pic in pictograms:
if pic not in pictogram_map:
pictogram_map[pic] = {"chemicals": [], "p_codes": set()}
pictogram_map[pic]["chemicals"].append(product.product_name)
pictogram_map[pic]["p_codes"].update(p_codes)
# Convert to response items
items = []
for code, data in sorted(pictogram_map.items()):
ghs_info = GHS_PICTOGRAMS.get(code, {"name": code, "hazard": "Unknown"})
items.append(GHSPictogramTrainingItem(
pictogram_code=code,
pictogram_name=ghs_info["name"],
hazard_description=ghs_info["hazard"],
chemicals_with_pictogram=data["chemicals"],
chemical_count=len(data["chemicals"]),
precautionary_statements=sorted(data["p_codes"]),
))
return items
# ========================================================================
# Private Helpers
# ========================================================================
def _get_site_hazard_summary(
self, company_id: UUID, site_id: Optional[UUID], site_name: str
) -> SiteHazardSummary:
"""Collect hazard summary from ChemIQ inventory for a site."""
query = self.db.query(CompanyProductCatalog).filter(
CompanyProductCatalog.company_id == company_id,
)
if site_id:
product_ids = self.db.query(ChemIQInventory.company_product_id).filter(
ChemIQInventory.company_id == company_id,
ChemIQInventory.site_id == site_id,
).distinct()
query = query.filter(CompanyProductCatalog.company_product_id.in_(product_ids))
products = query.all()
hazard_categories = set()
ghs_pictograms = set()
signal_words = set()
chemicals_by_pictogram = {}
for product in products:
# From product catalog
if product.hazard_categories:
hazard_categories.update(product.hazard_categories)
if product.signal_word:
signal_words.add(product.signal_word)
# From SDS hazard info
if product.current_sds_id:
hazard_info = self.db.query(SDSHazardInfo).filter(
SDSHazardInfo.sds_id == product.current_sds_id
).first()
if hazard_info and hazard_info.ghs_pictograms:
for pic in hazard_info.ghs_pictograms:
ghs_pictograms.add(pic)
if pic not in chemicals_by_pictogram:
chemicals_by_pictogram[pic] = []
chemicals_by_pictogram[pic].append(product.product_name)
return SiteHazardSummary(
site_id=site_id or uuid.UUID(int=0),
site_name=site_name,
total_chemicals=len(products),
hazard_categories=sorted(hazard_categories),
ghs_pictograms=sorted(ghs_pictograms),
signal_words=sorted(signal_words),
chemicals_by_pictogram=chemicals_by_pictogram,
)
def _get_sds_content_blocks(
self, company_id: UUID, site_id: Optional[UUID]
) -> List[SDSContentBlock]:
"""Get SDS content blocks for sections 4, 6, 8 for all site chemicals."""
query = self.db.query(CompanyProductCatalog).filter(
CompanyProductCatalog.company_id == company_id,
CompanyProductCatalog.current_sds_id.isnot(None),
)
if site_id:
product_ids = self.db.query(ChemIQInventory.company_product_id).filter(
ChemIQInventory.company_id == company_id,
ChemIQInventory.site_id == site_id,
).distinct()
query = query.filter(CompanyProductCatalog.company_product_id.in_(product_ids))
products = query.all()
blocks = []
for product in products:
sds_sections = self.db.query(SDSSection).filter(
SDSSection.sds_id == product.current_sds_id,
SDSSection.section_number.in_([4, 6, 8]),
).all()
hazard_info = self.db.query(SDSHazardInfo).filter(
SDSHazardInfo.sds_id == product.current_sds_id
).first()
pictograms = hazard_info.ghs_pictograms if hazard_info else []
for section in sds_sections:
content = section.structured_content or {}
content_html = self._render_sds_section_html(product.product_name, section)
blocks.append(SDSContentBlock(
sds_id=product.current_sds_id,
product_name=product.product_name,
section_number=section.section_number,
section_title=section.section_title,
content_html=content_html,
hazard_pictograms=pictograms,
))
return blocks
def _build_lesson_previews(
self, request: HazComCourseGenerateRequest, summary: SiteHazardSummary
) -> List[GeneratedLessonPreview]:
"""Build preview list of lessons that would be generated."""
lessons = []
if request.include_ghs_overview:
lessons.append(GeneratedLessonPreview(
title="GHS Classification Overview",
lesson_type="text",
content_summary=f"Overview of {len(summary.ghs_pictograms)} GHS pictogram categories "
f"found across {summary.total_chemicals} chemicals at this site.",
sds_sections_used=["Hazard Info"],
chemical_count=summary.total_chemicals,
))
if request.include_ppe_lesson:
lessons.append(GeneratedLessonPreview(
title="Personal Protective Equipment (PPE) Requirements",
lesson_type="text",
content_summary="PPE requirements compiled from SDS Section 8 (Exposure Controls / PPE) "
"for all chemicals at this site.",
sds_sections_used=["Section 8"],
chemical_count=summary.total_chemicals,
))
if request.include_emergency_lesson:
lessons.append(GeneratedLessonPreview(
title="Emergency Procedures & Spill Response",
lesson_type="text",
content_summary="Accidental release measures and emergency procedures from SDS Section 6 "
"for all chemicals at this site.",
sds_sections_used=["Section 6"],
chemical_count=summary.total_chemicals,
))
if request.include_first_aid_lesson:
lessons.append(GeneratedLessonPreview(
title="First Aid Measures",
lesson_type="text",
content_summary="First aid measures from SDS Section 4 for all chemicals at this site.",
sds_sections_used=["Section 4"],
chemical_count=summary.total_chemicals,
))
return lessons
def _get_quiz_topics(
self, request: HazComCourseGenerateRequest, summary: SiteHazardSummary
) -> List[str]:
"""Determine quiz topics based on request and hazard summary."""
topics = []
if request.include_ghs_overview:
topics.append("GHS pictogram identification")
topics.append("Hazard categories")
if request.include_ppe_lesson:
topics.append("PPE selection")
if request.include_emergency_lesson:
topics.append("Emergency procedures")
topics.append("Spill response")
if request.include_first_aid_lesson:
topics.append("First aid measures")
return topics
def _create_ghs_overview_lesson(
self, course_id: UUID, sort_order: int, summary: SiteHazardSummary
):
"""Create a GHS overview lesson with pictogram information."""
content = {
"type": "hazcom_ghs_overview",
"pictograms": [],
"hazard_categories": summary.hazard_categories,
"signal_words": summary.signal_words,
}
for pic_code in summary.ghs_pictograms:
pic_info = GHS_PICTOGRAMS.get(pic_code, {"name": pic_code, "hazard": "Unknown"})
chemicals = summary.chemicals_by_pictogram.get(pic_code, [])
content["pictograms"].append({
"code": pic_code,
"name": pic_info["name"],
"hazard": pic_info["hazard"],
"chemicals": chemicals[:10], # Limit to first 10 for readability
"total_chemicals": len(chemicals),
})
lesson = SafePathLesson(
lesson_id=uuid.uuid4(),
course_id=course_id,
title="GHS Classification Overview",
lesson_type="text",
content=content,
sort_order=sort_order,
completion_threshold_percent=80,
locale="en",
)
self.db.add(lesson)
def _create_sds_section_lesson(
self,
course_id: UUID,
sort_order: int,
title: str,
section_number: int,
sds_content_blocks: List[SDSContentBlock],
):
"""Create a lesson from a specific SDS section across all chemicals."""
relevant_blocks = [b for b in sds_content_blocks if b.section_number == section_number]
content = {
"type": f"hazcom_sds_section_{section_number}",
"section_number": section_number,
"chemicals": [],
}
for block in relevant_blocks:
content["chemicals"].append({
"product_name": block.product_name,
"sds_id": str(block.sds_id),
"content_html": block.content_html,
"pictograms": block.hazard_pictograms,
})
lesson = SafePathLesson(
lesson_id=uuid.uuid4(),
course_id=course_id,
title=title,
lesson_type="text",
content=content,
sort_order=sort_order,
completion_threshold_percent=80,
locale="en",
)
self.db.add(lesson)
def _create_hazcom_quiz(
self, course_id: UUID, summary: SiteHazardSummary, question_count: int
) -> int:
"""Generate a quiz with questions about GHS, PPE, and emergency procedures."""
quiz = SafePathQuiz(
quiz_id=uuid.uuid4(),
course_id=course_id,
title="HazCom Knowledge Check",
sort_order=999, # Always at end
randomize_questions=True,
)
self.db.add(quiz)
questions_created = 0
# Generate GHS pictogram identification questions
for pic_code in summary.ghs_pictograms[:5]: # Max 5 pictogram questions
if questions_created >= question_count:
break
pic_info = GHS_PICTOGRAMS.get(pic_code)
if not pic_info:
continue
# Create MCQ: "What hazard does the [pictogram] GHS pictogram represent?"
wrong_answers = [
v["hazard"] for k, v in GHS_PICTOGRAMS.items()
if k != pic_code
][:3]
question = SafePathQuizQuestion(
question_id=uuid.uuid4(),
quiz_id=quiz.quiz_id,
question_text=f"What hazard does the {pic_info['name']} ({pic_code}) GHS pictogram represent?",
question_type="mcq_single",
options=[
{"id": "a", "text": pic_info["hazard"], "is_correct": True},
{"id": "b", "text": wrong_answers[0] if len(wrong_answers) > 0 else "No specific hazard", "is_correct": False},
{"id": "c", "text": wrong_answers[1] if len(wrong_answers) > 1 else "General warning", "is_correct": False},
{"id": "d", "text": wrong_answers[2] if len(wrong_answers) > 2 else "Environmental impact only", "is_correct": False},
],
sort_order=questions_created + 1,
points=1,
)
self.db.add(question)
questions_created += 1
# Generate True/False questions about signal words
if "DANGER" in summary.signal_words and questions_created < question_count:
question = SafePathQuizQuestion(
question_id=uuid.uuid4(),
quiz_id=quiz.quiz_id,
question_text="DANGER is the more severe signal word compared to WARNING in GHS classification.",
question_type="true_false",
options=[
{"id": "true", "text": "True", "is_correct": True},
{"id": "false", "text": "False", "is_correct": False},
],
sort_order=questions_created + 1,
points=1,
)
self.db.add(question)
questions_created += 1
return questions_created
def _render_sds_section_html(self, product_name: str, section) -> str:
"""Render SDS section content as HTML for lesson display."""
content = section.structured_content or {}
html = f"<h3>{product_name}</h3>"
html += f"<h4>Section {section.section_number}: {section.section_title}</h4>"
if isinstance(content, dict):
for key, value in content.items():
html += f"<p><strong>{key}:</strong> {value}</p>"
elif isinstance(content, str):
html += f"<p>{content}</p>"
return html
def _get_site_name(self, site_id: UUID) -> str:
"""Get site name by ID."""
site = self.db.query(CompanySite).filter(
CompanySite.site_id == site_id
).first()
return site.site_name if site else "Unknown Site"
def _log_event(
self, company_id: UUID, user_id: UUID, event_type: str,
entity_type: str, entity_id: UUID, details: dict
):
"""Write to SafePath audit log."""
log = SafePathAuditLog(
log_id=uuid.uuid4(),
company_id=company_id,
event_type=event_type,
entity_type=entity_type,
entity_id=entity_id,
user_id=user_id,
details=details,
)
self.db.add(log)
4. API Endpoints
File: app/api/v1/safepath/hazcom_integration.py
"""SafePath HazCom Integration API
Endpoints for generating courses from HazCom plans, handling chemical-change
retraining triggers, and querying SDS-linked training content.
"""
from uuid import UUID
from typing import Optional, List
from fastapi import APIRouter, Depends, HTTPException, status, Query
from sqlalchemy.orm import Session
from app.db.session import get_db
from app.api.dependencies.permissions import get_user_context
from app.services.safepath.hazcom_integration_service import HazComIntegrationService
from app.schemas.safepath.hazcom_integration import (
HazComCourseGenerateRequest,
HazComCoursePreviewResponse,
HazComCourseGenerateResponse,
ChemicalChangeEvent,
RetrainingTriggerResponse,
PlanVersionRetrainingRequest,
PlanVersionRetrainingResponse,
GHSPictogramTrainingItem,
SiteHazardSummary,
)
router = APIRouter(prefix="/safepath/hazcom", tags=["SafePath HazCom Integration"])
# ============================================================================
# Course Generation from HazCom Plan
# ============================================================================
@router.post(
"/generate-course/preview",
response_model=HazComCoursePreviewResponse,
summary="Preview auto-generated course from HazCom plan",
)
async def preview_hazcom_course(
request: HazComCourseGenerateRequest,
db: Session = Depends(get_db),
user_ctx=Depends(get_user_context),
):
"""Preview what an auto-generated HazCom training course would contain.
Reads the approved/active HazCom plan and SDS data for chemicals in scope,
then returns a preview of lessons, quiz, and hazard summary.
The course is NOT created — use the generate endpoint to create it.
"""
service = HazComIntegrationService(db)
preview = service.preview_course_generation(request, user_ctx.company_id)
if not preview:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="HazCom plan not found or not in approved/active status.",
)
return preview
@router.post(
"/generate-course",
response_model=HazComCourseGenerateResponse,
summary="Generate a draft course from HazCom plan",
)
async def generate_hazcom_course(
request: HazComCourseGenerateRequest,
db: Session = Depends(get_db),
user_ctx=Depends(get_user_context),
):
"""Generate a draft SafePath course from an approved/active HazCom plan.
Creates a course with auto-generated lessons (GHS overview, PPE, emergency,
first aid) and quiz questions. Course is saved in Draft status for
trainer/coordinator review before publishing.
Requires: SAFEPATH + CHEMIQ modules enabled. Standard or Pro tier.
"""
service = HazComIntegrationService(db)
result = service.generate_course(request, user_ctx.company_id, user_ctx.user_id)
if not result:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="HazCom plan not found or not in approved/active status.",
)
return result
# ============================================================================
# Chemical-Change Retraining
# ============================================================================
@router.post(
"/chemical-change",
response_model=RetrainingTriggerResponse,
summary="Process chemical-change retraining trigger",
)
async def handle_chemical_change(
event: ChemicalChangeEvent,
db: Session = Depends(get_db),
user_ctx=Depends(get_user_context),
):
"""Process a chemical-change event from ChemIQ.
When a new chemical is added to a site, this endpoint checks for active
auto-assignment rules with trigger_type='chemical_added' and creates
retraining assignments for affected site employees.
Typically called internally by ChemIQ service after inventory update.
"""
service = HazComIntegrationService(db)
return service.handle_chemical_change(event)
# ============================================================================
# Plan-Version Retraining
# ============================================================================
@router.post(
"/plan-retraining",
response_model=PlanVersionRetrainingResponse,
summary="Process plan-version retraining",
)
async def handle_plan_version_retraining(
request: PlanVersionRetrainingRequest,
db: Session = Depends(get_db),
user_ctx=Depends(get_user_context),
):
"""Process retraining after a new HazCom plan version is published.
Optionally creates a new version of the linked course and auto-assigns
it to employees who completed the previous version.
"""
service = HazComIntegrationService(db)
return service.handle_plan_version_published(
request, user_ctx.company_id, user_ctx.user_id
)
# ============================================================================
# SDS-Linked Training Content
# ============================================================================
@router.get(
"/site-hazard-summary/{site_id}",
response_model=SiteHazardSummary,
summary="Get hazard summary for a site",
)
async def get_site_hazard_summary(
site_id: UUID,
db: Session = Depends(get_db),
user_ctx=Depends(get_user_context),
):
"""Get a hazard summary for a site based on ChemIQ inventory.
Returns hazard categories, GHS pictograms, signal words, and chemical
counts — used by the course generation UI to show what will be covered.
"""
service = HazComIntegrationService(db)
summary = service.get_site_hazard_summary(user_ctx.company_id, site_id)
if not summary:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Site not found.",
)
return summary
@router.get(
"/ghs-pictogram-training",
response_model=List[GHSPictogramTrainingItem],
summary="Get GHS pictogram training items for a site",
)
async def get_ghs_pictogram_training(
site_id: Optional[UUID] = Query(None),
db: Session = Depends(get_db),
user_ctx=Depends(get_user_context),
):
"""Get GHS pictogram training items based on chemicals at a site.
Returns which GHS pictograms are present, which chemicals have them,
and precautionary statements — used for auto-generated GHS lessons.
"""
service = HazComIntegrationService(db)
return service.get_ghs_pictogram_training_items(user_ctx.company_id, site_id)
5. Integration Hooks
5.1 Hook in HazCom Plan Service — publish_plan()
File: app/services/hazcom/hazcom_plan_service.py
Add after the plan is set to "active" status (approximately line 1148):
# --- SafePath integration hook ---
# After publishing, check if SafePath + ChemIQ modules are enabled.
# If so, notify SafePath to flag retraining for linked courses.
try:
from app.services.safepath.hazcom_integration_service import HazComIntegrationService
from app.schemas.safepath.hazcom_integration import PlanVersionRetrainingRequest
integration_service = HazComIntegrationService(self.uow._session)
retraining_request = PlanVersionRetrainingRequest(
plan_id=plan.plan_id,
previous_plan_id=previous_active_plan_id,
auto_assign=False, # Flag for review, don't auto-assign
due_date_offset_days=30,
)
integration_service.handle_plan_version_published(
retraining_request, company_id, audit_ctx.user_id if audit_ctx else None
)
except ImportError:
pass # SafePath module not available — skip
except Exception as e:
# Log but don't fail plan publishing
import logging
logging.getLogger(__name__).warning(f"SafePath integration hook failed: {e}")
5.2 Hook in ChemIQ Inventory Service — on chemical added
File: app/services/chemiq/chemiq_inventory_service.py (or equivalent)
Add after a new inventory item is successfully created:
# --- SafePath integration hook ---
# When a new chemical is added, check for auto-assignment rules
try:
from app.services.safepath.hazcom_integration_service import HazComIntegrationService
from app.schemas.safepath.hazcom_integration import ChemicalChangeEvent
integration_service = HazComIntegrationService(self.db)
event = ChemicalChangeEvent(
event_type="chemical_added",
company_id=company_id,
site_id=inventory_item.site_id,
product_catalog_id=inventory_item.company_product_id,
product_name=product.product_name,
hazard_categories=product.hazard_categories or [],
ghs_pictograms=[], # Populated from SDS if available
sds_id=product.current_sds_id,
)
integration_service.handle_chemical_change(event)
except ImportError:
pass # SafePath module not available — skip
except Exception as e:
import logging
logging.getLogger(__name__).warning(f"SafePath chemical-change hook failed: {e}")
6. Frontend Changes
6.1 TypeScript Interfaces
File: src/types/safepath.ts (add to existing file)
// ============================================================================
// HazCom Integration Types
// ============================================================================
export interface HazComCourseGenerateRequest {
plan_id: string;
site_id?: string;
include_ghs_overview: boolean;
include_ppe_lesson: boolean;
include_emergency_lesson: boolean;
include_first_aid_lesson: boolean;
include_quiz: boolean;
quiz_question_count: number;
title_override?: string;
}
export interface GeneratedLessonPreview {
title: string;
lesson_type: string;
content_summary: string;
sds_sections_used: string[];
chemical_count: number;
}
export interface GeneratedQuizPreview {
question_count: number;
question_types: string[];
topics_covered: string[];
}
export interface HazComCoursePreviewResponse {
plan_id: string;
plan_name: string;
plan_version: number;
site_name: string | null;
suggested_title: string;
chemical_count: number;
hazard_categories_found: string[];
ghs_pictograms_found: string[];
lessons: GeneratedLessonPreview[];
quiz: GeneratedQuizPreview | null;
estimated_duration_minutes: number;
}
export interface HazComCourseGenerateResponse {
course_id: string;
title: string;
status: string;
lesson_count: number;
quiz_question_count: number;
plan_version_id: string;
message: string;
}
export interface SiteHazardSummary {
site_id: string;
site_name: string;
total_chemicals: number;
hazard_categories: string[];
ghs_pictograms: string[];
signal_words: string[];
chemicals_by_pictogram: Record<string, string[]>;
}
export interface GHSPictogramTrainingItem {
pictogram_code: string;
pictogram_name: string;
hazard_description: string;
chemicals_with_pictogram: string[];
chemical_count: number;
precautionary_statements: string[];
}
6.2 API Methods
File: src/services/api/safepath.api.ts (add to existing file)
// ============================================================================
// HazCom Integration API
// ============================================================================
export const previewHazComCourse = async (
request: HazComCourseGenerateRequest,
token: string,
userId: string,
companyId: string,
): Promise<HazComCoursePreviewResponse> => {
const res = await fetch(`${API_URL}/api/v1/safepath/hazcom/generate-course/preview`, {
method: 'POST',
headers: authHeaders(token, userId, companyId),
body: JSON.stringify(request),
});
if (!res.ok) throw new Error('Failed to preview HazCom course');
return res.json();
};
export const generateHazComCourse = async (
request: HazComCourseGenerateRequest,
token: string,
userId: string,
companyId: string,
): Promise<HazComCourseGenerateResponse> => {
const res = await fetch(`${API_URL}/api/v1/safepath/hazcom/generate-course`, {
method: 'POST',
headers: authHeaders(token, userId, companyId),
body: JSON.stringify(request),
});
if (!res.ok) throw new Error('Failed to generate HazCom course');
return res.json();
};
export const getSiteHazardSummary = async (
siteId: string,
token: string,
userId: string,
companyId: string,
): Promise<SiteHazardSummary> => {
const res = await fetch(`${API_URL}/api/v1/safepath/hazcom/site-hazard-summary/${siteId}`, {
headers: authHeaders(token, userId, companyId),
});
if (!res.ok) throw new Error('Failed to get site hazard summary');
return res.json();
};
export const getGHSPictogramTraining = async (
siteId: string | undefined,
token: string,
userId: string,
companyId: string,
): Promise<GHSPictogramTrainingItem[]> => {
const params = siteId ? `?site_id=${siteId}` : '';
const res = await fetch(`${API_URL}/api/v1/safepath/hazcom/ghs-pictogram-training${params}`, {
headers: authHeaders(token, userId, companyId),
});
if (!res.ok) throw new Error('Failed to get GHS pictogram training');
return res.json();
};
6.3 Frontend Page — HazCom Course Generator
File: src/pages/safepath/courses/hazcom-generate.tsx
This page provides a wizard-style UI for generating a HazCom course:
- Step 1 — Select Plan: Choose an approved/active HazCom plan from dropdown
- Step 2 — Configure: Toggle lesson types (GHS overview, PPE, emergency, first aid), set quiz options
- Step 3 — Preview: Show preview of what will be generated (lesson list, quiz topics, chemical count)
- Step 4 — Generate: Create the draft course, redirect to course detail page for editing
The page should:
- Use
SafePathPageHeaderwith breadcrumb back to course list - Show site hazard summary (GHS pictograms found, chemical count, hazard categories)
- Color-code GHS pictograms with standard colors
- Display preview cards for each lesson that will be generated
- Show estimated duration
- "Generate Draft Course" button → creates course → redirects to
/safepath/courses/{id}
6.4 Course List Modification
File: src/pages/safepath/courses/index.tsx
Add a "Generate from HazCom Plan" button next to "Create Course":
- Only visible when company has both SafePath and ChemIQ modules enabled
- Check via module entitlement (existing pattern)
- Links to
/safepath/courses/hazcom-generate - Use a beaker/flask icon from Lucide (
FlaskConical)
7. Module Entitlement Checks
7.1 Backend
Before executing any HazCom integration endpoint, verify:
# Check that company has both SAFEPATH and CHEMIQ (or HAZCOM_PLAN) modules enabled
from app.db.models.module import CompanyEnabledModule
safepath_enabled = db.query(CompanyEnabledModule).filter(
CompanyEnabledModule.company_id == company_id,
CompanyEnabledModule.module_code == "SAFEPATH",
CompanyEnabledModule.is_active == True,
).first()
chemiq_enabled = db.query(CompanyEnabledModule).filter(
CompanyEnabledModule.company_id == company_id,
CompanyEnabledModule.module_code.in_(["CHEMIQ", "HAZCOM_PLAN"]),
CompanyEnabledModule.is_active == True,
).first()
if not safepath_enabled or not chemiq_enabled:
raise HTTPException(status_code=403, detail="SafePath and ChemIQ modules required.")
7.2 Frontend
// Check module availability before showing HazCom integration features
const hasChemIQ = enabledModules.some(m => m.module_code === 'CHEMIQ');
const hasSafePath = enabledModules.some(m => m.module_code === 'SAFEPATH');
const showHazComIntegration = hasChemIQ && hasSafePath;
8. Verification Checklist
Backend
-
HazComIntegrationServiceinstantiates without errors -
POST /safepath/hazcom/generate-course/previewreturns lesson previews for a valid plan -
POST /safepath/hazcom/generate-coursecreates a draft course with lessons and quiz - Generated course has
plan_version_idset to the source plan ID - Generated lessons have structured
contentJSONB with SDS data - Generated quiz has correctly structured MCQ and true/false questions
-
POST /safepath/hazcom/chemical-changecreates assignments when rules exist -
POST /safepath/hazcom/chemical-changereturns 0 assignments when no rules exist -
POST /safepath/hazcom/plan-retrainingcreates new course version when auto_assign=true -
GET /safepath/hazcom/site-hazard-summary/{site_id}returns correct chemical counts -
GET /safepath/hazcom/ghs-pictogram-trainingreturns pictogram training items - Integration hook in
publish_plan()fires without breaking plan publishing - Integration hook in ChemIQ fires without breaking inventory creation
- Module entitlement checks block access when modules are not enabled
Frontend
- "Generate from HazCom Plan" button appears only when ChemIQ is enabled
- HazCom course generator wizard loads approved plans
- Preview step shows correct lesson count and hazard summary
- Course generation creates draft and redirects to course detail
- GHS pictograms display with correct icons and color coding
- Site hazard summary loads for selected site
9. Future Enhancements (Phase 3)
- AI Explainer — Contextual chat during HazCom courses that retrieves plan/SDS snippets
- AI Course Builder — Use LLM to generate richer lesson content from SDS data
- AI Quiz Generator — Use LLM to generate more varied and contextual quiz questions
- Multi-language — Auto-translate generated courses to Spanish
- Delta retraining — Compare old vs new plan version and only retrain on changed sections