Skip to main content

Implementation Plan: Timeseries Processing Upgrade with LPS Tags

Overview​

This implementation plan covers the development of parse_timeseries_csv_v2.py with two major upgrades:

  1. Frequency-based tag creation: Dynamically detect reading frequency and create appropriate tags
  2. LPS demand tags: Create parallel liters-per-second tags for hydraulic modeling

Prerequisites​

1. Environment Setup​

  • Ensure Python 3.8+ is installed
  • Update requirements.txt with new dependencies:
    pandas>=2.0.0
    numpy>=1.24.0
    psycopg2-binary>=2.9.0
  • Run pip install -r requirements.txt

2. Database Access​

  • Verify PostgreSQL connection details
  • Ensure write access to DeviceTags and DeviceTagData tables
  • Consider using test database for initial runs

3. Input Data​

  • Identify timeseries CSV file location
  • Verify CSV encoding (cp1252)
  • Confirm required columns exist:
    • REG_SERIAL
    • READING_YEAR
    • READING_PERIOD
    • CONSUMPTION

Phase 1: Analysis Script (1-2 days)​

Objective​

Create a read-only analysis script to profile existing data and validate design assumptions.

Tasks​

1.1 Create v2/analyze_frequency_patterns.py​

"""
Analysis script to profile timeseries data frequency patterns.
NO database writes - read-only analysis.
"""

Note: All new scripts will be created in the v2/ subdirectory. Legacy scripts remain in v1/ for reference.

Functions to implement:

  • load_csv(): Load CSV with proper encoding
  • analyze_frequencies(): Count periods per device per year
  • classify_frequency(): Map period counts to frequency types
  • generate_summary_report(): Create human-readable report
  • identify_anomalies(): Flag unusual period counts (3, 5-11, >12)
  • find_mixed_frequency_devices(): List devices with multiple frequencies

1.2 Generate Analysis Report​

  • Run analysis on full dataset: python v2/analyze_frequency_patterns.py path/to/timeseries.csv
  • Review output report covering:
    • Total devices analyzed
    • Frequency distribution (annual/semiannual/quarterly/monthly counts)
    • Devices with mixed frequencies (% and count)
    • Anomalies (unexpected period counts)
    • Sample device timelines (show frequency transitions)

1.3 Validate Assumptions​

  • Confirm frequency classification logic matches real data
  • Verify timestamp conversion will work for all frequencies
  • Identify any data quality issues requiring special handling

Deliverables:

  • v2/analyze_frequency_patterns.py
  • Analysis report (text or CSV)
  • Decision on handling anomalies

Phase 2: Core Implementation (3-5 days)​

Objective​

Develop v2/parse_timeseries_csv_v2.py with frequency detection and consumption tag creation.

Architecture​

v2/parse_timeseries_csv_v2.py
├── Configuration
│   ├── Database connection
│   ├── Table names
│   └── Constants (conversion factors)
├── Data Loading
│   └── load_csv_to_dataframe()
├── Frequency Analysis
│   ├── analyze_frequency_patterns()
│   ├── classify_frequency()
│   └── validate_frequency_data()
├── DeviceTag Creation
│   ├── identify_device_tag_needs()
│   ├── generate_device_tags()
│   └── bulk_create_device_tags()
├── DeviceTagDatum Creation
│   ├── convert_reading_to_unix_time()
│   ├── prepare_datum_records()
│   └── bulk_create_device_tag_data()
├── Logging & Reporting
│   ├── setup_logging()
│   ├── log_progress()
│   └── generate_summary_report()
└── Main Execution
    ├── Dry-run mode (--dry-run)
    └── Production mode

Tasks​

2.1 Setup & Configuration​

  • Create v2/parse_timeseries_csv_v2.py

  • Import dependencies (pandas, psycopg2, datetime, logging)

  • Define configuration constants:

    # Database tables
    DEVICES_TABLE = "Devices"
    DEVICE_TAGS_TABLE = "DeviceTags"
    DEVICE_TAG_DATA_TABLE = "DeviceTagData"

    # Frequency classification
    FREQUENCY_MAP = {
      1: 'annual',
      2: 'semiannual',
      4: 'quarterly',
      12: 'monthly'
    }

  • Setup logging with appropriate verbosity levels

2.2 Data Loading Module​

  • Implement load_csv_to_dataframe(csv_path)
    • Handle cp1252 encoding
    • Validate required columns exist
    • Handle missing/null values
    • Log data summary (rows, devices, year range)

2.3 Frequency Analysis Module​

  • Implement analyze_frequency_patterns(df)

    • Group by REG_SERIAL and READING_YEAR
    • Count unique READING_PERIOD per group
    • Classify frequency using FREQUENCY_MAP
    • Return frequency analysis dataframe

  • Implement classify_frequency(period_count)

    • Map period count to frequency type
    • Log warnings for unexpected counts
    • Return frequency string or 'unknown_X'

  • Implement validate_frequency_data(df)

    • Check for duplicate (device, year, period) combinations
    • Identify gaps in period sequences
    • Log data quality issues

2.4 DeviceTag Creation Module​

  • Implement identify_device_tag_needs(df, frequency_df)

    • Merge frequency analysis with main dataframe
    • Get unique (device, frequency) combinations
    • Return dataframe of needed tags

  • Implement fetch_device_mappings(cursor)

    • Query Devices table for Id, SerialNumber, Name
    • Return as pandas DataFrame
    • Log devices not found in database

  • Implement generate_device_tags(unique_combinations, devices_df)

    • Create consumption tag records:
      • Name: {DeviceName}_{frequency}_consumption
      • TagType: {frequency}_consumption
      • Unit: cubic feet
      • DeviceId: from devices_df
    • Return list of tag dictionaries

  • Implement bulk_create_device_tags(cursor, device_tags)

    • Use executemany() or execute_values() for performance
    • Handle ON CONFLICT if tags already exist
    • Log number of tags created
    • Return created tag IDs

2.5 Timestamp Conversion Module​

  • Implement convert_reading_to_unix_time(year, period)
    • Map period to month (quarterly mapping as default):
      • Period 1 → January 15
      • Period 2 → April 15
      • Period 3 → July 15
      • Period 4 → October 15
    • For monthly data: period = month
    • Use UTC timezone
    • Return Unix timestamp (integer)

2.6 DeviceTagDatum Creation Module​

  • Implement prepare_datum_records(df, frequency_df, device_tags_db)

    • Merge frequency type back to main dataframe
    • Merge device mappings
    • Merge DeviceTag IDs
    • Convert to Unix timestamps
    • Return list of (DeviceTagId, UnixTime, Value) tuples

  • Implement bulk_create_device_tag_data(cursor, datum_records)

    • Use executemany() with ON CONFLICT
    • Batch inserts if dataset is large (chunks of 10k)
    • Log progress periodically
    • Return number of records created/updated

2.7 Reporting Module​

  • Implement generate_summary_report(stats)
    • Total devices processed
    • Total DeviceTags created
    • Frequency distribution
    • Total DeviceTagDatum records created
    • Processing time
    • Any errors or warnings
    • Save to file and log to console

2.8 Main Execution Flow​

  • Implement main() function:

    1. Parse command-line arguments (--csv, --dry-run, --db-config)
    2. Setup logging
    3. Load CSV data
    4. Analyze frequencies
    5. Identify device tag needs
    6. (If not dry-run) Create DeviceTags
    7. (If not dry-run) Create DeviceTagDatum records
    8. Generate summary report
    9. Close database connection

  • Add dry-run mode:

    • All analysis runs normally
    • No database writes
    • Print what would be created
    • Show sample records

Deliverables:

  • v2/parse_timeseries_csv_v2.py (functional, with consumption tags only)
  • Unit tests for key functions
  • Documentation strings for all functions

Phase 3: LPS Tag Integration (2-3 days)​

Objective​

Add parallel LPS (liters per second) tag creation to the existing implementation.

Tasks​

3.1 Add Conversion Constants​

  • Add to configuration section:

    # Unit conversion constants
    CUBIC_FEET_TO_LITERS = 28.3168

    # Time period constants (seconds)
    SECONDS_PER_MONTH = 30.44 * 24 * 3600  # 2,629,440
    SECONDS_PER_QUARTER = 3 * SECONDS_PER_MONTH  # 7,888,320
    SECONDS_PER_SEMIANNUAL = 6 * SECONDS_PER_MONTH  # 15,776,640
    SECONDS_PER_ANNUAL = 365.25 * 24 * 3600  # 31,557,600

    PERIOD_SECONDS = {
      'monthly': SECONDS_PER_MONTH,
      'quarterly': SECONDS_PER_QUARTER,
      'semiannual': SECONDS_PER_SEMIANNUAL,
      'annual': SECONDS_PER_ANNUAL
    }

3.2 Implement Conversion Function​

  • Create convert_cuft_to_lps(consumption_cuft, frequency_type)
    • Convert cubic feet to liters
    • Get period seconds from PERIOD_SECONDS dict
    • Divide liters by seconds
    • Return lps value (float)
    • Handle unknown frequency types gracefully

3.3 Modify DeviceTag Generation​

  • Update generate_device_tags() to create both tag types:
    • For each (device, frequency) combination:
      • Create consumption tag (existing logic)
      • Create lps tag:
        • Name: {DeviceName}_{frequency}_demand_lps
        • TagType: {frequency}_demand_lps
        • Unit: liters/second
        • DeviceId: same as consumption tag
    • Return combined list (2x tags)

3.4 Modify Datum Record Preparation​

  • Update prepare_datum_records() to handle both tag types:
    • Fetch all DeviceTags (consumption + lps) from database
    • Separate into two dataframes by TagType suffix
    • Prepare consumption records (existing logic)
    • Prepare lps records:
      • Apply convert_cuft_to_lps() to each record
      • Use same UnixTime as consumption records
      • Store converted value
    • Return combined list of records

3.5 Update Reporting​

  • Modify generate_summary_report() to show:
    • Total consumption tags created
    • Total lps tags created
    • Ratio should be 1:1
    • Sample conversion calculations for validation

3.6 Add Validation Tests​

  • Create validate_lps_conversions() function:
    • Spot-check sample conversions
    • Compare against manual calculations
    • Verify lps values are reasonable (not negative, not absurdly large)
    • Log any suspicious values

Deliverables:

  • Updated v2/parse_timeseries_csv_v2.py with LPS support
  • Validation tests for conversion accuracy
  • Updated documentation

Phase 4: Testing (2-3 days)​

Objective​

Thoroughly test the implementation on progressively larger datasets.

Test Strategy​

4.1 Unit Testing​

  • Test frequency classification with edge cases:

    • Single period per year → annual
    • Two periods → semiannual
    • Four periods → quarterly
    • Twelve periods → monthly
    • Unexpected counts (3, 5-11, >12)

  • Test timestamp conversion:

    • Verify January 15th for period 1
    • Verify quarterly mappings
    • Test year boundaries (2000, 2023, etc.)

  • Test LPS conversion:

    • Annual: 1000 cu ft over 1 year → expected lps
    • Quarterly: 250 cu ft over 3 months → expected lps
    • Monthly: 83.33 cu ft over 1 month → expected lps
    • Verify conversion factor is correct

4.2 Integration Testing with Subset​

  • Create test CSV with known data:

    • Device A: Annual readings (5 years)
    • Device B: Quarterly readings (2 years)
    • Device C: Mixed (annual → quarterly transition)
    • Device D: Monthly readings (1 year)

  • Run script on test data (dry-run mode)

    • Verify correct frequency classification
    • Verify tag names are correct
    • Verify datum count matches expectations

  • Run script on test database

    • Verify DeviceTags created correctly
    • Verify DeviceTagDatum records created
    • Query database to validate data
    • Verify consumption and lps values are paired correctly

4.3 Validation Queries​

  • Write SQL queries to validate results:

    -- Check tag pairs (should be 1:1 ratio)
    SELECT
      COUNT(CASE WHEN "TagType" LIKE '%_consumption' THEN 1 END) as consumption_tags,
      COUNT(CASE WHEN "TagType" LIKE '%_demand_lps' THEN 1 END) as lps_tags
    FROM "DeviceTags";

    -- Check datum count per tag type
    SELECT
      dt."TagType",
      COUNT(*) as datum_count
    FROM "DeviceTagData" dtd
    JOIN "DeviceTags" dt ON dtd."DeviceTagId" = dt."Id"
    GROUP BY dt."TagType"
    ORDER BY dt."TagType";

    -- Verify consumption and lps have same timestamps for a device
    SELECT
      d."Name",
      dt_cons."Name" as consumption_tag,
      dt_lps."Name" as lps_tag,
      dtd_cons."UnixTime",
      dtd_lps."UnixTime",
      dtd_cons."Value" as cuft,
      dtd_lps."Value" as lps
    FROM "Devices" d
    JOIN "DeviceTags" dt_cons ON d."Id" = dt_cons."DeviceId"
      AND dt_cons."TagType" LIKE '%_consumption'
    JOIN "DeviceTags" dt_lps ON d."Id" = dt_lps."DeviceId"
      AND dt_lps."TagType" LIKE '%_demand_lps'
      AND REPLACE(dt_cons."TagType", '_consumption', '') = REPLACE(dt_lps."TagType", '_demand_lps', '')
    JOIN "DeviceTagData" dtd_cons ON dt_cons."Id" = dtd_cons."DeviceTagId"
    JOIN "DeviceTagData" dtd_lps ON dt_lps."Id" = dtd_lps."DeviceTagId"
      AND dtd_cons."UnixTime" = dtd_lps."UnixTime"
    WHERE d."Name" = 'WTRMN00345'
    ORDER BY dtd_cons."UnixTime";

  • Manually verify sample calculations:

    • Pick 5 random records
    • Calculate expected lps by hand
    • Compare with database values
    • Tolerance: ±0.01% due to floating point

4.4 Performance Testing​

  • Test on medium dataset (1000 devices, ~10k records)

    • Measure processing time
    • Monitor memory usage
    • Target: < 1 minute

  • Test on large dataset (10k devices, ~100k records)

    • Measure processing time
    • Monitor memory usage
    • Target: < 10 minutes

  • Identify bottlenecks if performance is poor:

    • Database inserts (try batch sizing)
    • DataFrame merges (optimize join keys)
    • Conversion calculations (vectorize with numpy)

Deliverables:

  • Test suite with passing tests
  • Test report documenting results
  • Performance benchmarks

Phase 5: Production Deployment (1-2 days)​

Objective​

Run the script on production data and verify results.

Pre-Deployment Checklist​

  • All tests passing
  • Code reviewed
  • Documentation complete
  • Database backup created
  • Rollback plan documented

Deployment Steps​

5.1 Pre-Production Validation​

  • Run analysis script on production CSV

    • Compare results with Phase 1 analysis
    • Verify no unexpected changes in data

  • Run main script in dry-run mode

    • Review what will be created
    • Verify tag counts match expectations
    • Check for any errors or warnings

5.2 Production Run (Option A: Clean Database)​

If starting fresh or in test environment:

  • Clear existing DeviceTag and DeviceTagDatum records (if appropriate)
  • Run parse_timeseries_csv_v2.py on full dataset
  • Monitor progress (watch logs)
  • Wait for completion
  • Review summary report

5.3 Production Run (Option B: Incremental Update)​

If adding to existing data:

  • Ensure ON CONFLICT handling is correct
  • Run script with appropriate conflict resolution
  • Monitor for conflicts/duplicates
  • Review update counts

5.4 Post-Deployment Validation​

  • Run validation SQL queries (from Phase 4.3)

  • Verify total counts:

    • DeviceTags created = 2x unique (device, frequency) combinations
    • DeviceTagDatum records = 2x (one for consumption, one for lps)

  • Spot-check devices:

    • Select 10 random devices
    • Verify they have correct tags
    • Verify datum values are reasonable
    • Verify consumption/lps pairs match

  • Check for anomalies:

    • Devices with missing tags
    • Orphaned datum records
    • Unexpected tag types

5.5 Generate Final Report​

  • Create production run report including:
    • Total devices processed
    • Total tags created (by type)
    • Total datum records created
    • Processing time
    • Data quality issues encountered
    • Validation results
    • Sample queries with results

Deliverables:

  • Production-ready v2/parse_timeseries_csv_v2.py
  • Production run report
  • Updated documentation

Phase 6: Documentation & Handoff (1 day)​

Objective​

Document the new system for future maintenance and use.

Tasks​

6.1 Update README.md​

  • Add section on frequency-based processing

  • Document LPS tag functionality

  • Add usage examples:

    # Analysis only
    python v2/analyze_frequency_patterns.py timeseries.csv

    # Dry run
    python v2/parse_timeseries_csv_v2.py --csv timeseries.csv --dry-run

    # Production run
    python v2/parse_timeseries_csv_v2.py --csv timeseries.csv

  • Document command-line arguments

  • Add troubleshooting section

6.2 Code Documentation​

  • Ensure all functions have docstrings
  • Add inline comments for complex logic
  • Document conversion formulas with references

6.3 Create Operational Guide​

  • Document when to run the script (quarterly? annually?)
  • Document monitoring procedures
  • Document common issues and solutions
  • Document validation procedures

6.4 Knowledge Transfer​

  • Walk through code with team
  • Demonstrate dry-run and production modes
  • Show how to read logs and reports
  • Show validation queries

Deliverables:

  • Updated README.md
  • Operational guide document
  • Code with complete documentation
  • Knowledge transfer session

Timeline Summary​

PhaseDurationDependencies
1. Analysis Script1-2 daysNone
2. Core Implementation3-5 daysPhase 1 complete
3. LPS Integration2-3 daysPhase 2 complete
4. Testing2-3 daysPhase 3 complete
5. Production Deployment1-2 daysPhase 4 complete, approvals
6. Documentation & Handoff1 dayPhase 5 complete
Total10-16 days~2-3 weeks

Risk Mitigation​

Technical Risks​

RiskImpactLikelihoodMitigation
Large dataset causes memory issuesHighMediumUse pandas chunking, process in batches
Conversion calculations are incorrectHighLowExtensive testing, manual validation
Database locks during bulk insertsMediumMediumUse smaller batch sizes, run during off-hours
Existing data conflictsMediumMediumImplement robust ON CONFLICT handling
Unexpected frequency patterns in dataMediumMediumAnalysis phase identifies issues early

Process Risks​

RiskImpactLikelihoodMitigation
Timeline slipsMediumMediumBuilt-in buffer time, prioritize core features
Requirements change mid-developmentMediumLowModular design allows easy adjustments
Data quality issues discovered lateHighLowPhase 1 analysis catches issues early
Production deployment failsHighLowDry-run validation, backup/rollback plan

Success Criteria​

Functional Requirements​

  • Document updated with LPS specification
  • Analysis script successfully profiles data
  • Main script correctly classifies frequencies
  • Consumption tags created with proper naming
  • LPS tags created with proper naming
  • Datum records created for both tag types
  • Conversion calculations are accurate
  • Script handles mixed-frequency devices

Performance Requirements​

  • Process 100k records in < 10 minutes
  • Memory usage stays reasonable (< 2GB)
  • Database inserts use efficient bulk operations

Quality Requirements​

  • All unit tests pass
  • Integration tests pass on sample data
  • Manual validation confirms accuracy
  • No data loss or corruption
  • Comprehensive logging for debugging
  • Clear error messages

Documentation Requirements​

  • Code is well-documented
  • README.md is updated
  • Operational guide is complete
  • Team is trained on new system

Rollback Plan​

If issues are discovered after production deployment:

  1. Identify the issue

    • Review logs
    • Check validation queries
    • Identify affected records

  2. Assess impact

    • How many devices affected?
    • Is data corrupted or just incomplete?
    • Can we fix in-place or need to rollback?

  3. Rollback procedure (if needed)

    • Restore database from backup
    • Re-run with corrected script
    • Validate results

  4. Fix-in-place procedure (if possible)

    • Write corrective SQL script
    • Test on subset
    • Apply to production
    • Validate results

Appendix A: Key Files​

Directory Structure​

devices/
├── v1/                           # Legacy scripts (reference only)
│   ├── parse_timeseries_csv.py
│   ├── parse_timeseries_csv_bulk.py
│   └── [other legacy scripts]
├── v2/                           # New implementation
│   ├── analyze_frequency_patterns.py
│   └── parse_timeseries_csv_v2.py
├── DEVICE_TIMESERIES_UPGRADE_OCTOBER_2025.md
├── IMPLEMENTATION_PLAN.md
└── README.md

New Files Created​

  • v2/analyze_frequency_patterns.py - Analysis script
  • v2/parse_timeseries_csv_v2.py - Main processing script
  • IMPLEMENTATION_PLAN.md - This document

Modified Files​

  • DEVICE_TIMESERIES_UPGRADE_OCTOBER_2025.md - Updated with LPS specification
  • README.md - Updated with new functionality
  • requirements.txt - Updated with pandas/numpy dependencies

Database Tables​

  • Devices - Read for SerialNumber to Id mapping
  • DeviceTags - Created/updated by script
  • DeviceTagData - Created/updated by script

Appendix B: Conversion Formula Reference​

Cubic Feet to Liters​

1 cubic foot = 28.3168 liters

Seconds per Period​

Month (30.44 days avg):     2,629,440 seconds
Quarter (3 months):         7,888,320 seconds
Semi-annual (6 months):    15,776,640 seconds
Annual (365.25 days):      31,557,600 seconds

LPS Calculation Example​

Given:
- Consumption: 1000 cubic feet
- Frequency: Quarterly

Step 1: Convert to liters
1000 cu ft × 28.3168 L/cu ft = 28,316.8 liters

Step 2: Divide by seconds in quarter
28,316.8 L ÷ 7,888,320 seconds = 0.003589 L/s

Result: 0.003589 lps (average demand for that quarter)

Appendix C: Example Validation Calculations​

Device: WTRMN00345, Annual Reading​

Input:

  • Year: 2015
  • Period: 1
  • Consumption: 15,000 cubic feet
  • Frequency: Annual

Expected Consumption Tag:

  • DeviceTag: WTRMN00345_annual_consumption
  • UnixTime: 1421280000 (2015-01-15 00:00:00 UTC)
  • Value: 15000.0
  • Unit: cubic feet

Expected LPS Tag:

  • DeviceTag: WTRMN00345_annual_demand_lps
  • UnixTime: 1421280000 (2015-01-15 00:00:00 UTC)
  • Value: 15000 × 28.3168 ÷ 31,557,600 = 0.01346 L/s
  • Unit: liters/second

Device: WTRMN00789, Quarterly Reading​

Input:

  • Year: 2023
  • Period: 2 (Q2)
  • Consumption: 3,500 cubic feet
  • Frequency: Quarterly

Expected Consumption Tag:

  • DeviceTag: WTRMN00789_quarterly_consumption
  • UnixTime: 1681516800 (2023-04-15 00:00:00 UTC)
  • Value: 3500.0
  • Unit: cubic feet

Expected LPS Tag:

  • DeviceTag: WTRMN00789_quarterly_demand_lps
  • UnixTime: 1681516800 (2023-04-15 00:00:00 UTC)
  • Value: 3500 × 28.3168 ÷ 7,888,320 = 0.01256 L/s
  • Unit: liters/second

Questions & Decisions Log​

Q1: Should we delete existing tags before creating new ones?​

Decision: TBD based on environment (test vs production)

  • Test environment: Yes, clean slate
  • Production: Depends on existing data state

Q2: How do we handle devices not in the Devices table?​

Decision: Log them and export to CSV for investigation. Don't create orphan tags.

Q3: What if period count doesn't match any expected frequency?​

Decision: Create tag with type unknown_{count}_consumption and log warning. Manual review needed.

Q4: Should monthly data use period = month for timestamp?​

Decision: Yes, for monthly data period directly maps to month (1=Jan, 2=Feb, etc.)

Q5: How precise should LPS values be stored?​

Decision: Use float/double precision in database. Display with 5 decimal places in reports.

Q6: Should we create lps tags for unknown frequencies?​

Decision: Yes, but log warning. Any frequency can be converted to lps with appropriate period seconds.

Contact & Support​

For questions or issues with this implementation:

  • Review documentation in DEVICE_TIMESERIES_UPGRADE_OCTOBER_2025.md
  • Check logs in v2/parse_timeseries_csv_v2.log
  • Review validation queries in this plan
  • Compare with legacy implementation in v1/ directory

End of Implementation Plan

Last updated on by Jake