From d7492a1fd24e74b45daeb2c3df26c9345810d7bd Mon Sep 17 00:00:00 2001 From: Georgie Kennedy Date: Mon, 14 Sep 2026 14:35:26 +1000 Subject: [PATCH 1/3] add amt hierarchy --- src/pbs_client/toolkit/core/__init__.py | 6 ++ src/pbs_client/toolkit/core/service.py | 122 +++++++++++++++++++++ tests/test_query.py | 138 ++++++++++++++++++++++++ 3 files changed, 266 insertions(+) diff --git a/src/pbs_client/toolkit/core/__init__.py b/src/pbs_client/toolkit/core/__init__.py index ec67a7c..caf78d9 100644 --- a/src/pbs_client/toolkit/core/__init__.py +++ b/src/pbs_client/toolkit/core/__init__.py @@ -4,10 +4,12 @@ BenefitTypeCode, IndicationText, ItemExpansion, + MpComponentSplit, RestrictionExpansion, expand_item, find_items, get_item, + get_item_amt_hierarchy, get_item_atc_codes, get_item_indication_text, get_item_restrictions, @@ -15,16 +17,19 @@ item_restrictions, lookup_item, resolve_schedule, + split_mp_components, ) __all__ = [ "BenefitTypeCode", "IndicationText", "ItemExpansion", + "MpComponentSplit", "RestrictionExpansion", "expand_item", "find_items", "get_item", + "get_item_amt_hierarchy", "get_item_atc_codes", "get_item_indication_text", "get_item_restrictions", @@ -32,4 +37,5 @@ "item_restrictions", "lookup_item", "resolve_schedule", + "split_mp_components", ] diff --git a/src/pbs_client/toolkit/core/service.py b/src/pbs_client/toolkit/core/service.py index f5a172e..1425942 100644 --- a/src/pbs_client/toolkit/core/service.py +++ b/src/pbs_client/toolkit/core/service.py @@ -2,6 +2,7 @@ from __future__ import annotations +import re from dataclasses import dataclass, field from datetime import date, datetime from enum import Enum @@ -15,6 +16,7 @@ ATC, Indication, Item, + ItemAmt, ItemAtcRltd, ItemRestrictionRltd, PrescribingTxt, @@ -47,6 +49,21 @@ class IndicationText: severity: str | None = None +@dataclass(frozen=True, slots=True) +class MpComponentSplit: + """Conservative component-text interpretation of one AMT MP row.""" + + schedule_code: int + li_item_id: str + pbs_concept_id: int + source_field: Literal["preferred_term", "pbs_preferred_term"] | None + source_term: str | None + component_terms: tuple[str, ...] + status: Literal["split", "single_term", "missing", "unsupported", "divergent"] + reason: str | None = None + field_divergence: tuple[str, str] | None = None + + @dataclass(frozen=True, slots=True) class RestrictionExpansion: restriction: RestrictionText @@ -76,6 +93,10 @@ def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None self.parts.append(" ") +_AMT_CONCEPT_TYPE_ORDER = {"MP": 0, "MPUU": 1, "MPP": 2, "TPP": 3, "TPUU": 4} +_AMT_COMPONENT_CONNECTOR = re.compile(r"\b(?:and|with)\b", re.IGNORECASE) + + def _clean_html(value: str | None) -> str | None: if not value: return None @@ -318,6 +339,104 @@ def get_item_atc_codes(session: Session, item: Item) -> list[ATC]: ] +def get_item_amt_hierarchy(session: Session, item: Item) -> list[ItemAmt]: + """Return every linked AMT row, keeping its PBS row identity intact.""" + + rows = session.scalars( + select(ItemAmt).where( + ItemAmt.schedule_code == item.schedule_code, + ItemAmt.li_item_id == item.li_item_id, + ) + ).all() + rows.sort( + key=lambda row: ( + _AMT_CONCEPT_TYPE_ORDER.get(row.concept_type_code, len(_AMT_CONCEPT_TYPE_ORDER)), + row.concept_type_code, + row.pbs_concept_id, + ) + ) + return rows + + +def _parse_mp_component_text( + text: str, +) -> tuple[Literal["split", "single_term", "unsupported"], tuple[str, ...], str | None]: + if "(&)" in text: + return "unsupported", (), 'unsupported "(&)" separator' + if _AMT_COMPONENT_CONNECTOR.search(text): + return "unsupported", (), '"with"/"and" text is not treated as a combination' + if "+" not in text: + return "single_term", (text,), None + + parts = tuple(part.strip() for part in text.split(" + ")) + if len(parts) < 2 or any(not part or "+" in part for part in parts): + return "unsupported", (), "unrecognized or empty combination component" + return "split", parts, None + + +def _select_mp_source( + concept: ItemAmt, +) -> tuple[ + Literal["preferred_term", "pbs_preferred_term"] | None, + str | None, + str | None, + tuple[str, str] | None, +]: + preferred = concept.preferred_term + fallback = concept.pbs_preferred_term + preferred_text = preferred.strip() if preferred and preferred.strip() else None + fallback_text = fallback.strip() if fallback and fallback.strip() else None + + if preferred and fallback and preferred_text and fallback_text: + normalized_preferred = " ".join(preferred_text.split()).casefold() + normalized_fallback = " ".join(fallback_text.split()).casefold() + if normalized_preferred != normalized_fallback: + return None, None, None, (preferred, fallback) + + if preferred_text: + return "preferred_term", preferred, preferred_text, None + if fallback_text: + return "pbs_preferred_term", fallback, fallback_text, None + return None, None, None, None + + +def _split_mp_concept(concept: ItemAmt) -> MpComponentSplit: + source_field, source_term, text, field_divergence = _select_mp_source(concept) + component_terms: tuple[str, ...] = () + reason: str | None = None + status: Literal["split", "single_term", "missing", "unsupported", "divergent"] + + if field_divergence: + status = "divergent" + reason = "preferred_term and pbs_preferred_term differ" + elif text: + status, component_terms, reason = _parse_mp_component_text(text) + else: + status = "missing" + + return MpComponentSplit( + schedule_code=concept.schedule_code, + li_item_id=concept.li_item_id, + pbs_concept_id=concept.pbs_concept_id, + source_field=source_field, + source_term=source_term, + component_terms=component_terms, + status=status, + reason=reason, + field_divergence=field_divergence, + ) + + +def split_mp_components(session: Session, item: Item) -> list[MpComponentSplit]: + """Return one conservative component-text result per linked MP row.""" + + return [ + _split_mp_concept(concept) + for concept in get_item_amt_hierarchy(session, item) + if concept.concept_type_code == "MP" + ] + + def expand_item(session: Session, item: Item) -> ItemExpansion: """Return the complete convenience expansion for an item.""" @@ -337,10 +456,12 @@ def expand_item(session: Session, item: Item) -> ItemExpansion: "BenefitTypeCode", "IndicationText", "ItemExpansion", + "MpComponentSplit", "RestrictionExpansion", "expand_item", "find_items", "get_item", + "get_item_amt_hierarchy", "get_item_atc_codes", "get_item_indication_text", "get_item_restrictions", @@ -348,4 +469,5 @@ def expand_item(session: Session, item: Item) -> ItemExpansion: "item_restrictions", "lookup_item", "resolve_schedule", + "split_mp_components", ] diff --git a/tests/test_query.py b/tests/test_query.py index d53737c..9c47e7b 100644 --- a/tests/test_query.py +++ b/tests/test_query.py @@ -4,6 +4,7 @@ ATC, Indication, Item, + ItemAmt, ItemAtcRltd, ItemRestrictionRltd, PrescribingTxt, @@ -14,10 +15,13 @@ from pbs_client.toolkit.analytics import indication_candidates from pbs_client.toolkit.core import ( BenefitTypeCode, + MpComponentSplit, expand_item, find_items, + get_item_amt_hierarchy, get_item_indication_text, resolve_schedule, + split_mp_components, ) @@ -195,6 +199,140 @@ def test_indication_text_uses_clean_fallback_and_excludes_notes(session_factory) assert indications[0].prescribing_txt_id is None +def test_item_amt_hierarchy_orders_and_preserves_source_rows(session_factory): + amt_rows = [ + (40, "MPP", "shared-code", "Same term"), + (20, "MP", "shared-code", "Same term"), + (30, "MPUU", "unit-code", None), + (10, "MP", "shared-code", "Same term"), + ] + with session_factory() as session: + session.add(Schedule(schedule_code=20, effective_date="2026-06-01", effective_year=2026)) + session.add_all( + [ + Item(schedule_code=20, li_item_id="li-amt", pbs_code="AMT"), + Item(schedule_code=20, li_item_id="li-no-mp", pbs_code="NO-MP"), + ] + ) + session.commit() + session.add_all( + ItemAmt( + schedule_code=20, + li_item_id="li-amt", + pbs_concept_id=concept_id, + concept_type_code=concept_type, + amt_code=amt_code, + preferred_term=term, + ) + for concept_id, concept_type, amt_code, term in amt_rows + ) + session.add( + ItemAmt( + schedule_code=20, + li_item_id="li-no-mp", + pbs_concept_id=50, + concept_type_code="MPP", + ) + ) + session.commit() + + item = session.get(Item, (20, "li-amt")) + hierarchy = get_item_amt_hierarchy(session, item) + no_amt_rows = get_item_amt_hierarchy(session, Item(schedule_code=20, li_item_id="absent")) + no_mp_rows = split_mp_components(session, session.get(Item, (20, "li-no-mp"))) + + assert all(isinstance(concept, ItemAmt) for concept in hierarchy) + assert [(concept.concept_type_code, concept.pbs_concept_id) for concept in hierarchy] == [ + ("MP", 10), + ("MP", 20), + ("MPUU", 30), + ("MPP", 40), + ] + assert [concept.amt_code for concept in hierarchy[:2]] == ["shared-code", "shared-code"] + assert no_amt_rows == [] + assert no_mp_rows == [] + + +def test_split_mp_components_uses_unambiguous_fields_and_preserves_terms(session_factory): + source_rows = [ + (1, " Aspirin + metformin ", "aspirin + METFORMIN"), + (2, None, "One + Two"), + (3, "Floxacillin", "Flucloxacillin"), + (8, None, None), + (9, "Single component", "Single component"), + (10, "Repeated + Repeated", "Repeated + Repeated"), + ] + with session_factory() as session: + session.add(Schedule(schedule_code=21, effective_date="2026-06-01", effective_year=2026)) + session.add(Item(schedule_code=21, li_item_id="li-components", pbs_code="COMPONENTS")) + session.flush() + session.add_all( + [ + ItemAmt( + schedule_code=21, + li_item_id="li-components", + pbs_concept_id=concept_id, + concept_type_code="MP", + preferred_term=preferred, + pbs_preferred_term=pbs_preferred, + ) + for concept_id, preferred, pbs_preferred in source_rows + ] + ) + session.flush() + results = split_mp_components(session, session.get(Item, (21, "li-components"))) + + assert all(isinstance(result, MpComponentSplit) for result in results) + by_concept_id = {result.pbs_concept_id: result for result in results} + assert [result.status for result in results] == [ + "split", + "split", + "divergent", + "missing", + "single_term", + "split", + ] + assert by_concept_id[1].source_field == "preferred_term" + assert by_concept_id[1].source_term == " Aspirin + metformin " + assert by_concept_id[1].component_terms == ("Aspirin", "metformin") + assert by_concept_id[2].source_field == "pbs_preferred_term" + assert by_concept_id[2].component_terms == ("One", "Two") + assert by_concept_id[3].field_divergence == ("Floxacillin", "Flucloxacillin") + assert by_concept_id[9].component_terms == ("Single component",) + assert by_concept_id[10].component_terms == ("Repeated", "Repeated") + + +def test_split_mp_components_rejects_unverified_separators(session_factory): + source_rows = [ + (4, "Formula with vitamins and minerals"), + (5, "Estradiol (&) estradiol + dydrogesterone"), + (6, "Drug A+Drug B"), + (7, "Drug + "), + ] + with session_factory() as session: + session.add(Schedule(schedule_code=22, effective_date="2026-06-01", effective_year=2026)) + session.add(Item(schedule_code=22, li_item_id="li-unsupported", pbs_code="UNSUPPORTED")) + session.flush() + session.add_all( + [ + ItemAmt( + schedule_code=22, + li_item_id="li-unsupported", + pbs_concept_id=concept_id, + concept_type_code="MP", + preferred_term=term, + pbs_preferred_term=term, + ) + for concept_id, term in source_rows + ] + ) + session.flush() + results = split_mp_components(session, session.get(Item, (22, "li-unsupported"))) + + assert [result.status for result in results] == ["unsupported"] * len(source_rows) + assert all(not result.component_terms and result.reason for result in results) + + def test_item_lookup_with_unknown_date_returns_no_items(session_factory): with session_factory() as session: session.add(Schedule(schedule_code=7, effective_date="2026-01-01", effective_year=2026)) From ee46237811000e5e01434218d24dae360b8df171 Mon Sep 17 00:00:00 2001 From: Georgie Kennedy Date: Mon, 14 Sep 2026 14:51:51 +1000 Subject: [PATCH 2/3] ingredient and supplier handling --- src/pbs_client/toolkit/__init__.py | 2 +- src/pbs_client/toolkit/analytics/__init__.py | 18 +- .../toolkit/analytics/indications.py | 190 ++++++++++++++- src/pbs_client/toolkit/core/__init__.py | 4 + src/pbs_client/toolkit/core/service.py | 30 +++ src/pbs_client/toolkit/pricing.py | 80 +++++++ tests/test_query.py | 223 +++++++++++++++++- 7 files changed, 540 insertions(+), 7 deletions(-) create mode 100644 src/pbs_client/toolkit/pricing.py diff --git a/src/pbs_client/toolkit/__init__.py b/src/pbs_client/toolkit/__init__.py index c8f4747..56dc5b8 100644 --- a/src/pbs_client/toolkit/__init__.py +++ b/src/pbs_client/toolkit/__init__.py @@ -1 +1 @@ -"""PBS-native navigation and indication-candidate helpers.""" +"""PBS-native navigation, eligibility, indication, and pricing helpers.""" diff --git a/src/pbs_client/toolkit/analytics/__init__.py b/src/pbs_client/toolkit/analytics/__init__.py index 73c798f..4c724c8 100644 --- a/src/pbs_client/toolkit/analytics/__init__.py +++ b/src/pbs_client/toolkit/analytics/__init__.py @@ -1,5 +1,19 @@ """PBS-native composed analytics over toolkit core primitives.""" -from pbs_client.toolkit.analytics.indications import IndicationCandidate, indication_candidates +from pbs_client.toolkit.analytics.indications import ( + CriteriaText, + IndicationCandidate, + ItemCriteriaBreakdown, + ParameterText, + get_item_criteria_breakdown, + indication_candidates, +) -__all__ = ["IndicationCandidate", "indication_candidates"] +__all__ = [ + "CriteriaText", + "IndicationCandidate", + "ItemCriteriaBreakdown", + "ParameterText", + "get_item_criteria_breakdown", + "indication_candidates", +] diff --git a/src/pbs_client/toolkit/analytics/indications.py b/src/pbs_client/toolkit/analytics/indications.py index 88d01e2..7e27d78 100644 --- a/src/pbs_client/toolkit/analytics/indications.py +++ b/src/pbs_client/toolkit/analytics/indications.py @@ -5,10 +5,26 @@ from dataclasses import dataclass from datetime import date, datetime +from sqlalchemy import and_, select from sqlalchemy.orm import Session -from pbs_client.db.model import Item, Schedule -from pbs_client.toolkit.core import IndicationText, find_items, get_item_indication_text +from pbs_client.db.model import ( + Criteria, + CriteriaParameterRltd, + DispensingRule, + Item, + ItemPrescribingTxtRltd, + Parameter, + PrescribingTxt, + Program, + Schedule, +) +from pbs_client.toolkit.core import ( + IndicationText, + find_items, + get_item_dispensing_rule_links, + get_item_indication_text, +) @dataclass(frozen=True, slots=True) @@ -20,6 +36,167 @@ class IndicationCandidate: indication: IndicationText +@dataclass(frozen=True, slots=True) +class ParameterText: + """A parameter text and all structured PBS parameter rows that describe it.""" + + relationship: CriteriaParameterRltd + prescribing_text: PrescribingTxt + parameters: tuple[Parameter, ...] + + +@dataclass(frozen=True, slots=True) +class CriteriaText: + """One item-linked eligibility criterion with its linked parameter detail.""" + + item_relationship: ItemPrescribingTxtRltd + criteria: Criteria + prescribing_text: PrescribingTxt + parameters: tuple[ParameterText, ...] + + +@dataclass(frozen=True, slots=True) +class ItemCriteriaBreakdown: + """Clinical criteria plus separate funding and dispensing context for an item.""" + + criteria: tuple[CriteriaText, ...] + program: Program | None + dispensing_rules: tuple[DispensingRule, ...] + + +def get_item_criteria_breakdown(session: Session, item: Item) -> ItemCriteriaBreakdown: + """Return an item's linked eligibility text and separate program/rule context. + + Criteria and parameter text remain PBS-native source rows with their + relationship metadata intact. Program and dispensing-rule values are + returned as administrative context, not interpreted as clinical criteria. + """ + + criteria_rows = [] + if item.pbs_code is not None: + criteria_rows = session.execute( + select(ItemPrescribingTxtRltd, Criteria, PrescribingTxt) + .join( + PrescribingTxt, + and_( + PrescribingTxt.schedule_code == ItemPrescribingTxtRltd.schedule_code, + PrescribingTxt.prescribing_txt_id == ItemPrescribingTxtRltd.prescribing_txt_id, + ), + ) + .join( + Criteria, + and_( + Criteria.schedule_code == PrescribingTxt.schedule_code, + Criteria.criteria_prescribing_txt_id == PrescribingTxt.prescribing_txt_id, + ), + ) + .where( + ItemPrescribingTxtRltd.schedule_code == item.schedule_code, + ItemPrescribingTxtRltd.pbs_code == item.pbs_code, + PrescribingTxt.prescribing_type == "CRITERIA", + ) + .order_by( + ItemPrescribingTxtRltd.pt_position, + ItemPrescribingTxtRltd.prescribing_txt_id, + ) + ).all() + + criteria_ids = sorted( + {criteria.criteria_prescribing_txt_id for _, criteria, _ in criteria_rows} + ) + parameter_rows = [] + parameters_by_text_id: dict[int, list[Parameter]] = {} + if criteria_ids: + parameter_rows = session.execute( + select(CriteriaParameterRltd, PrescribingTxt) + .join( + PrescribingTxt, + and_( + PrescribingTxt.schedule_code == CriteriaParameterRltd.schedule_code, + PrescribingTxt.prescribing_txt_id + == CriteriaParameterRltd.parameter_prescribing_txt_id, + ), + ) + .where( + CriteriaParameterRltd.schedule_code == item.schedule_code, + CriteriaParameterRltd.criteria_prescribing_txt_id.in_(criteria_ids), + ) + .order_by( + CriteriaParameterRltd.criteria_prescribing_txt_id, + CriteriaParameterRltd.pt_position, + CriteriaParameterRltd.parameter_prescribing_txt_id, + ) + ).all() + parameter_ids = sorted( + {relationship.parameter_prescribing_txt_id for relationship, _ in parameter_rows} + ) + if parameter_ids: + for parameter in session.scalars( + select(Parameter) + .where( + Parameter.schedule_code == item.schedule_code, + Parameter.parameter_prescribing_txt_id.in_(parameter_ids), + ) + .order_by( + Parameter.parameter_prescribing_txt_id, + Parameter.assessment_type, + ) + ).all(): + parameters_by_text_id.setdefault( + parameter.parameter_prescribing_txt_id, [] + ).append(parameter) + + parameters_by_criteria_id: dict[int, list[ParameterText]] = {} + for relationship, prescribing_text in parameter_rows: + parameters_by_criteria_id.setdefault( + relationship.criteria_prescribing_txt_id, [] + ).append( + ParameterText( + relationship=relationship, + prescribing_text=prescribing_text, + parameters=tuple( + parameters_by_text_id.get(relationship.parameter_prescribing_txt_id, ()) + ), + ) + ) + + criteria = tuple( + CriteriaText( + item_relationship=item_relationship, + criteria=criterion, + prescribing_text=prescribing_text, + parameters=tuple( + parameters_by_criteria_id.get(criterion.criteria_prescribing_txt_id, ()) + ), + ) + for item_relationship, criterion, prescribing_text in criteria_rows + ) + program = ( + session.get(Program, (item.schedule_code, item.program_code)) + if item.program_code is not None + else None + ) + dispensing_rule_links = get_item_dispensing_rule_links(session, item) + rule_mnemonics = {link.dispensing_rule_mnem for link in dispensing_rule_links} + dispensing_rules = ( + session.scalars( + select(DispensingRule) + .where( + DispensingRule.schedule_code == item.schedule_code, + DispensingRule.dispensing_rule_mnem.in_(rule_mnemonics), + ) + .order_by(DispensingRule.dispensing_rule_mnem) + ).all() + if rule_mnemonics + else [] + ) + return ItemCriteriaBreakdown( + criteria=criteria, + program=program, + dispensing_rules=tuple(dispensing_rules), + ) + + def indication_candidates( session: Session, item_code: str, @@ -46,4 +223,11 @@ def indication_candidates( return candidates -__all__ = ["IndicationCandidate", "indication_candidates"] +__all__ = [ + "CriteriaText", + "IndicationCandidate", + "ItemCriteriaBreakdown", + "ParameterText", + "get_item_criteria_breakdown", + "indication_candidates", +] diff --git a/src/pbs_client/toolkit/core/__init__.py b/src/pbs_client/toolkit/core/__init__.py index caf78d9..c4868ac 100644 --- a/src/pbs_client/toolkit/core/__init__.py +++ b/src/pbs_client/toolkit/core/__init__.py @@ -11,7 +11,9 @@ get_item, get_item_amt_hierarchy, get_item_atc_codes, + get_item_dispensing_rule_links, get_item_indication_text, + get_item_manufacturer, get_item_restrictions, item_atc_codes, item_restrictions, @@ -31,7 +33,9 @@ "get_item", "get_item_amt_hierarchy", "get_item_atc_codes", + "get_item_dispensing_rule_links", "get_item_indication_text", + "get_item_manufacturer", "get_item_restrictions", "item_atc_codes", "item_restrictions", diff --git a/src/pbs_client/toolkit/core/service.py b/src/pbs_client/toolkit/core/service.py index 1425942..49b76e9 100644 --- a/src/pbs_client/toolkit/core/service.py +++ b/src/pbs_client/toolkit/core/service.py @@ -18,7 +18,9 @@ Item, ItemAmt, ItemAtcRltd, + ItemDispensingRuleRltd, ItemRestrictionRltd, + Organisation, PrescribingTxt, RestrictionText, RstrctnPrscrbngTxtRltd, @@ -180,6 +182,19 @@ def get_item(session: Session, item_code: str, **kwargs: Any) -> Item | None: return next(iter(find_items(session, item_code, **kwargs)), None) +def get_item_manufacturer(session: Session, item: Item) -> Organisation | None: + """Return the manufacturer PBS links directly to this item, if present. + + The Item resource's ``organisation_id`` identifies its manufacturer. + ``ItemOrganisationRltd`` describes a wholesaling relationship instead + and is deliberately not used here. + """ + + if item.organisation_id is None: + return None + return session.get(Organisation, (item.schedule_code, item.organisation_id)) + + def get_item_restrictions(session: Session, item: Item) -> list[RestrictionExpansion]: """Expand an item through restrictions, prescribing text, and indications.""" @@ -339,6 +354,19 @@ def get_item_atc_codes(session: Session, item: Item) -> list[ATC]: ] +def get_item_dispensing_rule_links(session: Session, item: Item) -> list[ItemDispensingRuleRltd]: + """Return the dispensing-rule relationship rows attached to an item.""" + + return session.scalars( + select(ItemDispensingRuleRltd) + .where( + ItemDispensingRuleRltd.schedule_code == item.schedule_code, + ItemDispensingRuleRltd.li_item_id == item.li_item_id, + ) + .order_by(ItemDispensingRuleRltd.dispensing_rule_mnem) + ).all() + + def get_item_amt_hierarchy(session: Session, item: Item) -> list[ItemAmt]: """Return every linked AMT row, keeping its PBS row identity intact.""" @@ -463,7 +491,9 @@ def expand_item(session: Session, item: Item) -> ItemExpansion: "get_item", "get_item_amt_hierarchy", "get_item_atc_codes", + "get_item_dispensing_rule_links", "get_item_indication_text", + "get_item_manufacturer", "get_item_restrictions", "item_atc_codes", "item_restrictions", diff --git a/src/pbs_client/toolkit/pricing.py b/src/pbs_client/toolkit/pricing.py new file mode 100644 index 0000000..d38bda4 --- /dev/null +++ b/src/pbs_client/toolkit/pricing.py @@ -0,0 +1,80 @@ +"""PBS pricing inputs, kept separate from clinical indication analytics.""" + +from __future__ import annotations + +from dataclasses import dataclass + +from sqlalchemy import select +from sqlalchemy.orm import Session + +from pbs_client.db.model import ( + Copayment, + Fee, + Item, + ItemDispensingRuleRltd, + ItemPricingEvent, + MarkupBand, +) +from pbs_client.toolkit.core import get_item_dispensing_rule_links + + +@dataclass(frozen=True, slots=True) +class ItemPricingBreakdown: + """Raw schedule-level and item-level inputs relevant to PBS pricing.""" + + item: Item + copayment: Copayment | None + fee: Fee | None + dispensing_rule_links: tuple[ItemDispensingRuleRltd, ...] + markup_bands: tuple[MarkupBand, ...] + pricing_events: tuple[ItemPricingEvent, ...] + + +def get_item_pricing_breakdown(session: Session, item: Item) -> ItemPricingBreakdown: + """Return linked PBS pricing inputs without estimating a patient's payment. + + The final amount depends on patient category, dispensing circumstances, + and applicable premiums or thresholds; this function exposes source rows + rather than implying one universal out-of-pocket price. + """ + + copayment = session.get(Copayment, item.schedule_code) + fee = ( + session.get(Fee, (item.schedule_code, item.program_code)) + if item.program_code is not None + else None + ) + dispensing_rule_links = tuple(get_item_dispensing_rule_links(session, item)) + rule_mnemonics = {link.dispensing_rule_mnem for link in dispensing_rule_links} + markup_bands = ( + session.scalars( + select(MarkupBand) + .where( + MarkupBand.schedule_code == item.schedule_code, + MarkupBand.program_code == item.program_code, + MarkupBand.dispensing_rule_mnem.in_(rule_mnemonics), + ) + .order_by(MarkupBand.dispensing_rule_mnem, MarkupBand.markup_band_code) + ).all() + if item.program_code is not None and rule_mnemonics + else [] + ) + pricing_events = session.scalars( + select(ItemPricingEvent) + .where( + ItemPricingEvent.schedule_code == item.schedule_code, + ItemPricingEvent.li_item_id == item.li_item_id, + ) + .order_by(ItemPricingEvent.event_type_code, ItemPricingEvent.percentage_applied) + ).all() + return ItemPricingBreakdown( + item=item, + copayment=copayment, + fee=fee, + dispensing_rule_links=dispensing_rule_links, + markup_bands=tuple(markup_bands), + pricing_events=tuple(pricing_events), + ) + + +__all__ = ["ItemPricingBreakdown", "get_item_pricing_breakdown"] diff --git a/tests/test_query.py b/tests/test_query.py index 9c47e7b..1ca8912 100644 --- a/tests/test_query.py +++ b/tests/test_query.py @@ -2,27 +2,43 @@ from pbs_client.db.model import ( ATC, + Copayment, + Criteria, + CriteriaParameterRltd, + DispensingRule, + Fee, Indication, Item, ItemAmt, ItemAtcRltd, + ItemDispensingRuleRltd, + ItemOrganisationRltd, + ItemPrescribingTxtRltd, + ItemPricingEvent, ItemRestrictionRltd, + MarkupBand, + Organisation, + Parameter, PrescribingTxt, + Program, RestrictionText, RstrctnPrscrbngTxtRltd, Schedule, ) -from pbs_client.toolkit.analytics import indication_candidates +from pbs_client.toolkit.analytics import get_item_criteria_breakdown, indication_candidates from pbs_client.toolkit.core import ( BenefitTypeCode, MpComponentSplit, expand_item, find_items, get_item_amt_hierarchy, + get_item_dispensing_rule_links, get_item_indication_text, + get_item_manufacturer, resolve_schedule, split_mp_components, ) +from pbs_client.toolkit.pricing import get_item_pricing_breakdown def test_offline_item_restriction_and_atc_expansion(session_factory): @@ -378,3 +394,208 @@ def test_unknown_benefit_type_is_preserved(session_factory): indications = get_item_indication_text(session, session.get(Item, (12, "li-5"))) assert indications[0].benefit_type_code == "Z" + + +def test_item_manufacturer_uses_item_manufacturer_link_not_wholesaler_link(session_factory): + with session_factory() as session: + session.add(Schedule(schedule_code=50, effective_date="2026-09-01", effective_year=2026)) + session.commit() + session.add_all( + [ + Organisation(schedule_code=50, organisation_id=1, name="Item manufacturer"), + Organisation(schedule_code=50, organisation_id=2, name="Wholesaler"), + ] + ) + session.commit() + item = Item( + schedule_code=50, + li_item_id="li-manufacturer", + pbs_code="M1", + organisation_id=1, + ) + session.add(item) + session.commit() + session.add( + ItemOrganisationRltd(schedule_code=50, pbs_code="M1", organisation_id=2) + ) + session.commit() + + manufacturer = get_item_manufacturer(session, item) + + assert manufacturer is not None + assert manufacturer.name == "Item manufacturer" + + +def test_item_criteria_breakdown_preserves_parameter_program_and_rule_context(session_factory): + with session_factory() as session: + session.add(Schedule(schedule_code=51, effective_date="2026-09-01", effective_year=2026)) + session.commit() + session.add_all( + [ + Program(schedule_code=51, program_code="CT", program_title="Chemotherapy"), + DispensingRule( + schedule_code=51, + dispensing_rule_mnem="HOSP", + dispensing_rule_title="Public hospital", + community_pharmacy_indicator="N", + ), + PrescribingTxt( + schedule_code=51, + prescribing_txt_id=100, + prescribing_type="CRITERIA", + prescribing_txt="Patient has the required clinical condition.", + ), + PrescribingTxt( + schedule_code=51, + prescribing_txt_id=200, + prescribing_type="PARAMETER", + prescribing_txt="Document the patient's clinical status.", + ), + PrescribingTxt( + schedule_code=51, + prescribing_txt_id=300, + prescribing_type="INDICATION", + prescribing_txt="Indication text is kept separate.", + ), + ] + ) + session.commit() + session.add_all( + [ + Criteria( + schedule_code=51, + criteria_prescribing_txt_id=100, + criteria_type="CLINICAL_PATIENT", + parameter_relationship="AND", + ), + Parameter( + schedule_code=51, + assessment_type="CLINICAL", + parameter_prescribing_txt_id=200, + parameter_type="CLINICAL_PATIENT", + ), + ] + ) + session.commit() + session.add( + Item( + schedule_code=51, + li_item_id="li-criteria", + pbs_code="C1", + program_code="CT", + ) + ) + session.commit() + session.add_all( + [ + ItemPrescribingTxtRltd( + schedule_code=51, + pbs_code="C1", + prescribing_txt_id=100, + pt_position=1, + ), + ItemPrescribingTxtRltd( + schedule_code=51, + pbs_code="C1", + prescribing_txt_id=300, + pt_position=2, + ), + CriteriaParameterRltd( + schedule_code=51, + criteria_prescribing_txt_id=100, + parameter_prescribing_txt_id=200, + pt_position=1, + ), + ItemDispensingRuleRltd( + schedule_code=51, + li_item_id="li-criteria", + dispensing_rule_mnem="HOSP", + ), + ] + ) + session.commit() + + item = session.get(Item, (51, "li-criteria")) + result = get_item_criteria_breakdown(session, item) + links = get_item_dispensing_rule_links(session, item) + + assert len(result.criteria) == 1 + criterion = result.criteria[0] + assert criterion.prescribing_text.prescribing_txt.startswith("Patient has") + assert criterion.criteria.criteria_type == "CLINICAL_PATIENT" + assert len(criterion.parameters) == 1 + assert criterion.parameters[0].prescribing_text.prescribing_txt.startswith("Document") + assert criterion.parameters[0].parameters[0].parameter_type == "CLINICAL_PATIENT" + assert result.program.program_title == "Chemotherapy" + assert [rule.dispensing_rule_mnem for rule in result.dispensing_rules] == ["HOSP"] + assert [link.dispensing_rule_mnem for link in links] == ["HOSP"] + + +def test_item_pricing_breakdown_returns_source_inputs_without_calculating_patient_amount( + session_factory, +): + with session_factory() as session: + session.add(Schedule(schedule_code=52, effective_date="2026-09-01", effective_year=2026)) + session.commit() + session.add_all( + [ + Program(schedule_code=52, program_code="CT", program_title="Chemotherapy"), + DispensingRule( + schedule_code=52, + dispensing_rule_mnem="HOSP", + dispensing_rule_title="Public hospital", + ), + ] + ) + session.commit() + session.add_all( + [ + Copayment(schedule_code=52, general=31.60, concessional=7.70), + Fee(schedule_code=52, program_code="CT", dispensing_fee_ready_prepared=1.00), + MarkupBand( + schedule_code=52, + program_code="CT", + dispensing_rule_mnem="HOSP", + markup_band_code="A", + limit=100.0, + variable=0.1, + offset=0.0, + fixed=1.0, + ), + ] + ) + session.commit() + session.add( + Item( + schedule_code=52, + li_item_id="li-price", + pbs_code="P1", + program_code="CT", + ) + ) + session.commit() + session.add_all( + [ + ItemDispensingRuleRltd( + schedule_code=52, + li_item_id="li-price", + dispensing_rule_mnem="HOSP", + special_patient_contribution=2.0, + ), + ItemPricingEvent( + schedule_code=52, + li_item_id="li-price", + percentage_applied=5.0, + event_type_code="APRIL_ADJUSTMENT", + ), + ] + ) + session.commit() + + result = get_item_pricing_breakdown(session, session.get(Item, (52, "li-price"))) + + assert result.copayment.general == 31.60 + assert result.fee.dispensing_fee_ready_prepared == 1.00 + assert result.dispensing_rule_links[0].special_patient_contribution == 2.0 + assert result.markup_bands[0].markup_band_code == "A" + assert result.pricing_events[0].event_type_code == "APRIL_ADJUSTMENT" From caa38d06d209583144d20ec64d817e37f6cbbc6a Mon Sep 17 00:00:00 2001 From: Georgie Kennedy Date: Mon, 14 Sep 2026 15:29:46 +1000 Subject: [PATCH 3/3] linting --- src/pbs_client/toolkit/core/service.py | 32 +++++++++++++++----------- 1 file changed, 18 insertions(+), 14 deletions(-) diff --git a/src/pbs_client/toolkit/core/service.py b/src/pbs_client/toolkit/core/service.py index 49b76e9..f669295 100644 --- a/src/pbs_client/toolkit/core/service.py +++ b/src/pbs_client/toolkit/core/service.py @@ -357,25 +357,29 @@ def get_item_atc_codes(session: Session, item: Item) -> list[ATC]: def get_item_dispensing_rule_links(session: Session, item: Item) -> list[ItemDispensingRuleRltd]: """Return the dispensing-rule relationship rows attached to an item.""" - return session.scalars( - select(ItemDispensingRuleRltd) - .where( - ItemDispensingRuleRltd.schedule_code == item.schedule_code, - ItemDispensingRuleRltd.li_item_id == item.li_item_id, - ) - .order_by(ItemDispensingRuleRltd.dispensing_rule_mnem) - ).all() + return list( + session.scalars( + select(ItemDispensingRuleRltd) + .where( + ItemDispensingRuleRltd.schedule_code == item.schedule_code, + ItemDispensingRuleRltd.li_item_id == item.li_item_id, + ) + .order_by(ItemDispensingRuleRltd.dispensing_rule_mnem) + ).all() + ) def get_item_amt_hierarchy(session: Session, item: Item) -> list[ItemAmt]: """Return every linked AMT row, keeping its PBS row identity intact.""" - rows = session.scalars( - select(ItemAmt).where( - ItemAmt.schedule_code == item.schedule_code, - ItemAmt.li_item_id == item.li_item_id, - ) - ).all() + rows = list( + session.scalars( + select(ItemAmt).where( + ItemAmt.schedule_code == item.schedule_code, + ItemAmt.li_item_id == item.li_item_id, + ) + ).all() + ) rows.sort( key=lambda row: ( _AMT_CONCEPT_TYPE_ORDER.get(row.concept_type_code, len(_AMT_CONCEPT_TYPE_ORDER)),