|
#!/usr/bin/env python3 |
|
|
|
import re |
|
import asyncio |
|
from argparse import ArgumentParser, Namespace |
|
from enum import Enum |
|
import csv |
|
from typing import Any, Iterable, Coroutine, NamedTuple |
|
from aiohttp import ClientSession |
|
from yarl import URL |
|
|
|
|
|
__version__: str = '0.1.0' |
|
__author__: str = 'rysson' |
|
|
|
JSON = dict[str, Any] |
|
|
|
|
|
class Location(NamedTuple): |
|
lat: float |
|
lon: float |
|
|
|
@classmethod |
|
def parse(cls, v: str) -> 'Location': |
|
return Location(*re.split(r'\s*[,; ]\s*', v)) |
|
|
|
|
|
class MountingPlace(Enum): |
|
FREE = 'free' |
|
BUILDING = 'building' |
|
|
|
|
|
class Mode(Enum): |
|
MONTH = 'month' |
|
SUM = 'sum' |
|
MIN = 'min' |
|
|
|
|
|
async def download(location: Location, slope: int = 35, azimuth: int = 0, power: float = 1, *, |
|
loss: int = 14, mounting_place: MountingPlace = MountingPlace.FREE) -> JSON: |
|
url = URL('https://re.jrc.ec.europa.eu/api/v5_2/PVcalc') |
|
url = url % { |
|
'lat': location.lat, |
|
'lon': location.lon, |
|
'addatabase': 'PVGIS-SARAH2', |
|
'browser': 1, |
|
'userhorizon': '', |
|
'usehorizon': 1, |
|
'outputformat': 'json', |
|
'js': '1', |
|
'select_database_grid': 'PVGIS-SARAH2', |
|
'pvtechchoice': 'crystSi', |
|
'peakpower': power, |
|
'loss': loss, |
|
'mountingplace': mounting_place.value, |
|
'angle': slope, |
|
'aspect': azimuth, |
|
} |
|
|
|
async with ClientSession() as sess: |
|
async with sess.get(url) as resp: |
|
if resp.status >= 400: |
|
print(f'ERROR: download failed {resp.status} for {slope=}, {azimuth=}') |
|
return await resp.json() |
|
|
|
|
|
async def get_kwh(location: Location, slope: int | range = 35, azimuth: int | range = 0, power: float = 1, *, |
|
loss: int = 14, mounting_place: MountingPlace = MountingPlace.FREE) -> list[float]: |
|
|
|
async def down(slope: int, azimuth: int): |
|
data = await download(location, slope=slope, azimuth=azimuth, power=power, |
|
loss=loss, mounting_place=mounting_place) |
|
months = [d['E_m'] for d in data['outputs']['monthly']['fixed']] |
|
row = [slope, azimuth, *months] |
|
row.append(sum(months)) |
|
row.append(min(months)) |
|
return row |
|
|
|
if isinstance(slope, (int, float)): |
|
slope = [slope] |
|
if isinstance(azimuth, (int, float)): |
|
azimuth = [azimuth] |
|
|
|
# max 10 connections at once |
|
return await execute(10, [down(slope=sl, azimuth=az) for az in azimuth for sl in slope]) |
|
|
|
|
|
async def execute(worker_count: int, coros: Iterable[Coroutine], *, return_exceptions: bool = True) -> None: |
|
async def worker(coro): |
|
async with sem: |
|
return await coro |
|
|
|
sem = asyncio.Semaphore(worker_count) |
|
return await asyncio.gather(*(worker(coro) for coro in coros), return_exceptions=return_exceptions) |
|
|
|
|
|
async def run(args: Namespace) -> None: |
|
kwargs = {k: getattr(args, k) for k in ('power', 'loss', 'mounting_place')} |
|
with open(args.output, 'w', newline='') as f: |
|
writer = csv.writer(f, delimiter=',', quotechar='"', quoting=csv.QUOTE_MINIMAL) |
|
if args.mode == Mode.MONTH: |
|
columns = ('nachylenie', 'azymut', 'styczeń', 'luty', 'marzec', 'kwiecień', 'maj', 'czerwiec', |
|
'lipiec', 'sierpień', 'wrzesień', 'październik', 'listopad', 'grudzień', 'Σ', 'min') |
|
writer.writerow(columns) |
|
for row in await get_kwh(args.location, slope=args.slope, azimuth=args.azimuth, **kwargs): |
|
writer.writerow(row) |
|
elif args.mode in (Mode.MIN, Mode.SUM): |
|
index: int = -1 if args.mode == Mode.MIN else -2 |
|
data = {(row[0], row[1]): row[index] |
|
for row in await get_kwh(args.location, slope=args.slope, azimuth=args.azimuth, **kwargs)} |
|
writer.writerow((r'↓ azymut \ kąt →', *args.slope)) |
|
for azimuth in args.azimuth: |
|
writer.writerow((azimuth, *(data[slope, azimuth] for slope in args.slope))) |
|
|
|
|
|
def parse_range(s: str) -> range | list[int]: |
|
if ':' in s: |
|
start, stop, *step = map(int, s.split(':')) |
|
return range(start, stop + 1, step[0] if step else 5) |
|
return [int(s)] |
|
|
|
|
|
def main(argv: list[str] = None) -> None: |
|
p = ArgumentParser(description='--slope and --azimuth take numerical values or START:STOP[:STEP] range') |
|
p.add_argument('--version', action='version', version=__version__) |
|
p.add_argument('location', metavar='LAN,LON', type=Location.parse, |
|
help='location (lan,lon)') |
|
p.add_argument('-s', '--slope', type=parse_range, default='35', |
|
help='slope, angle of the PV modules from the horizontal plane [35]') |
|
p.add_argument('-z', '--azimuth', type=parse_range, default='0', |
|
help='azimuth, -90° is East, 0° is South and 90° is West [0]') |
|
p.add_argument('-p', '--power', type=float, default=1., |
|
help='PV power in kWp [1]') |
|
p.add_argument('--loss', type=int, default=14, |
|
help='the estimated system losses (cables, power inverters, dirt/snow) [14]') |
|
p.add_argument('--mounting-place', choices=MountingPlace, type=MountingPlace, default=MountingPlace.FREE, |
|
help='mounting place: free or building [free]') |
|
p.add_argument('-m', '--mode', choices=Mode, type=Mode, default=Mode.MONTH, |
|
help='Output mode (month - rows, sum or min - 2D slope/azimuth) [month]') |
|
p.add_argument('-o', '--output', default='pv.csv', metavar='PATH', |
|
help='output CSV file [pv.csv]') |
|
args: Namespace = p.parse_args(argv) |
|
return asyncio.run(run(args)) |
|
|
|
|
|
if __name__ == '__main__': |
|
main() |
Energy of 1 kWp PV on South in slope
0:90:5
in Poland.Last column is most interesting in a off-grid installation. It's minimal month energy.
It looks that the best slope is 60-70°. 10% less then max in a year, 3% less then could be in December.