Skip to main content

Device Timeseries Processing Upgrade - October 2025

Executive Summary​

This document outlines a major redesign of the timeseries CSV parsing logic to handle devices with mixed data frequencies over time. The current implementation assumes a single frequency per device based on ROUTE_CATEGORY or BILLING_CYCLE, but actual data shows devices transitioning from annual → semi-annual → quarterly readings over their lifetime.

Problem Statement​

Current Implementation Issues​

  1. Frequency determination relies on unreliable fields:

    • ROUTE_CATEGORY (e.g., "METERED - ANNUAL STRATAS")
    • BILLING_CYCLE
    • These fields don't accurately reflect actual reading frequency

  2. Real-world data patterns (observed in annual metered devices):

    • 2013-2018: Truly annual readings (period 1 only)
    • One transition year: Semi-annual readings (2 periods per year)
    • 2019-2023: Quarterly readings (4 periods per year)

  3. One DeviceTag per device assumption:

    • Current design creates one tag per device based on route category
    • Cannot handle mixed frequencies within the same device's history

Data Example​

REG_SERIAL: 70093784A
Year  | Periods | Actual Frequency
------|---------|------------------
2013  | 1       | Annual
2014  | 1       | Annual
2015  | 1       | Annual
2016  | 1       | Annual
2017  | 1       | Annual
2018  | 1       | Annual
2019  | 2       | Semi-annual
2020  | 4       | Quarterly
2021  | 4       | Quarterly
2022  | 4       | Quarterly
2023  | 4       | Quarterly

Proposed Solution​

Core Design Principles​

  1. Infer frequency from actual data, not metadata fields
  2. Support multiple DeviceTags per device for different time frequencies
  3. Use pandas for efficient data grouping and analysis
  4. Determine frequency by counting periods per year per device

Frequency Classification​

Based on period count per year:

Periods/YearFrequencyTag Type
1Annualannual_consumption
2Semi-annualsemiannual_consumption
4Quarterlyquarterly_consumption
12Monthlymonthly_consumption

Note: "Semi-annual" is the proper term for twice per year (biannual can mean every two years, so avoid it).

DeviceTag Naming Convention​

Format: {Device.Name}_{frequency_type}

Examples:

  • WTRMN00345_annual_consumption
  • WTRMN00345_semiannual_consumption
  • WTRMN00345_quarterly_consumption
  • WTRMN00345_monthly_consumption

Implementation Design​

Step 1: Load Data into Pandas DataFrame​

import pandas as pd
import psycopg2
from datetime import datetime, timezone

# Load CSV into DataFrame
df = pd.read_csv(csv_file_path, encoding='cp1252')

# Required columns
# - REG_SERIAL: Device serial number (join key)
# - READING_YEAR: Year of reading
# - READING_PERIOD: Period number (1-4 for quarterly, 1-12 for monthly, etc.)
# - CONSUMPTION: Consumption value
# - (optional) ROUTE_CATEGORY, BILLING_CYCLE for logging/debugging

Step 2: Analyze Frequency per Device per Year​

# Group by device and year, count distinct periods
frequency_analysis = df.groupby(['REG_SERIAL', 'READING_YEAR']).agg({
    'READING_PERIOD': 'nunique',  # Count unique periods
    'CONSUMPTION': 'count'         # Count total records
}).reset_index()

frequency_analysis.columns = ['REG_SERIAL', 'READING_YEAR', 'PERIOD_COUNT', 'RECORD_COUNT']

# Classify frequency
def classify_frequency(period_count):
    if period_count == 1:
        return 'annual'
    elif period_count == 2:
        return 'semiannual'
    elif period_count == 4:
        return 'quarterly'
    elif period_count == 12:
        return 'monthly'
    else:
        # Log warning for unexpected counts
        return f'unknown_{period_count}'

frequency_analysis['FREQUENCY_TYPE'] = frequency_analysis['PERIOD_COUNT'].apply(classify_frequency)

Step 3: Identify Unique Device-Frequency Combinations​

# Get all unique device-frequency combinations
device_tags_needed = df.merge(frequency_analysis[['REG_SERIAL', 'READING_YEAR', 'FREQUENCY_TYPE']],
                               on=['REG_SERIAL', 'READING_YEAR'])

# Get unique combinations (one DeviceTag per combination)
unique_tags = device_tags_needed[['REG_SERIAL', 'FREQUENCY_TYPE']].drop_duplicates()

# Summary: How many tags per device?
tags_per_device = unique_tags.groupby('REG_SERIAL').size().reset_index(name='TAG_COUNT')
print(f"Devices with multiple frequencies: {(tags_per_device['TAG_COUNT'] > 1).sum()}")

Step 4: Bulk Create DeviceTags​

# Join with Devices table to get Device IDs and Names
cursor.execute(f'SELECT "Id", "SerialNumber", "Name" FROM "{DEVICES_TABLE}"')
devices = pd.DataFrame(cursor.fetchall(), columns=['DeviceId', 'SerialNumber', 'DeviceName'])

# Merge to get device IDs
unique_tags = unique_tags.merge(devices,
                                 left_on='REG_SERIAL',
                                 right_on='SerialNumber',
                                 how='inner')

# Generate DeviceTag records
device_tags = []
for _, row in unique_tags.iterrows():
    tag_name = f"{row['DeviceName']}_{row['FREQUENCY_TYPE']}_consumption"
    tag_type = f"{row['FREQUENCY_TYPE']}_consumption"

    device_tags.append({
        'Name': tag_name,
        'TagType': tag_type,
        'Unit': 'cubic feet',
        'DeviceId': row['DeviceId']
    })

# Bulk insert DeviceTags
# Use executemany or execute_values for performance

Step 5: Bulk Create DeviceTagDatum Records​

# Merge frequency type back to main dataframe
df_with_freq = df.merge(frequency_analysis[['REG_SERIAL', 'READING_YEAR', 'FREQUENCY_TYPE']],
                         on=['REG_SERIAL', 'READING_YEAR'])

# Merge with device info
df_with_freq = df_with_freq.merge(devices,
                                   left_on='REG_SERIAL',
                                   right_on='SerialNumber',
                                   how='inner')

# Get DeviceTag IDs (query what we just created)
cursor.execute(f'SELECT "Id", "DeviceId", "TagType" FROM "{DEVICE_TAGS_TABLE}"')
device_tags_db = pd.DataFrame(cursor.fetchall(), columns=['DeviceTagId', 'DeviceId', 'TagType'])

# Merge to get DeviceTagId
df_with_tags = df_with_freq.merge(device_tags_db,
                                   left_on=['DeviceId', 'FREQUENCY_TYPE'],
                                   right_on=['DeviceId', 'TagType'],
                                   how='inner')

# Convert to Unix timestamps
def convert_reading_to_unix_time(year, period):
    """
    Convert reading year and period to Unix timestamp.
    Uses quarterly mapping and 15th day of month.
    """
    if period <= 4:
        month = (period - 1) * 3 + 1  # Q1=Jan, Q2=Apr, Q3=Jul, Q4=Oct
    else:
        month = 1

    dt = datetime(year, month, 15, tzinfo=timezone.utc)
    return int(dt.timestamp())

df_with_tags['UnixTime'] = df_with_tags.apply(
    lambda row: convert_reading_to_unix_time(row['READING_YEAR'], row['READING_PERIOD']),
    axis=1
)

# Prepare DeviceTagDatum records
datum_records = df_with_tags[['DeviceTagId', 'UnixTime', 'CONSUMPTION']].values.tolist()

# Bulk insert with ON CONFLICT handling
cursor.executemany(f"""
    INSERT INTO "{DEVICE_TAG_DATA_TABLE}" ("DeviceTagId", "UnixTime", "Value")
    VALUES (%s, %s, %s)
    ON CONFLICT ("DeviceTagId", "UnixTime")
    DO UPDATE SET "Value" = EXCLUDED."Value"
""", datum_records)

Data Quality Considerations​

Validation Rules​

  1. Period count validation:

    • Log warnings for unexpected period counts (e.g., 3, 5-11, >12)
    • These may indicate data quality issues

  2. Duplicate detection:

    • Check for duplicate (REG_SERIAL, READING_YEAR, READING_PERIOD) combinations
    • Keep most recent or highest value (TBD)

  3. Missing devices:

    • Track devices in CSV that don't exist in Devices table
    • Export to CSV for investigation

Edge Cases​

  1. Devices with single year of data:

    • Still classify based on period count
    • A device with only 2023 data and 4 periods = quarterly

  2. Gaps in years:

    • Device may have readings in 2015, skip 2016-2018, resume 2019
    • Each year classified independently

  3. Period count changes mid-year:

    • Shouldn't happen (year is grouping unit)
    • If it does, log as data quality issue

Migration Strategy​

Phase 1: Analysis (Pre-Implementation)​

  1. Run analysis script to profile existing data:

    python analyze_frequency_patterns.py timeseries.csv

  2. Generate report showing:

    • Devices with mixed frequencies
    • Distribution of frequency types
    • Any unusual patterns

Phase 2: New Script Development​

  1. Create parse_timeseries_csv_v2.py with pandas-based logic
  2. Include dry-run mode to preview what will be created
  3. Add comprehensive logging

Phase 3: Testing​

  1. Test on subset of data (e.g., 10 devices with mixed frequencies)
  2. Verify DeviceTag creation
  3. Verify DeviceTagDatum records
  4. Compare results with expected behavior

Phase 4: Full Processing​

  1. Clear existing DeviceTag and DeviceTagDatum records (or use separate test DB)
  2. Run new script on full dataset
  3. Validate results

New Script Structure​

parse_timeseries_csv_v2.py​

parse_timeseries_csv_v2.py
├── load_csv_to_dataframe()
├── analyze_frequency_patterns()
│   ├── group_by_device_year()
│   ├── count_periods()
│   └── classify_frequency()
├── identify_device_tag_needs()
├── bulk_create_device_tags()
├── prepare_datum_records()
├── bulk_create_device_tag_data()
└── generate_summary_report()

Dependencies​

Add to requirements.txt:

pandas>=2.0.0
numpy>=1.24.0
psycopg2-binary>=2.9.0

Expected Benefits​

  1. Accurate frequency classification: Based on actual data, not metadata
  2. Support for evolving devices: Devices that change frequency over time
  3. Better data organization: Each frequency type has its own tag
  4. Improved query performance: Can query specific frequency types
  5. Data quality insights: Analysis reveals patterns and anomalies

Potential Challenges​

  1. Increased DeviceTag count: Some devices may have 2-3 tags instead of 1
  2. UI considerations: UI may need to handle multiple tags per device
  3. Query complexity: Queries may need to join across multiple tags
  4. Historical data: Existing data may need migration/cleanup

Example Output​

Before (Current Implementation)​

Device: WTRMN00345
  └── DeviceTag: "METERED - ANNUAL STRATAS Consumption" (annual_consumption)
      └── 58 DeviceTagDatum records (mixed frequencies, timestamps misaligned)

After (New Implementation)​

Device: WTRMN00345
  ├── DeviceTag: "WTRMN00345_annual_consumption"
  │   └── 6 DeviceTagDatum records (2013-2018, period 1)
  ├── DeviceTag: "WTRMN00345_semiannual_consumption"
  │   └── 2 DeviceTagDatum records (2019, periods 1-2)
  └── DeviceTag: "WTRMN00345_quarterly_consumption"
      └── 16 DeviceTagDatum records (2020-2023, periods 1-4 each year)

Timestamp Conversion (Unchanged)​

The timestamp conversion logic remains the same as current implementation:

  • All data treated as QUARTERLY for timestamp purposes
  • Period 1-4 → Q1-Q4 (Jan, Apr, Jul, Oct)
  • Uses 15th day of month to avoid timezone display issues
  • Formula: datetime(year, month, 15, tzinfo=timezone.utc)

This works because:

  • Annual (period 1) → Jan 15
  • Semi-annual (periods 1-2) → Jan 15, Apr 15
  • Quarterly (periods 1-4) → Jan 15, Apr 15, Jul 15, Oct 15
  • Monthly needs custom logic (period = month)

Success Criteria​

  • All devices successfully processed
  • DeviceTags accurately reflect frequency types
  • No timestamp misalignments
  • Data quality report generated
  • Performance acceptable (< 2 minutes for 50K records)
  • Clear logging for debugging
  • Summary statistics match expectations

Additional Upgrade: Liters Per Second (LPS) Tags​

Overview​

In addition to the consumption tags (in cubic feet), we will create a parallel set of DeviceTags that represent average demand in liters per second (lps) for each time period. These lps tags enable demand-based analysis and integration with hydraulic models that require flow rates rather than cumulative consumption.

Rationale​

  • Consumption tags show total volume consumed over a period (cubic feet)
  • LPS tags show average flow rate during that period (liters/second)
  • This dual representation supports both billing/consumption analysis and hydraulic modeling

Conversion Logic​

For each consumption reading, calculate average lps:

# Conversion constants
CUBIC_FEET_TO_LITERS = 28.3168

# Seconds per period
SECONDS_PER_MONTH = 30.44 * 24 * 3600  # Average month: ~2,629,440 seconds
SECONDS_PER_QUARTER = 3 * SECONDS_PER_MONTH  # ~7,888,320 seconds
SECONDS_PER_SEMIANNUAL = 6 * SECONDS_PER_MONTH  # ~15,776,640 seconds
SECONDS_PER_ANNUAL = 365.25 * 24 * 3600  # ~31,557,600 seconds

def convert_cuft_to_lps(consumption_cuft, frequency_type):
  """
  Convert cubic feet consumption to average liters per second.

  Args:
    consumption_cuft: Consumption value in cubic feet
    frequency_type: 'monthly', 'quarterly', 'semiannual', or 'annual'

  Returns:
    Average flow rate in liters per second
  """
  liters = consumption_cuft * CUBIC_FEET_TO_LITERS

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

  seconds = period_seconds.get(frequency_type, SECONDS_PER_QUARTER)
  return liters / seconds

Tag Naming Convention​

For each existing frequency tag, create a corresponding lps tag:

Original Tag TypeUnitLPS Tag TypeUnit
annual_consumptioncubic feetannual_demand_lpsliters/second
semiannual_consumptioncubic feetsemiannual_demand_lpsliters/second
quarterly_consumptioncubic feetquarterly_demand_lpsliters/second
monthly_consumptioncubic feetmonthly_demand_lpsliters/second

DeviceTag Examples​

For a device WTRMN00345 with multiple frequencies:

Consumption Tags:

  • WTRMN00345_annual_consumption (cubic feet)
  • WTRMN00345_quarterly_consumption (cubic feet)

Demand Tags:

  • WTRMN00345_annual_demand_lps (liters/second)
  • WTRMN00345_quarterly_demand_lps (liters/second)

Implementation Notes​

  1. Parallel tag creation: For every consumption DeviceTag created, create a corresponding demand_lps DeviceTag
  2. Same timestamps: Both tags use identical UnixTime values for their datum records
  3. Conversion during bulk insert: Apply conversion formula when creating DeviceTagDatum records for lps tags
  4. Unit field: Set Unit to "liters/second" for lps tags vs "cubic feet" for consumption tags

Modified Step 4: Bulk Create DeviceTags (With LPS Tags)​

# Generate DeviceTag records (both consumption and lps)
device_tags = []
for _, row in unique_tags.iterrows():
  device_name = row['DeviceName']
  freq_type = row['FREQUENCY_TYPE']
  device_id = row['DeviceId']

  # Consumption tag (cubic feet)
  consumption_tag_name = f"{device_name}_{freq_type}_consumption"
  device_tags.append({
    'Name': consumption_tag_name,
    'TagType': f"{freq_type}_consumption",
    'Unit': 'cubic feet',
    'DeviceId': device_id
  })

  # Demand tag (liters per second)
  lps_tag_name = f"{device_name}_{freq_type}_demand_lps"
  device_tags.append({
    'Name': lps_tag_name,
    'TagType': f"{freq_type}_demand_lps",
    'Unit': 'liters/second',
    'DeviceId': device_id
  })

# Bulk insert DeviceTags (now 2x as many tags)

Modified Step 5: Bulk Create DeviceTagDatum Records (With LPS Values)​

# Get DeviceTag IDs for both consumption and lps tags
cursor.execute(f'SELECT "Id", "DeviceId", "TagType" FROM "{DEVICE_TAGS_TABLE}"')
device_tags_db = pd.DataFrame(cursor.fetchall(), columns=['DeviceTagId', 'DeviceId', 'TagType'])

# Separate consumption and lps tags
consumption_tags = device_tags_db[device_tags_db['TagType'].str.endswith('_consumption')]
lps_tags = device_tags_db[device_tags_db['TagType'].str.endswith('_demand_lps')]

# Merge to get consumption DeviceTagId
df_with_consumption_tags = df_with_freq.merge(
  consumption_tags,
  left_on=['DeviceId', 'FREQUENCY_TYPE'],
  right_on=['DeviceId', device_tags_db['TagType'].str.replace('_consumption', '')],
  how='inner'
)

# Merge to get lps DeviceTagId
df_with_lps_tags = df_with_freq.merge(
  lps_tags,
  left_on=['DeviceId', 'FREQUENCY_TYPE'],
  right_on=['DeviceId', device_tags_db['TagType'].str.replace('_demand_lps', '')],
  how='inner'
)

# Convert readings to Unix timestamps (same for both)
df_with_consumption_tags['UnixTime'] = df_with_consumption_tags.apply(
  lambda row: convert_reading_to_unix_time(row['READING_YEAR'], row['READING_PERIOD']),
  axis=1
)
df_with_lps_tags['UnixTime'] = df_with_lps_tags.apply(
  lambda row: convert_reading_to_unix_time(row['READING_YEAR'], row['READING_PERIOD']),
  axis=1
)

# Prepare consumption datum records (cubic feet)
consumption_records = df_with_consumption_tags[['DeviceTagId', 'UnixTime', 'CONSUMPTION']].values.tolist()

# Convert and prepare lps datum records
df_with_lps_tags['LPS_VALUE'] = df_with_lps_tags.apply(
  lambda row: convert_cuft_to_lps(row['CONSUMPTION'], row['FREQUENCY_TYPE']),
  axis=1
)
lps_records = df_with_lps_tags[['DeviceTagId', 'UnixTime', 'LPS_VALUE']].values.tolist()

# Bulk insert both sets of records
cursor.executemany(f"""
  INSERT INTO "{DEVICE_TAG_DATA_TABLE}" ("DeviceTagId", "UnixTime", "Value")
  VALUES (%s, %s, %s)
  ON CONFLICT ("DeviceTagId", "UnixTime")
  DO UPDATE SET "Value" = EXCLUDED."Value"
""", consumption_records + lps_records)

Updated Example Output​

Device: WTRMN00345
  ├── DeviceTag: "WTRMN00345_annual_consumption" (cubic feet)
  │   └── 6 DeviceTagDatum records (2013-2018, period 1)
  ├── DeviceTag: "WTRMN00345_annual_demand_lps" (liters/second)
  │   └── 6 DeviceTagDatum records (2013-2018, period 1)
  ├── DeviceTag: "WTRMN00345_semiannual_consumption" (cubic feet)
  │   └── 2 DeviceTagDatum records (2019, periods 1-2)
  ├── DeviceTag: "WTRMN00345_semiannual_demand_lps" (liters/second)
  │   └── 2 DeviceTagDatum records (2019, periods 1-2)
  ├── DeviceTag: "WTRMN00345_quarterly_consumption" (cubic feet)
  │   └── 16 DeviceTagDatum records (2020-2023, periods 1-4 each year)
  └── DeviceTag: "WTRMN00345_quarterly_demand_lps" (liters/second)
      └── 16 DeviceTagDatum records (2020-2023, periods 1-4 each year)

Impact on Success Criteria​

  • All consumption tags have corresponding lps tags
  • LPS conversion calculations validated against manual calculations
  • Both tag types use identical timestamps for matching periods
  • Total DeviceTag count is 2x expected (consumption + lps)

Future Enhancements​

  1. Automatic frequency detection on import: Real-time classification
  2. Frequency transition alerts: Notify when device changes frequency
  3. Data quality dashboard: Visualize patterns and anomalies
  4. Historical analysis: Compare consumption across frequency types
  5. Peak demand analysis: Use lps tags to identify peak demand patterns
  6. Hydraulic model integration: Export lps data for EPANET or other hydraulic models

References​

  • Current implementation (v1): v1/parse_timeseries_csv.py and v1/parse_timeseries_csv_bulk.py
  • New implementation (v2): Will be in v2/ directory
  • Timestamp logic: Lines 194-228 in v1/parse_timeseries_csv.py
  • Device models: Models/ directory
  • Documentation: README.md in python/devices/

Document History​

DateAuthorChanges
2025-10-22AIInitial design document
2025-11-03AIAdded LPS (liters per second) tag upgrade specification

Next Steps​

  1. Review and approve design (including LPS upgrade)
  2. Create analysis script to profile current data
  3. Implement parse_timeseries_csv_v2.py with:
    • Frequency-based tag creation (consumption tags)
    • Parallel LPS tag creation (demand tags)
    • Conversion logic for cubic feet to liters/second
  4. Test on subset (validate both consumption and lps values)
  5. Verify conversion calculations:
    • Spot-check LPS values against manual calculations
    • Confirm period-appropriate divisors (monthly, quarterly, etc.)
  6. Run on full dataset
  7. Update documentation
Last updated on by Jake