Ejemplos de Uso
Esta sección muestra ejemplos prácticos para casos de uso comunes con ambas versiones del cliente.
Nota
IMPORTANTE sobre formatos de fecha: La API de Datadis requiere fechas en formato mensual (YYYY/MM) para los endpoints de consumo y potencia máxima. NO se permiten fechas con días específicos (YYYY/MM/DD).
Análisis de Consumo Anual
Cliente V1 - Análisis Básico
from datadis_python.client.v1.simple_client import SimpleDatadisClientV1
from datetime import datetime
import json
def analizar_consumo_anual_v1(username, password, cups, distributor_code, year="2024"):
"""Analiza el consumo energético de todo un año usando cliente V1"""
with SimpleDatadisClientV1(username, password) as client:
print(f"Analizando consumo anual {year} con cliente V1")
# Obtener datos de todo el año (formato mensual requerido)
consumption = client.get_consumption(
cups=cups,
distributor_code=distributor_code,
date_from=f"{year}/01", # Enero
date_to=f"{year}/12" # Diciembre
)
if not consumption:
print("No se encontraron datos de consumo")
return None
# Análisis por meses
consumo_mensual = {}
for registro in consumption:
# Extraer mes del timestamp (YYYY/MM/DD HH:MM:SS)
mes = registro.date[:7] # YYYY/MM
if mes not in consumo_mensual:
consumo_mensual[mes] = 0
if registro.consumptionKWh:
consumo_mensual[mes] += registro.consumptionKWh
# Estadísticas anuales
total_anual = sum(consumo_mensual.values())
promedio_mensual = total_anual / 12 if consumo_mensual else 0
mes_mayor_consumo = max(consumo_mensual, key=consumo_mensual.get) if consumo_mensual else None
mes_menor_consumo = min(consumo_mensual, key=consumo_mensual.get) if consumo_mensual else None
# Resultados
print(f"Consumo total: {total_kwh:.2f} kWh")
print(f"Consumo promedio diario: {consumo_promedio:.2f} kWh")
print(f"Consumo máximo diario: {consumo_maximo:.2f} kWh")
print(f"Consumo mínimo diario: {consumo_minimo:.2f} kWh")
print(f"Días con datos: {len(consumo_diario)}")
return {
"total_anual": total_anual,
"promedio_mensual": promedio_mensual,
"consumo_mensual": consumo_mensual,
"mes_mayor_consumo": mes_mayor_consumo,
"mes_menor_consumo": mes_menor_consumo,
"registros_totales": len(consumption)
}
# Uso del ejemplo - NOTA: usar fechas en formato mensual YYYY/MM
resultado = analizar_consumo_anual_v1(
username="tu_nif",
password="tu_contraseña",
cups="ES1234000000000001JN0F",
distributor_code="2",
year="2024"
)
Cliente V2 - Análisis Robusto con Manejo de Errores
from datadis_python.client.v2.simple_client import SimpleDatadisClientV2
def analizar_consumo_anual_v2(username, password, cups, distributor_code, year="2024"):
"""Analiza el consumo energético con manejo robusto de errores usando cliente V2"""
with SimpleDatadisClientV2(username, password) as client:
print(f"Analizando consumo anual {year} con cliente V2")
# Obtener datos con manejo de errores mejorado
consumption_response = client.get_consumption(
cups=cups,
distributor_code=distributor_code,
date_from=f"{year}/01",
date_to=f"{year}/12"
)
print(f"Procesando período: {periodo}")
consumption = consumption_response.time_curve
if not consumption:
print("No se encontraron datos de consumo")
return None
# Análisis detallado
consumo_mensual = {}
datos_por_metodo = {"Real": 0, "Estimada": 0}
# Mostrar comparación
print("\nComparación de períodos:")
for periodo, datos in resultados.items():
print(f"{periodo}: {datos['total_kwh']:.2f} kWh ({datos['registros']} registros)")
if registro.consumptionKWh:
consumo_mensual[mes]["consumo"] += registro.consumptionKWh
if registro.surplusEnergyKWh:
consumo_mensual[mes]["excedentes"] += registro.surplusEnergyKWh
if registro.selfConsumptionKWh:
consumo_mensual[mes]["autoconsumo"] += registro.selfConsumptionKWh
consumo_mensual[mes]["registros"] += 1
# Contar método de obtención
if registro.obtainMethod:
datos_por_metodo[registro.obtainMethod] = datos_por_metodo.get(registro.obtainMethod, 0) + 1
# Estadísticas anuales
total_consumo = sum(mes_data["consumo"] for mes_data in consumo_mensual.values())
total_excedentes = sum(mes_data["excedentes"] for mes_data in consumo_mensual.values())
total_autoconsumo = sum(mes_data["autoconsumo"] for mes_data in consumo_mensual.values())
print(f"Consumo total anual: {total_consumo:.2f} kWh")
print(f"Excedentes totales: {total_excedentes:.2f} kWh")
print(f"Autoconsumo total: {total_autoconsumo:.2f} kWh")
print(f"Datos reales: {datos_por_metodo.get('Real', 0)} registros")
print(f"Datos estimados: {datos_por_metodo.get('Estimada', 0)} registros")
return {
"total_consumo": total_consumo,
"total_excedentes": total_excedentes,
"total_autoconsumo": total_autoconsumo,
"consumo_mensual": consumo_mensual,
"datos_por_metodo": datos_por_metodo,
"errores_distribuidor": len(consumption_response.distributor_error)
}
print("Exportando datos completos...")
# 1. Consumo
print("Obteniendo consumo...")
consumo = client.get_consumption(
cups=cups,
distributor_code=distributor_code,
date_from=f"{year}/01",
date_to=f"{year}/12"
)
# 2. Potencia máxima
print("Obteniendo potencia máxima...")
potencia = client.get_max_power(
cups=cups,
distributor_code=distributor_code,
date_from=f"{year}/01",
date_to=f"{year}/12"
)
# 3. Contratos
print("Obteniendo contratos...")
contratos = client.get_contract_detail(
cups=cups,
distributor_code=distributor_code
)
# Preparar datos para análisis
datos_consumo = []
for registro in consumption_response.time_curve:
datos_consumo.append({
"fecha": registro.date,
"consumo_kwh": registro.consumptionKWh or 0,
"excedentes_kwh": registro.surplusEnergyKWh or 0,
"autoconsumo_kwh": registro.selfConsumptionKWh or 0,
"metodo": registro.obtainMethod
})
print(f"Datos exportados a: {filename}")
return filename
# Análisis mensual
resumen_mensual = df.groupby('mes').agg({
'consumo_kwh': 'sum',
'excedentes_kwh': 'sum',
'autoconsumo_kwh': 'sum'
}).round(2)
# Información del contrato
contrato_info = {}
if contract_response.contract:
contrato = contract_response.contract[0]
contrato_info = {
"potencia_contratada_kw": contrato.contractedPowerkW,
"tarifa": contrato.accessFare,
"tipo_punto": contrato.pointType,
"tension": contrato.voltage,
"comercializadora": contrato.marketer
}
# Generar gráficos
fig, axes = plt.subplots(2, 2, figsize=(15, 12))
# Gráfico 1: Consumo mensual
resumen_mensual['consumo_kwh'].plot(kind='bar', ax=axes[0,0], color='steelblue')
axes[0,0].set_title('Consumo Mensual (kWh)')
axes[0,0].set_ylabel('kWh')
if not suministros:
print("No se encontraron puntos de suministro")
return
print(f"Procesando {len(suministros)} puntos de suministro...")
# Gráfico 4: Distribución horaria (si hay datos horarios)
if len(df) > 100: # Solo si hay suficientes datos
df['hora'] = df['fecha'].dt.hour
consumo_por_hora = df.groupby('hora')['consumo_kwh'].mean()
consumo_por_hora.plot(kind='bar', ax=axes[1,1], color='orange')
axes[1,1].set_title('Patrón de Consumo por Hora')
axes[1,1].set_ylabel('kWh promedio')
plt.tight_layout()
plt.savefig(f'informe_consumo_{cups}_{year}.png', dpi=300, bbox_inches='tight')
print(f"Gráficos guardados en: informe_consumo_{cups}_{year}.png")
for i, suministro in enumerate(suministros, 1):
print(f"\nProcesando suministro {i}/{len(suministros)}: {suministro.cups}")
# Guardar informe completo
with open(f'informe_detallado_{cups}_{year}.json', 'w', encoding='utf-8') as f:
json.dump(estadisticas, f, indent=2, ensure_ascii=False, default=str)
print("Informe detallado generado:")
print(f"- Consumo total: {estadisticas['consumo_total_kwh']} kWh")
print(f"- Promedio mensual: {estadisticas['consumo_promedio_mensual']} kWh")
print(f"- Mayor consumo: {estadisticas['mes_mayor_consumo']}")
print(f"- Tiene autoconsumo: {estadisticas['tiene_autoconsumo']}")
return estadisticas
# Ejemplo de uso completo
if __name__ == "__main__":
# Configuración
USERNAME = "tu_nif"
PASSWORD = "tu_contraseña"
YEAR = "2024"
resultados.append(resultado)
print(f"Consumo: {total_kwh:.2f} kWh ({len(consumo)} registros)")
except Exception as e:
print(f"Error procesando {suministro.cups}: {e}")
resultados.append({
"cups": suministro.cups,
"error": str(e)
})
# Resumen
print(f"\nResumen de {len(resultados)} suministros:")
total_general = 0
for resultado in resultados:
if "error" not in resultado:
print(f" {resultado['cups']}: {resultado['total_kwh']:.2f} kWh")
total_general += resultado['total_kwh']
else:
print(f" {resultado['cups']}: ERROR - {resultado['error']}")
print(f"\nConsumo total de todos los suministros: {total_general:.2f} kWh")
return resultados
Validación y Limpieza de Datos
from datadis_python.client.v1.simple_client import SimpleDatadisClientV1
from datetime import datetime
def validar_y_limpiar_datos(username, password, cups, distributor_code, fecha_inicio, fecha_fin):
"""Valida y limpia los datos obtenidos de la API"""
with SimpleDatadisClientV1(username, password) as client:
print("Obteniendo y validando datos...")
consumo = client.get_consumption(
cups=cups,
distributor_code=distributor_code,
date_from=fecha_inicio,
date_to=fecha_fin
)
print(f"Datos originales: {len(consumo)} registros")
# Validaciones
datos_validos = []
errores = {
"consumo_negativo": 0,
"fecha_invalida": 0,
"valores_extremos": 0
}
for registro in consumo:
try:
# Validar consumo no negativo
if registro.consumptionKWh and registro.consumptionKWh < 0:
errores["consumo_negativo"] += 1
continue
# Validar fecha válida
datetime.strptime(registro.date, "%Y/%m/%d %H:%M:%S")
# Validar valores no extremos (más de 100 kWh por hora es sospechoso)
if registro.consumptionKWh and registro.consumptionKWh > 100:
errores["valores_extremos"] += 1
continue
datos_validos.append(registro)
except ValueError:
errores["fecha_invalida"] += 1
except Exception:
continue
# Resultados de validación
print(f"Datos válidos: {len(datos_validos)}")
print(f"Errores encontrados:")
for tipo_error, cantidad in errores.items():
if cantidad > 0:
print(f" - {tipo_error}: {cantidad}")
# Estadísticas de datos limpios
if datos_validos:
consumos = [d.consumption_kwh for d in datos_validos]
print(f"\nEstadísticas de datos limpios:")
print(f" - Total: {sum(consumos):.2f} kWh")
print(f" - Promedio: {sum(consumos)/len(consumos):.2f} kWh")
print(f" - Máximo: {max(consumos):.2f} kWh")
print(f" - Mínimo: {min(consumos):.2f} kWh")
return datos_validos, errores
Uso con Configuración Personalizada
from datadis_python.client.v1.simple_client import SimpleDatadisClientV1
from datadis_python.exceptions import DatadisError
class DatadisManager:
"""Clase wrapper para gestionar múltiples operaciones con Datadis"""
def __init__(self, username, password, timeout=180, retries=5):
self.username = username
self.password = password
self.timeout = timeout
self.retries = retries
self._client = None
def __enter__(self):
self._client = SimpleDatadisClientV1(
username=self.username,
password=self.password,
timeout=self.timeout,
retries=self.retries
)
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if self._client:
self._client.close()
def obtener_resumen_completo(self):
"""Obtiene un resumen completo de la cuenta"""
if not self._client:
raise DatadisError("Cliente no inicializado")
resumen = {
"distribuidores": [],
"suministros": [],
"contratos": [],
"estado": "ok"
}
try:
# Distribuidores
resumen["distribuidores"] = self._client.get_distributors()
# Suministros
resumen["suministros"] = self._client.get_supplies()
# Contratos para cada suministro
for suministro in resumen["suministros"]:
if resumen["distribuidores"] and resumen["distribuidores"][0].distributor_codes:
codigo_dist = resumen["distribuidores"][0].distributor_codes[0]
contratos = self._client.get_contract_detail(
cups=suministro.cups,
distributor_code=codigo_dist
)
resumen["contratos"].extend(contratos)
except Exception as e:
resumen["estado"] = f"error: {e}"
return resumen
# Uso
with DatadisManager("tu_nif", "tu_contraseña", timeout=240, retries=3) as manager:
resumen = manager.obtener_resumen_completo()
print(f"Estado: {resumen['estado']}")
print(f"Distribuidores: {len(resumen['distribuidores'])}")
print(f"Suministros: {len(resumen['suministros'])}")
print(f"Contratos: {len(resumen['contratos'])}")