pss-datapool 0.1.0

Unified PSS system state management interface for Redis (hot) and SQLite (cold)

pss-datapool

Unified PSS system state management interface — Redis (hot) + SQLite (cold) transparently encapsulated.

Installation

pip install pss-datapool

Or from source:

pip install -e .

Quick Start

from pss.datapool import DataPool, DataPoolError

# Initialize (Redis + SQLite auto-connected)
pool = DataPool()

# Write data
pool.put("battery", "BATT-01", {"soc_pct": 85.5, "soh_pct": 98.2})

# Read data
battery = pool.get("battery", "BATT-01")
print(battery["soc_pct"])  # 85.5

# Batch read multiple channels
state = pool.get_batch(["battery", "charge", "order"])

# Snapshot & persist
snap = pool.snapshot()
pool.persist_snapshot()

# Restore from previous run
pool.load_from_db("run-20240626-001")

# Subscribe to changes
def on_change(channel: str):
    print(f"{channel} updated")

pool.subscribe("battery", on_change)

Configuration

All parameters are optional. Defaults read from environment variables:

Env Variable	Default	Description
PSS_REDIS_HOST	localhost	Redis host
PSS_REDIS_PORT	6379	Redis port
PSS_REDIS_DB	0	Redis DB number
PSS_REDIS_TIMEOUT_S	3	Redis socket timeout
PSS_SQLITE_PATH	pss_snapshot.db	SQLite database path
PSS_KEY_PREFIX	pss	Redis key namespace

Or pass explicitly:

pool = DataPool(
    redis_host="192.168.1.100",
    redis_port=6380,
    sqlite_path="/data/snapshots.db",
)

API Reference

CRUD

Method	Description
put(channel, key, data)	Write data to channel
get(channel, key)	Read single value
get_batch(channels)	Batch read multiple channels
query(channel, filter_fn)	Filter channel values
delete(channel, key)	Remove a key
list_all(channel)	List all key-value pairs

Snapshot

Method	Description
snapshot()	Export all channels as dict
restore(snapshot)	Restore from dict
persist_snapshot()	Write to SQLite
load_from_db(run_id)	Restore from SQLite

Pub/Sub

Method	Description
subscribe(channel, callback)	Subscribe to changes
unsubscribe(channel, callback)	Remove subscription

Commands

Method	Description
send_command(target, command, params)	Send control command
poll_commands(target)	List pending commands
ack_command(cmd_id, result)	Acknowledge completion

Error Handling

try:
    pool.put("battery", "BATT-01", data)
except DataPoolError as e:
    if e.code == "E001":
        print("Redis down — triggering emergency stop")
    elif e.code == "E010":
        print(f"Data corruption: {e.detail}")

Requirements

• Python >= 3.9
• Redis >= 5.0 (server)
• redis Python package (installed automatically)

License

MIT