Skip to content

Commit

Permalink
Remove old params shortcuts to parameter groups
Browse files Browse the repository at this point in the history
  • Loading branch information
Kane610 committed Dec 21, 2023
1 parent 7aa847e commit 8e47bc9
Show file tree
Hide file tree
Showing 4 changed files with 81 additions and 157 deletions.
85 changes: 0 additions & 85 deletions axis/vapix/interfaces/parameters/param_cgi.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,12 +7,7 @@

from typing import TYPE_CHECKING, Any

from ...models.parameters.brand import BrandParam
from ...models.parameters.image import ImageParam
from ...models.parameters.param_cgi import ParamRequest, params_to_dict
from ...models.parameters.properties import PropertyParam
from ...models.parameters.stream_profile import StreamProfileParam
from ...models.stream_profile import StreamProfile
from ..api_handler import ApiHandler
from .brand import BrandParameterHandler
from .image import ImageParameterHandler
Expand Down Expand Up @@ -86,83 +81,3 @@ async def update(self, group: str = "") -> None:
def get_param(self, group: str) -> dict[str, Any]:
"""Get parameter group."""
return self._items.get("root", {}).get(group, {})

# Brand

async def update_brand(self) -> None:
"""Update brand group of parameters."""
await self.brand_handler.update()

@property
def brand(self) -> BrandParam | None:
"""Provide brand parameters."""
return self.brand_handler.get_params().get("0")

# Image

async def update_image(self) -> None:
"""Update image group of parameters."""
await self.image_handler.update()

@property
def image_params(self) -> ImageParam:
"""Provide image parameters."""
return self.image_handler.get_params().get("0")
return ImageParam.decode(self.get_param("Image"))

@property
def image_sources(self) -> dict[str, Any]:
"""Image source information."""
data = {}
if params := self.image_handler.get_params().get("0"):
data = params.data
return data

# Properties

async def update_properties(self) -> None:
"""Update properties group of parameters."""
await self.property_handler.update()
# await self.update("Properties")

@property
def properties(self) -> PropertyParam:
"""Provide property parameters."""
return self.property_handler.get_params().get("0")
# data = {}
# if params := self.property_handler.get_params().get("0"):
# data = params
# return data
# return PropertyParam.decode(self.get_param("Properties"))

# PTZ

async def update_ptz(self) -> None:
"""Update PTZ group of parameters."""
await self.update("PTZ")

@property
def ptz_data(self) -> dict[str, Any]:
"""Create a smaller dictionary containing all PTZ information."""
return self.get_param("PTZ")

# Stream profiles

async def update_stream_profiles(self) -> None:
"""Update stream profiles group of parameters."""
await self.stream_profile_handler.update()

@property
def stream_profiles_params(self) -> StreamProfileParam:
"""Provide stream profiles parameters."""
return self.stream_profile_handler.get_params().get("0")

@property
def stream_profiles_max_groups(self) -> int:
"""Maximum number of supported stream profiles."""
return self.stream_profile_handler.get_params().get("0").max_groups

@property
def stream_profiles(self) -> list[StreamProfile]:
"""Return a list of stream profiles."""
return self.stream_profile_handler.get_params().get("0").stream_profiles
40 changes: 25 additions & 15 deletions axis/vapix/vapix.py
Original file line number Diff line number Diff line change
Expand Up @@ -83,28 +83,36 @@ def firmware_version(self) -> str:
"""Firmware version of device."""
if self.basic_device_info.supported():
return self.basic_device_info.version
return self.params.properties.firmware_version
if self.params.property_handler.supported():
return self.params.property_handler.get_params()["0"].firmware_version
return ""

@property
def product_number(self) -> str:
"""Product number of device."""
if self.basic_device_info.supported():
return self.basic_device_info.prodnbr
return self.params.brand.prodnbr
if self.params.brand_handler.supported():
return self.params.brand_handler.get_params()["0"].prodnbr
return ""

@property
def product_type(self) -> str:
"""Product type of device."""
if self.basic_device_info.supported():
return self.basic_device_info.prodtype
return self.params.brand.prodtype
if self.params.brand_handler.supported():
return self.params.brand_handler.get_params()["0"].prodtype
return ""

@property
def serial_number(self) -> str:
"""Device serial number."""
if self.basic_device_info.supported():
return self.basic_device_info.serialnumber
return self.params.properties.system_serialnumber
if self.params.property_handler.supported():
return self.params.property_handler.get_params()["0"].system_serialnumber
return ""

@property
def access_rights(self) -> SecondaryGroup:
Expand All @@ -118,7 +126,9 @@ def streaming_profiles(self) -> list:
"""List streaming profiles."""
if self.stream_profiles.supported():
return list(self.stream_profiles.values())
return self.params.stream_profiles
if self.params.stream_profile_handler.supported():
return self.params.stream_profile_handler.get_params()["0"].stream_profiles
return []

@property
def ports(self) -> IoPortManagement | Ports:
Expand Down Expand Up @@ -189,27 +199,30 @@ async def initialize_param_cgi(self, preload_data: bool = True) -> None:
tasks.append(self.params.update())

else:
tasks.append(self.params.update_properties())
tasks.append(self.params.update_ptz())
tasks.append(self.params.property_handler.update())
tasks.append(self.params.ptz_handler.update())

if not self.basic_device_info.supported():
tasks.append(self.params.update_brand())
tasks.append(self.params.brand_handler.update())

if not self.io_port_management.supported():
tasks.append(self.port_cgi.update())

if not self.stream_profiles.supported():
tasks.append(self.params.update_stream_profiles())
tasks.append(self.params.stream_profile_handler.update())

if self.view_areas.supported():
tasks.append(self.params.update_image())
tasks.append(self.params.image_handler.update())

await asyncio.gather(*tasks)

if not self.params.property_handler.supported():
return

if not self.light_control.supported() and self.params.properties.light_control:
if (
not self.light_control.supported()
and self.params.property_handler.get_params()["0"].light_control
):
try:
await self.light_control.update()
except Unauthorized: # Probably a viewer account
Expand All @@ -218,14 +231,11 @@ async def initialize_param_cgi(self, preload_data: bool = True) -> None:
if not self.io_port_management.supported():
self.port_cgi._items = self.port_cgi.process_ports()

if self.params.properties.ptz:
self.ptz._items = self.ptz.process_ptz()

async def initialize_applications(self) -> None:
"""Load data for applications on device."""
self.applications = Applications(self)
if self.params.property_handler.supported() and version.parse(
self.params.properties.embedded_development
self.params.property_handler.get_params()["0"].embedded_development
) >= version.parse(APPLICATIONS_MINIMUM_VERSION):
try:
await self.applications.update()
Expand Down
Loading

0 comments on commit 8e47bc9

Please sign in to comment.