You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
287 lines
9.6 KiB
Python
287 lines
9.6 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import KW_ONLY, field
|
|
from typing import * # type: ignore
|
|
|
|
import rio
|
|
import datetime
|
|
from mac_vendor_lookup import AsyncMacLookup
|
|
|
|
from db_classes import *
|
|
from .. import components as comps
|
|
|
|
import asyncio
|
|
|
|
class AddPage(rio.Component):
|
|
|
|
"""
|
|
A set of fields for adding/editing items.
|
|
"""
|
|
|
|
partnum: str = ""
|
|
mac: str = ""
|
|
serial: str = ""
|
|
fwver: str = ""
|
|
code: str = ""
|
|
popup_message: str = ""
|
|
popup_show: bool = False
|
|
popup_color: str = 'warning'
|
|
date: datetime.date = datetime.date.today()
|
|
tz: datetime.tzinfo = datetime.tzinfo()
|
|
time: datetime.datetime = datetime.datetime.now()
|
|
time_start: str = datetime.datetime.now().strftime(format="%H:%M")
|
|
macvendor: str = ""
|
|
manu: str = ""
|
|
manufield: str = ""
|
|
office: str = ""
|
|
description: str = ""
|
|
location: str = ""
|
|
location_code: str = ""
|
|
|
|
@rio.event.periodic(1)
|
|
def set_office_init(self):
|
|
self.office = self.session[comps.Settings].office
|
|
#print("Populated:", self.office)
|
|
|
|
async def check_mac(self,mac):
|
|
print("Checking", mac)
|
|
try:
|
|
macvendor = await AsyncMacLookup().lookup(mac)
|
|
if self.manufield == "" or self.manufield == self.macvendor: # blank or set by MAC already
|
|
self.manu = macvendor
|
|
self.manufield = macvendor
|
|
else:
|
|
self.manu = self.manufield
|
|
self.macvendor = macvendor
|
|
#print(macvendor)
|
|
except:
|
|
pass
|
|
# not valid MAC?
|
|
|
|
async def check_all(self):
|
|
await self.check_mac(self.mac)
|
|
# check part number
|
|
# lookup in PL_Export_rel
|
|
|
|
async def add_part(self):
|
|
await self.check_all()
|
|
if self.code == "":
|
|
# FAIL
|
|
self.popup_message = "\n Missing barcode! \n\n"
|
|
self.popup_show = True
|
|
self.popup_color = 'danger'
|
|
await asyncio.sleep(2)
|
|
self.popup_show = False
|
|
else:
|
|
# OK, add part
|
|
if self.location == "":
|
|
self.location_code = ""
|
|
if create_item(self.partnum, self.serial, self.office, self.code, locationid=self.location_code, description=self.description, manufacturer=self.manu, mac=self.mac, fwver=self.fwver) == False:
|
|
self.popup_message = "\n Duplicate barcode! \n\n"
|
|
self.popup_show = True
|
|
self.popup_color = 'warning'
|
|
await asyncio.sleep(2)
|
|
self.popup_show = False
|
|
else:
|
|
self.popup_message = "\n Part added! \n\n"
|
|
self.popup_show = True
|
|
self.popup_color = 'success'
|
|
self.name: str = ""
|
|
self.partnum: str = ""
|
|
self.mac: str = ""
|
|
self.serial: str = ""
|
|
self.fwver: str = ""
|
|
self.code: str = ""
|
|
self.macvendor: str = ""
|
|
self.manu: str = ""
|
|
self.manufield: str = ""
|
|
self.description: str = ""
|
|
self.location: str = ""
|
|
self.location_code: str = ""
|
|
await asyncio.sleep(2)
|
|
self.popup_show = False
|
|
|
|
|
|
async def _add_part_enter(self, event: rio.TextInputConfirmEvent):
|
|
await self.add_part()
|
|
|
|
async def _add_part_button(self):
|
|
await self.add_part()
|
|
|
|
@rio.event.periodic(1)
|
|
def update_time_view(self):
|
|
self.time_start = self.session.timezone.fromutc(datetime.datetime.now()).now().strftime(format="%m/%d/%Y %H:%M:%S")
|
|
self.time = datetime.datetime.now()
|
|
|
|
def _set_time(self, event: rio.TextInputChangeEvent):
|
|
time_str = event.text
|
|
try:
|
|
time_obj = datetime.datetime.strptime(time_str, "%H:%M").time()
|
|
dt_with_time = datetime.datetime.combine(self.date, time_obj)
|
|
dt_with_tz = dt_with_time.replace(tzinfo=self.session.timezone)
|
|
self.time = dt_with_tz
|
|
#event.text =
|
|
#print(self.time)
|
|
except ValueError:
|
|
pass
|
|
|
|
async def _update_mac(self, event: rio.TextInputChangeEvent):
|
|
await self.check_mac(event.text)
|
|
|
|
async def _update_location(self, event: rio.TextInputChangeEvent):
|
|
print("Checking " + self.location)
|
|
if get_location_id(self.location) != False:
|
|
self.location_code = self.location
|
|
print("Found location " + get_location_id(self.location).name)
|
|
self.location = get_location_id(self.location).name
|
|
|
|
|
|
def _update_partnum(self, event: rio.TextInputChangeEvent):
|
|
def __find_hm_header_static(txt):
|
|
searchlist = ["RSPS", "RSPE", "RSP", "RSB", "LRS", "RS", "OS", "RED", "MSP", "MSM", "MS", "MM", "EESX", "EES", "OZD", "OBR"]
|
|
for header in searchlist:
|
|
if txt.find(header) >= 0:
|
|
return txt.find(header) + len(header) + 2
|
|
|
|
def __find_header_general(txt):
|
|
acount = 0
|
|
ncount = 0
|
|
dash = False
|
|
for char in txt:
|
|
if char == '-':
|
|
dash = True
|
|
break
|
|
if char.isdigit():
|
|
ncount += 1
|
|
if char.isalpha():
|
|
acount += 1
|
|
|
|
if dash and acount <= 5 and ncount <= 5:
|
|
return acount+ncount
|
|
elif dash and acount >5 and acount <= 10 and ncount == 0:
|
|
return acount+ncount
|
|
return -1
|
|
|
|
def __find_hm_fwver(txt):
|
|
a = txt.find(".")
|
|
if a > 5:
|
|
return a-2
|
|
return -1
|
|
|
|
def __find_hm_fwmode(txt):
|
|
a = __find_hm_fwver(txt)
|
|
#print(a)
|
|
if txt.find("BRS") == 0:
|
|
a -= 1
|
|
if txt[a] in ['S', 'A']:
|
|
return '2' + txt[a]
|
|
a = __find_hm_fwver(txt)
|
|
if txt.find("GRS") == 0:
|
|
a -= 4
|
|
if txt[a:a+2] in ['2S', '2A', '3S', '3A']: # 1040
|
|
return txt[a:a+2]
|
|
elif txt[a+2:a+4] in ['2S', '2A', '3S', '3A']: # 1020/30
|
|
return txt[a+2:a+4]
|
|
a = __find_hm_fwver(txt)
|
|
if txt.find("RSP") == 0 or txt.find("OS") == 0 or txt.find("MSP") == 0 or txt.find("EESX") == 0 or txt.find("RED"):
|
|
a -= 2
|
|
if txt[a:a+2] in ['2S', '2A', '3S', '3A']:
|
|
return txt[a:a+2]
|
|
a = __find_hm_fwver(txt)
|
|
if txt.find("RS") == 0 or txt.find("MS") == 0:
|
|
a -= 3
|
|
print(txt,txt[a])
|
|
|
|
if txt[a] in ['P', 'E']:
|
|
return '2' + txt[a]
|
|
a = __find_hm_fwver(txt)
|
|
if txt.find("EAGLE") == 0:
|
|
a -= 2
|
|
if txt[a:a+2] in ['3F', 'MB', 'IN', 'UN', 'OP', '01', 'SU', 'NF']:
|
|
return txt[a:a+2]
|
|
print("Failed to match software level", repr(txt))
|
|
return ""
|
|
|
|
pn = event.text
|
|
self.name = ""
|
|
if __find_header_general(pn) >= 0:
|
|
# hirschmann switch detected
|
|
self.name = pn[0:__find_header_general(pn)]
|
|
if __find_hm_fwver(pn) >= 0:
|
|
self.fwver = pn[__find_hm_fwver(pn):]
|
|
if len(__find_hm_fwmode(pn)) > 0:
|
|
self.fwver += " SW-L" + __find_hm_fwmode(pn)
|
|
|
|
def build(self) -> rio.Component:
|
|
return rio.Column(
|
|
rio.Popup(
|
|
anchor=rio.Text(
|
|
text="Add a part below:",
|
|
style='heading1',
|
|
align_x = 0.5
|
|
),
|
|
color=self.bind().popup_color,
|
|
is_open=self.bind().popup_show,
|
|
content=rio.Text(
|
|
text=self.bind().popup_message,
|
|
),
|
|
|
|
),
|
|
rio.TextInput(
|
|
label="Barcode",
|
|
text=self.bind().code
|
|
),
|
|
rio.TextInput(
|
|
label="Full part number",
|
|
text=self.bind().partnum,
|
|
on_change=self._update_partnum
|
|
),
|
|
rio.TextInput(
|
|
label="Serial",
|
|
text=self.bind().serial,
|
|
on_confirm=self._add_part_enter
|
|
),
|
|
rio.TextInput(
|
|
label="MAC",
|
|
text=self.bind().mac,
|
|
on_change=self._update_mac,
|
|
on_confirm=self._add_part_enter
|
|
),
|
|
rio.TextInput(
|
|
label="Location (optional)",
|
|
text=self.bind().location,
|
|
on_change=self._update_location,
|
|
on_confirm=self._add_part_enter
|
|
),
|
|
rio.TextInput(
|
|
label="Manufacturer",
|
|
text=self.bind().manufield,
|
|
on_confirm=self._add_part_enter
|
|
),
|
|
rio.TextInput(
|
|
label="FW Ver",
|
|
text=self.bind().fwver
|
|
),
|
|
rio.MultiLineTextInput(
|
|
label="Description (optional)",
|
|
text=self.bind().description
|
|
),
|
|
rio.TextInput(
|
|
label="Timestamp",
|
|
is_sensitive=False,
|
|
text = self.bind().time_start,
|
|
#on_change=self._set_time
|
|
),
|
|
rio.Button(
|
|
content="Add",
|
|
on_press=self._add_part_button
|
|
),
|
|
|
|
spacing=2,
|
|
min_width=60,
|
|
margin_bottom=4,
|
|
align_x=0.5,
|
|
align_y=0,
|
|
)
|
|
|