nmelone/HenryPump-Drivers
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

added public ip lookup

Browse Source
This commit is contained in:
Nico Melone
2025-07-09 13:49:03 -05:00
parent 24d15140f2
commit 9ab7aede7c
7 changed files with 83 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
flow-monitor/config.txt
View File

@@ -6,6 +6,6 @@
},
"deviceName": "flowmonitor",
"driverId": "0140",
"releaseVersion": "20",
"releaseVersion": "22",
"driverFileName": "flow-monitor.py"
}

19
flow-monitor/flow-monitor.py
View File

@@ -8,7 +8,7 @@ import sys
import os
from device_base import deviceBase
import persistence
from utilities import get_public_ip_address
from utilities import get_public_ip_address, get_private_ip_address
# LOGGING SETUP
from logging.handlers import RotatingFileHandler
@@ -184,7 +184,7 @@ class start(threading.Thread, deviceBase):
self.force_send = False
self.daemon = True
self.version = "20"
self.version = "22"
self.finished = threading.Event()
threading.Thread.start(self)
@@ -238,8 +238,19 @@ class start(threading.Thread, deviceBase):
##############################################
# Determine public IP address and send to Meshify
public_ip_address = get_public_ip_address()
self.sendtodb('public_ip_address', public_ip_address, 0)
public_ip_address = "0.0.0.0"
private_ip_address = "0.0.0.0"
try:
public_ip_address = get_public_ip_address()
self.sendtodb('public_ip_address', public_ip_address, 0)
except Exception as e:
self.sendtodb('error', 'could not get public ip', 0)
try:
private_ip_address = get_private_ip_address()
self.sendtodb('private_ip_address', private_ip_address, 0)
except Exception as e:
self.sendtodb('error', 'could not get private ip', 0)
ip_checked_time = time.time()
# Attempt to retrieve data stored in the database

23
flow-monitor/flow-monitorv2/flowmonitor.py
View File

@@ -275,16 +275,19 @@ class start(threading.Thread, deviceBase):
def _check_ip_address(self):
"""Check the public IP address and send to Meshify if changed."""
self.public_ip_address_last_checked = time.time()
test_public_ip = get_public_ip_address()
test_public_ip = test_public_ip
test_private_ip = get_private_ip_address()
if not test_public_ip == self.public_ip_address and not test_public_ip == "0.0.0.0":
self.sendtodb('public_ip_address', test_public_ip, 0)
self.public_ip_address = test_public_ip
if not test_private_ip == self.private_ip_address:
self.sendtodb('private_ip_address', test_private_ip, 0)
self.private_ip_address = test_private_ip
try:
self.public_ip_address_last_checked = time.time()
test_public_ip = get_public_ip_address()
test_public_ip = test_public_ip
test_private_ip = get_private_ip_address()
if not test_public_ip == self.public_ip_address and not test_public_ip == "0.0.0.0":
self.sendtodb('public_ip_address', test_public_ip, 0)
self.public_ip_address = test_public_ip
if not test_private_ip == self.private_ip_address:
self.sendtodb('private_ip_address', test_private_ip, 0)
self.private_ip_address = test_private_ip
except Exception as e:
self.sendtodb('error', e, 0)
def flowmonitor_sync(self, name, value):

21
flow-monitor/utilities.py
View File

@@ -1,11 +1,26 @@
"""Utility functions for the driver."""
import socket
try:
import urllib
import contextlib
except:
pass
def get_public_ip_address():
"""Find the public IP Address of the host device."""
def get_private_ip_address():
"""Find the private IP Address of the host device."""
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.connect(("8.8.8.8", 80))
public_ip = sock.getsockname()[0]
sock.close()
return public_ip
def get_public_ip_address():
"""Find the public IP Address of the host device."""
ip_address = "0.0.0.0"
try:
with contextlib.closing(urllib.urlopen("http://checkip.amazonaws.com")) as url:
ip_address = url.read()
except Exception as e:
print("could not resolve check IP: {}".format(e))
return ip_address
return ip_address[:-1]

2
promagmbs/config.txt
View File

@@ -6,6 +6,6 @@
},
"deviceName": "promagmbs",
"driverId": "0199",
"releaseVersion": "4",
"releaseVersion": "5",
"driverFileName": "promagmbs.py"
}

20
promagmbs/promagmbs.py
View File

@@ -2,7 +2,7 @@
import threading
from device_base import deviceBase
from utilities import get_public_ip_address
from utilities import get_public_ip_address, get_private_ip_address
import time
import logging
import sys
@@ -34,7 +34,7 @@ class start(threading.Thread, deviceBase):
deviceBase.__init__(self, name=name, number=number, mac=mac, Q=Q, mcu=mcu, companyId=companyId, offset=offset, mqtt=mqtt, Nodes=Nodes)
self.daemon = True
self.version = "4"
self.version = "5"
self.finished = threading.Event()
self.forceSend = False
threading.Thread.start(self)
@@ -55,9 +55,19 @@ class start(threading.Thread, deviceBase):
logger.info("promagmbs driver will start in {} seconds".format(wait_sec - i))
time.sleep(1)
logger.warning("BOOM! Starting promagmbs driver...")
public_ip_address = get_public_ip_address()
self.sendtodbDev(1, 'public_ip_address', public_ip_address, 0, 'promagmbs')
public_ip_address = "0.0.0.0"
private_ip_address = "0.0.0.0"
try:
public_ip_address = get_public_ip_address()
self.sendtodbDev(1, 'public_ip_address', public_ip_address, 0, 'promagmbs')
except:
self.sendtodbDev(1, 'error', "error in public ip", 0, 'promagmbs')
try:
private_ip_address = get_private_ip_address()
self.sendtodbDev(1, 'private_ip_address', private_ip_address, 0, 'promagmbs')
except:
self.sendtodbDev(1, 'error', "error in private ip", 0, 'promagmbs')
send_loops = 0
watchdog_loops = 0

25
promagmbs/utilities.py
View File

@@ -1,15 +1,30 @@
"""Utility functions for the driver."""
import socket
import struct
try:
import urllib
import contextlib
except:
pass
def get_private_ip_address():
"""Find the private IP Address of the host device."""
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.connect(("8.8.8.8", 80))
public_ip = sock.getsockname()[0]
sock.close()
return public_ip
def get_public_ip_address():
"""Find the public IP Address of the host device."""
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.connect(("8.8.8.8", 80))
ip = s.getsockname()[0]
s.close()
return ip
ip_address = "0.0.0.0"
try:
with contextlib.closing(urllib.urlopen("http://checkip.amazonaws.com")) as url:
ip_address = url.read()
except Exception as e:
print("could not resolve check IP: {}".format(e))
return ip_address
return ip_address[:-1]
def int_to_float16(int_to_convert):