def archive_e(charger_id, transaction_id, meter_values):
"""
Store MeterValues for a charger/transaction as a dated file for graphing.
"""
date_str = datetime.utcnow().strftime("%Y-%m-%d")
base = gw.resource("work", "etron", "graphs", charger_id)
os.makedirs(base, exist_ok=True)
# File name: <date>_<txn_id>.json (add .json for safety)
file_path = os.path.join(base, f"{date_str}_{transaction_id}.json")
with open(file_path, "w") as f:
json.dump(meter_values, f, indent=2)
return file_path
def authorize_balance(**record):
"""
Default OCPP RFID secondary validator: Only authorize if balance >= 1.
The RFID needs to exist already for this to be called in the first place.
"""
try:
return float(record.get("balance", "0")) >= 1
except Exception:
return False
def extract_meter(tx):
"""
Return the latest Energy.Active.Import.Register (kWh) from MeterValues or meterStop.
"""
if not tx:
return "-"
# Try meterStop first
if tx.get("meterStop") is not None:
try:
return float(tx["meterStop"]) / 1000.0 # assume Wh, convert to kWh
except Exception:
return tx["meterStop"]
# Try MeterValues: last entry, find Energy.Active.Import.Register
mv = tx.get("MeterValues", [])
if mv:
last_mv = mv[-1]
for sv in last_mv.get("sampledValue", []):
if sv.get("measurand") == "Energy.Active.Import.Register":
return sv.get("value")
return "-"
def is_abnormal_status(status: str, error_code: str) -> bool:
"""Determine if a status/errorCode is 'abnormal' per OCPP 1.6."""
status = (status or "").capitalize()
error_code = (error_code or "").capitalize()
# Available/NoError or Preparing are 'normal'
if status in ("Available", "Preparing") and error_code in ("Noerror", "", None):
return False
# All Faulted, Unavailable, Suspended, etc. are abnormal
if status in ("Faulted", "Unavailable", "Suspendedev", "Suspended", "Removed"):
return True
if error_code not in ("Noerror", "", None):
return True
return False
def power_consumed(tx):
"""Calculate power consumed in kWh from transaction's meter values (Energy.Active.Import.Register)."""
if not tx:
return 0.0
# Try to use MeterValues if present and well-formed
meter_values = tx.get("MeterValues", [])
energy_vals = []
for entry in meter_values:
# entry should be a dict with sampledValue: [...]
for sv in entry.get("sampledValue", []):
if sv.get("measurand") == "Energy.Active.Import.Register":
val = sv.get("value")
# Parse value as float (from string), handle missing
try:
val_f = float(val)
if sv.get("unit") == "Wh":
val_f = val_f / 1000.0
# else assume kWh
energy_vals.append(val_f)
except Exception:
pass
if energy_vals:
start = energy_vals[0]
end = energy_vals[-1]
return round(end - start, 3)
# Fallback to meterStart/meterStop if no sampled values
meter_start = tx.get("meterStart")
meter_stop = tx.get("meterStop")
# handle int or float or None
try:
if meter_start is not None and meter_stop is not None:
return round(float(meter_stop) / 1000.0 - float(meter_start) / 1000.0, 3)
if meter_start is not None:
return 0.0 # no consumption measured
except Exception:
pass
return 0.0
def redirect(url, code=None):
""" Aborts execution and causes a 303 or 302 redirect, depending on
the HTTP protocol version. """
if not code:
code = 303 if request.get('SERVER_PROTOCOL') == "HTTP/1.1" else 302
res = response.copy(cls=HTTPResponse)
res.status = code
res.body = ""
res.set_header('Location', urljoin(request.url, url))
raise res
def view_energy_graph(*, charger_id=None, date=None, **_):
"""
Render a page with a graph for a charger's session by date.
"""
import glob
from datetime import datetime
html = ['<link rel="stylesheet" href="/static/styles/charger_status.css">']
html.append('<h1>Charger Transaction Graph</h1>')
# Form for charger/date selector
graph_dir = gw.resource("work", "etron", "graphs")
charger_dirs = sorted(os.listdir(graph_dir)) if os.path.isdir(graph_dir) else []
txn_files = []
if charger_id:
cdir = os.path.join(graph_dir, charger_id)
if os.path.isdir(cdir):
txn_files = sorted(glob.glob(os.path.join(cdir, "*.json")))
html.append('<form method="get" action="/ocpp/csms/energy-graph" style="margin-bottom:2em;">')
html.append('<label>Charger: <select name="charger_id">')
html.append('<option value="">(choose)</option>')
for cid in charger_dirs:
sel = ' selected' if cid == charger_id else ''
html.append(f'<option value="{cid}"{sel}>{cid}</option>')
html.append('</select></label> ')
if txn_files:
html.append('<label>Transaction Date: <select name="date">')
html.append('<option value="">(choose)</option>')
for fn in txn_files:
# Filename: YYYY-MM-DD_<txn_id>.json
dt = os.path.basename(fn).split("_")[0]
sel = ' selected' if dt == date else ''
html.append(f'<option value="{dt}"{sel}>{dt}</option>')
html.append('</select></label> ')
html.append('<button type="submit">Show</button></form>')
# Load and render the graph if possible
graph_data = []
if charger_id and date:
base = os.path.join(graph_dir, charger_id)
match = glob.glob(os.path.join(base, f"{date}_*.json"))
if match:
with open(match[0]) as f:
graph_data = json.load(f)
# Graph placeholder: (replace with your JS plotting lib)
html.append('<div style="background:#222;border-radius:1em;padding:1.5em;min-height:320px;">')
if graph_data:
html.append('<h3>Session kWh Over Time</h3>')
html.append('<pre style="color:#fff;font-size:1.02em;">')
# Show simple table (replace with a chart)
html.append("Time | kWh\n---------------------|------\n")
for mv in graph_data:
ts = mv.get("timestampStr", "-")
kwh = "-"
for sv in mv.get("sampledValue", []):
if sv.get("measurand") == "Energy.Active.Import.Register":
kwh = sv.get("value")
html.append(f"{ts:21} | {kwh}\n")
html.append('</pre>')
else:
html.append("<em>No data available for this session.</em>")
html.append('</div>')
return "".join(html)