Returns a session unique identified for CompartmentProperties
string containing session unique identified for CompartmentProperties
str
+Returns a compartment location for CompartmentProperties
compartment location for CompartmentProperties
str
+Returns a compartment status for CompartmentProperties
compartment location for CompartmentProperties
CompartmentActualStatus
+Returns an open code for Parcel
Open code for Parcel
str
+Returns a QR image for Parcel
QR image for Parcel
BytesIO
+Returns a compartment properties for Parcel
Compartment properties for Parcel
Returns a compartment properties for Parcel
Compartment properties for Parcel
Returns a compartment location for Parcel
Compartment location for Parcel
Returns a compartment status for Parcel
Compartment status for Parcel
CompartmentActualStatus
+Returns a mocked location for PickupPoint
tuple containing location for PickupPoint
tuple
+
+from aiohttp import ClientSession
+from typing import List
+import logging
+
+from inpost.static import *
+
+
+[docs]class Inpost:
+ """Python representation of an Inpost app. Essentially implements methods to manage all incoming parcels"""
+
+[docs] def __init__(self):
+ """Constructor method"""
+ self.phone_number: str | None = None
+ self.sms_code: str | None = None
+ self.auth_token: str | None = None
+ self.refr_token: str | None = None
+ self.sess: ClientSession = ClientSession()
+ self.parcel: Parcel | None = None
+ self._log: logging.Logger | None = None
+
+ def __repr__(self):
+ return f'Phone number: {self.phone_number}\nToken: {self.auth_token}'
+
+[docs] @classmethod
+ async def from_phone_number(cls, phone_number: str | int):
+ """`Classmethod` to initialize :class:`Inpost` object with phone number
+
+ :param phone_number: User's Inpost phone number
+ :type phone_number: str | int"""
+ if isinstance(phone_number, int):
+ phone_number = str(phone_number)
+ inp = cls()
+ await inp.set_phone_number(phone_number=phone_number)
+ inp._log.info(f'initialized by from_phone_number')
+ return inp
+
+[docs] async def set_phone_number(self, phone_number: str | int) -> bool:
+ """Set :class:`Inpost` phone number required for verification
+
+ :param phone_number: User's Inpost phone number
+ :type phone_number: str | int
+ :return: True if `Inpost.phone_number` is set
+ :rtype: bool
+ :raises PhoneNumberError: Wrong phone number format"""
+ if isinstance(phone_number, int):
+ phone_number = str(phone_number)
+
+ if len(phone_number) == 9 and phone_number.isdigit():
+ self._log = logging.getLogger(f'{__class__.__name__}.{phone_number}')
+ self._log.setLevel(level=logging.DEBUG)
+ self._log.info(f'initializing inpost object with phone number {phone_number}')
+ self.phone_number = phone_number
+ return True
+
+ raise PhoneNumberError(f'Wrong phone number format: {phone_number} (should be 9 digits)')
+
+[docs] async def send_sms_code(self) -> bool:
+ """Sends sms code to `Inpost.phone_number`
+
+ :return: True if sms code sent
+ :rtype: bool
+ :raises PhoneNumberError: Missing phone number
+ :raises UnauthorizedError: Unauthorized access to inpost services,
+ :raises NotFoundError: Phone number not found
+ :raises UnidentifiedAPIError: Unexpected things happened
+ """
+ if not self.phone_number: # can't log it cuz if there's no phone number no logger initialized @shrug
+ raise PhoneNumberError('Phone number missing')
+
+ self._log.info(f'sending sms code')
+ async with await self.sess.post(url=send_sms_code,
+ json={
+ 'phoneNumber': f'{self.phone_number}'
+ }) as phone:
+ match phone.status:
+ case 200:
+ self._log.debug(f'sms code sent')
+ return True
+ case 401:
+ self._log.error(f'could not send sms code, unauthorized')
+ raise UnauthorizedError(reason=phone)
+ case 404:
+ self._log.error(f'could not send sms code, not found')
+ raise NotFoundError(reason=phone)
+ case _:
+ self._log.error(f'could not send sms code, unhandled status')
+
+ raise UnidentifiedAPIError(reason=phone)
+
+ # if phone.status == 200:
+ # self._log.debug(f'sms code sent')
+ # return True
+ # else:
+ # self._log.error(f'could not sent sms code')
+ # raise PhoneNumberError(reason=phone)
+
+[docs] async def confirm_sms_code(self, sms_code: str | int) -> bool:
+ """Confirms sms code sent to `Inpost.phone_number` and fetches tokens
+
+ :param sms_code: sms code sent to Inpost.phone_number device
+ :type sms_code: str | int
+ :return: True if sms code gets confirmed and tokens fetched
+ :rtype: bool
+ :raises SmsCodeError: Wrong sms code format
+ :raises UnauthorizedError: Unauthorized access to inpost services,
+ :raises NotFoundError: Phone number not found
+ :raises UnidentifiedAPIError: Unexpected thing happened
+ """
+
+ if not self.phone_number: # can't log it cuz if there's no phone number no logger initialized @shrug
+ raise PhoneNumberError('Phone number missing')
+
+ if isinstance(sms_code, int):
+ sms_code = str(sms_code)
+
+ if len(sms_code) != 6 or not sms_code.isdigit():
+ raise SmsCodeError(reason=f'Wrong sms code format: {sms_code} (should be 6 digits)')
+
+ self._log.info(f'confirming sms code')
+ async with await self.sess.post(url=confirm_sms_code,
+ headers=appjson,
+ json={
+ "phoneNumber": self.phone_number,
+ "smsCode": sms_code,
+ "phoneOS": "Android"
+ }) as confirmation:
+ match confirmation.status:
+ case 200:
+ resp = await confirmation.json()
+ self.sms_code = sms_code
+ self.refr_token = resp['refreshToken']
+ self.auth_token = resp['authToken']
+ self._log.debug(f'sms code confirmed')
+ return True
+ case 401:
+ self._log.error(f'could not confirm sms code, unauthorized')
+ raise UnauthorizedError(reason=confirmation)
+ case 404:
+ self._log.error(f'could not confirm sms code, not found')
+ raise NotFoundError(reason=confirmation)
+ case _:
+ self._log.error(f'could not confirm sms code, unhandled status')
+
+ raise UnidentifiedAPIError(reason=confirmation)
+
+ # if confirmation.status == 200:
+ # resp = await confirmation.json()
+ # self.sms_code = sms_code
+ # self.refr_token = resp['refreshToken']
+ # self.auth_token = resp['authToken']
+ # self._log.debug(f'sms code confirmed')
+ # return True
+ # else:
+ # self._log.error(f'could not confirm sms code')
+ # raise SmsCodeConfirmationError(reason=confirmation)
+
+[docs] async def refresh_token(self) -> bool:
+ """Refreshes authorization token using refresh token
+
+ :return: True if Inpost.auth_token gets refreshed
+ :rtype: bool
+ :raises RefreshTokenError: Missing refresh token
+ :raises UnauthorizedError: Unauthorized access to inpost services,
+ :raises NotFoundError: Phone number not found
+ :raises UnidentifiedAPIError: Unexpected thing happened
+ """
+ self._log.info(f'refreshing token')
+
+ if not self.refr_token:
+ self._log.error(f'refresh token missing')
+ raise RefreshTokenError(reason='Refresh token missing')
+
+ async with await self.sess.post(url=refresh_token,
+ headers=appjson,
+ json={
+ "refreshToken": self.refr_token,
+ "phoneOS": "Android"
+ }) as confirmation:
+ match confirmation.status:
+ case 200:
+ resp = await confirmation.json()
+ if resp['reauthenticationRequired']:
+ self._log.error(f'could not refresh token, log in again')
+ raise ReAuthenticationError(reason='You need to log in again!')
+
+ self.auth_token = resp['authToken']
+ self._log.debug(f'token refreshed')
+ return True
+ case 401:
+ self._log.error(f'could not refresh token, unauthorized')
+ raise UnauthorizedError(reason=confirmation)
+ case 404:
+ self._log.error(f'could not refresh token, not found')
+ raise NotFoundError(reason=confirmation)
+ case _:
+ self._log.error(f'could not refresh token, unhandled status')
+
+ raise UnidentifiedAPIError(reason=confirmation)
+
+ # if confirmation.status == 200:
+ # resp = await confirmation.json()
+ # if resp['reauthenticationRequired']:
+ # self._log.error(f'could not refresh token, log in again')
+ # raise ReAuthenticationError(reason='You need to log in again!')
+ # self.auth_token = resp['authToken']
+ # self._log.debug(f'token refreshed')
+ # return True
+ #
+ # else:
+ # self._log.error(f'could not refresh token')
+ # raise RefreshTokenException(reason=confirmation)
+
+[docs] async def logout(self) -> bool:
+ """Logouts user from inpost api service
+
+ :return: True if the user is logged out
+ :rtype: bool
+ :raises NotAuthenticatedError: User not authenticated in inpost service
+ :raises UnauthorizedError: Unauthorized access to inpost services,
+ :raises NotFoundError: Phone number not found
+ :raises UnidentifiedAPIError: Unexpected thing happened"""
+ self._log.info(f'logging out')
+
+ if not self.auth_token:
+ self._log.error(f'authorization token missing')
+ raise NotAuthenticatedError(reason='Not logged in')
+
+ async with await self.sess.post(url=logout,
+ headers={'Authorization': self.auth_token}) as resp:
+ match resp.status:
+ case 200:
+ self.phone_number = None
+ self.refr_token = None
+ self.auth_token = None
+ self.sms_code = None
+ self._log.debug('logged out')
+ return True
+ case 401:
+ self._log.error('could not log out, unauthorized')
+ raise UnauthorizedError(reason=resp)
+ case 404:
+ self._log.error('could not log out, not found')
+ raise NotFoundError(reason=resp)
+ case _:
+ self._log.error('could not log out, unhandled status')
+
+ raise UnidentifiedAPIError(reason=resp)
+
+ # if resp.status == 200:
+ # self.phone_number = None
+ # self.refr_token = None
+ # self.auth_token = None
+ # self.sms_code = None
+ # self._log.debug('logged out')
+ # return True
+ # else:
+ # self._log.error('could not log out')
+ # raise UnidentifiedAPIError(reason=resp)
+
+[docs] async def disconnect(self) -> bool:
+ """Simplified method to logout and close user's session
+
+ :return: True if user is logged out and session is closed else False
+ :raises NotAuthenticatedError: User not authenticated in inpost service"""
+ self._log.info(f'disconnecting')
+ if not self.auth_token:
+ self._log.error(f'authorization token missing')
+ raise NotAuthenticatedError(reason='Not logged in')
+
+ if await self.logout():
+ await self.sess.close()
+ self._log.debug(f'disconnected')
+ return True
+
+ self._log.error('could not disconnect')
+ return False
+
+[docs] async def get_parcel(self, shipment_number: int | str, parse=False) -> dict | Parcel:
+ """Fetches single parcel from provided shipment number
+
+ :param shipment_number: Parcel's shipment number
+ :type shipment_number: int | str
+ :param parse: if set to True method will return :class:`Parcel` else :class:`dict`
+ :type parse: bool
+ :return: Fetched parcel data
+ :rtype: dict | Parcel
+ :raises NotAuthenticatedError: User not authenticated in inpost service
+ :raises UnauthorizedError: Unauthorized access to inpost services,
+ :raises NotFoundError: Phone number not found
+ :raises UnidentifiedAPIError: Unexpected thing happened"""
+ self._log.info(f'getting parcel with shipment number: {shipment_number}')
+
+ if not self.auth_token:
+ self._log.error(f'authorization token missing')
+ raise NotAuthenticatedError(reason='Not logged in')
+
+ async with await self.sess.get(url=f"{parcel}{shipment_number}",
+ headers={'Authorization': self.auth_token},
+ ) as resp:
+ match resp.status:
+ case 200:
+ self._log.debug(f'parcel with shipment number {shipment_number} received')
+ return await resp.json() if not parse else Parcel(await resp.json(), logger=self._log)
+ case 401:
+ self._log.error(f'could not get parcel with shipment number {shipment_number}, unauthorized')
+ raise UnauthorizedError(reason=resp)
+ case 404:
+ self._log.error(f'could not get parcel with shipment number {shipment_number}, not found')
+ raise NotFoundError(reason=resp)
+ case _:
+ self._log.error(f'could not get parcel with shipment number {shipment_number}, unhandled status')
+
+ raise UnidentifiedAPIError(reason=resp)
+ # if resp.status == 200:
+ # self._log.debug(f'parcel with shipment number {shipment_number} received')
+ # return await resp.json() if not parse else Parcel(await resp.json(), logger=self._log)
+ #
+ # else:
+ # self._log.error(f'could not get parcel with shipment number {shipment_number}')
+ # raise UnidentifiedAPIError(reason=resp)
+
+[docs] async def get_parcels(self,
+ parcel_type: ParcelType = ParcelType.TRACKED,
+ status: ParcelStatus | List[ParcelStatus] | None = None,
+ pickup_point: str | List[str] | None = None,
+ shipment_type: ParcelShipmentType | List[ParcelShipmentType] | None = None,
+ parcel_size: ParcelLockerSize | ParcelCarrierSize | None = None,
+ parse: bool = False) -> List[dict] | List[Parcel]:
+ """Fetches all available parcels for set `Inpost.phone_number` and optionally filters them
+
+ :param parcel_type: Parcel type (e.g. received, sent, returned)
+ :type parcel_type: ParcelType
+ :param status: status that each fetched parcels has to be in
+ :type status: ParcelStatus | list[ParcelStatus] | None
+ :param pickup_point: Fetched parcels have to be picked from this pickup point (e.g. `GXO05M`)
+ :type pickup_point: str | list[str] | None
+ :param shipment_type: Fetched parcels have to be shipped that way
+ :type shipment_type: ParcelShipmentType | list[ParcelShipmentType] | None
+ :param parcel_size: Fetched parcels have to be this size
+ :type parcel_size: ParcelLockerSize | ParcelCarrierSize | None
+ :param parse: if set to True method will return list[:class:`Parcel`] else list[:class:`dict`]
+ :type parse: bool
+ :return: fetched parcels data
+ :rtype: list[dict] | list[Parcel]
+ :raises NotAuthenticatedError: User not authenticated in inpost service
+ :raises ParcelTypeError: Unknown parcel type selected
+ :raises UnauthorizedError: Unauthorized access to inpost services,
+ :raises NotFoundError: Phone number not found
+ :raises UnidentifiedAPIError: Unexpected thing happened"""
+ self._log.info('getting parcels')
+
+ if not self.auth_token:
+ self._log.error(f'authorization token missing')
+ raise NotAuthenticatedError(reason='Not logged in')
+
+ if not isinstance(parcel_type, ParcelType):
+ self._log.error(f'wrong parcel type {parcel_type}')
+ raise ParcelTypeError(reason=f'Unknown parcel type: {parcel_type}')
+
+ match parcel_type:
+ case ParcelType.TRACKED:
+ self._log.debug(f'getting parcel type {parcel_type}')
+ url = parcels
+ case ParcelType.SENT:
+ self._log.debug(f'getting parcel type {parcel_type}')
+ url = sent
+ case ParcelType.RETURNS:
+ self._log.debug(f'getting parcel type {parcel_type}')
+ url = returns
+ case _:
+ self._log.error(f'wrong parcel type {parcel_type}')
+ raise ParcelTypeError(reason=f'Unknown parcel type: {parcel_type}')
+
+ async with await self.sess.get(url=url,
+ headers={'Authorization': self.auth_token},
+ ) as resp:
+ match resp.status:
+ case 200:
+ self._log.debug(f'received {parcel_type} parcels')
+ _parcels = (await resp.json())['parcels']
+
+ if status is not None:
+ if isinstance(status, ParcelStatus):
+ status = [status]
+
+ _parcels = (_parcel for _parcel in _parcels if ParcelStatus[_parcel['status']] in status)
+
+ if pickup_point is not None:
+ if isinstance(pickup_point, str):
+ pickup_point = [pickup_point]
+
+ _parcels = (_parcel for _parcel in _parcels if _parcel['pickUpPoint']['name'] in pickup_point)
+
+ if shipment_type is not None:
+ if isinstance(shipment_type, ParcelShipmentType):
+ shipment_type = [shipment_type]
+
+ _parcels = (_parcel for _parcel in _parcels if
+ ParcelShipmentType[_parcel['shipmentType']] in shipment_type)
+
+ if parcel_size is not None:
+ if isinstance(parcel_size, ParcelCarrierSize):
+ parcel_size = [parcel_size]
+
+ _parcels = (_parcel for _parcel in _parcels if
+ ParcelCarrierSize[_parcel['parcelSize']] in parcel_size)
+
+ if isinstance(parcel_size, ParcelLockerSize):
+ parcel_size = [parcel_size]
+
+ _parcels = (_parcel for _parcel in _parcels if
+ ParcelLockerSize[_parcel['parcelSize']] in parcel_size)
+
+ return _parcels if not parse else [Parcel(parcel_data=data, logger=self._log) for data in _parcels]
+ case 401:
+ self._log.error(f'could not get parcels, unauthorized')
+ raise UnauthorizedError(reason=resp)
+ case 404:
+ self._log.error(f'could not get parcels, not found')
+ raise NotFoundError(reason=resp)
+ case _:
+ self._log.error(f'could not get parcels, unhandled status')
+
+ raise UnidentifiedAPIError(reason=resp)
+
+ # if resp.status == 200:
+ # self._log.debug(f'received {parcel_type} parcels')
+ # _parcels = (await resp.json())['parcels']
+ #
+ # if status is not None:
+ # if isinstance(status, ParcelStatus):
+ # status = [status]
+ #
+ # _parcels = (_parcel for _parcel in _parcels if ParcelStatus[_parcel['status']] in status)
+ #
+ # if pickup_point is not None:
+ # if isinstance(pickup_point, str):
+ # pickup_point = [pickup_point]
+ #
+ # _parcels = (_parcel for _parcel in _parcels if _parcel['pickUpPoint']['name'] in pickup_point)
+ #
+ # if shipment_type is not None:
+ # if isinstance(shipment_type, ParcelShipmentType):
+ # shipment_type = [shipment_type]
+ #
+ # _parcels = (_parcel for _parcel in _parcels if
+ # ParcelShipmentType[_parcel['shipmentType']] in shipment_type)
+ #
+ # if parcel_size is not None:
+ # if isinstance(parcel_size, ParcelCarrierSize):
+ # parcel_size = [parcel_size]
+ #
+ # _parcels = (_parcel for _parcel in _parcels if
+ # ParcelCarrierSize[_parcel['parcelSize']] in parcel_size)
+ #
+ # if isinstance(parcel_size, ParcelLockerSize):
+ # parcel_size = [parcel_size]
+ #
+ # _parcels = (_parcel for _parcel in _parcels if
+ # ParcelLockerSize[_parcel['parcelSize']] in parcel_size)
+ #
+ # return _parcels if not parse else [Parcel(parcel_data=data, logger=self._log) for data in _parcels]
+ #
+ # else:
+ # self._log.error(f'could not get parcels')
+ # raise UnidentifiedAPIError(reason=resp)
+
+[docs] async def collect_compartment_properties(self, shipment_number: str | int | None = None,
+ parcel_obj: Parcel | None = None, location: dict | None = None) -> bool:
+ """Validates sent data and fetches required compartment properties for opening
+
+ :param shipment_number: Parcel's shipment number
+ :type shipment_number: int | str | None
+ :param parcel_obj: :class:`Parcel` object to obtain data from
+ :type parcel_obj: Parcel | None
+ :param location: Fetched parcels have to be picked from this pickup point (e.g. `GXO05M`)
+ :type location: dict | None
+ :return: fetched parcels data
+ :rtype: bool
+ :raises SingleParamError: Fields shipment_number and parcel_obj filled in but only one of them is required
+ :raises NotAuthenticatedError: User not authenticated in inpost service
+ :raises UnauthorizedError: Unauthorized access to inpost services,
+ :raises NotFoundError: Phone number not found
+ :raises UnidentifiedAPIError: Unexpected thing happened
+
+ .. warning:: you must fill in only one parameter - shipment_number or parcel_obj!"""
+
+ self._log.info(f'collecting compartment properties for {shipment_number}')
+
+ if shipment_number and parcel_obj:
+ self._log.error(f'shipment_number and parcel_obj filled in')
+ raise SingleParamError(reason='Fields shipment_number and parcel_obj filled in! Choose one!')
+
+ if not self.auth_token:
+ self._log.error(f'authorization token missing')
+ raise NotAuthenticatedError(reason='Not logged in')
+
+ if shipment_number is not None and parcel_obj is None:
+ self._log.debug(f'parcel_obj not provided, getting from shipment number {shipment_number}')
+ parcel_obj = await self.get_parcel(shipment_number=shipment_number, parse=True)
+
+ async with await self.sess.post(url=collect,
+ headers={'Authorization': self.auth_token},
+ json={
+ 'parcel': parcel_obj.compartment_open_data,
+ 'geoPoint': location if location is not None else parcel_obj.mocked_location
+ }) as collect_resp:
+ match collect_resp.status:
+ case 200:
+ self._log.debug(f'collected compartment properties for {shipment_number}')
+ parcel_obj.compartment_properties = await collect_resp.json()
+ self.parcel = parcel_obj
+ return True
+ case 401:
+ self._log.error(f'could not collect compartment properties for {shipment_number}, unauthorized')
+ raise UnauthorizedError(reason=collect_resp)
+ case 404:
+ self._log.error(f'could not collect compartment properties for {shipment_number}, not found')
+ raise NotFoundError(reason=collect_resp)
+ case _:
+ self._log.error(f'could not collect compartment properties for {shipment_number}, unhandled status')
+
+ raise UnidentifiedAPIError(reason=collect_resp)
+
+ # if collect_resp.status == 200:
+ # self._log.debug(f'collected compartment properties for {shipment_number}')
+ # parcel_obj.compartment_properties = await collect_resp.json()
+ # self.parcel = parcel_obj
+ # return True
+ #
+ # else:
+ # self._log.error(f'could not collect compartment properties for {shipment_number}')
+ # raise UnidentifiedAPIError(reason=collect_resp)
+
+[docs] async def open_compartment(self) -> bool:
+ """Opens compartment for `Inpost.parcel` object
+
+ :return: True if compartment gets opened
+ :rtype: bool
+ :raises NotAuthenticatedError: User not authenticated in inpost service
+ :raises UnauthorizedError: Unauthorized access to inpost services,
+ :raises NotFoundError: Phone number not found
+ :raises UnidentifiedAPIError: Unexpected thing happened"""
+ self._log.info(f'opening compartment for {self.parcel.shipment_number}')
+
+ if not self.auth_token:
+ self._log.debug(f'authorization token missing')
+ raise NotAuthenticatedError(reason='Not logged in')
+
+ async with await self.sess.post(url=compartment_open,
+ headers={'Authorization': self.auth_token},
+ json={
+ 'sessionUuid': self.parcel.compartment_properties.session_uuid
+ }) as compartment_open_resp:
+ match compartment_open_resp.status:
+ case 200:
+ self._log.debug(f'opened comaprtment for {self.parcel.shipment_number}')
+ self.parcel.compartment_properties.location = await compartment_open_resp.json()
+ return True
+ case 401:
+ self._log.error(f'could not open compartment for {self.parcel.shipment_number}, unauthorized')
+ raise UnauthorizedError(reason=compartment_open_resp)
+ case 404:
+ self._log.error(f'could not open compartment for {self.parcel.shipment_number}, not found')
+ raise NotFoundError(reason=compartment_open_resp)
+ case _:
+ self._log.error(f'could not open compartment for {self.parcel.shipment_number}, unhandled status')
+
+ raise UnidentifiedAPIError(reason=compartment_open_resp)
+
+ # if compartment_open_resp.status == 200:
+ # self._log.debug(f'opened comaprtment for {self.parcel.shipment_number}')
+ # self.parcel.compartment_properties.location = await compartment_open_resp.json()
+ # return True
+ #
+ # else:
+ # self._log.error(f'could not open compartment for {self.parcel.shipment_number}')
+ # raise UnidentifiedAPIError(reason=compartment_open_resp)
+
+[docs] async def check_compartment_status(self,
+ expected_status: CompartmentExpectedStatus = CompartmentExpectedStatus.OPENED) -> bool:
+ """Checks and compare compartment status (e.g. opened, closed) with expected status
+
+ :param expected_status: Compartment expected status
+ :type expected_status: CompartmentExpectedStatus
+ :return: True if actual status equals expected status else False
+ :rtype: bool
+ :raises NotAuthenticatedError: User not authenticated in inpost service
+ :raises UnauthorizedError: Unauthorized access to inpost services,
+ :raises NotFoundError: Phone number not found
+ :raises UnidentifiedAPIError: Unexpected thing happened"""
+ self._log.info(f'checking compartment status for {self.parcel.shipment_number}')
+
+ if not self.auth_token:
+ self._log.debug(f'authorization token missing')
+ raise NotAuthenticatedError(reason='Not logged in')
+
+ if not self.parcel:
+ self._log.debug(f'parcel missing')
+ raise NoParcelError(reason='Parcel is not set')
+
+ async with await self.sess.post(url=compartment_status,
+ headers={'Authorization': self.auth_token},
+ json={
+ 'sessionUuid': self.parcel.compartment_properties.session_uuid,
+ 'expectedStatus': expected_status.name
+ }) as compartment_status_resp:
+ match compartment_status_resp.status:
+ case 200:
+ self._log.debug(f'checked compartment status for {self.parcel.shipment_number}')
+ return CompartmentExpectedStatus[
+ (await compartment_status_resp.json())['status']] == expected_status
+ case 401:
+ self._log.error(
+ f'could not check compartment status for {self.parcel.shipment_number}, unauthorized')
+ raise UnauthorizedError(reason=compartment_status_resp)
+ case 404:
+ self._log.error(f'could not check compartment status for {self.parcel.shipment_number}, not found')
+ raise NotFoundError(reason=compartment_status_resp)
+ case _:
+ self._log.error(
+ f'could not check compartment status for {self.parcel.shipment_number}, unhandled status')
+
+ raise UnidentifiedAPIError(reason=compartment_status_resp)
+
+ # if compartment_status_resp.status == 200:
+ # self._log.debug(f'checked compartment status for {self.parcel.shipment_number}')
+ # return CompartmentExpectedStatus[(await compartment_status_resp.json())['status']] == expected_status
+ # else:
+ # self._log.error(f'could not check compartment status for {self.parcel.shipment_number}')
+ # raise UnidentifiedAPIError(reason=compartment_status_resp)
+
+[docs] async def terminate_collect_session(self) -> bool:
+ """Terminates user session in inpost api service
+
+ :return: True if the user session is terminated
+ :rtype: bool
+ :raises NotAuthenticatedError: User not authenticated in inpost service
+ :raises UnauthorizedError: Unauthorized access to inpost services,
+ :raises NotFoundError: Phone number not found
+ :raises UnidentifiedAPIError: Unexpected thing happened"""
+ self._log.info(f'terminating collect session for {self.parcel.shipment_number}')
+
+ if not self.auth_token:
+ self._log.debug(f'authorization token missing')
+ raise NotAuthenticatedError(reason='Not logged in')
+
+ async with await self.sess.post(url=terminate_collect_session,
+ headers={'Authorization': self.auth_token},
+ json={
+ 'sessionUuid': self.parcel.compartment_properties.session_uuid
+ }) as terminate_resp:
+ match terminate_resp.status:
+ case 200:
+ self._log.debug(f'terminated collect session for {self.parcel.shipment_number}')
+ return True
+ case 401:
+ self._log.error(
+ f'could not terminate collect session for {self.parcel.shipment_number}, unauthorized')
+ raise UnauthorizedError(reason=terminate_resp)
+ case 404:
+ self._log.error(f'could not terminate collect session for {self.parcel.shipment_number}, not found')
+ raise NotFoundError(reason=terminate_resp)
+ case _:
+ self._log.error(
+ f'could not terminate collect session for {self.parcel.shipment_number}, unhandled status')
+
+ raise UnidentifiedAPIError(reason=terminate_resp)
+
+ # if terminate_resp.status == 200:
+ # self._log.debug(f'terminated collect session for {self.parcel.shipment_number}')
+ # return True
+ # else:
+ # self._log.error(f'could not terminate collect session for {self.parcel.shipment_number}')
+ # raise UnidentifiedAPIError(reason=terminate_resp)
+
+[docs] async def collect(self, shipment_number: str | None = None, parcel_obj: Parcel | None = None,
+ location: dict | None = None) -> bool:
+ """Simplified method to open compartment
+
+ :param shipment_number: Parcel's shipment number
+ :type shipment_number: int | str | None
+ :param parcel_obj: :class:`Parcel` object to obtain data from
+ :type parcel_obj: Parcel | None
+ :param location: Fetched parcels have to be picked from this pickup point (e.g. `GXO05M`)
+ :type location: dict | None
+ :return: fetched parcels data
+ :rtype: bool
+ :raises SingleParamError: Fields shipment_number and parcel_obj filled in but only one of them is required
+ :raises NotAuthenticatedError: User not authenticated in inpost service
+ :raises UnauthorizedError: Unauthorized access to inpost services,
+ :raises NotFoundError: Phone number not found
+ :raises UnidentifiedAPIError: Unexpected thing happened
+
+ .. warning:: you must fill in only one parameter - shipment_number or parcel_obj!"""
+
+ self._log.info(f'collecing parcel with shipment number {self.parcel.shipment_number}')
+
+ if shipment_number and parcel_obj:
+ self._log.error(f'shipment_number and parcel_obj filled in')
+ raise SingleParamError(reason='Fields shipment_number and parcel_obj filled! Choose one!')
+
+ if not self.auth_token:
+ self._log.error(f'authorization token missing')
+ raise NotAuthenticatedError(reason='Not logged in')
+
+ if shipment_number is not None and parcel_obj is None:
+ parcel_obj = await self.get_parcel(shipment_number=shipment_number, parse=True)
+
+ if await self.collect_compartment_properties(parcel_obj=parcel_obj, location=location):
+ if await self.open_compartment():
+ if await self.check_compartment_status():
+ return True
+
+ return False
+
+[docs] async def close_compartment(self) -> bool:
+ """Checks whether actual compartment status and expected one matches then notifies inpost api that compartment is closed
+
+ :return: True if compartment status is closed and successfully terminates user's session else False
+ :rtype: bool"""
+ self._log.info(f'closing compartment for {self.parcel.shipment_number}')
+
+ if await self.check_compartment_status(expected_status=CompartmentExpectedStatus.CLOSED):
+ if await self.terminate_collect_session():
+ return True
+
+ return False
+
+[docs] async def get_prices(self) -> dict:
+ """Fetches prices for inpost services
+
+ :return: :class:`dict` of prices for inpost services
+ :rtype: dict
+ :raises NotAuthenticatedError: User not authenticated in inpost service
+ :raises UnauthorizedError: Unauthorized access to inpost services,
+ :raises NotFoundError: Phone number not found
+ :raises UnidentifiedAPIError: Unexpected thing happened"""
+ self._log.info(f'getting parcel prices')
+
+ if not self.auth_token:
+ self._log.debug(f'authorization token missing')
+ raise NotAuthenticatedError(reason='Not logged in')
+
+ async with await self.sess.get(url=parcel_prices,
+ headers={'Authorization': self.auth_token}) as resp:
+ match resp.status:
+ case 200:
+ self._log.debug(f'got parcel prices')
+ return await resp.json()
+ case 401:
+ self._log.error('could not get parcel prices, unauthorized')
+ raise UnauthorizedError(reason=resp)
+ case 404:
+ self._log.error('could not get parcel prices, not found')
+ raise NotFoundError(reason=resp)
+ case _:
+ self._log.error('could not get parcel prices, unhandled status')
+
+ raise UnidentifiedAPIError(reason=resp)
+
+ # if resp.status == 200:
+ # self._log.debug(f'got parcel prices')
+ # return await resp.json()
+ #
+ # else:
+ # self._log.error('could not get parcel prices')
+ # raise UnidentifiedAPIError(reason=resp)
+
+import logging
+import random
+from io import BytesIO
+from typing import List, Tuple
+
+import qrcode
+from arrow import get, arrow
+
+from inpost.static.statuses import *
+
+
+[docs]class Parcel:
+ """Object representation of :class:`inpost.api.Inpost` compartment properties
+
+ :param parcel_data: :class:`dict` containing all parcel data
+ :type parcel_data: dict
+ :param logger: :class:`logging.Logger` parent instance
+ :type logger: logging.Logger"""
+
+[docs] def __init__(self, parcel_data: dict, logger: logging.Logger):
+ """Constructor method"""
+ self.shipment_number: str = parcel_data['shipmentNumber']
+ self._log: logging.Logger = logger.getChild(f'{__class__.__name__}.{self.shipment_number}')
+ self.shipment_type: ParcelShipmentType = ParcelShipmentType[parcel_data['shipmentType']]
+ self._open_code: str | None = parcel_data['openCode'] if 'openCode' in parcel_data else None
+ self._qr_code: QRCode | None = QRCode(qrcode_data=parcel_data['qrCode'], logger=self._log) \
+ if 'qrCode' in parcel_data else None
+ self.stored_date: arrow | None = get(parcel_data['storedDate']) if 'storedDate' in parcel_data else None
+ self.pickup_date: arrow | None = get(parcel_data['pickUpDate']) if 'pickUpDate' in parcel_data else None
+ self.parcel_size: ParcelLockerSize | ParcelCarrierSize = ParcelLockerSize[parcel_data['parcelSize']] \
+ if self.shipment_type == ParcelShipmentType.parcel else ParcelCarrierSize[parcel_data['parcelSize']]
+ self.receiver: Receiver = Receiver(receiver_data=parcel_data['receiver'], logger=self._log)
+ self.sender: Sender = Sender(sender_data=parcel_data['sender'], logger=self._log)
+ self.pickup_point: PickupPoint = PickupPoint(pickuppoint_data=parcel_data['pickUpPoint'], logger=self._log) \
+ if 'pickUpPoint' in parcel_data else None
+ self.multi_compartment: MultiCompartment | None = MultiCompartment(
+ parcel_data['multiCompartment'], logger=self._log) if 'multiCompartment' in parcel_data else None
+ self.is_end_off_week_collection: bool = parcel_data['endOfWeekCollection']
+ self.operations: Operations = Operations(operations_data=parcel_data['operations'], logger=self._log)
+ self.status: ParcelStatus = ParcelStatus[parcel_data['status']]
+ self.event_log: List[EventLog] = [EventLog(eventlog_data=event, logger=self._log)
+ for event in parcel_data['eventLog']]
+ self.avizo_transaction_status: str = parcel_data['avizoTransactionStatus']
+ self.shared_to: List[SharedTo] = [SharedTo(sharedto_data=person, logger=self._log)
+ for person in parcel_data['sharedTo']]
+ self.ownership_status: ParcelOwnership = ParcelOwnership[parcel_data['ownershipStatus']]
+ self._compartment_properties: CompartmentProperties | None = None
+
+ self._log.debug(f'created parcel with shipment number {self.shipment_number}')
+
+ # log all unexpected things, so you can make an issue @github
+ if self.shipment_type == ParcelShipmentType.UNKNOWN:
+ self._log.debug(f'unexpected shipment_type: {parcel_data["shipmentType"]}')
+
+ if self.parcel_size == ParcelCarrierSize.UNKNOWN or self.parcel_size == ParcelLockerSize.UNKNOWN:
+ self._log.debug(f'unexpected parcel_size: {parcel_data["parcelSize"]}')
+
+ if self.status == ParcelStatus.UNKNOWN:
+ self._log.debug(f'unexpected parcel status: {parcel_data["status"]}')
+
+ if self.ownership_status == ParcelOwnership.UNKNOWN:
+ self._log.debug(f'unexpected ownership status: {parcel_data["ownershipStatus"]}')
+
+ def __str__(self):
+ return f"Sender: {str(self.sender)}\n" \
+ f"Shipment number: {self.shipment_number}\n" \
+ f"Status: {self.status}\n" \
+ f"Pickup point: {self.pickup_point}"
+
+ @property
+ def open_code(self) -> str | None:
+ """Returns an open code for :class:`Parcel`
+
+ :return: Open code for :class:`Parcel`
+ :rtype: str"""
+ self._log.debug('getting open code')
+ if self.shipment_type == ParcelShipmentType.parcel:
+ self._log.debug('got open code')
+ return self._open_code
+
+ self._log.debug('wrong ParcelShipmentType')
+ return None
+
+ @property
+ def generate_qr_image(self) -> BytesIO | None:
+ """Returns a QR image for :class:`Parcel`
+
+ :return: QR image for :class:`Parcel`
+ :rtype: BytesIO"""
+ self._log.debug('generating qr image')
+ if self.shipment_type == ParcelShipmentType.parcel:
+ self._log.debug('got qr image')
+ return self._qr_code.qr_image
+
+ self._log.debug('wrong ParcelShipmentType')
+ return None
+
+ @property
+ def compartment_properties(self):
+ """Returns a compartment properties for :class:`Parcel`
+
+ :return: Compartment properties for :class:`Parcel`
+ :rtype: CompartmentProperties"""
+ self._log.debug('getting comparment properties')
+ if self.shipment_type == ParcelShipmentType.parcel:
+ self._log.debug('got compartment properties')
+ return self._compartment_properties
+
+ self._log.debug('wrong ParcelShipmentType')
+ return None
+
+ @compartment_properties.setter
+ def compartment_properties(self, compartmentproperties_data: dict):
+ """Set compartment properties for :class:`Parcel`
+
+ :param compartmentproperties_data: :class:`dict` containing compartment properties data for :class:`Parcel`
+ :type compartmentproperties_data: CompartmentProperties"""
+ self._log.debug(f'setting compartment properties with {compartmentproperties_data}')
+ if self.shipment_type == ParcelShipmentType.parcel:
+ self._log.debug('compartment properties set')
+ self._compartment_properties = CompartmentProperties(compartmentproperties_data=compartmentproperties_data,
+ logger=self._log)
+
+ self._log.debug('wrong ParcelShipmentType')
+
+ @property
+ def compartment_location(self):
+ """Returns a compartment location for :class:`Parcel`
+
+ :return: Compartment location for :class:`Parcel`
+ :rtype: CompartmentLocation"""
+ self._log.debug('getting compartment location')
+ if self.shipment_type == ParcelShipmentType.parcel:
+ self._log.debug('got compartment location')
+ return self._compartment_properties.location
+
+ self._log.debug('wrong ParcelShipmentType')
+ return None
+
+ # @compartment_location.setter
+ # def compartment_location(self, location_data):
+ # """Set compartment location for :class:`Parcel`
+ # :param location_data: :class:`dict` containing `compartment properties` data for :class:`Parcel`
+ # :type location_data: CompartmentProperties"""
+ # self._log.debug('setting compartment location')
+ # if self.shipment_type == ParcelShipmentType.parcel:
+ # self._log.debug('compartment location set')
+ # self._compartment_properties.location = location_data
+ #
+ # self._log.debug('wrong ParcelShipmentType')
+
+ @property
+ def compartment_status(self) -> CompartmentActualStatus | None:
+ """Returns a compartment status for :class:`Parcel`
+
+ :return: Compartment status for :class:`Parcel`
+ :rtype: CompartmentActualStatus"""
+ self._log.debug('getting compartment status')
+ if self.shipment_type == ParcelShipmentType.parcel:
+ self._log.debug('got compartment status')
+ return self._compartment_properties.status
+
+ self._log.debug('wrong ParcelShipmentType')
+ return None
+
+ # @compartment_status.setter
+ # def compartment_status(self, status):
+ # self._log.debug('setting compartment status')
+ # if self.shipment_type == ParcelShipmentType.parcel:
+ # self._log.debug('compartment status set')
+ # self._compartment_properties.status = status
+ #
+ # self._log.debug('wrong ParcelShipmentType')
+
+ @property
+ def compartment_open_data(self):
+ """Returns a compartment open data for :class:`Parcel`
+
+ :return: dict containing compartment open data for :class:`Parcel`
+ :rtype: dict"""
+ self._log.debug('getting compartment open data')
+ if self.shipment_type == ParcelShipmentType.parcel:
+ self._log.debug('got compartment open data')
+ return {
+ 'shipmentNumber': self.shipment_number,
+ 'openCode': self._open_code,
+ 'receiverPhoneNumber': self.receiver.phone_number
+ }
+
+ self._log.debug('wrong ParcelShipmentType')
+ return None
+
+ @property
+ def mocked_location(self):
+ """Returns a mocked location for :class:`Parcel`
+
+ :return: dict containing mocked location for :class:`Parcel`
+ :rtype: dict"""
+ self._log.debug('getting mocked location')
+ if self.shipment_type == ParcelShipmentType.parcel:
+ self._log.debug('got mocked location')
+ return {
+ 'latitude': round(self.pickup_point.latitude + random.uniform(-0.00005, 0.00005), 6),
+ 'longitude': round(self.pickup_point.longitude + random.uniform(-0.00005, 0.00005), 6),
+ 'accuracy': round(random.uniform(1, 4), 1)
+ }
+
+ self._log.debug('wrong ParcelShipmentType')
+ return None
+
+
+[docs]class Receiver:
+ """Object representation of :class:`Parcel` receiver
+
+ :param receiver_data: :class:`dict` containing sender data for :class:`Parcel`
+ :type receiver_data: dict
+ :param logger: :class:`logging.Logger` parent instance
+ :type logger: logging.Logger"""
+[docs] def __init__(self, receiver_data: dict, logger: logging.Logger):
+ """Constructor method"""
+ self.email: str = receiver_data['email']
+ self.phone_number: str = receiver_data['phoneNumber']
+ self.name: str = receiver_data['name']
+ self._log: logging.Logger = logger.getChild(__class__.__name__)
+
+ self._log.debug('created')
+
+
+[docs]class Sender:
+ """Object representation of :class:`Parcel` sender
+
+ :param sender_data: :class:`dict` containing sender data for :class:`Parcel`
+ :type sender_data: dict
+ :param logger: :class:`logging.Logger` parent instance
+ :type logger: logging.Logger"""
+
+[docs] def __init__(self, sender_data: dict, logger: logging.Logger):
+ """Constructor method"""
+ self.sender_name: str = sender_data['name']
+ self._log: logging.Logger = logger.getChild(__class__.__name__)
+
+ self._log.debug('created')
+
+ def __str__(self) -> str:
+ return self.sender_name
+
+
+[docs]class PickupPoint:
+ """Object representation of :class:`Parcel` pickup point
+
+ :param pickuppoint_data: :class:`dict` containing pickup point data for :class:`Parcel`
+ :type pickuppoint_data: dict
+ :param logger: :class:`logging.Logger` parent instance
+ :type logger: logging.Logger"""
+
+[docs] def __init__(self, pickuppoint_data: dict, logger: logging.Logger):
+ """Constructor method"""
+ self.name: str = pickuppoint_data['name']
+ self.latitude: float = pickuppoint_data['location']['latitude']
+ self.longitude: float = pickuppoint_data['location']['longitude']
+ self.description: str = pickuppoint_data['locationDescription']
+ self.opening_hours: str = pickuppoint_data['openingHours']
+ self.post_code: str = pickuppoint_data['addressDetails']['postCode']
+ self.city: str = pickuppoint_data['addressDetails']['city']
+ self.province: str = pickuppoint_data['addressDetails']['province']
+ self.street: str = pickuppoint_data['addressDetails']['street']
+ self.building_number: str = pickuppoint_data['addressDetails']['buildingNumber']
+ self.virtual: int = pickuppoint_data['virtual']
+ self.point_type: str = pickuppoint_data['pointType']
+ self.type: List[ParcelDeliveryType] = [ParcelDeliveryType[data] for data in pickuppoint_data['type']]
+ self.location_round_the_clock: bool = pickuppoint_data['location247']
+ self.doubled: bool = pickuppoint_data['doubled']
+ self.image_url: str = pickuppoint_data['imageUrl']
+ self.easy_access_zone: bool = pickuppoint_data['easyAccessZone']
+ self.air_sensor: bool = pickuppoint_data['airSensor']
+
+ self._log: logging.Logger = logger.getChild(__class__.__name__)
+ self._log.debug('created')
+
+ if ParcelDeliveryType.UNKNOWN in self.type:
+ self._log.debug(f'unknown delivery type: {pickuppoint_data["type"]}')
+
+ def __str__(self) -> str:
+ return self.name
+
+ @property
+ def location(self) -> Tuple[float, float]:
+ """Returns a mocked location for :class:`PickupPoint`
+
+ :return: tuple containing location for :class:`PickupPoint`
+ :rtype: tuple"""
+ self._log.debug('getting location')
+ return self.latitude, self.longitude
+
+
+[docs]class MultiCompartment:
+ """Object representation of :class:`Parcel` `multicompartment`
+
+ :param multicompartment_data: :class:`dict` containing multicompartment data for :class:`Parcel`
+ :type multicompartment_data: dict
+ :param logger: :class:`logging.Logger` parent instance
+ :type logger: logging.Logger"""
+
+[docs] def __init__(self, multicompartment_data: dict, logger: logging.Logger):
+ """Constructor method"""
+ self.uuid = multicompartment_data['uuid']
+ self.shipment_numbers: List[str] | None = multicompartment_data['shipmentNumbers'] \
+ if 'shipmentNumbers' in multicompartment_data else None
+ self.presentation: bool = multicompartment_data['presentation']
+ self.collected: bool = multicompartment_data['collected']
+
+ self._log: logging.Logger = logger.getChild(__class__.__name__)
+ self._log.debug('created')
+
+
+[docs]class Operations:
+ """Object representation of :class:`Parcel` `operations`
+
+ :param operations_data: :class:`dict` containing operations data for :class:`Parcel`
+ :type operations_data: dict
+ :param logger: :class:`logging.Logger` parent instance
+ :type logger: logging.Logger"""
+
+[docs] def __init__(self, operations_data: dict, logger: logging.Logger):
+ """Constructor method"""
+ self.manual_archive: bool = operations_data['manualArchive']
+ self.auto_archivable_since: arrow | None = get(
+ operations_data['autoArchivableSince']) if 'autoArchivableSince' in operations_data else None
+ self.delete: bool = operations_data['delete']
+ self.collect: bool = operations_data['collect']
+ self.expand_avizo: bool = operations_data['expandAvizo']
+ self.highlight: bool = operations_data['highlight']
+ self.refresh_until: arrow = get(operations_data['refreshUntil'])
+ self.request_easy_access_zone: str = operations_data['requestEasyAccessZone']
+ self.is_voicebot: bool = operations_data['voicebot']
+ self.can_share_to_observe: bool = operations_data['canShareToObserve']
+ self.can_share_open_code: bool = operations_data['canShareOpenCode']
+ self.can_share_parcel: bool = operations_data['canShareParcel']
+
+ self._log: logging.Logger = logger.getChild(__class__.__name__)
+ self._log.debug('created')
+
+
+[docs]class EventLog:
+ """Object representation of :class:`Parcel` single eventlog
+
+ :param eventlog_data: :class:`dict` containing single eventlog data for :class:`Parcel`
+ :type eventlog_data: dict
+ :param logger: :class:`logging.Logger` parent instance
+ :type logger: logging.Logger"""
+
+[docs] def __init__(self, eventlog_data: dict, logger: logging.Logger):
+ """Constructor method"""
+ self.type: str = eventlog_data['type']
+ self.name: ParcelStatus = ParcelStatus[eventlog_data['name']]
+ self.date: arrow = get(eventlog_data['date'])
+
+ self._log: logging.Logger = logger.getChild(__class__.__name__)
+ self._log.debug('created')
+
+ if self.name == ParcelStatus.UNKNOWN:
+ self._log.debug(f'unknown parcel status: {eventlog_data["name"]}')
+
+
+
+
+
+[docs]class QRCode:
+ """Object representation of :class:`Parcel` QRCode
+
+ :param qrcode_data: :class:`str` containing qrcode data for :class:`Parcel`
+ :type qrcode_data: str
+ :param logger: :class:`logging.Logger` parent instance
+ :type logger: logging.Logger"""
+
+[docs] def __init__(self, qrcode_data: str, logger: logging.Logger):
+ """Constructor method"""
+ self._qr_code = qrcode_data
+
+ self._log: logging.Logger = logger.getChild(__class__.__name__)
+ self._log.debug('created')
+
+ @property
+ def qr_image(self) -> BytesIO:
+ """Returns a generated QR image for :class:`QRCode`
+
+ :return: QR Code image
+ :rtype: BytesIO"""
+ self._log.debug('generating qr image')
+ qr = qrcode.QRCode(
+ version=3,
+ error_correction=qrcode.constants.ERROR_CORRECT_H,
+ box_size=20,
+ border=4,
+ mask_pattern=5
+ )
+
+ qr.add_data(self._qr_code)
+ qr.make(fit=False)
+ img1 = qr.make_image(fill_color="black", back_color="white")
+ bio = BytesIO()
+ bio.name = 'qr.png'
+ img1.save(bio, 'PNG')
+ bio.seek(0)
+ self._log.debug('generated qr image')
+ return bio
+
+
+[docs]class CompartmentLocation:
+ """Object representation of :class:`CompartmentProperties` compartment location
+
+ :param compartmentlocation_data: :class:`dict` containing compartment location data for :class:`Parcel`
+ :type compartmentlocation_data: dict
+ :param logger: :class:`logging.Logger` parent instance
+ :type logger: logging.Logger"""
+
+[docs] def __init__(self, compartmentlocation_data: dict, logger: logging.Logger):
+ """Constructor method"""
+ self.name: str = compartmentlocation_data['compartment']['name']
+ self.side: str = compartmentlocation_data['compartment']['location']['side']
+ self.column: str = compartmentlocation_data['compartment']['location']['column']
+ self.row: str = compartmentlocation_data['compartment']['location']['row']
+ self.open_compartment_waiting_time: int = compartmentlocation_data['openCompartmentWaitingTime']
+ self.action_time: int = compartmentlocation_data['actionTime']
+ self.confirm_action_time: int = compartmentlocation_data['confirmActionTime']
+
+ self._log: logging.Logger = logger.getChild(__class__.__name__)
+ self._log.debug('created')
+
+
+[docs]class CompartmentProperties:
+ """Object representation of :class:`Parcel` compartment properties
+
+ :param compartmentproperties_data: :class:`dict` containing compartment properties data for :class:`Parcel`
+ :type compartmentproperties_data: dict
+ :param logger: :class:`logging.Logger` parent instance
+ :type logger: logging.Logger"""
+
+[docs] def __init__(self, compartmentproperties_data: dict, logger: logging.Logger):
+ """Constructor method"""
+ self._session_uuid: str = compartmentproperties_data['sessionUuid']
+ self._session_expiration_time: int = compartmentproperties_data['sessionExpirationTime']
+ self._location: CompartmentLocation | None = None
+ self._status: CompartmentActualStatus | None = None
+
+ self._log: logging.Logger = logger.getChild(__class__.__name__)
+ self._log.debug('created')
+
+ @property
+ def session_uuid(self):
+ """Returns a session unique identified for :class:`CompartmentProperties`
+
+ :return: string containing session unique identified for :class:`CompartmentProperties`
+ :rtype: str"""
+ self._log.debug('getting session uuid')
+ return self._session_uuid
+
+ @property
+ def location(self):
+ """Returns a compartment location for :class:`CompartmentProperties`
+
+ :return: compartment location for :class:`CompartmentProperties`
+ :rtype: str"""
+ self._log.debug('getting location')
+ return self._location
+
+ @location.setter
+ def location(self, location_data: dict):
+ """Set a compartment location for :class:`CompartmentProperties`
+
+ :param location_data: dict containing compartment location data for :class:`CompartmentProperties`
+ :type location_data: dict"""
+ self._log.debug('setting location')
+ self._location = CompartmentLocation(location_data, self._log)
+
+ @property
+ def status(self):
+ """Returns a compartment status for :class:`CompartmentProperties`
+
+ :return: compartment location for :class:`CompartmentProperties`
+ :rtype: CompartmentActualStatus"""
+ self._log.debug('getting status')
+ return self._status
+
+ # @status.setter
+ # def status(self, status_data: str | CompartmentActualStatus):
+ # self._log.debug('setting status')
+ # self._status = status_data if isinstance(status_data, CompartmentActualStatus) \
+ # else CompartmentActualStatus[status_data]
+ #
+ # if self._status == CompartmentActualStatus.UNKNOWN and isinstance(status_data, str):
+ # self._log.debug(f'unexpected compartment actual status: {status_data}')
+