justin/new-site
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

mcs150: enrich intake from FMCSA carrier census before PDF fill

Browse source 
The MCS-150 biennial update re-confirms the carrier's existing FMCSA
record. Previously the PDF filler only had whatever the intake form
collected; rescued/sparse orders (or orders where the carrier's data
lives in FMCSA, not the intake) produced near-empty forms. Now we pull
the carrier census (legal name, address, EIN, fleet counts) from the
FMCSA carrier API and merge it under any customer-provided intake values
(customer edits win), so the form is pre-filled with the carrier's
current registered data. Refactored the FMCSA fetch into a shared
_fetch_fmcsa_carrier helper used by both enrichment and status check.
This commit is contained in:
justin 2026-06-10 12:35:43 -05:00
parent 7e5946d65a
commit d5e66786a2
8 changed files with 294 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

3
scripts/workers/services/__init__.py
View file

@ -107,6 +107,9 @@ SERVICE_HANDLERS: dict[str, type] = {
# ── Foreign qualification (Certificate of Authority) ─────────────────
"foreign-qualification-single": ForeignQualificationHandler,
"foreign-qualification-multi": ForeignQualificationHandler, # same handler, fans out per-state
# ── Business name reservation (admin-assisted; files SOS name hold) ──
"name-reservation-tx": MCS150UpdateHandler, # admin-assisted: file Form 501 via SOSDirect
"name-reservation-nv": MCS150UpdateHandler, # admin-assisted: file NV reservation via SilverFlume
# ── State PUC/PSC registration ────────────────────────────────────
"state-puc": StatePucFilingHandler,
# ── CDR storage tier add-ons (quota bumps, not filings) ────────────

79
scripts/workers/services/ink_signature_plotter.py
View file

@ -91,6 +91,85 @@ class PlotterConfig:
servo_up_angle: int = 90
# Line-us mapping (only used when dialect == "lineus").
lineus: "LineUsConfig | None" = None
# Pen-specific tuning (ink flow, dwell, line look). See PenProfile / PENS.
pen: "PenProfile | None" = None
@dataclass(frozen=True)
class PenProfile:
"""Per-pen tuning for a writing instrument mounted in the plotter.
Different pens lay ink differently when the plotter drops the tip STRAIGHT
DOWN (no human wrist angle/pressure modulation). Wet gel/rollerball pens read
as a genuine signature but need (a) a short dwell at pen-down so the ink wets
the paper before the first move (avoids a starting gap), and (b) a slower draw
feed so the line lays down evenly without skipping on fast reversals. These
values are applied by the emitters on top of the machine PlotterConfig.
Recommended default for legal filings: uni-ball Signo (UM-151 0.38mm) —
archival, waterproof, fraud-resistant gel ink that reads as original wet ink.
pen_down_dwell_ms : pause after the pen contacts paper, before the first
drawing move, to let wet ink start cleanly (G4 dwell).
pen_up_dwell_ms : brief pause after lifting (lets gel/rollerball tips
"snap" off cleanly without a tail). Usually small/0.
draw_feed : per-pen drawing feed (mm/min). Overrides the machine
default when set; wet pens like ~600-900.
pen_down_bias_mm : small extra downward offset (mm) added to pen_down so a
spring holder keeps positive contact for this tip.
tip_mm : nominal tip width (mm), informational / preview only.
"""
name: str = "uniball-signo"
label: str = "uni-ball Signo UM-151 0.38 (gel)"
pen_down_dwell_ms: int = 120
pen_up_dwell_ms: int = 0
draw_feed: float = 750.0
pen_down_bias_mm: float = 0.0
tip_mm: float = 0.38
# Tuned presets for the pens we keep on hand. Default = uni-ball Signo.
PENS: dict[str, PenProfile] = {
"uniball-signo": PenProfile(
name="uniball-signo",
label="uni-ball Signo UM-151 0.38 (gel, archival/waterproof)",
pen_down_dwell_ms=120, pen_up_dwell_ms=0, draw_feed=750.0,
pen_down_bias_mm=0.0, tip_mm=0.38,
),
"pilot-g2": PenProfile(
name="pilot-g2",
label="Pilot G-2 0.7 (gel, hand-written look)",
# G-2 can blob on a long pen-down dwell; keep dwell short.
pen_down_dwell_ms=80, pen_up_dwell_ms=0, draw_feed=800.0,
pen_down_bias_mm=0.0, tip_mm=0.7,
),
"energel": PenProfile(
name="energel",
label="Pentel EnerGel 0.5 (liquid-gel, fast-dry batch)",
# Liquid-gel starts instantly and dries fast — minimal dwell, faster feed.
pen_down_dwell_ms=60, pen_up_dwell_ms=0, draw_feed=900.0,
pen_down_bias_mm=0.0, tip_mm=0.5,
),
"fountain": PenProfile(
name="fountain",
label="Fountain pen + document ink (max ink character)",
# Fountain nibs need a touch more settle time and a gentle, slower line;
# add a small downward bias so the nib keeps consistent flow.
pen_down_dwell_ms=180, pen_up_dwell_ms=40, draw_feed=600.0,
pen_down_bias_mm=-0.1, tip_mm=0.5,
),
}
DEFAULT_PEN = "uniball-signo"
def load_pen(name: str | None) -> PenProfile:
"""Return a PenProfile for a named pen (default uni-ball Signo)."""
key = (name or DEFAULT_PEN).lower()
if key not in PENS:
raise ValueError(f"unknown pen '{name}'; choices: {sorted(PENS)}")
return PENS[key]
@dataclass(frozen=True)

75
scripts/workers/services/mcs150_update.py
View file

@ -97,6 +97,22 @@ class MCS150UpdateHandler:
# Check current MCS-150 status via FMCSA API
mcs150_status = self._check_current_status(dot_number)
# Enrich the intake with the carrier's CURRENT registered data from the
# FMCSA carrier API. The MCS-150 biennial update re-confirms the carrier's
# existing FMCSA record, so the authoritative source for the form is the
# FMCSA census (legal name, address, EIN, fleet counts) -- the intake form
# only collects the DOT number + any changes. Customer-provided intake
# values take precedence over the census (so edits/changes win).
census = self._fetch_carrier_record(dot_number)
if census:
merged = {**census, **{k: v for k, v in intake.items() if v not in (None, "")}}
intake = merged
LOG.info("[%s] Enriched intake from FMCSA census (legal_name=%s)",
order_number, intake.get("legal_name"))
else:
LOG.warning("[%s] No FMCSA census data for DOT %s -- form may be sparse",
order_number, dot_number)
# Step 1: Fill the official MCS-150 PDF
pdf_path = None
try:
@ -321,30 +337,59 @@ class MCS150UpdateHandler:
return [minio_path] if minio_path else []
def _check_current_status(self, dot_number: str) -> str:
"""Check current MCS-150 status via FMCSA API."""
def _fetch_fmcsa_carrier(self, dot_number: str) -> dict:
"""Fetch the raw FMCSA carrier census record for a DOT number."""
try:
import urllib.request
api_key = os.environ.get("FMCSA_API_KEY", "")
if not api_key:
return "API key not configured"
return {}
url = f"https://mobile.fmcsa.dot.gov/qc/services/carriers/{dot_number}?webKey={api_key}"
req = urllib.request.Request(url, headers={"Accept": "application/json"})
with urllib.request.urlopen(req, timeout=10) as resp:
data = json.loads(resp.read())
carrier = data.get("content", {}).get("carrier", {})
outdated = carrier.get("mcs150Outdated", "?")
status = carrier.get("statusCode", "?")
allowed = carrier.get("allowedToOperate", "?")
return (
f"Status: {status}, Allowed: {allowed}, "
f"MCS-150 Outdated: {outdated}"
)
return data.get("content", {}).get("carrier", {}) or {}
except Exception as exc:
return f"Could not check: {exc}"
LOG.warning("FMCSA carrier fetch failed for %s: %s", dot_number, exc)
return {}
def _fetch_carrier_record(self, dot_number: str) -> dict:
"""Return the carrier's current FMCSA data mapped to the intake keys the
PDF filler expects (so the biennial-update form is pre-filled with the
carrier's existing registered data). Empty dict if unavailable.
"""
c = self._fetch_fmcsa_carrier(dot_number)
if not c:
return {}
def s(v):
return "" if v is None else str(v)
out: dict = {
"legal_name": s(c.get("legalName")),
"dba_name": s(c.get("dbaName")),
"dot_number": s(c.get("dotNumber") or dot_number),
"ein": s(c.get("ein")),
"address_street": s(c.get("phyStreet")),
"address_city": s(c.get("phyCity")),
"address_state": s(c.get("phyState")),
"address_zip": s(c.get("phyZipcode")),
"phone": s(c.get("telephone") or c.get("phone")),
"power_units": s(c.get("totalPowerUnits")),
"drivers": s(c.get("totalDrivers")),
}
return {k: v for k, v in out.items() if v}
def _check_current_status(self, dot_number: str) -> str:
"""Check current MCS-150 status via FMCSA API."""
c = self._fetch_fmcsa_carrier(dot_number)
if not c:
return "Could not check FMCSA status"
return (
f"Status: {c.get('statusCode', '?')}, "
f"Allowed: {c.get('allowedToOperate', '?')}, "
f"MCS-150 Outdated: {c.get('mcs150Outdated', '?')}"
)
def _create_pending_signature_todo(self, order_number, entity_name, dot_number,
slug, minio_path, customer_email):