resistance.treatments
resistance.treatments
Strain-aware TB treatment.
TxR (product) holds the regimen’s per-strain efficacy, agent-level adherence, and acquisition-on-failure rules. TxDeliveryR (delivery) initiates treatment from the active-disease states at state-specific rates (or from a custom eligibility callable), freezes the outcome at initiation, and resolves it after a fixed course. It also supports treatment monitoring / regimen switching via interrupt + supersedes (see treatment_monitoring_eligibility).
Per the reference treatment operator (the ODE’s π(m→s)): each carried strain is cured independently (efficacy reduced for strains resistant to regimen drugs), the whole course’s outcomes are correlated through a single agent-level adherence draw, failures return to the state treatment was initiated from with the surviving strains, and each surviving treatment-susceptible strain independently acquires resistance to each regimen drug with probability q_acq[drug] (× a per-state RR) as replacement — so multiple strains may acquire resistance in one course, mirroring de-novo acquisition, with multiplicity transferred source→target. This reproduces the ODE’s π(m → s) outcome table exactly for the single-strain-per-agent two-strain case.
Classes
| Name | Description |
|---|---|
| TxDeliveryR | Rate-based strain-aware treatment delivery. |
| TxR | Strain-aware treatment product. |
TxDeliveryR
resistance.treatments.TxDeliveryR(
product,
rate_asym=ss.peryear(0.0),
rate_sym=ss.peryear(2.0),
dur_treatment=ss.months(6),
eligibility=None,
supersedes=None,
retreat_after=None,
treat_latent=False,
**kwargs,
)Rate-based strain-aware treatment delivery.
Each step, active-TB agents not already on treatment start a course at state-specific rates (rate_asym from ASYMPTOMATIC, rate_sym from SYMPTOMATIC), or from a custom eligibility callable (e.g. a DST-routed regimen). The per-strain outcome is frozen at initiation and resolved after dur_treatment: fully cleared courses go to CLEARED (with post-treatment reinfection protection); otherwise the agent returns to the state it was treated from carrying the surviving (and possibly newly resistant) strains.
Treatment monitoring / regimen switching: a delivery given supersedes=[name, ...] will interrupt any ongoing course on those deliveries for its eligible agents before starting them, so a second-line regimen can take over an in-progress first-line course.
Parameters
| Name | Type | Description | Default |
|---|---|---|---|
| product | TxR | the strain-aware treatment product. | required |
| rate_asym | ss.rate |
treatment initiation rate from ASYMPTOMATIC. | ss.peryear(0.0) |
| rate_sym | ss.rate |
treatment initiation rate from SYMPTOMATIC. | ss.peryear(2.0) |
| dur_treatment | ss.dur |
course duration (fixed). | ss.months(6) |
| eligibility | callable | optional sim -> uids override; if given, those agents start treatment instead of the rate-based rule. |
None |
| supersedes | str / list | name(s) of other TxDeliveryR whose ongoing course is interrupted for eligible agents before this delivery starts them (regimen switching). |
None |
| retreat_after | ss.dur |
refractory period after a course ends before the same agent may be re-treated by this delivery. Prevents a failing agent from being re-treated on every resolution off a single stale DST result (L1). Default None = no guard. |
None |
| treat_latent | bool | how latent (INFECTION) agents selected for treatment are handled. False (default): strain-aware sterilization — every carried strain susceptible to all regimen drugs is cleared with certainty, any regimen-resistant strain persists (the agent stays latent carrying it), and an agent left carrying no strain moves to CLEARED. No course is run and no resistance is acquired. For a pan-susceptible agent this clears the only strain → CLEARED, matching the single-strain behavior of base tbsim.TxDelivery. True: run through a full course that can fail / select for resistance. See L3 / implementation-decisions.md D-L3. (The default rate-based eligibility never selects latent agents, so this only affects custom / DST-routed eligibilities.) |
False |
Methods
| Name | Description |
|---|---|
| failure_case_eligibility | Classify a later treatment episode as treatment failure/retreatment vs a new case. |
| interrupt | Prematurely stop this delivery’s ongoing courses for uids (treatment monitoring / switch). |
| step | Resolve completed courses, then initiate new ones. |
failure_case_eligibility
resistance.treatments.TxDeliveryR.failure_case_eligibility(
within,
base=None,
new_case=False,
)Classify a later treatment episode as treatment failure/retreatment vs a new case.
Implements the spec’s requirement (§Diagnostics) to “track time since last treatment initiation to inform whether later treatment is managed as treatment failure, with need for DST/second-line treatments, or as a new case.” Reads the durable, cross-regimen tb.ti_last_treatment written by every :class:TxDeliveryR at initiation.
Returns a sim -> uids eligibility callable selecting agents whose most recent treatment initiation was within within (an ss.dur) of the current step — i.e. to be managed as a treatment failure (route to DST / second-line). Pass new_case=True for the complement (agents with no treatment within the window → managed as a new case). base optionally restricts the candidate pool (default: current active TB), e.g. base=dst.matches(RIF=True).
Feed as eligibility= to a DST or TxDeliveryR (optionally via :func:eligibility_all)::
failed = TxDeliveryR.failure_case_eligibility(within=ss.years(2))
second_line = tbsim.TxDeliveryR(eligibility=failed, supersedes=['first'], product=...)interrupt
resistance.treatments.TxDeliveryR.interrupt(uids)Prematurely stop this delivery’s ongoing courses for uids (treatment monitoring / switch).
Reverts interrupted agents to the state treatment was initiated from, keeping their current strains, so another delivery can re-treat them. Only agents actually on this delivery’s course (on_course) are affected — the flag is delivery-scoped, so a shared TREATMENT state never lets one delivery interrupt another delivery’s patients (#447).
step
resistance.treatments.TxDeliveryR.step()Resolve completed courses, then initiate new ones.
TxR
resistance.treatments.TxR(
strains,
base_efficacy=0.85,
resist_penalty=None,
efficacy_by_strain=None,
adherence=1.0,
q_acq=None,
acq_state_rr=None,
regimen_drugs=None,
**kwargs,
)Strain-aware treatment product.
Parameters
| Name | Type | Description | Default |
|---|---|---|---|
| strains | Strains | the strain registry (e.g. tb.strains). |
required |
| base_efficacy | float | per-course cure probability for a treatment-susceptible strain. | 0.85 |
| resist_penalty | dict | per-drug multiplicative efficacy penalty applied for each regimen drug a strain is resistant to, e.g. {'TX': 0.333} gives a resistant strain 1/3 the cure probability. Resistance to a drug outside the regimen does not reduce efficacy. |
None |
| efficacy_by_strain | array | optional explicit per-strain cure-probability vector T_l = {t_1,l, ..., t_m,l} (length strains.m, spec §“Treatment efficacy”). When given it is the per-strain efficacy and base_efficacy/resist_penalty are ignored; use it when the constrained base × ∏penalty parameterization cannot express the desired vector. |
None |
| adherence | float or callable | per-course completion probability correlating outcomes across an agent’s strains (non-completers clear no strains this course). A float applies one regimen-level probability to every agent; a callable uids -> per-agent probability makes adherence a regimen-level distribution that varies by agent and is applied across all that agent’s strains (spec §“Treatment efficacy”), e.g. adherence=lambda uids: my_dist.rvs(uids). |
1.0 |
| q_acq | dict | per-drug probability {drug: prob} that a surviving treatment-susceptible strain acquires resistance to that regimen drug on a failed course (always replacement). Default none (off). |
None |
| acq_state_rr | dict | per-TB-state multiplier on q_acq at time of failure (spec’s RR on q). Default 1 for ASYMPTOMATIC/SYMPTOMATIC, 0 for all other states. |
None |
| regimen_drugs | list | drugs the regimen acts on; determines which strains can be cured/acquire resistance. Default: all drugs in strains. |
None |
Methods
| Name | Description |
|---|---|
| acquire_counts | Apply acquisition-on-failure (replacement) to the surviving counts of failed courses. |
| roll_survivors | Pre-roll the surviving strain mask for each treated agent (adherence, then per-strain cure). |
acquire_counts
resistance.treatments.TxR.acquire_counts(tb, uids, counts0, states=None)Apply acquisition-on-failure (replacement) to the surviving counts of failed courses.
Each surviving carried strain independently rolls, once per regimen drug it is susceptible to, whether it acquires that resistance (prob q_acq[drug] × the per-agent state RR acq_state_rr; default 0 outside ASYMPTOMATIC/SYMPTOMATIC). Hits mutate the strain to its resistant counterpart, transferring multiplicity source→target via :meth:TBResistant._apply_acquisition. Multiple strains may acquire resistance in one course.
Returns (counts, mask, n_events) for the treated agents.
roll_survivors
resistance.treatments.TxR.roll_survivors(tb, uids)Pre-roll the surviving strain mask for each treated agent (adherence, then per-strain cure).
Functions
| Name | Description |
|---|---|
| eligibility_all | Combinator (L5): return a sim -> uids selecting agents returned by all of callables |
| eligibility_any | Combinator (L5): return a sim -> uids selecting agents returned by any of callables |
| treatment_monitoring_eligibility | Eligibility callable selecting agents on tx_name’s course for at least after_steps steps. |
| will_fail | Eligibility factory (L5): on-treatment agents on tx_name whose pre-rolled course outcome is a |
eligibility_all
resistance.treatments.eligibility_all(*callables)Combinator (L5): return a sim -> uids selecting agents returned by all of callables (set intersection). E.g. eligibility_all(treatment_monitoring_eligibility('first'), dst.matches(RIF=True)) switches only agents that are both far enough into first-line and observed RIF-resistant.
eligibility_any
resistance.treatments.eligibility_any(*callables)Combinator (L5): return a sim -> uids selecting agents returned by any of callables (set union).
treatment_monitoring_eligibility
resistance.treatments.treatment_monitoring_eligibility(
tx_name,
after_steps=4,
every_steps=None,
require=None,
)Eligibility callable selecting agents on tx_name’s course for at least after_steps steps.
Feed as eligibility= to a monitoring DxDelivery (to flag still-bacteriologically-positive agents by TB state), or — combined with supersedes=[tx_name] on a second-line TxDeliveryR — to switch regimens mid-course (spec §“Treatment monitoring”).
Parameters
| Name | Type | Description | Default |
|---|---|---|---|
| tx_name | str | the name of the TxDeliveryR to monitor. |
required |
| after_steps | int | minimum sim steps since ti_treatment_start before eligibility. Default 4. |
4 |
| every_steps | int / None | if given, re-test every N steps after the first; else a single test at after_steps. |
None |
| require | callable | optional additional sim -> uids AND-ed in (sugar over :func:eligibility_all), e.g. require=dst.matches(RIF=True, exclude_on_treatment=False) to make monitoring contingent on an observed DST profile (L5). Note monitored agents are on treatment, so pass exclude_on_treatment=False to a matches used here. |
None |
will_fail
resistance.treatments.will_fail(tx_name)Eligibility factory (L5): on-treatment agents on tx_name whose pre-rolled course outcome is a failure (pending_surv != 0). Because the outcome is frozen at initiation, this is an oracle — it selects agents whose course will fail before it completes — useful for constructing failure-contingent regimen-switch scenarios.