-
Notifications
You must be signed in to change notification settings - Fork 185
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #569 from ggardet/master
Add WiFi Dual Meter device
- Loading branch information
Showing
3 changed files
with
310 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,243 @@ | ||
| # TinyTuya WiFi Dual Meter Device | ||
| # -*- coding: utf-8 -*- | ||
| """ | ||
| Python module to interface with Tuya WiFi Dual Meter Devices | ||
| Author: Guillaume Gardet | ||
| Local Control Classes | ||
| WiFiDualMeterDevice(...) | ||
| See OutletDevice() for constructor arguments | ||
| Functions | ||
| WiFiDualMeterDevice: | ||
| get_current_b() | ||
| get_total_power() | ||
| get_voltage_calibration() | ||
| get_current_calibration_a() | ||
| get_power_calibration_a() | ||
| get_energy_calibration_a() | ||
| get_power_factor_b() | ||
| get_current_calibration_b() | ||
| get_power_calibration_b() | ||
| get_energy_calibration_b() | ||
| get_energy_reverse_calibration_a() | ||
| get_energy_reverse_calibration_b() | ||
| get_report_rate() | ||
| Inherited | ||
| json = status() # returns json payload | ||
| set_version(version) # 3.1 [default] or 3.3 | ||
| set_socketPersistent(False/True) # False [default] or True | ||
| set_socketNODELAY(False/True) # False or True [default] | ||
| set_socketRetryLimit(integer) # retry count limit [default 5] | ||
| set_socketTimeout(timeout) # set connection timeout in seconds [default 5] | ||
| set_dpsUsed(dps_to_request) # add data points (DPS) to request | ||
| add_dps_to_request(index) # add data point (DPS) index set to None | ||
| set_retry(retry=True) # retry if response payload is truncated | ||
| set_status(on, switch=1, nowait) # Set status of switch to 'on' or 'off' (bool) | ||
| set_value(index, value, nowait) # Set int value of any index. | ||
| heartbeat(nowait) # Send heartbeat to device | ||
| updatedps(index=[1], nowait) # Send updatedps command to device | ||
| turn_on(switch=1, nowait) # Turn on device / switch # | ||
| turn_off(switch=1, nowait) # Turn off | ||
| set_timer(num_secs, nowait) # Set timer for num_secs | ||
| set_debug(toggle, color) # Activate verbose debugging output | ||
| set_sendWait(num_secs) # Time to wait after sending commands before pulling response | ||
| detect_available_dps() # Return list of DPS available from device | ||
| generate_payload(command, data) # Generate TuyaMessage payload for command with data | ||
| send(payload) # Send payload to device (do not wait for response) | ||
| receive() | ||
| """ | ||
|
|
||
| from ..core import Device | ||
|
|
||
| class WiFiDualMeterDevice(Device): | ||
|
|
||
| DPS_FORWARD_ENERGY_TOTAL = '1' | ||
| DPS_REVERSE_ENERGY_TOTAL = '2' | ||
| DPS_POWER_A = '101' | ||
| DPS_DIR_CUR_A = '102' | ||
| DPS_DIR_CUR_B = '104' | ||
| DPS_POWER_B = '105' | ||
| DPS_ENERGY_FORWARD_A = '106' | ||
| DPS_ENERGY_REVERSE_A = '107' | ||
| DPS_ENERGY_FORWARD_B = '108' | ||
| DPS_ENERGY_REVERSE_B = '109' | ||
| DPS_POWER_FACTOR_A = '110' | ||
| DPS_FREQ = '111' | ||
| DPS_VOLTAGE = '112' | ||
| DPS_CURRENT_A = '113' | ||
| DPS_CURRENT_B = '114' | ||
| DPS_TOTAL_POWER = '115' | ||
| DPS_VOLTAGE_CALIBRATION = '116' | ||
| DPS_CURRENT_CALIBRATION_A = '117' | ||
| DPS_POWER_CALIBRATION_A = '118' | ||
| DPS_ENERGY_CALIBRATION_A = '119' | ||
| DPS_POWER_FACTOR_B = '121' | ||
| DPS_FREQUENCY_CALIBRATION = '122' | ||
| DPS_CURRENT_CALIBRATION_B = '123' | ||
| DPS_POWER_CALIBRATION_B = '124' | ||
| DPS_ENERGY_CALIBRATION_B = '125' | ||
| DPS_ENERGY_CALIBRATION_REVERSE_A = '127' | ||
| DPS_ENERGY_CALIBRATION_REVERSE_B = '128' | ||
| DPS_REPORT_RATE = '129' | ||
|
|
||
| dps_data = { | ||
| DPS_FORWARD_ENERGY_TOTAL: { 'name': 'forward_energy_total', 'unit': 'kWh', 'scale': 100 }, | ||
| DPS_REVERSE_ENERGY_TOTAL: { 'name': 'reverse_energy_total', 'unit': 'kWh', 'scale': 100 }, | ||
| DPS_POWER_A: { 'name': 'power_a', 'unit': 'W', 'scale': 10 }, | ||
| DPS_DIR_CUR_A: { 'name': 'dir_curent_a', 'enum': ['FORWARD', 'REVERSE'] }, | ||
| DPS_DIR_CUR_B: { 'name': 'dir_current_b', 'enum': ['FORWARD', 'REVERSE'] }, | ||
| DPS_POWER_B: { 'name': 'power_b', 'unit': 'W', 'scale': 10 }, | ||
| DPS_ENERGY_FORWARD_A: { 'name': 'forward_energy_a', 'unit': 'kWh', 'scale': 100 }, | ||
| DPS_ENERGY_REVERSE_A: { 'name': 'reverse_energy_a', 'unit': 'kWh', 'scale': 100 }, | ||
| DPS_ENERGY_FORWARD_B: { 'name': 'forward_energy_b', 'unit': 'kWh', 'scale': 100 }, | ||
| DPS_ENERGY_REVERSE_B: { 'name': 'reverse_energy_b', 'unit': 'kWh', 'scale': 100 }, | ||
| DPS_POWER_FACTOR_A: { 'name': 'power_factor_a', 'scale': 100 }, | ||
| DPS_FREQ: { 'name': 'ac_frequency', 'unit': 'Hz', 'scale': 100 }, | ||
| DPS_VOLTAGE: { 'name': 'ac_voltage', 'unit': 'V', 'scale': 10 }, | ||
| DPS_CURRENT_A: { 'name': 'current_a', 'unit': 'mA'}, | ||
| DPS_CURRENT_B: { 'name': 'current_b', 'unit': 'mA'}, | ||
| DPS_TOTAL_POWER: { 'name': 'total_power', 'unit': 'W', 'scale': 10 }, | ||
| DPS_VOLTAGE_CALIBRATION : { 'name': 'voltage_calibration', 'scale': 1000 }, | ||
| DPS_CURRENT_CALIBRATION_A: { 'name': 'current_calibration_a', 'scale': 1000 }, | ||
| DPS_POWER_CALIBRATION_A: { 'name': 'power_calibration_a', 'scale': 1000 }, | ||
| DPS_ENERGY_CALIBRATION_A: { 'name': 'energy_calibration_a', 'scale': 1000 }, | ||
| DPS_POWER_FACTOR_B: { 'name': 'power_factor_b', 'scale': 100 }, | ||
| DPS_CURRENT_CALIBRATION_B: { 'name': 'current_calibration_b', 'scale': 1000 }, | ||
| DPS_POWER_CALIBRATION_B: { 'name': 'power_calibration_b', 'scale': 1000 }, | ||
| DPS_ENERGY_CALIBRATION_B: { 'name': 'energy_calibration_b', 'scale': 1000 }, | ||
| DPS_ENERGY_CALIBRATION_REVERSE_A: { 'name': 'energy_calibration_reverse_a', 'scale': 1000 }, | ||
| DPS_ENERGY_CALIBRATION_REVERSE_B: { 'name': 'energy_calibration_reverse_b', 'scale': 1000 }, | ||
| DPS_REPORT_RATE: { 'name': 'report_rate', 'unit': 's' }, | ||
| } | ||
|
|
||
| def get_value(self, dps_code, status_data=None): | ||
| if status_data is None: | ||
| status_data = self.status() | ||
| name = self.dps_data[dps_code]['name'] | ||
| try: | ||
| scale = self.dps_data[dps_code]['scale'] | ||
| except KeyError: | ||
| scale = 1 | ||
| try: | ||
| unit = self.dps_data[dps_code]['unit'] | ||
| except KeyError: | ||
| unit = "" | ||
| val = status_data['dps'][dps_code] | ||
| if isinstance(val, int): | ||
| val = val / scale | ||
| return {name+'_raw': val, | ||
| name+'_fmt': str(val) + ' '+ unit} | ||
|
|
||
| def get_forward_energy_total(self, status_data=None): | ||
| return self.get_value(dps_code=self.DPS_FORWARD_ENERGY_TOTAL) | ||
|
|
||
| def get_reverse_energy_total(self, status_data=None): | ||
| return self.get_value(dps_code=self.DPS_REVERSE_ENERGY_TOTAL) | ||
|
|
||
| def get_power_a(self, status_data=None): | ||
| return self.get_value(dps_code=self.DPS_POWER_A) | ||
|
|
||
| def get_dir_cur_a(self, status_data=None): | ||
| return self.get_value(dps_code=self.DPS_DIR_CUR_A) | ||
|
|
||
| def get_dir_cur_b(self, status_data=None): | ||
| return self.get_value(dps_code=self.DPS_DIR_CUR_B) | ||
|
|
||
| def get_power_b(self, status_data=None): | ||
| return self.get_value(dps_code=self.DPS_POWER_B) | ||
|
|
||
| def get_energy_forward_a(self, status_data=None): | ||
| return self.get_value(dps_code=self.DPS_ENERGY_FORWARD_A) | ||
|
|
||
| def get_energy_reverse_a(self, status_data=None): | ||
| return self.get_value(dps_code=self.DPS_ENERGY_REVERSE_A) | ||
|
|
||
| def get_energy_forward_b(self, status_data=None): | ||
| return self.get_value(dps_code=self.DPS_ENERGY_FORWARD_B) | ||
|
|
||
| def get_energy_reverse_b(self, status_data=None): | ||
| return self.get_value(dps_code=self.DPS_ENERGY_REVERSE_B) | ||
|
|
||
| def get_power_factor_a(self, status_data=None): | ||
| return self.get_value(dps_code=self.DPS_POWER_FACTOR_A) | ||
|
|
||
| def get_freq(self, status_data=None): | ||
| return self.get_value(dps_code=self.DPS_FREQ) | ||
|
|
||
| def get_voltage(self, status_data=None): | ||
| return self.get_value(dps_code=self.DPS_VOLTAGE) | ||
|
|
||
| def get_current_a(self, status_data=None): | ||
| return self.get_value(dps_code=self.DPS_CURRENT_A) | ||
|
|
||
| def get_current_b(self, status_data=None): | ||
| return self.get_value(dps_code=self.DPS_CURRENT_B) | ||
|
|
||
| def get_total_power(self, status_data=None): | ||
| return self.get_value(dps_code=self.DPS_TOTAL_POWER) | ||
|
|
||
| def get_voltage_calibration(self, status_data=None): | ||
| return self.get_value(dps_code=self.DPS_VOLTAGE_CALIBRATION) | ||
|
|
||
| def get_current_calibration_a(self, status_data=None): | ||
| return self.get_value(dps_code=self.DPS_CURRENT_CALIBRATION_A) | ||
|
|
||
| def get_power_calibration_a(self, status_data=None): | ||
| return self.get_value(dps_code=self.DPS_POWER_CALIBRATION_A) | ||
|
|
||
| def get_energy_calibration_a(self, status_data=None): | ||
| return self.get_value(dps_code=self.DPS_ENERGY_CALIBRATION_A) | ||
|
|
||
| def get_power_factor_b(self, status_data=None): | ||
| return self.get_value(dps_code=self.DPS_POWER_FACTOR_B) | ||
|
|
||
| def get_current_calibration_b(self, status_data=None): | ||
| return self.get_value(dps_code=self.DPS_CURRENT_CALIBRATION_B) | ||
|
|
||
| def get_power_calibration_b(self, status_data=None): | ||
| return self.get_value(dps_code=self.DPS_POWER_CALIBRATION_B) | ||
|
|
||
| def get_energy_calibration_b(self, status_data=None): | ||
| return self.get_value(dps_code=self.DPS_ENERGY_CALIBRATION_B) | ||
|
|
||
| def get_energy_reverse_calibration_a(self, status_data=None): | ||
| return self.get_value(dps_code=self.DPS_ENERGY_CALIBRATION_REVERSE_A) | ||
|
|
||
| def get_energy_reverse_calibration_b(self, status_data=None): | ||
| return self.get_value(dps_code=self.DPS_ENERGY_CALIBRATION_REVERSE_B) | ||
|
|
||
| def get_report_rate(self, status_data=None): | ||
| return self.get_value(dps_code=self.DPS_REPORT_RATE) | ||
|
|
||
| def print_all(self, status_data=None): | ||
| if status_data is None: | ||
| status_data = self.status() | ||
| print(self.get_forward_energy_total(status_data)) | ||
| print(self.get_reverse_energy_total(status_data)) | ||
| print(self.get_power_a(status_data)) | ||
| print(self.get_dir_cur_a(status_data)) | ||
| print(self.get_dir_cur_b(status_data)) | ||
| print(self.get_power_b(status_data)) | ||
| print(self.get_energy_forward_a(status_data)) | ||
| print(self.get_energy_reverse_a(status_data)) | ||
| print(self.get_energy_forward_b(status_data)) | ||
| print(self.get_energy_reverse_b(status_data)) | ||
| print(self.get_power_factor_a(status_data)) | ||
| print(self.get_freq(status_data)) | ||
| print(self.get_voltage(status_data)) | ||
| print(self.get_current_a(status_data)) | ||
| print(self.get_current_b(status_data)) | ||
| print(self.get_total_power(status_data)) | ||
| print(self.get_voltage_calibration(status_data)) | ||
| print(self.get_current_calibration_a(status_data)) | ||
| print(self.get_power_calibration_a(status_data)) | ||
| print(self.get_energy_calibration_a(status_data)) | ||
| print(self.get_power_factor_b(status_data)) | ||
| print(self.get_power_calibration_b(status_data)) | ||
| print(self.get_energy_calibration_b(status_data)) | ||
| print(self.get_energy_reverse_calibration_a(status_data)) | ||
| print(self.get_energy_reverse_calibration_b(status_data)) | ||
| print(self.get_report_rate(status_data)) | ||
|
|