nemo_voice_agent.evaluation.tools.tau2_telecom_predicates

Telecom DB-state predicate ports for the db_state_assertions framework.

Six predicates total — four user-side and two agent-side. Each one is registered under domain="tau2_telecom" via the flat (domain, func_name) predicate registry; the runner dispatches by func_name only (predicates are side-agnostic, the caller picks the right DB).

User-side predicates (read shared_state["db"] on the user bot):

assert_mobile_data_status(db, expected_status) — single derived bool summarizing whether mobile data is functional. Combines 6 device fields via _get_mobile_data_working.
assert_internet_speed(db, expected_speed, expected_desc=None) — runs a deterministic simulated speed test against device + surroundings state; checks simulated_speed >= expected_speed (and optionally that the speed-category label matches).
assert_service_status(db, expected_status) — direct check on device.network_connection_status.
assert_can_send_mms(db, expected_status) — multi-field check (mobile data + network tech + MMSC config + messaging-app permissions).

Agent-side predicates (read shared_state["db"] on the agent bot):

assert_data_refueling_amount(db, customer_id, line_id, expected_amount) — checks the line’s data_refueling_gb field against an expected value (within float tolerance).
assert_no_overdue_bill(db, overdue_bill_id) — true if the named bill is either absent from the DB or marked PAID.

Each predicate returns bool and never raises (the dispatcher catches, but we don’t rely on that — upstream’s semantics for these is “false on unexpected conditions”). The matching assert_value in the upstream task JSON is what determines pass/fail; the predicate just produces the underlying signal.

Module Contents

Functions

Name	Description
`_can_send_mms`	Mirror of upstream’s `TelecomUserTools._can_send_mms`.
`_get_bill_by_id`	Iterate `db["bills"]` looking for `bill_id`. Returns `None` if not found.
`_get_customer_by_id`	Iterate `db["customers"]` looking for `customer_id`. Returns
`_get_line_by_id`	Iterate `db["lines"]` looking for `line_id`.
`_get_mobile_data_working`	Return True if mobile data is currently functional, False otherwise.
`_get_target_line`	Lookup a line scoped to a customer.
`_run_speed_test`	Simulate a deterministic speed test against current device state.
`assert_can_send_mms`	Assert that the messaging-app MMS-send capability matches `expected_status`.
`assert_data_refueling_amount`	Assert that the named line’s `data_refueling_gb` matches `expected_amount`.
`assert_internet_speed`	Assert that the simulated speed meets a minimum threshold.
`assert_mobile_data_status`	Assert that mobile data working status matches `expected_status`.
`assert_no_overdue_bill`	Assert that `overdue_bill_id` is either absent or paid.
`assert_service_status`	Assert that `device.network_connection_status` matches `expected_status`.

API

nemo_voice_agent.evaluation.tools.tau2_telecom_predicates._can_send_mms(
    db: typing.Dict[str, typing.Any]
) -> bool

Mirror of upstream’s TelecomUserTools._can_send_mms.

MMS sending requires all of:

mobile data working (delegates to _get_mobile_data_working)
network technology is 3G or higher (NOT 2G)
WiFi-calling-with-MMS-over-WiFi is OFF (carrier limitation)
MMSC URL is configured (non-None)
The messaging app exists in app_statuses and has both storage and sms permissions

nemo_voice_agent.evaluation.tools.tau2_telecom_predicates._get_bill_by_id(
    db: typing.Dict[str, typing.Any],
    bill_id: str
) -> typing.Optional[typing.Dict[str, typing.Any]]

Iterate db["bills"] looking for bill_id. Returns None if not found.

Upstream raises ValueError here; we return None because assert_no_overdue_bill treats “bill not found” as success and the None-check is simpler than try/except for that semantic.

nemo_voice_agent.evaluation.tools.tau2_telecom_predicates._get_customer_by_id(
    db: typing.Dict[str, typing.Any],
    customer_id: str
) -> typing.Optional[typing.Dict[str, typing.Any]]

Iterate db["customers"] looking for customer_id. Returns None if not found (upstream raises). Same rationale as _get_bill_by_id — caller decides the missing-customer semantic.

nemo_voice_agent.evaluation.tools.tau2_telecom_predicates._get_line_by_id(
    db: typing.Dict[str, typing.Any],
    line_id: str
) -> typing.Optional[typing.Dict[str, typing.Any]]

Iterate db["lines"] looking for line_id.

nemo_voice_agent.evaluation.tools.tau2_telecom_predicates._get_mobile_data_working(
    db: typing.Dict[str, typing.Any]
) -> bool

Return True if mobile data is currently functional, False otherwise.

Mirrors upstream’s TelecomUserTools._get_mobile_data_working. Mobile data is NOT working when any of the following are true (early-return short-circuit, in this order):

Airplane mode is on
Network signal strength is NONE
Network connection status is NO_SERVICE
User is abroad AND (roaming not enabled on device OR roaming not allowed by location)
Master data switch is off
Surroundings flag says data usage exceeded

nemo_voice_agent.evaluation.tools.tau2_telecom_predicates._get_target_line(
    db: typing.Dict[str, typing.Any],
    customer_id: str,
    line_id: str
) -> typing.Optional[typing.Dict[str, typing.Any]]

Lookup a line scoped to a customer.

Mirrors upstream’s _get_target_line: verifies line_id is in customer.line_ids before returning the line. Returns None if either the customer or the line isn’t found, or if the line isn’t associated with the customer.

nemo_voice_agent.evaluation.tools.tau2_telecom_predicates._run_speed_test(
    db: typing.Dict[str, typing.Any]
) -> typing.Tuple[typing.Optional[float], typing.Optional[str]]

Simulate a deterministic speed test against current device state.

Mirrors upstream’s TelecomUserTools._run_speed_test. Returns (speed_mbps, description) where:

speed_mbps is None when mobile data isn’t working (description = “No Connection” in that case).
Otherwise speed is computed as (min + max) / 2 * signal_factor * base_speed_factor with:
- (min, max) from the network-technology speed-range table
- signal_factor from the signal-strength multiplier table
- base_speed_factor reduced by VPN-with-poor-performance (×0.1) and data-saver-mode (×0.2)
description is a verbal label (“Very Poor” / “Poor” / “Fair” / “Good” / “Excellent”) bucketed from the computed speed.

nemo_voice_agent.evaluation.tools.tau2_telecom_predicates.assert_can_send_mms(
    db: typing.Dict[str, typing.Any],
    expected_status: bool
) -> bool

Assert that the messaging-app MMS-send capability matches expected_status.

nemo_voice_agent.evaluation.tools.tau2_telecom_predicates.assert_data_refueling_amount(
    db: typing.Dict[str, typing.Any],
    customer_id: str,
    line_id: str,
    expected_amount: float
) -> bool

Assert that the named line’s data_refueling_gb matches expected_amount.

Float-tolerance compare (abs(diff) < 1e-6). Returns False if the line cannot be located (vs upstream which raises) — predicate-as-bool semantics are simpler for the dispatcher.

nemo_voice_agent.evaluation.tools.tau2_telecom_predicates.assert_internet_speed(
    db: typing.Dict[str, typing.Any],
    expected_speed: float,
    expected_desc: typing.Optional[str] = None
) -> bool

Assert that the simulated speed meets a minimum threshold.

Parameters:

expected_speed

float

Minimum acceptable Mbps. The simulated speed must be >= expected_speed for the assertion to pass. When the speed test fails entirely (mobile data not working), the speed is treated as 0.0 — assertion passes only if expected_speed is also <= 0.

expected_desc

Optional[str]" default="None

Optional verbal-bucket constraint. When provided, the simulated speed’s category label (case-insensitive) must equal this value AND the numeric threshold must also be met. Common values from upstream tasks: "excellent", "good".

nemo_voice_agent.evaluation.tools.tau2_telecom_predicates.assert_mobile_data_status(
    db: typing.Dict[str, typing.Any],
    expected_status: bool
) -> bool

Assert that mobile data working status matches expected_status.

nemo_voice_agent.evaluation.tools.tau2_telecom_predicates.assert_no_overdue_bill(
    db: typing.Dict[str, typing.Any],
    overdue_bill_id: str
) -> bool

Assert that overdue_bill_id is either absent or paid.

Mirrors upstream: returns True when (a) the bill isn’t in the DB at all (missing means “no longer overdue”), or (b) the bill exists and its status is PAID. Returns False otherwise (e.g. still overdue, draft, awaiting payment, disputed).

nemo_voice_agent.evaluation.tools.tau2_telecom_predicates.assert_service_status(
    db: typing.Dict[str, typing.Any],
    expected_status: str
) -> bool

Assert that device.network_connection_status matches expected_status.

expected_status is the string value of a NetworkStatus enum (e.g. "connected", "no_service").