zdzichu/ha-addon-energa-meter-gh-fork
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

modified: .dockerignore

Browse Source 
modified:   src/main.py
	modified:   src/moj_licznik.py

	modified:   .dockerignore
	modified:   src/CHANGELOG.md
	modified:   src/config.yaml
	modified:   src/main.py
	modified:   src/moj_licznik.py
This commit is contained in:
TC
2023-10-22 15:09:06 +02:00
parent cdc3f93e56
commit d210d3108b
5 changed files with 55 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.dockerignore
View File

@@ -2,6 +2,7 @@ env/
__pycache__ __pycache__
src/database.sqlite src/database.sqlite
src/config.ini src/config.ini
config.ini
import.py import.py
check.py check.py
Dockerfile Dockerfile

4
src/CHANGELOG.md
View File

@@ -1,2 +1,4 @@
## v0.1.0 [2023-10-21] ## v0.1.0 [2023-10-21]
- Wersja beta - Wersja beta
## v0.1.1 [2023-10-22]
- Dodano obsługe błędnego logowania

2
src/config.yaml
View File

@@ -1,6 +1,6 @@
name: "Energa meter" name: "Energa meter"
description: "Energa meter addon" description: "Energa meter addon"
version: "0.1.0" version: "0.1.1"
slug: "energa_meter" slug: "energa_meter"
init: false init: false
options: options:

19
src/main.py
View File

@@ -20,15 +20,16 @@ def main():
mojLicznik = MojLicznik() mojLicznik = MojLicznik()
print(f"Logowanie...", username) print(f"Logowanie...", username)
mojLicznik.login(username, password) mojLicznik.login(username, password)
print(f"Aktualizacja liczników...") if mojLicznik.loginStatus:
mojLicznik.uppdate_measurments() print(f"Aktualizacja liczników...")
print(f"Wyszukiwanie najstarszych danych...") mojLicznik.uppdate_measurments()
mojLicznik.update_first_date() print(f"Wyszukiwanie najstarszych danych...")
print(f"Pobieranie danych...") mojLicznik.update_first_date()
mojLicznik.download_charts(True) print(f"Pobieranie danych...")
mojLicznik.update_last_days() mojLicznik.download_charts(True)
mojLicznik.set_daily_zones() mojLicznik.update_last_days()
mojLicznik.logout() mojLicznik.set_daily_zones()
mojLicznik.logout()
if __name__ == "__main__": if __name__ == "__main__":
main() main()

69
src/moj_licznik.py
View File

@@ -110,39 +110,50 @@ class MojLicznik:
try: try:
response = self.session.post(login_url, data=login_data) response = self.session.post(login_url, data=login_data)
response.raise_for_status() response.raise_for_status()
self.loginStatus = True
print(f"Zalogowano")
except HTTPError as e: except HTTPError as e:
print(f"Wystąpił błąd HTTP: {e}") print(f"Wystąpił błąd HTTP: {e}")
soup = BeautifulSoup(response.text, 'html.parser') soup = BeautifulSoup(response.text, 'html.parser')
select_elements = soup.find_all('script', type='text/javascript')
meter_isd = [] login_error_text = 'Użytkownik lub hasło niepoprawne'
for el in select_elements: login_error = soup.find('div', text=login_error_text)
pattern = r"id:\s+(\d+),[\s\S]*?ppe:\s+'([\d\s]+)',[\s\S]*?tariffCode:\s+'([^']+)',[\s\S]*?name:\s+'([^']+)'"
matches = re.search(pattern, el.text) if login_error:
if matches: self.loginStatus = False
id_value = matches.group(1) print(login_error_text)
ppe_value = matches.group(2) else:
tariffCode_value = matches.group(3) self.loginStatus = True
name_value = matches.group(4) print(f"Zalogowano")
meter_isd.append(id_value)
retrieved_record = PPETable.get_or_none(id=id_value) select_elements = soup.find_all('script', type='text/javascript')
if retrieved_record: meter_isd = []
print(f"Licznik {id_value} istnieje w systemie.") for el in select_elements:
if not retrieved_record.is_active: pattern = r"id:\s+(\d+),[\s\S]*?ppe:\s+'([\d\s]+)',[\s\S]*?tariffCode:\s+'([^']+)',[\s\S]*?name:\s+'([^']+)'"
retrieved_record.is_active = True matches = re.search(pattern, el.text)
retrieved_record.save() if matches:
else: id_value = matches.group(1)
print(f"Licznik {id_value} nie istnieje w systemie.") ppe_value = matches.group(2)
data = PPETable.create( tariffCode_value = matches.group(3)
id=id_value, name_value = matches.group(4)
ppe=ppe_value, meter_isd.append(id_value)
tariffCode=tariffCode_value, retrieved_record = PPETable.get_or_none(id=id_value)
name=name_value if retrieved_record:
) print(f"Licznik {id_value} istnieje w systemie.")
update_query = PPETable.update(is_active=0).where(PPETable.id.not_in(meter_isd)) if not retrieved_record.is_active:
update_query.execute() retrieved_record.is_active = True
retrieved_record.save()
else:
print(f"Licznik {id_value} nie istnieje w systemie.")
data = PPETable.create(
id=id_value,
ppe=ppe_value,
tariffCode=tariffCode_value,
name=name_value
)
update_query = PPETable.update(is_active=0).where(PPETable.id.not_in(meter_isd))
update_query.execute()
def logout(self): def logout(self):
logout_url = f"{self.meter_url}/dp/MainLogout.go" logout_url = f"{self.meter_url}/dp/MainLogout.go"