ingest command new version with unique identifir in notification message

projmon/commands.py

@@ -1,19 +1,24 @@
 from datetime import datetime, time, date
 
-from typing import Dict, List, Any, ClassVar, Optional
+from typing import Dict, List, Any, ClassVar, Optional, Union, Generic, TypeVar
 from uuid import UUID
-from pydantic import BaseModel, ConfigDict, model_validator, field_validator, Field
+from pydantic import BaseModel, ConfigDict, model_validator, field_validator, Field, field_serializer
 
 class MessageHeaderWrapper(BaseModel):
     Id: int
     Type: str
-    Timestamp: datetime
+    Timestamp: datetime = Field(default_factory=datetime.now)
     Source: str
 
     @field_validator('Timestamp', mode='before')
     @classmethod
     def to_datetime(cls, value: str) -> datetime:
         return datetime.fromtimestamp(int(value)/1000)
+    
+    @field_serializer('Timestamp')
+    @classmethod
+    def to_timestamp(cls, value: datetime) -> int:
+        return int(value.timestamp() * 1000)
 
     def __eq__(self, other):
         if not isinstance(other, MessageHeaderWrapper):
@@ -23,13 +28,19 @@ class MessageHeaderWrapper(BaseModel):
             self.model_dump(exclude={"Timestamp"})
             == other.model_dump(exclude={"Timestamp"})
         )
+        
+class MessageBodyWrapper(BaseModel):
+    ...
+    
+V = TypeVar("V", bound=MessageBodyWrapper)
 
-class SMSMessage(BaseModel):
+class SMSMessageWrapper(BaseModel, Generic[V]):
     model_config = ConfigDict(arbitrary_types_allowed=True)
     path: ClassVar[List[str]]
     params: ClassVar[Dict[str, str]] = {}
 
     MessageHeader: MessageHeaderWrapper
+    MessageBody: Optional[V] = None
     
 class DCPInfoWrapper(BaseModel):
     ID: UUID
@@ -43,7 +54,7 @@ class DCPInfoWrapper(BaseModel):
     IsPlayable: bool
     IsTransferred: bool
 
-class DCPInfoList(SMSMessage):
+class DCPInfoList(SMSMessageWrapper):
     DCPInfo: List[DCPInfoWrapper] | None 
     
     path = ['content', 'dcp', 'info', 'list']
@@ -53,7 +64,7 @@ class PowerStatusWrapper(BaseModel):
     Device: str
     State: str
 
-class PowerStatus(SMSMessage):
+class PowerStatus(SMSMessageWrapper):
     PowerStatus: List[PowerStatusWrapper]
     
     path = ['status', 'sms', 'powerstatus']
@@ -76,7 +87,7 @@ class ShowStatusDetailWrapper(BaseModel):
     def extract(cls, data: str) -> List[int]:
         return [int(v) for v in data.split(',')]
 
-class ShowStatus(SMSMessage):
+class ShowStatus(SMSMessageWrapper):
     PlayState: str
     ShowStatusDetail: ShowStatusDetailWrapper
     PlayBackMode: str
@@ -121,14 +132,14 @@ type JobProgressElem = JobProgress
 class JobProgressWrapper(BaseModel):
     JobProgress: List[JobProgressElem] | JobProgressElem
 
-class DCPImportJobList(SMSMessage):
+class DCPImportJobList(SMSMessageWrapper):
     IsPaused: bool
     JobProgressList: Optional[JobProgressWrapper] = None
 
     path = ['content', 'dcp', 'command']
     params = {'action': 'ListImportJobs'}
 
-class StorageInfo(SMSMessage):
+class StorageInfo(SMSMessageWrapper):
     FileSystem: str
     TotalCapacity: int
     SpaceInUse: int
@@ -168,12 +179,12 @@ class SPLMetadata(BaseModel):
         return time.fromisoformat(value)
 
 type SPLMetadataElem = SPLMetadata
-class SPLList(SMSMessage):
+class SPLList(SMSMessageWrapper):
     path = ['show','playlist','info','list']
     
     SPLMetadata: List[SPLMetadataElem]
 
-class PrjLampInformation(SMSMessage):
+class PrjLampInformation(SMSMessageWrapper):
     path = ['command', 'projector', 'maintenance', 'lamp', 'info', 'get']
 
     ProjectorId: Optional[int] = None
@@ -195,7 +206,7 @@ class HDDStatusWrapper(BaseModel):
     Status: str
 
 type HDDStatusElem = HDDStatusWrapper
-class RAIDStatusDetail(SMSMessage):
+class RAIDStatusDetail(SMSMessageWrapper):
     RAIDStatus: str
     HDDStatus: HDDStatusElem
 
@@ -219,8 +230,57 @@ class ShowScheduleEvent(BaseModel):
     IsRecurringEvent: bool
     RecurrenceRule: Optional[RecurrenceWrapper] = None
 
-class ShowScheduleEventList(SMSMessage):
+class ShowScheduleEventList(SMSMessageWrapper):
     path = ['show', 'schedule', 'info', 'events', 'list']
 
     ShowScheduleEventList: List[ShowScheduleEvent]
+    
+    
+###### POST COMMANDS ##### 
+class SMSMessage(BaseModel):
+    SMSMessage: SMSMessageWrapper
 
+class ParameterWrapper(BaseModel):
+    Name: str
+    Value: Union[str, int, bool]
+    
+class ParameterListWrapper(BaseModel):
+    Parameter: List[ParameterWrapper]
+    
+class CommandWrapper(BaseModel):
+    Operation: str
+    ParameterList: ParameterListWrapper
+    
+class CommandBodyWrapper(MessageBodyWrapper):
+    Command: CommandWrapper
+    
+class AddImportJob(SMSMessageWrapper):
+    path = ['content', 'dcp', 'command']
+    
+    MessageHeader: MessageHeaderWrapper = MessageHeaderWrapper (
+        Id=-1,
+        Type="Command",
+        Source="LSM-100v2"
+    )
+    
+    MessageBody: Optional[MessageBodyWrapper] = CommandBodyWrapper(
+        Command = CommandWrapper(
+            Operation="AddImportJob", 
+            ParameterList=ParameterListWrapper(
+                Parameter=[
+                ParameterWrapper(Name="Resume", Value=True),
+                ParameterWrapper(Name="Validate", Value=True)
+            ])
+    ))
+    
+class ErrorResponseContentWrapper(BaseModel):
+    Severity: str
+    Code: int
+    Message: str
+    Solution: Optional[str]
+    
+class ErrorResponseWrapper(BaseModel):
+    ErrorResponse: ErrorResponseContentWrapper
+    
+class ErrorResponse(SMSMessageWrapper):
+    MessageBody: Optional[ErrorResponseWrapper] = None

projmon/ftpexplorer.py

@@ -1,17 +1,18 @@
 from ftplib import FTP, error_perm
 import os
 from pydantic import BaseModel, Field
-from typing import Optional, List
+from typing import Optional, List, Dict
 from datetime import datetime
 from time import sleep
 from paho.mqtt import client as mqtt_client
 from logging import Logger
+from uuid import uuid5, NAMESPACE_OID
 
 class FTPDirInfo(BaseModel):
     name: str
     modified_time: datetime = Field(default_factory=datetime.now)
-    unique: str = Field(default_factory=lambda: str(datetime.now().timestamp()))
-
+    
+type DirDescriptors = Dict[str, FTPDirInfo]
 class FTPServer(BaseModel):
     name: str
     host: str
@@ -20,7 +21,7 @@ class FTPServer(BaseModel):
     password: Optional[str] = None
     cwd: Optional[str] = None
     timeout: int = 15
-    dirs: List[FTPDirInfo] = Field(default_factory=list)
+    dirs: DirDescriptors = Field(default_factory=dict)
     notify_filters: List[str] = Field(default_factory=list)
     
     def connect(self) -> FTP:
@@ -40,8 +41,12 @@ class FTPServers(BaseModel):
 class NotificationMessage(BaseModel):
     server: str
     name: str
+    unique: str
     modified_time: datetime
     
+class NotificationResponse(BaseModel):
+    unique: str
+    
 def load_ftp_servers_from_json(file_path: str) -> FTPServers:
     """Loads FTP server configurations from a JSON file."""
     with open(file_path, 'r') as f:
@@ -75,12 +80,12 @@ def initialize_ftp_servers(file_path: str) -> FTPServers:
     save_ftp_servers_to_json(dcp_servers, file_path)
     return dcp_servers
 
-def list_new_directories(ftp_server: FTPServer) -> List[FTPDirInfo]:
+def list_new_directories(ftp_server: FTPServer) -> DirDescriptors:
     """Lists new directories on the FTP server that are not already known."""
     with ftp_server.connect() as ftp_connection:
         if ftp_server.cwd:
             ftp_connection.cwd(ftp_server.cwd)
-        new_dirs: List[FTPDirInfo] = []
+        new_dirs: DirDescriptors = {}
         try:
             # main method, retrieve directories with details
             for d in ftp_connection.mlsd():  # List directory contents
@@ -89,16 +94,17 @@ def list_new_directories(ftp_server: FTPServer) -> List[FTPDirInfo]:
                 modified_time_str: str = dir_info.get('modify', '')
                 modified_time: datetime = datetime.strptime(modified_time_str, '%Y%m%d%H%M%S') if modified_time_str else datetime.now()
                 unique: str = dir_info.get('unique', str(datetime.now().timestamp()))
-                new_dirs.append(FTPDirInfo(name=dir_name, modified_time=modified_time, unique=unique))
+                new_dirs[unique]=FTPDirInfo(name=dir_name, 
+                                           modified_time=modified_time)
         except error_perm:
             # revert to simple directory listing mode
             for d in ftp_connection.nlst():
                 dir_name = d
-                new_dirs.append(FTPDirInfo(name=dir_name))
-                
-        new_dirs.sort(key=lambda x: x.name, reverse=False)  # Sort directories by name
+                unique = uuid5(namespace=NAMESPACE_OID,name=dir_name).hex.upper()[:11]
+                new_dirs[unique]=FTPDirInfo(name=dir_name)
+                                
         
-        diff_dirs: List[FTPDirInfo] = list(filter(lambda x: x.name not in [d.name for d in ftp_server.dirs], new_dirs))
+        diff_dirs: DirDescriptors = { k:v for k,v in new_dirs.items() if k not in ftp_server.dirs }
         ftp_server.dirs = new_dirs  # Update the server's directory list with the new directories
         return diff_dirs  # Return both new directories and the difference
                          
@@ -113,21 +119,22 @@ def check_new_titles(mqtt: mqtt_client.Client, topic: str, file_path: str, logge
         
     for server in dcp_servers.servers:
         # find differences between register and FTP server           
-        diff_dirs: List[FTPDirInfo] = list_new_directories(server)
+        diff_dirs: DirDescriptors = list_new_directories(server)
         # do not publish complete list with garbage
         if first_run:
             logger.info("DCP first check on %s, %d DCPs found", server.name, len(diff_dirs))
             continue
-        for dir in diff_dirs:
-            logger.info(f"New DCP found on {server.name}: {dir.name} (Modified: {dir.modified_time}, Unique: {dir.unique})")           
+        for unique, dir in diff_dirs.items():
+            logger.info(f"New DCP found on {server.name}: {dir.name} (Modified: {dir.modified_time}, Unique: {unique})")           
             for filter_str in server.notify_filters:
                 if filter_str in dir.name:
                     notification = NotificationMessage(
                         server=server.name, 
                         name=dir.name, 
-                        modified_time=dir.modified_time
+                        modified_time=dir.modified_time,
+                        unique=unique
                     )
-                    mqtt.publish(topic, notification.model_dump_json(), qos=2)
+                    mqtt.publish(topic, notification.model_dump_json(), qos=1)
                     sleep(1)  # Small delay to ensure message is sent
                     
     # Save server information with updated DCP list

projmon/projmon.py

@@ -2,26 +2,38 @@ import logging
 import os
 import sys
 import time
-from typing import List, Type, TypeVar, Set, Any
+import json
+import xmltodict
+from typing import List, Type, TypeVar, Set, Any, Dict
 from uuid import UUID
 
 import projrequest
-from commands import ShowStatus, SMSMessage, SPLList, SPLMetadata, StorageInfo, PrjLampInformation
+from commands import ShowStatus, SMSMessageWrapper, SMSMessage, CommandBodyWrapper, SPLList, SPLMetadata, StorageInfo, PrjLampInformation, AddImportJob, ParameterWrapper, ErrorResponse
 
 from paho.mqtt import client as mqtt_client
 from paho.mqtt.enums import CallbackAPIVersion
 from requests import Timeout, ConnectionError
 from pyutils.utils import SignalHandler
 from homeassistant import HAdevice, HAsensor, HAbutton
-from ftpexplorer import check_new_titles
+from ftpexplorer import check_new_titles, FTPServers, FTPServer, NotificationResponse
 
-T = TypeVar("T", bound=SMSMessage)
+T = TypeVar("T", bound=SMSMessageWrapper)
 
 # Get environment variables
 env = dict(os.environ)
 LOGGER: logging.Logger
 
 DISCOVERY_PREFIX = "homeassistant"
+INTERVAL = int(env['INTERVAL'])
+DCP_INTERVAL = int(env['DCP_INTERVAL'])
+DCP_SERVER_FILE = str(env['DCP_SERVER_FILE'])
+DCP_NEW_TOPIC = str(env['DCP_NEW_TOPIC'])
+
+projector: projrequest.ProjectorConnection = projrequest.ProjectorConnection(
+    ip=env['PROJECTOR_IP'], 
+    username=env['PROJECTOR_USER'], 
+    password=env['PROJECTOR_PASSWORD']
+)
 
 def on_shutwown(client: mqtt_client.Client, userdata: Any, msg: mqtt_client.MQTTMessage) -> None:
     if msg and msg.payload.decode() == "shutdown":
@@ -33,14 +45,53 @@ def on_shutwown(client: mqtt_client.Client, userdata: Any, msg: mqtt_client.MQTT
         )
         projector.shutdown()
 
+def on_command(client: mqtt_client.Client, userdata: Any, msg: mqtt_client.MQTTMessage) -> None:
+    try:
+        content: Dict[str, Any] = json.loads(msg.payload.decode('utf-8'))
+        cmd: str = content["cmd"]
+        params: str = content["params"]
+        match cmd:
+            case "ingest":
+                notify: NotificationResponse = NotificationResponse.model_validate(params)
+                with open(DCP_SERVER_FILE, 'r') as fd:
+                    dcp_servers: FTPServers = FTPServers.model_validate_json(fd.read().encode('utf-8'))
+                    unique: str = notify.unique
+                    this_server: FTPServer = list(filter(lambda s: unique in s.dirs.keys(), dcp_servers.servers))[0]
+                    user_pswd: str = f"{this_server.username}:{this_server.password}" if this_server.username else "anonymous:"
+                    dir_name: str = f"{this_server.cwd}{"/" if this_server.cwd else ""}{this_server.dirs[unique].name}"
+                    ingest_command: AddImportJob = AddImportJob()
+                    assert isinstance(ingest_command.MessageBody, CommandBodyWrapper)
+                    ingest_command.MessageBody.Command.ParameterList.Parameter.insert(0,
+                        ParameterWrapper(
+                            Name="Folder",
+                            Value=f"ftp://{user_pswd}@{this_server.host}:{this_server.port}/{dir_name}"
+                        )
+                    )
+                    rv: Dict = projector.post(
+                        path=ingest_command.path,
+                        data=xmltodict.unparse(
+                        SMSMessage(
+                            SMSMessage=ingest_command
+                        ).model_dump(), indent=1)
+                    )
+                    error_response = ErrorResponse.model_validate(rv['SMSMessage'])
+                    if not error_response.MessageBody:
+                        LOGGER.warning("Invalid error response: [%s]", rv)
+                        return
+                    if error_response.MessageBody.ErrorResponse.Code != 100:
+                        LOGGER.error("Unable to ingest Title: [%s]", dir_name)
+                    else:
+                        LOGGER.info("Started ingest of Title: [%s] from [%s]", dir_name, this_server.name)   
+                
+    except json.JSONDecodeError as de:
+        LOGGER.error("Unable to decode command json: [%s]", de)
+    except KeyError as ke:
+        LOGGER.error("Unable to find key: [%s]", ke)
+
 ##################
 ###### MAIN ######
 ##################
 def main() -> int:
-    INTERVAL = int(env['INTERVAL'])
-    DCP_INTERVAL = int(env['DCP_INTERVAL'])
-    DCP_SERVER_FILE = str(env['DCP_SERVER_FILE'])
-    DCP_NEW_TOPIC = str(env['DCP_NEW_TOPIC'])
     try:
         mqtt = mqtt_client.Client(CallbackAPIVersion.VERSION2)
         mqtt.username_pw_set(username=env['MQTT_USER'], password=env['MQTT_PASSWORD'])
@@ -58,12 +109,7 @@ def main() -> int:
     last: float = 0
     last_title_check: float = 0
     handler: SignalHandler = SignalHandler(LOGGER)
-    projector: projrequest.ProjectorConnection = projrequest.ProjectorConnection(
-        ip=env['PROJECTOR_IP'], 
-        username=env['PROJECTOR_USER'], 
-        password=env['PROJECTOR_PASSWORD']
-    )
-
+    
     sony_device: HAdevice = HAdevice (
         identifiers=["sony_digitalcinema_projector"],
         name="Sony Digital Cinema Projector",
@@ -75,7 +121,7 @@ def main() -> int:
     sony_device.entities.append(HAbutton (
         name="Power Off",
         unique_id=f"{sony_device.short_name}_power_off",
-        command_topic=f"{sony_device.short_name}/commands",
+        command_topic=f"{sony_device.short_name}/shutdown",
         icon="mdi:power",
         payload_press="shutdown",
         entity_type="button",
@@ -243,8 +289,8 @@ def main() -> int:
 
         lamp_devices.append(lamp_device)
 
+    # Subscribe to MQTT topics for devices having associate actions
     all_devices: List[HAdevice] = [sony_device, *lamp_devices]
-
     for d in all_devices:
         for s in d.entities:
             # Publish config mesasge
@@ -257,13 +303,16 @@ def main() -> int:
                 if s.callback is not None:
                     mqtt.message_callback_add(sub=s.command_topic, callback=s.callback)
             time.sleep(0.2)
+            
+    # Subscrie to MQTT topics for generic commands
+    mqtt.subscribe(topic=f"{sony_device.short_name}/commands")
+    mqtt.message_callback_add(sub=f"{sony_device.short_name}/commands", callback=on_command)
 
     playlists: SPLList | None = None
-
     while handler.running:
         now: float = time.time()
         try:
-            resp: None | SMSMessage = None
+            resp: None | SMSMessageWrapper = None
             # update projector states
             lamp_num: int = 0
             for d in all_devices:
@@ -295,21 +344,22 @@ def main() -> int:
                             resp.ShowStatusDetail.CurrentEventTitle = "Unknown or CPL"
 
                     # send only one message for sensor with this type of request
-                    last_cls: Type = SMSMessage
+                    last_cls: Type = SMSMessageWrapper
                     # filter sensors of this device per this type of request
                     for sens in filter(lambda x: x.request == s, sensors):
                         if last_cls == sens.request:
                             continue
-                        if isinstance(resp, SMSMessage):
-                            sts: SMSMessage = resp
+                        if isinstance(resp, SMSMessageWrapper):
+                            sts: SMSMessageWrapper = resp
                             mqtt.publish(topic=sens.state_topic, 
                                             payload=sts.model_dump_json(
                                                 indent=1,
                                                 exclude={"MessageHeader"},
-                                                exclude_unset=True))
+                                                exclude_unset=True),
+                                            qos=1)
                         last_cls = sens.request
         except (Timeout, ConnectionError) as ce:
-            LOGGER.debug("Projector is probably shutdown: %s", ce)
+            LOGGER.debug("Projector is probably shutdown: [%s]", ce)
         except Exception as e:
             LOGGER.error("Unexpected exception: [%s]", e)
             return 1

projmon/projrequest.py

@@ -7,10 +7,10 @@ from os import getcwd
 from typing import Any, Dict, List, Type, TypeVar
 
 import xmltodict
-from commands import SMSMessage
+from commands import SMSMessageWrapper
 from requests.auth import HTTPBasicAuth
 
-T = TypeVar("T", bound=SMSMessage)
+T = TypeVar("T", bound=SMSMessageWrapper)
 
 TEST = False
 
@@ -25,6 +25,24 @@ class ProjectorConnection():
         self.username = username
         self.password = password
         self.auth = HTTPBasicAuth(username=self.username, password=self.password)
+        
+    def post(self, path: List[str], params: Dict[str, str] | None = None, data: str | None = None) -> Dict[str, Any]:
+        response: requests.Response = requests.post(
+            url=f"https://{self.ip}/{"/".join(path)}",
+            params=params,
+            headers={
+                "Accept": "application/xml"
+            },
+            auth=self.auth,
+            allow_redirects=False, 
+            timeout=5,
+            verify=False,
+            data=data
+        )
+        response.raise_for_status()
+        if response.status_code == 200: # HTTP ok response
+            return xmltodict.parse(response.text.replace("%2F", "/").replace("%3A", ":")) # Common containert for all messages
+        return {}
 
     def get(self, path: List[str], params: Dict[str, str] | None = None) -> Dict[str, Any]:
         response: requests.Response = requests.get(
@@ -56,6 +74,7 @@ class ProjectorConnection():
             rv.update(**rv['MessageBody'][tt])
             del rv['MessageBody']
             del rv['@xmlns']
+            rv.update(**rv)
             #print(json.dumps(rv, indent = 2))
             return cls.model_validate(rv)
         return None

projmon/xmlresponses/ingest_request.xml

@@ -0,0 +1,88 @@
+<!-- 
+{
+	"POST": {
+		"scheme": "https",
+		"host": "192.168.31.10",
+		"filename": "/content/dcp/command",
+		"remote": {
+			"Indirizzo": "192.168.31.10:443"
+		}
+	}
+}
+-->
+
+<!-- request -->
+<SMSMessage>
+    <MessageHeader>
+        <Id>-1</Id>
+        <Type>Command</Type>
+        <Timestamp>1789025739512</Timestamp>
+        <Source>LSM-100v2</Source>
+    </MessageHeader>
+    <MessageBody>
+        <Command>
+            <Operation>AddImportJob</Operation>
+            <ParameterList>
+                <Parameter>
+                    <Name>Folder</Name>
+                    <Value>ftp://dcpftp:dcpftp@192.168.31.16:21/1874853_LinkinParkU_TLR-Ext_F_EN-it_IT_51_4K_NXO_20260908_DLX_IOP_OV</Value>
+                </Parameter>
+                <Parameter>
+                    <Name>Resume</Name>
+                    <Value>true</Value>
+                </Parameter>
+                <Parameter>
+                    <Name>Validate</Name>
+                    <Value>true</Value>
+                </Parameter>
+            </ParameterList>
+        </Command>
+    </MessageBody>
+</SMSMessage>
+
+<!-- response -->
+<SMSMessage xmlns="http://xmlns.sony.net/d-cinema/sms/2007b">
+    <MessageHeader>
+        <Id>-1</Id>
+        <Type>ErrorResponse</Type>
+        <Timestamp>1789025967924</Timestamp>
+        <Source>LSM-100v2</Source>
+    </MessageHeader>
+    <MessageBody>
+        <ErrorResponse>
+            <Severity>INFO</Severity>
+            <Code>100</Code>
+            <Message>Request+successfully+processed.</Message>
+            <Solution></Solution>
+        </ErrorResponse>
+    </MessageBody>
+</SMSMessage>
+
+
+<!-- MY REQUEST -->
+<?xml version="1.0" encoding="utf-8"?>
+<SMSMessage>
+    <MessageHeader>
+        <Id>-1</Id>
+        <Type>Command</Type>
+        <Timestamp>1789038719365</Timestamp>
+        <Source>LSM-100v2</Source>
+    </MessageHeader>
+    <MessageBody>
+        <Command>
+            <Operation>AddImportJob</Operation>
+            <ParameterList>
+                <Name>Resume</Name>
+                <Value>true</Value>
+            </ParameterList>
+            <ParameterList>
+                <Name>Validate</Name>
+                <Value>true</Value>
+            </ParameterList>
+            <ParameterList>
+                <Name>Folder</Name>
+                <Value>ftp://dcpftp:dcpftp@192.168.31.16:21/Remain_TLR-F2_S_IT-XX_IT_51_4K_WR_20260907_PXL_SMPTE_OV</Value>
+            </ParameterList>
+        </Command>
+    </MessageBody>
+</SMSMessage>