-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathpriority.py
248 lines (203 loc) · 10.1 KB
/
priority.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
import json
import os
from flask import *
import threading
#flask constructor
priorityAPI = Flask(__name__)
class Device:
def __init__(self, name, priority, plug_status, energy):
self.name = name
self.priority = priority
self.plug_status = plug_status
self.energy = energy
self.plugip = "0.0.0.0" #ip has to be assigned after the fact
def __str__(self):
#object print method
return f"This device's name is '{self.name}'. The current priority is '{self.priority}', the plug status is currently '{self.plug_status}', and the current energy consumption is '{self.energy}'."
def __repr__(self):
#more readable representation method (for json import)
return f"Device object '{self.name}' at {hex(id(self))}"
def toJson(self):
return json.dumps(self.__dict__,)
def toggleState(self):
self.plug_status = not self.plug_status #boolean toggle using Not logical operator
def turnOn(self):
self.plug_status = 1
def turnOff(self):
self.plug_status = 0
def overwriteState(self, state):
self.plug_status = state
def overwritePriority(self, priority):
self.priority = priority
def overwriteConsumption(self, consumption):
self.energy = consumption
def overwritePlugIP(self, ipaddress):
self.plugip = ipaddress
def returnPlugIP(self):
return self.plugip
def returnConsumption(self):
return self.energy
def returnState(self):
return self.plug_status
def returnPriority(self):
return self.priority
class MonitoringSystem:
def __init__(self):
self.devices = {} #wouldnt it be better for this to be a list populated with objects of the Device class we just specified, instead of a dictionary?
def add_device(self, name, priority):
if name in self.devices:
print(f"Device '{name}' already exists. Please choose a different name.")
return
plug_status = input(f"Do you want to turn on the plug for device '{name}'? (1/0): ")
energy = float(input(f"Enter energy consumption for device '{name}' (in watts): "))
self.devices[name] = Device(name, priority, plug_status, energy)
print(f"Device '{name}' added with priority '{priority}', plug status '{plug_status}', and energy consumption '{energy}' watts.")
def set_priority(self, name, priority):
if name not in self.devices:
print(f"Device '{name}' does not exist.")
return
self.devices[name].priority = priority
print(f"Priority of device '{name}' set to '{priority}'.")
def turn_on_off_plug(self, name, plug_status):
if name not in self.devices:
print(f"Device '{name}' does not exist.")
return
self.devices[name].plug_status = plug_status
print(f"Plug status of device '{name}' set to '{plug_status}'.")
def list_devices(self):
print("List of Devices:")
for device in self.devices.values():
print(f"Name: {device.name}, Priority: {device.priority}, Plug Status: {device.plug_status}, Energy Consumption: {device.energy} watts")
def return_list(self):
return self.devices.values()
def json_export(list, file):
if not os.path.exists(file): #check file doesnt already exist first to prevent accidental overwrite
f = open(file, "w")
for device in list:
jsondata = json.dumps(device.__dict__) #convert each object into a json dictionary
print("Device Object" + jsondata)
f.write(jsondata + "\n") #write each object to a newline in json
f.close()
print("Export complete! Please find saved data at " + file)
else: #this runs if the first condition (file not already existing) fails
print("File already exists!")
x = input("Do you wish to overwrite? (y to overwrite, n to cancel operation): ")
if x == ("y"):
print("Overwriting json!")
#this is the same write code as above, but it is only called to overwrite whatever is already in file!
f = open(file, "w")
for device in list:
jsondata = json.dumps(device.__dict__) #convert each object into a json dictionary
print("Device Object" + jsondata)
f.write(jsondata + "\n") #write each object to a newline in json
f.close()
print("Export complete! Please find saved data at " + file)
else:
print("Cancelling operation!")
pass
#these json methods arent part of the energymonitoring system class so they will have to be exported separately
def json_parse(file):
if not os.path.exists(file):
print("File doesn't exist! Cancelling operation!") #prevent errors if a json file doesnt exist at provided directory
return -1
else:
device_list = []
print("Parsing device objects from json data!")
counter = 0
f = open(file, "r")
lines = f.readlines()
for line in lines: #read file contents line by line
print("Instantiating object " + str(counter) )
device_dictionary = json.loads(line) #may cause crashes if the json isn't structured appropriately, untested. device_dictionary holds current line contents.
print( str(device_dictionary) )
#instantiate device objects based on json dict contents
name = 'device_{}'.format(counter)
name = Device(**device_dictionary) #instantiate each json dictionary as device argument. multiple parameters are passed at once. it just works
print(name) #print object to verify attributes are set correctly
device_list.append(name) #append device object to the return list
counter += 1
f.close()
print("Device object creation complete! " + str(counter) + " objects created!")
return device_list
#constructor for energy system
monitoring_system = MonitoringSystem()
#priorityAPI functions below
@priorityAPI.route('/')
def priorityAPI_test():
return "priorityAPI Route works! Hello World! This text is being displayed with a Flask instance running asynchronously within a separate thread!"
@priorityAPI.route('/success') #generic return success
def success():
return "Operation complete!"
@priorityAPI.route('/devices/get', methods=['GET'])
def priorityAPI_getDevices():
#now the dictionary implementation is really painful
#we cannot directly export a dictionary of values. if it was a list of objects then we wouldn't have to do this badness.
#no good! too bad! this is a hack but it works and that is all that matters
dict = monitoring_system.return_list()
templist = []
for i in dict:
templist.append(i.toJson())
return jsonify(templist)
@priorityAPI.route('/devices/add', methods=['POST'])
def priorityAPI_addDevices(): #an example request: POST /devices/add?devicename=toaster&deviceconsumption=200&devicestatus=on&devicepriority=2
devicename = request.args.get("devicename")
deviceconsumption = request.args.get("deviceconsumption")
devicestatus = request.args.get("devicestatus")
devicepriority = request.args.get("devicepriority")
if devicename: #check that devicename exists before attempting input
newDevice = Device(devicename, devicepriority, devicestatus, deviceconsumption)
monitoring_system.devices[devicename] = newDevice
return("New device added!: " + devicename)
else:
return("Operation failed! Devicename is required!")
#program main loop
#priorityAPIThread = threading.Thread(target=priorityAPI.run)
#priorityAPIThread.start() #god i love multithreading
if __name__ == "__main__":
while True:
print("\n1. Add Device")
print("2. Set Priority")
print("3. Turn On/Off Plug")
print("4. List Devices")
print("5. Exit")
print("6. Export data to local JSON")
print("7. Import data from local JSON")
print("8. [DEBUG] View returned list")
choice = input("Enter your choice: ")
if choice == "1":
name = input("Enter device name: ")
priority = input("Enter priority (priority 1, priority 2, priority 3): ")
monitoring_system.add_device(name, priority)
elif choice == "2":
name = input("Enter device name: ")
priority = input("Enter new priority (priority 1, priority 2, priority 3): ")
monitoring_system.set_priority(name, priority)
elif choice == "3":
name = input("Enter device name: ")
plug_status = input("Turn on or off plug? (on/off): ")
monitoring_system.turn_on_off_plug(name, plug_status)
elif choice == "4":
monitoring_system.list_devices()
elif choice == "5":
print("Exiting...")
break
elif choice == "6":
print("Export data to JSON...")
filename = "exported_devices.json" #changing filename here changes the destination of export data
json_export( monitoring_system.return_list(), filename)
elif choice == "7":
print("Import data from JSON...")
filename = "exported_devices.json" #changing filename here changes the target of import data
if json_parse(filename) != -1:
devicelist = json_parse(filename) #this function returns a dictionary of device objects created from json data
print("Parsed objects: " + str(devicelist) ) #verify that returned objects are correct
for device in devicelist:
monitoring_system.devices[device.name] = Device(device.name, device.priority, device.plug_status, device.energy) #add new entry to the monitored device list as dictionary entries
#words dont exist for how much i dont like this implementation
else:
print("Import failed!")
elif choice == "8": #delete this later
returned = monitoring_system.return_list()
print(returned)
else:
print("Invalid choice. Please try again.")