Add Itron Roseville AMI adapter and shared S3 drop-off transport helpers - #224
Open
dmarulli wants to merge 7 commits into
Open
Add Itron Roseville AMI adapter and shared S3 drop-off transport helpers#224dmarulli wants to merge 7 commits into
dmarulli wants to merge 7 commits into
Conversation
First adapter sourcing from a utility-pushed S3 file drop rather than a vendor API/SFTP/DB. City of Roseville exports two Itron CSV views -- Register (cumulative reads, 8-hour cadence) and Interval (hourly consumption) -- via Informatica into a CaDC-owned S3 prefix. The bucket lives in the CaDC AWS account, so the adapter authenticates with a read-only IAM keypair from source secrets rather than the EC2 instance role. Key behaviors: - Lists top-level keys only (Delimiter='/'), so an archive/ subfolder of superseded files is never re-ingested; selects files whose filename date range (YYYYMM or YYYYMMDD_YYYYMMDD suffix) overlaps the extract range; processes files oldest-LastModified-first so re-delivered correction-window rows resolve newest-wins in READINGS. - Register+Interval streams merge onto shared GeneralMeterReads at exact (device_id, flowtime), xylem_datalake-style. Interval timestamps mark the END of the measured hour (verified empirically: hour-ending alignment reconciles 99.6% of 8-hour register windows exactly). - Excludes the Itron faulted-channel sentinel (4294967294 = 2**32-2) on both file types from transformed reads while preserving it in the raw base tables; skips reads with blank timestamps or non-numeric values (counted and logged, not fatal). - Timestamps are Pacific wall-clock 'MM/DD/YYYY HH:MM:SS.ffffff'; account_id is permanently unavailable in this feed by design -- Location_ID equals Cayenta's SERVICE_POINT and is the billing join key. Verified: 280-test suite passes; transform validated against the full 2026-07 production files (485,601 register + 3,873,912 interval rows -> 53,805 meters, 3,873,923 reads, exact expected merge/filter counts).
Following the xylem_moulton_niguel convention for utility-specific adapters: state explicitly that this adapter was built specially for Roseville and is not compatible with other utilities. The CSVs are Roseville's own database views, not Itron's native ChoiceConnect export -- an Itron utility onboarding via Itron's standard hosted-SFTP path would need a different adapter. Also record the device_id choice (Meter_Serial_Number; every read row carries the serial, 1:1 meter:endpoint:location, meter swaps start a new SCD2 device) and note that endpoint_id = Roseville's Itron radio OID, consistent with xylem_moulton_niguel's ert_id -> endpoint_id mapping.
Two changes following repo conventions for utility-specific adapters:
1. Rename roseville -> itron_roseville everywhere (file, class, source
type, config/secrets dataclasses, base tables, sql/docs/test/fixture
files), matching the {vendor}_{utility} convention established by
xylem_moulton_niguel. Safe to rename the base tables because they have
not been created in Snowflake yet.
2. Extract the transport-level pieces that are not Roseville-specific
into amiadapters/adapters/s3_drop.py (following connections.py's
precedent for shared transport helpers): cross-account client from
source-secret keys, paginated subfolder-safe listing, CaDC
filename-date-token parsing and overlap selection with LastModified
ordering, and BOM-safe CSV-to-dataclass download. The S3 drop-off is
CaDC's standard intake pattern for utilities that push files, so the
next S3-drop source can reuse these directly; the adapter keeps only
what is truly Roseville's (filename regex, row dataclasses, sentinel,
transform).
Also registers the new source type in the all-config/all-secrets combined
fixtures and their two consumers (test_can_create_adapters,
test_get_database_config), and fixes a typing.Pattern deprecation.
Verified: 288-test suite passes; transform re-validated against the full
2026-07 production files with identical exact counts.
Operator-specific context (whose bucket, whose naming spec, whose delivery mechanism) belongs at the adapter level, not in shared framework code. Rewrite the s3_drop module docstring to describe the generic S3-drop-source mechanics on their own terms; the operator- and utility-specific rationale stays in itron_roseville.py and its docs page.
dmarulli
changed the title
Same review lens as the billing parser: encode the generating mechanism, not the observed instances. 1. Fault-code filter is a threshold, not one magic number: Itron emits 32-bit max-value error codes (4294967294 = 2**32-2 observed; neighbors like 2**32-1 equally plausible). Real registers sit orders of magnitude lower (largest genuine: ~97.8M CF), so values >= 4e9 are excluded from transformed reads (still preserved in raw base tables). 2. Unit normalization strips the Itron commodity suffix (_WAT) generically instead of mapping only CF_WAT, with the GAL->GALLON alias map_reading's vocabulary requires (same aliasing as xylem_datalake). A future GAL_WAT maps cleanly; unknown units still raise. 3. Observability: blank Meter_Serial_Number rows are now counted (missing_device_id) like every other skip, and register-value overwrites from re-delivered corrections increment overwritten_differing symmetrically with the interval path. Verified: 291-test suite passes (new tests for threshold neighbors, GAL_WAT conversion, missing-serial counting); transform re-validated against the full 2026-07 production files with identical counts (53,805 meters / 3,873,923 reads / 485,294 merged / 296 fault rows excluded).
dmarulli
marked this pull request as ready for review
July 14, 2026 19:11
dmarulli
requested review from
christophertull,
jin-current,
mdowell12 and
pat-current
as code owners
July 14, 2026 19:11
mdowell12
reviewed
Jul 14, 2026
mdowell12
left a comment
mdowell12
left a comment
Collaborator
There was a problem hiding this comment.
👍 looking great! I left a couple nitpicks, feel free to ignore.
|
|
||
| logger = logging.getLogger(__name__) | ||
|
|
||
|
|
Collaborator
There was a problem hiding this comment.
Nits:
Some of this code (probably like 20%) is duplicated in amiadapters/outputs/s3.py, so maybe the perfect implementation would abstract what's shared. I think it's fine to duplicate here though.
Also, do we think another adapter will use this, filename conventions included? If that feels unlikely, maybe this code belongs inside the adapter.
| """ | ||
| Roseville pushes a file per day covering a rolling ~3-day window. We use | ||
| a 4-day extract window so a file pushed late (or a day skipped on our | ||
| side) is still picked up by filename-date overlap; re-processing is |
Collaborator
There was a problem hiding this comment.
Nit: I've noticed in my work that Claude is overzealous with code comments. This one feels longer than necessary, e.g. I don't think it needs to discuss idempotent upserts which are the norm for the whole pipeline. I've been trying to cut out that fluff to keep comments useful.
|
|
||
| def _get_s3_client(self): | ||
| if self._s3_client is None: | ||
| # The bucket lives in the CaDC AWS account, not the ami-connect |
Collaborator
Author
|
@mdowell12 - thanks for the notes! Always appreciated. I've made some tweaks based on them. On the proposed s3 module stuff, I went back and forth just a bit, but decided to bring that logic inside the Roseville adapter for now, until there's actually a second adapter that would use it. I'm just waiting on a final thumbs up from Roseville, then I will merge, deploy, etc. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What
New
itron_rosevilleadapter for City of Roseville's Itron AMI data — the first adapter whose source is a utility-pushed S3 file drop rather than a vendor API/SFTP/database. Roseville exports two CSV views (Register: cumulative reads at 8-hour cadence; Interval: hourly consumption) via Informatica into a CaDC-owned S3 prefix.The transport-level pieces that aren't Roseville-specific live in a new shared module
amiadapters/adapters/s3_drop.py(followingconnections.py's precedent for shared transport helpers): cross-account client from source-secret keys, paginated subfolder-safe listing, filename date-token parsing + overlap selection with LastModified ordering, BOM-safe CSV→dataclass download. The module's documentation is deliberately operator-neutral — operator- and utility-specific context (whose bucket, whose naming spec, delivery mechanism) lives at the adapter level. The adapter keeps what is truly Roseville's: filename regex, row dataclasses, error sentinel, transform.Naming follows the
{vendor}_{utility}convention established byxylem_moulton_niguel(file, class, source type, base tables, docs, fixtures). Like that adapter, this one was built specially for Roseville and is not compatible with other utilities — the CSVs are Roseville IT's own database views, not Itron's native ChoiceConnect export; an Itron utility onboarding via Itron's standard hosted-SFTP path would need a different adapter.Design decisions
ItronRosevilleSecretscarries a read-only IAM keypair — consistent with how every adapter treats source credentials. The boto3 client is built lazily in_extract(adapters are constructed at DAG-parse time).device_id = meter_id = Meter_Serial_Number(every read row carries the serial; 1:1 meter:endpoint:location; meter swaps start a new SCD2 device).endpoint_id = EndpointID(Itron's OID arc2.16.840.1.114416) — same convention asxylem_moulton_niguel'sert_id → endpoint_id.Delimiter="/"): anarchive/folder of superseded files is structurally invisible.LastModified-first so the newest delivered value wins per (meter, timestamp) in READINGS.4294967294(2³²−2) is excluded on both file types; blank-timestamp and non-numeric-value rows are counted+skipped; BOM-tolerant decoding;s3_prefixtrailing slash normalized (load-bearing withDelimiter="/"); tz-aware manual extract ranges normalized.account_idis permanently unavailable by design (Roseville's AMI system receives no account info from CIS).Location_ID= CayentaSERVICE_POINT, the billing join key — validated at 99.8% match against Roseville billing, both directions.Known limitations (documented in code + docs)
max()— lexicographic on VARCHAR; READINGS is the corrections-accurate record.Verification
s3_droptests, config/fixture registration incl. the all-config combined tests), black-formatted.Deploy checklist (not in this PR — requires prod access)
s3:GetObject/s3:ListBucketon the Roseville prefix.sql/itron-roseville-base.sqlin Snowflake (base tables must exist before first run).python cli.py config add-source cadc_roseville itron_roseville America/Los_Angeles --config s3_bucket=cadc-ami --config s3_prefix=rosevillecityof/ --config s3_region=us-east-1 --sinks <snowflake_sink>python cli.py config update-secret cadc_roseville --source-type itron_roseville --secret aws_access_key_id=... --secret aws_secret_access_key=...archive/.bash ./deploy.sh cadc).